Wire the composition root: flags, providers, runnables, manifests, samples, docs

Co-Authored-By: Claude <noreply@anthropic.com>

cmd/main.go

@@ -14,28 +14,48 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+// Package main is the composition root: it loads the provider config (fail
+// fast), assembles the provider registry, and wires the reconciler, health
+// engine, lease store, discovery API, orphan GC, and metrics onto one
+// controller-runtime manager.
 package main
 
 import (
+	"context"
 	"crypto/tls"
 	"flag"
 	"os"
+	"time"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
 	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
 	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/controller"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/discovery"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/gc"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/health"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/metrics"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/gcp"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kubernetes"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/registry"
 	// +kubebuilder:scaffold:imports
 )
 
@@ -60,6 +80,15 @@ func main() {
 	var secureMetrics bool
 	var enableHTTP2 bool
 	var tlsOpts []func(*tls.Config)
+
+	var providersConfig string
+	var discoveryAddr string
+	var proxyNamespace string
+	var healthWorkers int
+	var gcInterval, gcMinAge time.Duration
+	var gcAllowNamespaced bool
+	var leaseCooldown, maxLeaseTTL time.Duration
+
 	flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
 		"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
@@ -74,6 +103,28 @@ func main() {
 	flag.StringVar(&metricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.")
 	flag.BoolVar(&enableHTTP2, "enable-http2", false,
 		"If set, HTTP/2 will be enabled for the metrics server")
+
+	flag.StringVar(&providersConfig, "providers-config", "",
+		"Path to the providers config YAML. Required.")
+	flag.StringVar(&discoveryAddr, "discovery-addr", ":8090",
+		"Listen address of the discovery/lease HTTP API.")
+	flag.StringVar(&proxyNamespace, "proxy-namespace", "",
+		"Restrict the manager's cache to one namespace. Empty watches all namespaces. "+
+			"Restricting also disables orphan GC unless --gc-allow-namespaced is set.")
+	flag.IntVar(&healthWorkers, "health-workers", 8,
+		"Number of concurrent health-probe workers.")
+	flag.DurationVar(&gcInterval, "gc-interval", 10*time.Minute,
+		"Interval between orphan GC sweeps.")
+	flag.DurationVar(&gcMinAge, "gc-min-age", 10*time.Minute,
+		"Minimum instance age before orphan GC may delete it.")
+	flag.BoolVar(&gcAllowNamespaced, "gc-allow-namespaced", false,
+		"Allow orphan GC to run although the cache is namespace-restricted. Dangerous: proxies "+
+			"outside the namespace count as orphans and their instances get deleted.")
+	flag.DurationVar(&leaseCooldown, "lease-cooldown", 15*time.Minute,
+		"How long a reported proxy/target pair is excluded from lease selection.")
+	flag.DurationVar(&maxLeaseTTL, "max-lease-ttl", time.Hour,
+		"Maximum lease TTL a client may request.")
+
 	opts := zap.Options{
 		Development: true,
 	}
@@ -81,6 +132,31 @@ func main() {
 	flag.Parse()
 
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
+	ctx := ctrl.SetupSignalHandler()
+
+	// Providers load first and fail fast: a manager that comes up without
+	// its backends would just convert every Proxy into an error loop.
+	if providersConfig == "" {
+		setupLog.Error(nil, "--providers-config is required")
+		os.Exit(1)
+	}
+	cfg, err := provider.LoadConfigFile(providersConfig)
+	if err != nil {
+		setupLog.Error(err, "Failed to load providers config", "path", providersConfig)
+		os.Exit(1)
+	}
+	providers, err := registry.Build(ctx, cfg, map[string]registry.Constructor{
+		"kubernetes": kubernetes.New,
+		"gcp":        gcp.New,
+	})
+	if err != nil {
+		setupLog.Error(err, "Failed to build providers")
+		os.Exit(1)
+	}
+	m := metrics.New()
+	for name, p := range providers {
+		providers[name] = provider.WithMetrics(name, p, m)
+	}
 
 	// if the enable-http2 flag is false (the default), http/2 should be disabled
 	// due to its vulnerabilities. More specifically, disabling http/2 will
@@ -128,32 +204,91 @@ func main() {
 		metricsServerOptions.KeyName = metricsCertKey
 	}
 
+	// The Secret cache is restricted to labelled cloud-init Secrets: the
+	// operator has cluster-wide Secret read RBAC, and without the label
+	// selector it would cache every Secret in scope.
+	cacheOpts := cache.Options{
+		ByObject: map[client.Object]cache.ByObject{
+			&corev1.Secret{}: {
+				Label: labels.SelectorFromSet(labels.Set{crawlv1alpha1.LabelCloudInit: "true"}),
+			},
+		},
+	}
+	if proxyNamespace != "" {
+		cacheOpts.DefaultNamespaces = map[string]cache.Config{proxyNamespace: {}}
+	}
+
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:                 scheme,
 		Metrics:                metricsServerOptions,
 		HealthProbeBindAddress: probeAddr,
+		Cache:                  cacheOpts,
 		LeaderElection:         enableLeaderElection,
 		LeaderElectionID:       "b47711d1.example.com",
-		// LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily
-		// when the Manager ends. This requires the binary to immediately end when the
-		// Manager is stopped, otherwise, this setting is unsafe. Setting this significantly
-		// speeds up voluntary leader transitions as the new leader don't have to wait
-		// LeaseDuration time first.
-		//
-		// In the default scaffold provided, the program ends immediately after
-		// the manager stops, so would be fine to enable this option. However,
-		// if you are doing or is intended to do any operation such as perform cleanups
-		// after the manager stops then its usage might be unsafe.
-		// LeaderElectionReleaseOnCancel: true,
 	})
 	if err != nil {
 		setupLog.Error(err, "Failed to start manager")
 		os.Exit(1)
 	}
 
+	store := lease.NewStore(leaseCooldown)
+	if err := mgr.Add(store); err != nil {
+		setupLog.Error(err, "Failed to add lease store")
+		os.Exit(1)
+	}
+
+	engine := health.NewEngine(mgr.GetClient())
+	engine.Workers = healthWorkers
+	engine.Metrics = m
+	if err := mgr.Add(engine); err != nil {
+		setupLog.Error(err, "Failed to add health engine")
+		os.Exit(1)
+	}
+
+	if err := mgr.Add(&discovery.Server{
+		Reader:      mgr.GetClient(),
+		Store:       store,
+		Addr:        discoveryAddr,
+		Token:       os.Getenv("DISCOVERY_TOKEN"),
+		MaxLeaseTTL: maxLeaseTTL,
+		Metrics:     m,
+	}); err != nil {
+		setupLog.Error(err, "Failed to add discovery server")
+		os.Exit(1)
+	}
+
+	if err := mgr.Add(&gc.Sweeper{
+		Reader:              mgr.GetClient(),
+		Providers:           providers,
+		Interval:            gcInterval,
+		MinAge:              gcMinAge,
+		NamespaceRestricted: proxyNamespace != "",
+		AllowNamespaced:     gcAllowNamespaced,
+	}); err != nil {
+		setupLog.Error(err, "Failed to add orphan GC")
+		os.Exit(1)
+	}
+
+	if err := m.Register(ctrlmetrics.Registry,
+		proxyPhaseCounts(mgr.GetClient()),
+		func() int {
+			total := 0
+			for _, n := range store.Counts() {
+				total += n
+			}
+			return total
+		},
+	); err != nil {
+		setupLog.Error(err, "Failed to register metrics")
+		os.Exit(1)
+	}
+
 	if err := (&controller.ProxyReconciler{
-		Client: mgr.GetClient(),
-		Scheme: mgr.GetScheme(),
+		Client:       mgr.GetClient(),
+		Scheme:       mgr.GetScheme(),
+		Providers:    providers,
+		Health:       engine,
+		HealthEvents: engine.Events,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "Failed to create controller", "controller", "proxy")
 		os.Exit(1)
@@ -170,8 +305,31 @@ func main() {
 	}
 
 	setupLog.Info("Starting manager")
-	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
+	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "Failed to run manager")
 		os.Exit(1)
 	}
 }
+
+// proxyPhaseCounts reads phase counts from the cache at scrape time. Before
+// the cache has synced (or on any list error) it reports nothing rather
+// than something wrong.
+func proxyPhaseCounts(c client.Reader) func() map[string]int {
+	return func() map[string]int {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		var list crawlv1alpha1.ProxyList
+		if err := c.List(ctx, &list); err != nil {
+			return nil
+		}
+		counts := map[string]int{}
+		for i := range list.Items {
+			phase := string(list.Items[i].Status.Phase)
+			if phase == "" {
+				phase = string(crawlv1alpha1.PhasePending)
+			}
+			counts[phase]++
+		}
+		return counts
+	}
+}