egress-proxies-operator/cmd/main.go

/*
Copyright 2026.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
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"
	"fmt"
	"os"
	goruntime "runtime"
	"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"
	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/version"
	// +kubebuilder:scaffold:imports
)

var (
	scheme   = runtime.NewScheme()
	setupLog = ctrl.Log.WithName("setup")
)

func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))

	utilruntime.Must(crawlv1alpha1.AddToScheme(scheme))
	// +kubebuilder:scaffold:scheme
}

// nolint:gocyclo
func main() {
	var metricsAddr string
	var metricsCertPath, metricsCertName, metricsCertKey string
	var enableLeaderElection bool
	var probeAddr string
	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
	var showVersion bool
	var gcpWireFullPayloads bool

	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.")
	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
		"Enable leader election for controller manager. "+
			"Enabling this will ensure there is only one active controller manager.")
	flag.BoolVar(&secureMetrics, "metrics-secure", true,
		"If set, the metrics endpoint is served securely via HTTPS. Use --metrics-secure=false to use HTTP instead.")
	flag.StringVar(&metricsCertPath, "metrics-cert-path", "",
		"The directory that contains the metrics server certificate.")
	flag.StringVar(&metricsCertName, "metrics-cert-name", "tls.crt", "The name of the metrics server certificate file.")
	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.")
	flag.BoolVar(&showVersion, "version", false,
		"Print the commit the binary was built from and exit.")
	flag.BoolVar(&gcpWireFullPayloads, "gcp-wire-log-full-payloads", false,
		"Log GCP V(5) wire payloads verbatim instead of eliding fields larger than 1KiB.")

	opts := zap.Options{
		Development: true,
	}
	opts.BindFlags(flag.CommandLine)
	flag.Parse()

	if showVersion {
		fmt.Println(version.Resolve())
		os.Exit(0)
	}

	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
	setupLog.Info("Starting egress-proxies-operator",
		"commit", version.Resolve(), "goVersion", goruntime.Version())

	// Off (no-op spans, unchanged logs) unless OTEL_* env opts in; see
	// internal/tracing. Shutdown is called explicitly after mgr.Start
	// returns — the os.Exit paths below skip defers.
	tracingShutdown, err := tracing.Setup(context.Background(), setupLog,
		"egress-proxies-operator", version.Resolve())
	if err != nil {
		setupLog.Error(err, "Failed to set up tracing")
		os.Exit(1)
	}

	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": func(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, error) {
			// The kubernetes provider builds its own uncached client;
			// wrap its transport so its API calls join the caller's trace.
			return kubernetes.NewWithTransportWrapper(ctx, pc, tracing.RestConfigWrapper())
		},
		"gcp": func(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, error) {
			return gcp.NewWithWireOptions(ctx, pc, gcp.WireLogOptions{FullPayloads: gcpWireFullPayloads})
		},
	})
	if err != nil {
		setupLog.Error(err, "Failed to build providers")
		os.Exit(1)
	}
	m := metrics.New()
	for name, p := range providers {
		// Tracing outermost: the span covers the metrics recording too.
		providers[name] = provider.WithTracing(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
	// prevent from being vulnerable to the HTTP/2 Stream Cancellation and
	// Rapid Reset CVEs. For more information see:
	// - https://github.com/advisories/GHSA-qppj-fm5r-hxr3
	// - https://github.com/advisories/GHSA-4374-p667-p6c8
	disableHTTP2 := func(c *tls.Config) {
		setupLog.Info("Disabling HTTP/2")
		c.NextProtos = []string{"http/1.1"}
	}

	if !enableHTTP2 {
		tlsOpts = append(tlsOpts, disableHTTP2)
	}

	// Metrics endpoint is enabled in 'config/default/kustomization.yaml'. The Metrics options configure the server.
	// More info:
	// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.24.1/pkg/metrics/server
	// - https://book.kubebuilder.io/reference/metrics.html
	metricsServerOptions := metricsserver.Options{
		BindAddress:   metricsAddr,
		SecureServing: secureMetrics,
		TLSOpts:       tlsOpts,
	}

	if secureMetrics {
		// FilterProvider is used to protect the metrics endpoint with authn/authz.
		// These configurations ensure that only authorized users and service accounts
		// can access the metrics endpoint. The RBAC are configured in 'config/rbac/kustomization.yaml'. More info:
		// https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.24.1/pkg/metrics/filters#WithAuthenticationAndAuthorization
		metricsServerOptions.FilterProvider = filters.WithAuthenticationAndAuthorization
	}

	// If the certificate is not specified, controller-runtime will automatically
	// generate self-signed certificates for the metrics server. While convenient for development and testing,
	// this setup is not recommended for production. This project doesn't use cert-manager (no admission
	// webhooks, no other consumer of managed certs) -- pass real certs via the flags below if needed.
	if len(metricsCertPath) > 0 {
		setupLog.Info("Initializing metrics certificate watcher using provided certificates",
			"metrics-cert-path", metricsCertPath, "metrics-cert-name", metricsCertName, "metrics-cert-key", metricsCertKey)

		metricsServerOptions.CertDir = metricsCertPath
		metricsServerOptions.CertName = metricsCertName
		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: {}}
	}

	restCfg := ctrl.GetConfigOrDie()
	restCfg.Wrap(tracing.RestConfigWrapper())
	mgr, err := ctrl.NewManager(restCfg, ctrl.Options{
		Scheme:                 scheme,
		Metrics:                metricsServerOptions,
		HealthProbeBindAddress: probeAddr,
		Cache:                  cacheOpts,
		LeaderElection:         enableLeaderElection,
		LeaderElectionID:       "b47711d1.example.com",
	})
	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(),
		Providers:    providers,
		Health:       engine,
		HealthEvents: engine.Events,
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "Failed to create controller", "controller", "proxy")
		os.Exit(1)
	}
	// +kubebuilder:scaffold:builder

	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		setupLog.Error(err, "Failed to set up health check")
		os.Exit(1)
	}
	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
		setupLog.Error(err, "Failed to set up ready check")
		os.Exit(1)
	}

	setupLog.Info("Starting manager")
	startErr := mgr.Start(ctx)

	// Flush pending spans on the way out, error path included. Fresh
	// context: the signal ctx is already cancelled by the time Start
	// returns.
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	if err := tracingShutdown(shutdownCtx); err != nil {
		setupLog.Error(err, "Failed to flush traces on shutdown")
	}
	cancel()

	if startErr != nil {
		setupLog.Error(startErr, "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
	}
}