feat: add leader-election tuning (#2008)

* feat: add leader-election tuning Signed-off-by: Oliver Baehler <oliver@sudo-i.net> * feat: add leader-election tuning Signed-off-by: Oliver Baehler <oliver@sudo-i.net> * feat: add tracing for admission Signed-off-by: Oliver Baehler <oliver@sudo-i.net> * feat: add tracing for admission Signed-off-by: Oliver Baehler <oliver@sudo-i.net> * add tenant --------- Signed-off-by: Oliver Baehler <oliver@sudo-i.net>
2026-08-19 04:26:45 +00:00 · 2026-07-08 11:44:43 +02:00
parent f7c5523613
commit c17e4fa812
51 changed files with 5197 additions and 44 deletions
@@ -4,6 +4,7 @@
 package main

 import (
+	"context"
 	"crypto/tls"
 	goflag "flag"
 	"fmt"
@@ -122,16 +123,32 @@ func main() {

 		enableLeaderElection bool
 		enablePprof          bool
+		enableTracing        bool
 		version              bool
 		secureMetrics        bool
 		enableHTTP2          bool

+		tracingEndpoint              string
+		tracingSampleRatio           float64
+		tracingInsecure              bool
+		tracingHeaders               = tracingHeadersFlag{}
+		tracingBasicAuthUsername     string
+		tracingBasicAuthPassword     string
+		tracingTimeout               time.Duration
+		tracingCompression           string
+		tracingTLSServerName         string
+		tracingTLSInsecureSkipVerify bool
+
 		clientConnectionQPS   float32
 		clientConnectionBurst int32

 		webhookPort int

 		cacheSyncTimeout time.Duration
+
+		leaderElectionLeaseDuration time.Duration
+		leaderElectionRenewDeadline time.Duration
+		leaderElectionRetryPeriod   time.Duration
 	)

 	var goFlagSet goflag.FlagSet
@@ -152,6 +169,71 @@ func main() {
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.",
 	)
+	flag.BoolVar(
+		&enableTracing,
+		"enable-tracing",
+		false,
+		"Enable OpenTelemetry tracing for admission webhook requests.",
+	)
+	flag.StringVar(
+		&tracingEndpoint,
+		"tracing-otlp-endpoint",
+		"",
+		"OTLP gRPC endpoint for exporting traces, for example otel-collector.observability.svc:4317. If unset, OpenTelemetry environment variables are used.",
+	)
+	flag.BoolVar(
+		&tracingInsecure,
+		"tracing-otlp-insecure",
+		true,
+		"Disable transport security for the OTLP gRPC trace exporter.",
+	)
+	flag.Float64Var(
+		&tracingSampleRatio,
+		"tracing-sample-ratio",
+		1.0,
+		"Trace sampling ratio for admission webhook requests. Must be between 0 and 1.",
+	)
+	flag.Var(
+		tracingHeaders,
+		"tracing-otlp-header",
+		"OTLP gRPC metadata header in key=value format. Can be set multiple times.",
+	)
+	flag.StringVar(
+		&tracingBasicAuthUsername,
+		"tracing-otlp-basic-auth-username",
+		"",
+		"Basic auth username for the OTLP gRPC trace exporter. Can also be set with CAPSULE_TRACING_OTLP_BASIC_AUTH_USERNAME.",
+	)
+	flag.StringVar(
+		&tracingBasicAuthPassword,
+		"tracing-otlp-basic-auth-password",
+		"",
+		"Basic auth password for the OTLP gRPC trace exporter. Can also be set with CAPSULE_TRACING_OTLP_BASIC_AUTH_PASSWORD.",
+	)
+	flag.DurationVar(
+		&tracingTimeout,
+		"tracing-otlp-timeout",
+		0,
+		"Timeout for exporting a batch of spans. Empty or 0 uses OpenTelemetry's default.",
+	)
+	flag.StringVar(
+		&tracingCompression,
+		"tracing-otlp-compression",
+		"",
+		"Compression for OTLP gRPC trace exports. Supported value: gzip. Empty disables compression.",
+	)
+	flag.StringVar(
+		&tracingTLSServerName,
+		"tracing-otlp-tls-server-name",
+		"",
+		"TLS server name override for the OTLP gRPC trace exporter.",
+	)
+	flag.BoolVar(
+		&tracingTLSInsecureSkipVerify,
+		"tracing-otlp-tls-insecure-skip-verify",
+		false,
+		"Skip OTLP gRPC trace exporter TLS certificate verification. Not recommended for production.",
+	)
 	flag.IntVar(
 		&controllerConfig.Runtime.MaxConcurrentReconciles,
 		"workers",
@@ -164,6 +246,24 @@ func main() {
 		0,
 		"The timeout used when waiting for controller cache synchronization. If unset or 0, the controller-runtime default is used.",
 	)
+	flag.DurationVar(
+		&leaderElectionLeaseDuration,
+		"leader-election-lease-duration",
+		0,
+		"The duration that non-leader candidates wait to force acquire leadership. If unset or 0, the controller-runtime default is used.",
+	)
+	flag.DurationVar(
+		&leaderElectionRenewDeadline,
+		"leader-election-renew-deadline",
+		0,
+		"The duration that the acting leader retries refreshing leadership before giving up. If unset or 0, the controller-runtime default is used.",
+	)
+	flag.DurationVar(
+		&leaderElectionRetryPeriod,
+		"leader-election-retry-period",
+		0,
+		"The duration that leader election clients wait between retries. If unset or 0, the controller-runtime default is used.",
+	)
 	flag.StringVar(
 		&metricsAddr,
 		"metrics-addr",
@@ -464,6 +564,18 @@ func main() {
 		},
 	}

+	if leaderElectionLeaseDuration > 0 {
+		ctrlOpts.LeaseDuration = &leaderElectionLeaseDuration
+	}
+
+	if leaderElectionRenewDeadline > 0 {
+		ctrlOpts.RenewDeadline = &leaderElectionRenewDeadline
+	}
+
+	if leaderElectionRetryPeriod > 0 {
+		ctrlOpts.RetryPeriod = &leaderElectionRetryPeriod
+	}
+
 	if enablePprof {
 		ctrlOpts.PprofBindAddress = ":8082"
 	}
@@ -704,7 +816,11 @@ func main() {
 			ctrl.Log.WithName("capsule.ctrl").WithName("events"),
 			manager.GetEventRecorder("tenant-controller"),
 			cfg,
-		), webhooksList...); err != nil {
+		),
+		webhook.RegistrationOptions{
+			EnableTracing: enableTracing,
+		},
+		webhooksList...); err != nil {
 		setupLog.Error(err, "unable to setup webhooks")
 		os.Exit(1)
 	}
@@ -845,8 +961,49 @@ func main() {

 	setupLog.Info("starting manager")

-	if err = manager.Start(ctx); err != nil {
-		setupLog.Error(err, "problem running manager")
+	if tracingBasicAuthUsername == "" {
+		tracingBasicAuthUsername = os.Getenv("CAPSULE_TRACING_OTLP_BASIC_AUTH_USERNAME")
+	}
+
+	if tracingBasicAuthPassword == "" {
+		tracingBasicAuthPassword = os.Getenv("CAPSULE_TRACING_OTLP_BASIC_AUTH_PASSWORD")
+	}
+
+	tracingShutdown, err := setupTracing(ctx, tracingOptions{
+		enabled:               enableTracing,
+		endpoint:              tracingEndpoint,
+		insecure:              tracingInsecure,
+		sampleRatio:           tracingSampleRatio,
+		headers:               tracingHeaders,
+		basicAuthUsername:     tracingBasicAuthUsername,
+		basicAuthPassword:     tracingBasicAuthPassword,
+		timeout:               tracingTimeout,
+		compression:           tracingCompression,
+		tlsServerName:         tracingTLSServerName,
+		tlsInsecureSkipVerify: tracingTLSInsecureSkipVerify,
+	})
+	if err != nil {
+		setupLog.Error(err, "unable to initialize tracing")
 		os.Exit(1)
 	}
+
+	if err = manager.Start(ctx); err != nil {
+		setupLog.Error(err, "problem running manager")
+
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		if shutdownErr := tracingShutdown(shutdownCtx); shutdownErr != nil {
+			setupLog.Error(shutdownErr, "unable to shutdown tracing")
+		}
+
+		cancel()
+
+		os.Exit(1)
+	}
+
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	if err := tracingShutdown(shutdownCtx); err != nil {
+		setupLog.Error(err, "unable to shutdown tracing")
+	}
+
+	cancel()
 }
@@ -0,0 +1,152 @@
+// Copyright 2020-2026 Project Capsule Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/base64"
+	"fmt"
+	"maps"
+	"strings"
+	"time"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/sdk/resource"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.37.0"
+	"google.golang.org/grpc/credentials"
+
+	capsuleversion "github.com/projectcapsule/capsule/internal/version"
+)
+
+type tracingOptions struct {
+	enabled               bool
+	endpoint              string
+	insecure              bool
+	sampleRatio           float64
+	headers               map[string]string
+	basicAuthUsername     string
+	basicAuthPassword     string
+	timeout               time.Duration
+	compression           string
+	tlsServerName         string
+	tlsInsecureSkipVerify bool
+}
+
+func setupTracing(ctx context.Context, options tracingOptions) (func(context.Context) error, error) {
+	if !options.enabled {
+		return func(context.Context) error { return nil }, nil
+	}
+
+	if options.sampleRatio < 0 || options.sampleRatio > 1 {
+		return nil, fmt.Errorf("tracing sample ratio must be between 0 and 1, got %v", options.sampleRatio)
+	}
+
+	if (options.basicAuthUsername == "") != (options.basicAuthPassword == "") {
+		return nil, fmt.Errorf("tracing basic auth username and password must be configured together")
+	}
+
+	headers := make(map[string]string, len(options.headers)+1)
+	maps.Copy(headers, options.headers)
+
+	if options.basicAuthUsername != "" {
+		token := base64.StdEncoding.EncodeToString([]byte(options.basicAuthUsername + ":" + options.basicAuthPassword))
+		headers["authorization"] = "Basic " + token
+	}
+
+	exporterOptions := make([]otlptracegrpc.Option, 0, 6)
+
+	if options.endpoint != "" {
+		exporterOptions = append(exporterOptions, otlptracegrpc.WithEndpoint(options.endpoint))
+	}
+
+	if options.insecure {
+		exporterOptions = append(exporterOptions, otlptracegrpc.WithInsecure())
+	} else if options.tlsServerName != "" || options.tlsInsecureSkipVerify {
+		exporterOptions = append(exporterOptions, otlptracegrpc.WithTLSCredentials(credentials.NewTLS(&tls.Config{
+			ServerName:         options.tlsServerName,
+			InsecureSkipVerify: options.tlsInsecureSkipVerify, //nolint:gosec
+		})))
+	}
+
+	if len(headers) > 0 {
+		exporterOptions = append(exporterOptions, otlptracegrpc.WithHeaders(headers))
+	}
+
+	if options.timeout > 0 {
+		exporterOptions = append(exporterOptions, otlptracegrpc.WithTimeout(options.timeout))
+	}
+
+	if options.compression != "" {
+		if options.compression != "gzip" {
+			return nil, fmt.Errorf("unsupported tracing compression %q, supported values: gzip", options.compression)
+		}
+
+		exporterOptions = append(exporterOptions, otlptracegrpc.WithCompressor(options.compression))
+	}
+
+	exporter, err := otlptracegrpc.New(ctx, exporterOptions...)
+	if err != nil {
+		return nil, fmt.Errorf("create OTLP trace exporter: %w", err)
+	}
+
+	res, err := resource.Merge(
+		resource.Default(),
+		resource.NewWithAttributes(
+			semconv.SchemaURL,
+			semconv.ServiceName("capsule"),
+			semconv.ServiceVersion(capsuleversion.GitTag),
+		),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("create OpenTelemetry resource: %w", err)
+	}
+
+	provider := sdktrace.NewTracerProvider(
+		sdktrace.WithBatcher(exporter),
+		sdktrace.WithResource(res),
+		sdktrace.WithSampler(sdktrace.ParentBased(sdktrace.TraceIDRatioBased(options.sampleRatio))),
+	)
+
+	otel.SetTracerProvider(provider)
+	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
+		propagation.TraceContext{},
+		propagation.Baggage{},
+	))
+
+	return provider.Shutdown, nil
+}
+
+type tracingHeadersFlag map[string]string
+
+func (f tracingHeadersFlag) String() string {
+	if len(f) == 0 {
+		return ""
+	}
+
+	items := make([]string, 0, len(f))
+	for key, value := range f {
+		items = append(items, key+"="+value)
+	}
+
+	return strings.Join(items, ",")
+}
+
+func (f tracingHeadersFlag) Type() string {
+	return "key=value"
+}
+
+func (f tracingHeadersFlag) Set(value string) error {
+	key, headerValue, found := strings.Cut(value, "=")
+	if !found || key == "" {
+		return fmt.Errorf("tracing header must use key=value format")
+	}
+
+	f[key] = headerValue
+
+	return nil
+}