diff --git a/cmd/atecontroller/main.go b/cmd/atecontroller/main.go index f9a4bbd298..87ce83a43a 100644 --- a/cmd/atecontroller/main.go +++ b/cmd/atecontroller/main.go @@ -36,6 +36,7 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/fields" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/client-go/kubernetes" clientgoscheme "k8s.io/client-go/kubernetes/scheme" @@ -44,6 +45,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/healthz" ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.) _ "k8s.io/client-go/plugin/pkg/client/auth" @@ -91,7 +93,14 @@ func init() { utilruntime.Must(clientv1alpha1.AddToScheme(scheme)) // Register our CRD } -const serviceName = "atecontroller" +const ( + serviceName = "atecontroller" + + // metricsPullAddr is the manager's metrics listener, and metricsOffAddr + // turns it off. + metricsPullAddr = ":8080" + metricsOffAddr = "0" +) // logr verbosity V(n) maps to slog level -n, so V(1) stays below Info until // --log-level=debug. logr carries no context, so these records have no trace IDs. @@ -99,6 +108,34 @@ func newControllerRuntimeLogger(h slog.Handler) logr.Logger { return logr.FromSlogHandler(h) } +// metricsBindAddr is the manager's metrics address. servePull is whether +// OTEL_METRICS_EXPORTER leaves the listener on. +func metricsBindAddr(servePull bool) string { + if servePull { + return metricsPullAddr + } + return metricsOffAddr +} + +// managerOptions configures the controller-runtime manager. +func managerOptions(egressMITMCAPool types.NamespacedName, servePull bool) ctrl.Options { + return ctrl.Options{ + Scheme: scheme, + Metrics: metricsserver.Options{BindAddress: metricsBindAddr(servePull)}, + Cache: cache.Options{ + ByObject: map[client.Object]cache.ByObject{ + &corev1.Secret{}: { + Namespaces: map[string]cache.Config{ + egressMITMCAPool.Namespace: { + FieldSelector: fields.OneTermEqualSelector("metadata.name", egressMITMCAPool.Name), + }, + }, + }, + }, + }, + } +} + func main() { pflag.Parse() if *showVersion { @@ -131,13 +168,15 @@ func main() { defer serverboot.ShutdownProvider("TracerProvider", tp.Shutdown) // controller-runtime records reconcile, workqueue, and runtime metrics into its - // own Prometheus registry, which the manager serves. On the OTLP path the - // bridged queue histograms are padded so the Telemetry API accepts idle ones. - mp, err := serverboot.InitMetricsBridged(ctx, serviceName, ctrlmetrics.Registry, padEmptyExponentialHistograms) + // own Prometheus registry. OTEL_METRICS_EXPORTER picks where they go: the + // OTLP push, which pads the bridged queue histograms so the Telemetry API + // accepts idle ones, and the manager's scrape listener. + mp, servePull, err := serverboot.InitMetricsBridged(ctx, serviceName, ctrlmetrics.Registry, padEmptyExponentialHistograms) if err != nil { serverboot.Fatal(ctx, "Failed to initialize metrics", err) } defer serverboot.ShutdownProvider("MeterProvider", mp.Shutdown) + slog.InfoContext(ctx, "Metrics listener", slog.Bool("enabled", servePull), slog.String("addr", metricsBindAddr(servePull))) k8sConfig := ctrl.GetConfigOrDie() k8sClient, err := kubernetes.NewForConfig(k8sConfig) @@ -184,20 +223,7 @@ func main() { // EgressMITMTrustReconciler watches the Secret `egress-mitm-ca-pool`. systemNamespace := installdefaults.NamespaceFromPodEnv() egressMITMCAPool := controllers.EgressMITMCAPoolRef(systemNamespace) - mgr, err := ctrl.NewManager(k8sConfig, ctrl.Options{ - Scheme: scheme, - Cache: cache.Options{ - ByObject: map[client.Object]cache.ByObject{ - &corev1.Secret{}: { - Namespaces: map[string]cache.Config{ - egressMITMCAPool.Namespace: { - FieldSelector: fields.OneTermEqualSelector("metadata.name", egressMITMCAPool.Name), - }, - }, - }, - }, - }, - }) + mgr, err := ctrl.NewManager(k8sConfig, managerOptions(egressMITMCAPool, servePull)) if err != nil { setupLog.Error(err, "unable to start manager") os.Exit(1) diff --git a/cmd/atecontroller/main_test.go b/cmd/atecontroller/main_test.go index 7d95bff394..0068d361b5 100644 --- a/cmd/atecontroller/main_test.go +++ b/cmd/atecontroller/main_test.go @@ -21,7 +21,11 @@ import ( "testing" prombridge "go.opentelemetry.io/contrib/bridges/prometheus" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/cache" ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" ) func TestNewControllerRuntimeLoggerVerbosity(t *testing.T) { @@ -78,3 +82,61 @@ func TestBridgedRegistryProducesBeforeManagerStart(t *testing.T) { t.Errorf("no go_* family in %v", names) } } + +// The manager opens its metrics listener only when OTEL_METRICS_EXPORTER leaves +// the scrape endpoint on. +func TestManagerOptionsMetricsListener(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + servePull bool + wantAddr string + wantServed bool + }{ + {"serving", true, metricsPullAddr, true}, + {"disabled", false, metricsOffAddr, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + opts := managerOptions(types.NamespacedName{Namespace: "ate-system", Name: "pool"}, tt.servePull) + + if opts.Metrics.BindAddress != tt.wantAddr { + t.Errorf("metrics address = %q, want %q", opts.Metrics.BindAddress, tt.wantAddr) + } + srv, err := metricsserver.NewServer(opts.Metrics, nil, nil) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + if served := srv != nil; served != tt.wantServed { + t.Errorf("serves a metrics listener = %t, want %t", served, tt.wantServed) + } + }) + } +} + +func TestManagerOptionsScopeSecretCache(t *testing.T) { + t.Parallel() + + ref := types.NamespacedName{Namespace: "ate-system", Name: "pool"} + opts := managerOptions(ref, true) + + var byObject cache.ByObject + found := false + for obj, by := range opts.Cache.ByObject { + if _, isSecret := obj.(*corev1.Secret); isSecret { + byObject, found = by, true + } + } + if !found { + t.Fatal("no cache scope for Secrets") + } + cfg, ok := byObject.Namespaces[ref.Namespace] + if !ok || len(byObject.Namespaces) != 1 { + t.Fatalf("Secret cache namespaces = %v, want only %q", byObject.Namespaces, ref.Namespace) + } + if got := cfg.FieldSelector.String(); got != "metadata.name="+ref.Name { + t.Errorf("Secret field selector = %q, want %q", got, "metadata.name="+ref.Name) + } +} diff --git a/docs/dev/best-practices/metrics.md b/docs/dev/best-practices/metrics.md index d39232f572..9990b2a7e0 100644 --- a/docs/dev/best-practices/metrics.md +++ b/docs/dev/best-practices/metrics.md @@ -278,9 +278,9 @@ component name as the scope (`"atelet"`, `"ateapi"`, `"atecontroller"`): The meter provider itself is set up once per binary by `serverboot.InitMetrics` (Prometheus reader plus OTLP push), `serverboot.InitMetricsBridged` (OTLP push with a foreign Prometheus registry -bridged onto it, or that registry alone under `OTEL_METRICS_EXPORTER=none`; -atecontroller), or `serverboot.InitMetricsPushOnlyVia` (OTLP push only, over -the atelet relay when given one; ateom). +bridged onto it, or that registry alone under `OTEL_METRICS_EXPORTER=prometheus`, +and whether to serve it; atecontroller), or `serverboot.InitMetricsPushOnlyVia` +(OTLP push only, over the atelet relay when given one; ateom). A new component calls one of these and defers `ShutdownProvider`; a new package inside an existing component adds nothing there. diff --git a/docs/observability.md b/docs/observability.md index 042b8007ba..5cee8a8350 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -344,7 +344,7 @@ Weaver permits only `groups` and `imports` at the top level of a registry file, ### Bridged controller-runtime metrics (atecontroller) -atecontroller bridges controller-runtime's private Prometheus registry, which the manager serves on an unscraped `:8080`, onto its OTLP reader. So `controller_runtime_*`, `workqueue_*`, `certwatcher_*`, `rest_client_*`, `leader_election_*`, `go_*`, and `process_*` reach the collector too, keeping their Prometheus names because they are upstream instruments and renaming them would break existing controller-runtime dashboards. +atecontroller bridges controller-runtime's private Prometheus registry, which the manager serves on `:8080` only while `OTEL_METRICS_EXPORTER` leaves it on (see [Scraping instead of pushing](#scraping-instead-of-pushing)), onto its OTLP reader. So `controller_runtime_*`, `workqueue_*`, `certwatcher_*`, `rest_client_*`, `leader_election_*`, `go_*`, and `process_*` reach the collector too, keeping their Prometheus names because they are upstream instruments and renaming them would break existing controller-runtime dashboards. These can be used to answer whether the controller is keeping up, e.g. rising `workqueue_depth` or `workqueue_queue_duration_seconds` means reconciles are falling behind, and `controller_runtime_reconcile_errors_total` says which controller. @@ -356,7 +356,9 @@ A queue that has never processed an item bridges as an exponential histogram wit ### Scraping instead of pushing -ateapi, atelet, atenet-router and the credential provider also serve every instrument on their Prometheus `/metrics` endpoint. A cluster that scrapes those endpoints sets `OTEL_METRICS_EXPORTER=none` on the components, so each series reaches the backend once. With `none` the components install no OTLP metric reader and keep the Prometheus one; traces and logs are unaffected. atecontroller then registers its instruments (`ate.workerpool.*`) on controller-runtime's registry, so the manager's `:8080` serves them next to the controller-runtime families. ateom serves no endpoint of its own, so leave the variable unset on the worker pods, or it exports no metrics at all. The variable takes a comma-separated list, with the same rules as `OTEL_LOGS_EXPORTER`. `otlp` (the default) keeps the push, and `prometheus` or `none` stops it. `/metrics` stays on for each value. ateom serves no `/metrics`, so there `prometheus` is skipped as unknown with a warning, and the push stays on unless the variable is `none`. +ateapi, atelet, atenet-router and the credential provider also serve every instrument on their Prometheus `/metrics` endpoint. A cluster that scrapes those endpoints sets `OTEL_METRICS_EXPORTER=prometheus` on the components, so each series reaches the backend once. With `prometheus` the components install no OTLP metric reader and keep the Prometheus one; traces and logs are unaffected. `none` stops the push too. For now `/metrics` stays on for every value, including `otlp` and `none`. This is temporary until the probes move off `:9090`; `otlp` and `none` will then turn it off ([#2067](https://github.com/agent-substrate/substrate/issues/2067)). The variable takes a comma-separated list, with the same rules as `OTEL_LOGS_EXPORTER`. Unset is `otlp,prometheus`, or `otlp` alone on ateom. `otlp` keeps the push, and `prometheus` or `none` stops it. ateom serves no `/metrics`, so there `prometheus` is skipped as unknown with a warning, and the push stays on unless the variable is `none`. + +atecontroller's `:8080` listener follows the variable. When the variable is unset or `otlp,prometheus`, atecontroller pushes every metric over OTLP, and `:8080` serves only the controller-runtime families ([#2402](https://github.com/agent-substrate/substrate/issues/2402)). With `otlp`, atecontroller pushes every metric and closes `:8080`. With `prometheus`, atecontroller does not push, and `:8080` serves the controller-runtime families and the `ate.workerpool.*` instruments. With `none`, atecontroller does not push and closes `:8080`. ### Local Metrics with Prometheus (Kind Cluster) diff --git a/internal/serverboot/serverboot.go b/internal/serverboot/serverboot.go index f673b6ca7c..43eb34accf 100644 --- a/internal/serverboot/serverboot.go +++ b/internal/serverboot/serverboot.go @@ -208,17 +208,27 @@ const ( tracesExporterEnv = "OTEL_TRACES_EXPORTER" ) -// metricsPushEnabled applies OTEL_METRICS_EXPORTER with the shared exporter -// rules: the OTLP reader is installed while otlp is selected, as it is when the -// variable is unset. prometheus, the specification's pull exporter, is known -// only when pullServed, i.e. the component serves its own /metrics; elsewhere -// it is skipped as unknown, so a push-only component keeps exporting. -func metricsPushEnabled(ctx context.Context, pullServed bool) bool { +// metricsExporters applies OTEL_METRICS_EXPORTER with the shared exporter +// rules. prometheus, the specification's pull exporter, is known only when +// pullServed, i.e. the component serves its own /metrics; elsewhere it is +// skipped as unknown, so a push-only component keeps exporting. Unset selects +// every exporter the component has. +func metricsExporters(ctx context.Context, pullServed bool) Exporters { known := []string{ExporterOTLP} + def := Exporters{ExporterOTLP: true} if pullServed { known = append(known, exporterPrometheus) + def[exporterPrometheus] = true } - return resolveExportersEnv(ctx, metricsExporterEnv, known, Exporters{ExporterOTLP: true}).Has(ExporterOTLP) + exporters := resolveExportersEnv(ctx, metricsExporterEnv, known, def) + slog.InfoContext(ctx, "Metrics exporters resolved", slog.String("exporters", exporters.String())) + return exporters +} + +// metricsPushEnabled reports whether the OTLP reader is installed: while otlp +// is selected, as it is when the variable is unset. +func metricsPushEnabled(ctx context.Context, pullServed bool) bool { + return metricsExporters(ctx, pullServed).Has(ExporterOTLP) } // tracesPushEnabled applies OTEL_TRACES_EXPORTER with the shared exporter @@ -247,31 +257,45 @@ func InitMetrics(ctx context.Context, serviceName string) (*sdkmetric.MeterProvi // it does not own (atecontroller: controller-runtime's), so every metric // reaches the backend exactly once on whichever path is active. With the OTLP // export, reg is bridged onto the push and the OTel instruments stay off it, -// since the bridge would otherwise push them a second time. When -// OTEL_METRICS_EXPORTER does not select otlp, nothing is pushed and the OTel -// instruments register on reg, so its endpoint serves both. +// since the bridge would otherwise push them a second time. With prometheus +// alone, nothing is pushed and the OTel instruments register on reg, so its +// endpoint serves both. +// +// servePull reports whether OTEL_METRICS_EXPORTER leaves the scraped endpoint +// on, which the caller must then serve from reg: unset, prometheus, or a list +// with it. otlp and none turn it off, and with none no reader is installed. // // wrapProducer, when non-nil, wraps the bridge producer on the OTLP path, for // a rewrite the push backend needs but the scraped endpoint does not. func InitMetricsBridged(ctx context.Context, serviceName string, reg interface { promclient.Registerer promclient.Gatherer -}, wrapProducer func(sdkmetric.Producer) sdkmetric.Producer) (*sdkmetric.MeterProvider, error) { +}, wrapProducer func(sdkmetric.Producer) sdkmetric.Producer) (mp *sdkmetric.MeterProvider, servePull bool, err error) { if serviceName == "" { - return nil, fmt.Errorf("serviceName is required") + return nil, false, fmt.Errorf("serviceName is required") } - if metricsPushEnabled(ctx, true) { + exporters := metricsExporters(ctx, true) + servePull = exporters.Has(exporterPrometheus) + switch { + case exporters.Has(ExporterOTLP): producer := prombridge.NewMetricProducer(prombridge.WithGatherer(reg)) if wrapProducer != nil { producer = wrapProducer(producer) } - return newMeterProvider(ctx, serviceName, true, false, nil, []sdkmetric.Producer{producer}) + mp, err = newMeterProvider(ctx, serviceName, true, false, nil, []sdkmetric.Producer{producer}) + case servePull: + var promReader *prometheus.Exporter + if promReader, err = prometheus.New(prometheus.WithRegisterer(reg)); err != nil { + return nil, false, fmt.Errorf("create Prometheus metric exporter: %w", err) + } + mp, err = newMeterProvider(ctx, serviceName, false, false, nil, nil, promReader) + default: + mp, err = newMeterProvider(ctx, serviceName, false, false, nil, nil) } - promReader, err := prometheus.New(prometheus.WithRegisterer(reg)) if err != nil { - return nil, fmt.Errorf("create Prometheus metric exporter: %w", err) + return nil, false, err } - return newMeterProvider(ctx, serviceName, false, false, nil, nil, promReader) + return mp, servePull, nil } // InitMetricsPushOnlyVia is InitMetrics without the Prometheus reader, for a diff --git a/internal/serverboot/serverboot_test.go b/internal/serverboot/serverboot_test.go index b685a434b0..a7b41f2526 100644 --- a/internal/serverboot/serverboot_test.go +++ b/internal/serverboot/serverboot_test.go @@ -439,23 +439,32 @@ func bridgedRegistry(t *testing.T) *prometheus.Registry { return reg } -// With the OTLP export, the bridged family and the OTel instrument are each -// pushed once. The instrument must stay off the bridged registry: there the -// bridge would push it a second time, under its Prometheus name. +// With the OTLP export, whether selected or the default, the bridged family and +// the OTel instrument are each pushed once. The instrument must stay off the +// bridged registry: there the bridge would push it a second time, under its +// Prometheus name. func TestInitMetricsBridgedPushesEachMetricOnce(t *testing.T) { - collector := startMetricsCollector(t, "otlp") - reg := bridgedRegistry(t) - mp, err := InitMetricsBridged(t.Context(), "test-bridged-otlp", reg, nil) - if err != nil { - t.Fatalf("InitMetricsBridged: %v", err) - } - addOne(t, mp, "ate.test.bridged.count") - if gatheredNames(t, reg)["ate_test_bridged_count_total"] { - t.Error("the OTel instrument is registered on the bridged registry") - } - pushed := collector.flush(t, mp) - if pushed["ate.test.bridged.count"] != 1 || pushed["test_bridged_family"] != 1 || pushed["ate_test_bridged_count_total"] != 0 { - t.Errorf("want ate.test.bridged.count and test_bridged_family pushed once each and nothing else for them, got %v", pushed) + for _, tt := range []struct{ name, value string }{ + {"unset", ""}, + {"otlp", "otlp"}, + {"list", "otlp,prometheus"}, + } { + t.Run(tt.name, func(t *testing.T) { + collector := startMetricsCollector(t, tt.value) + reg := bridgedRegistry(t) + mp, _, err := InitMetricsBridged(t.Context(), "test-bridged-otlp", reg, nil) + if err != nil { + t.Fatalf("InitMetricsBridged: %v", err) + } + addOne(t, mp, "ate.test.bridged.count") + if gatheredNames(t, reg)["ate_test_bridged_count_total"] { + t.Error("the OTel instrument is registered on the bridged registry") + } + pushed := collector.flush(t, mp) + if pushed["ate.test.bridged.count"] != 1 || pushed["test_bridged_family"] != 1 || pushed["ate_test_bridged_count_total"] != 0 { + t.Errorf("want ate.test.bridged.count and test_bridged_family pushed once each and nothing else for them, got %v", pushed) + } + }) } } @@ -469,7 +478,7 @@ func TestInitMetricsBridgedWrapsPushedProducer(t *testing.T) { return inner.Produce(ctx) }) } - mp, err := InitMetricsBridged(t.Context(), "test-bridged-wrap", bridgedRegistry(t), wrap) + mp, _, err := InitMetricsBridged(t.Context(), "test-bridged-wrap", bridgedRegistry(t), wrap) if err != nil { t.Fatalf("InitMetricsBridged: %v", err) } @@ -486,27 +495,82 @@ type producerFunc func(context.Context) ([]metricdata.ScopeMetrics, error) func (f producerFunc) Produce(ctx context.Context) ([]metricdata.ScopeMetrics, error) { return f(ctx) } -// With OTEL_METRICS_EXPORTER=none, the OTel instrument is served from the +// With OTEL_METRICS_EXPORTER=prometheus, the OTel instrument is served from the // bridged registry next to its own families, and nothing is pushed. -func TestInitMetricsBridgedExporterNoneServesFromRegistry(t *testing.T) { - collector := startMetricsCollector(t, "none") +func TestInitMetricsBridgedPrometheusServesFromRegistry(t *testing.T) { + collector := startMetricsCollector(t, "prometheus") reg := bridgedRegistry(t) - mp, err := InitMetricsBridged(t.Context(), "test-bridged-none", reg, nil) + mp, servePull, err := InitMetricsBridged(t.Context(), "test-bridged-prometheus", reg, nil) if err != nil { t.Fatalf("InitMetricsBridged: %v", err) } - addOne(t, mp, "ate.test.bridgednone.count") + if !servePull { + t.Error("OTEL_METRICS_EXPORTER=prometheus turned the scrape endpoint off") + } + addOne(t, mp, "ate.test.bridgedprom.count") names := gatheredNames(t, reg) - if !names["ate_test_bridgednone_count_total"] || !names["test_bridged_family"] { + if !names["ate_test_bridgedprom_count_total"] || !names["test_bridged_family"] { t.Errorf("the bridged registry does not serve both the instrument and its own family: %v", names) } + if pushed := collector.flush(t, mp); pushed != nil { + t.Errorf("OTEL_METRICS_EXPORTER=prometheus still pushed %v", pushed) + } +} + +// With OTEL_METRICS_EXPORTER=none there is no push and no endpoint to serve, +// so the instrument is not registered anywhere. +func TestInitMetricsBridgedExporterNoneExportsNothing(t *testing.T) { + collector := startMetricsCollector(t, "none") + reg := bridgedRegistry(t) + mp, servePull, err := InitMetricsBridged(t.Context(), "test-bridged-none", reg, nil) + if err != nil { + t.Fatalf("InitMetricsBridged: %v", err) + } + if servePull { + t.Error("OTEL_METRICS_EXPORTER=none left the scrape endpoint on") + } + addOne(t, mp, "ate.test.bridgednone.count") + if gatheredNames(t, reg)["ate_test_bridgednone_count_total"] { + t.Error("the instrument is registered on the bridged registry") + } if pushed := collector.flush(t, mp); pushed != nil { t.Errorf("OTEL_METRICS_EXPORTER=none still pushed %v", pushed) } } +// servePull follows the policy table: unset and the list keep both paths. +func TestInitMetricsBridgedServePull(t *testing.T) { + tests := []struct { + name string + value string + wantPush bool + wantPull bool + }{ + {"unset", "", true, true}, + {"otlp", "otlp", true, false}, + {"prometheus", "prometheus", false, true}, + {"list", "otlp,prometheus", true, true}, + {"none", "none", false, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + collector := startMetricsCollector(t, tt.value) + mp, servePull, err := InitMetricsBridged(t.Context(), "test-bridged-table", bridgedRegistry(t), nil) + if err != nil { + t.Fatalf("InitMetricsBridged: %v", err) + } + if servePull != tt.wantPull { + t.Errorf("servePull = %t, want %t", servePull, tt.wantPull) + } + if pushed := collector.flush(t, mp) != nil; pushed != tt.wantPush { + t.Errorf("pushed = %t, want %t", pushed, tt.wantPush) + } + }) + } +} + func TestInitMetricsBridgedRequiresServiceName(t *testing.T) { - if _, err := InitMetricsBridged(t.Context(), "", prometheus.NewRegistry(), nil); err == nil { + if _, _, err := InitMetricsBridged(t.Context(), "", prometheus.NewRegistry(), nil); err == nil { t.Error("InitMetricsBridged(\"\") must return an error") } }