Add basic metrics for the controller engine

This'll let folks monitor things like:

* How many XR controllers are running
* How many watches are running
* Whether an XR controller is flapping
* Whether an XR controller is watching its XR and CompRev
* How many types of composed resource an XR controller is watching

Signed-off-by: Nic Cope <[email protected]>

Author: negz

cmd/crossplane/core/core.go

@@ -244,14 +244,14 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		return errors.Wrap(err, "cannot load client TLS certificates")
 	}
 
-	m := xfn.NewMetrics()
-	metrics.Registry.MustRegister(m)
+	xfnm := xfn.NewPrometheusMetrics()
+	metrics.Registry.MustRegister(xfnm)
 
 	// We want all XR controllers to share the same gRPC clients.
 	functionRunner := xfn.NewPackagedFunctionRunner(mgr.GetClient(),
 		xfn.WithLogger(log),
 		xfn.WithTLSConfig(clienttls),
-		xfn.WithInterceptorCreators(m),
+		xfn.WithInterceptorCreators(xfnm),
 	)
 
 	// Periodically remove clients for Functions that no longer exist.
@@ -375,6 +375,9 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		return errors.Wrap(err, "cannot create uncached client for API extension controllers")
 	}
 
+	cem := engine.NewPrometheusMetrics()
+	metrics.Registry.MustRegister(cem)
+
 	// It's important the engine's client is wrapped with unstructured.NewClient
 	// because controller-runtime always caches *unstructured.Unstructured, not
 	// our wrapper types like *composite.Unstructured. This client takes care of
@@ -384,6 +387,7 @@ func (c *startCommand) Run(s *runtime.Scheme, log logging.Logger) error { //noli
 		unstructured.NewClient(cached),
 		unstructured.NewClient(uncached),
 		engine.WithLogger(log),
+		engine.WithMetrics(cem),
 	)
 
 	// TODO(negz): Garbage collect informers for CRs that are still defined

internal/engine/engine.go

@@ -61,13 +61,12 @@ type ControllerEngine struct {
 	// exists.
 	uncached client.Client
 
-	log logging.Logger
-
-	// Protects everything below.
-	mx sync.RWMutex
-
-	// Running controllers, by name.
+	// Running controllers, by name. Protected by the mutex.
 	controllers map[string]*controller
+	mx          sync.RWMutex
+
+	log     logging.Logger
+	metrics Metrics
 }
 
 // TrackingInformers is a set of Informers. It tracks which are active.
@@ -76,15 +75,31 @@ type TrackingInformers interface {
 	ActiveInformers() []schema.GroupVersionKind
 }
 
+// Metrics for the controller engine.
+type Metrics interface {
+	// ControllerStarted records a controller start.
+	ControllerStarted(name string)
+
+	// ControllerStopped records a controller stop.
+	ControllerStopped(name string)
+
+	// WatchStarted records a watch start for a controller.
+	WatchStarted(name string, t WatchType)
+
+	// WatchStopped records a watch stop for a controller.
+	WatchStopped(name string, t WatchType)
+}
+
 // New creates a new controller engine.
 func New(mgr manager.Manager, infs TrackingInformers, c client.Client, nc client.Client, o ...ControllerEngineOption) *ControllerEngine {
 	e := &ControllerEngine{
 		mgr:         mgr,
 		infs:        infs,
 		cached:      c,
 		uncached:    nc,
-		log:         logging.NewNopLogger(),
 		controllers: make(map[string]*controller),
+		log:         logging.NewNopLogger(),
+		metrics:     &NopMetrics{},
 	}
 
 	for _, fn := range o {
@@ -104,6 +119,13 @@ func WithLogger(l logging.Logger) ControllerEngineOption {
 	}
 }
 
+// WithMetrics configures an Engine to expose metrics.
+func WithMetrics(m Metrics) ControllerEngineOption {
+	return func(e *ControllerEngine) {
+		e.metrics = m
+	}
+}
+
 type controller struct {
 	// The running controller.
 	ctrl kcontroller.Controller
@@ -213,6 +235,7 @@ func (e *ControllerEngine) Start(name string, o ...ControllerOption) error {
 		<-e.mgr.Elected()
 
 		e.log.Debug("Starting new controller", "controller", name)
+		e.metrics.ControllerStarted(name)
 
 		// Run the controller until its context is cancelled.
 		if err := c.Start(ctx); err != nil {
@@ -225,6 +248,7 @@ func (e *ControllerEngine) Start(name string, o ...ControllerOption) error {
 		}
 
 		e.log.Debug("Stopped controller", "controller", name)
+		e.metrics.ControllerStopped(name)
 	}()
 
 	if co.gc != nil {
@@ -281,7 +305,6 @@ func (e *ControllerEngine) Stop(ctx context.Context, name string) error {
 	c.cancel()
 	delete(e.controllers, name)
 
-	e.log.Debug("Stopped controller", "controller", name)
 	return nil
 }
 
@@ -441,6 +464,7 @@ func (e *ControllerEngine) StartWatches(ctx context.Context, name string, ws ...
 		c.sources[wid] = src
 
 		e.log.Debug("Started watching GVK", "controller", name, "watch-type", wid.Type, "watched-gvk", wid.GVK)
+		e.metrics.WatchStarted(name, wid.Type)
 	}
 
 	return nil
@@ -510,6 +534,7 @@ func (e *ControllerEngine) StopWatches(ctx context.Context, name string, ws ...W
 		}
 		delete(c.sources, wid)
 		e.log.Debug("Stopped watching GVK", "controller", name, "watch-type", wid.Type, "watched-gvk", wid.GVK)
+		e.metrics.WatchStopped(name, wid.Type)
 		stopped++
 	}
 

internal/engine/engine_metrics.go

@@ -0,0 +1,113 @@
+/*
+Copyright 2025 The Crossplane Authors.
+
+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 engine
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// NopMetrics does nothing.
+type NopMetrics struct{}
+
+// ControllerStarted does nothing.
+func (m *NopMetrics) ControllerStarted(_ string) {}
+
+// ControllerStopped does nothing.
+func (m *NopMetrics) ControllerStopped(_ string) {}
+
+// WatchStarted does nothing.
+func (m *NopMetrics) WatchStarted(_ string, _ WatchType) {}
+
+// WatchStopped does nothing.
+func (m *NopMetrics) WatchStopped(_ string, _ WatchType) {}
+
+// PrometheusMetrics for the controller engine.
+type PrometheusMetrics struct {
+	controllersStarted *prometheus.CounterVec
+	controllersStopped *prometheus.CounterVec
+
+	watchesStarted *prometheus.CounterVec
+	watchesStopped *prometheus.CounterVec
+}
+
+// NewPrometheusMetrics exposes controller engine metrics via Prometheus.
+func NewPrometheusMetrics() *PrometheusMetrics {
+	return &PrometheusMetrics{
+		controllersStarted: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "controllers_started_total",
+			Help:      "Total number of XR controllers started.",
+		}, []string{"controller"}),
+
+		controllersStopped: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "controllers_stopped_total",
+			Help:      "Total number of XR controllers stopped.",
+		}, []string{"controller"}),
+
+		watchesStarted: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "watches_started_total",
+			Help:      "Total number of watches started.",
+		}, []string{"controller", "type"}),
+
+		watchesStopped: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "composition",
+			Name:      "watches_stopped_total",
+			Help:      "Total number of watches stopped.",
+		}, []string{"controller", "type"}),
+	}
+}
+
+// ControllerStarted records a controller start.
+func (m *PrometheusMetrics) ControllerStarted(name string) {
+	m.controllersStarted.With(prometheus.Labels{"controller": name}).Inc()
+}
+
+// ControllerStopped records a controller stop.
+func (m *PrometheusMetrics) ControllerStopped(name string) {
+	m.controllersStopped.With(prometheus.Labels{"controller": name}).Inc()
+}
+
+// WatchStarted records a watch start for a controller.
+func (m *PrometheusMetrics) WatchStarted(name string, t WatchType) {
+	m.watchesStarted.With(prometheus.Labels{"controller": name, "type": string(t)}).Inc()
+}
+
+// WatchStopped records a watch stop for a controller.
+func (m *PrometheusMetrics) WatchStopped(name string, t WatchType) {
+	m.watchesStopped.With(prometheus.Labels{"controller": name, "type": string(t)}).Inc()
+}
+
+// Describe sends the super-set of all possible descriptors of metrics
+// collected by this Collector to the provided channel and returns once
+// the last descriptor has been sent.
+func (m *PrometheusMetrics) Describe(ch chan<- *prometheus.Desc) {
+	m.controllersStarted.Describe(ch)
+	m.controllersStopped.Describe(ch)
+	m.watchesStarted.Describe(ch)
+	m.watchesStopped.Describe(ch)
+}
+
+// Collect is called by the Prometheus registry when collecting
+// metrics. The implementation sends each collected metric via the
+// provided channel and returns once the last metric has been sent.
+func (m *PrometheusMetrics) Collect(ch chan<- prometheus.Metric) {
+	m.controllersStarted.Collect(ch)
+	m.controllersStopped.Collect(ch)
+	m.watchesStarted.Collect(ch)
+	m.watchesStopped.Collect(ch)
+}

internal/xfn/function_runner_metrics.go

@@ -26,17 +26,17 @@ import (
 	fnv1 "github.com/crossplane/crossplane/apis/apiextensions/fn/proto/v1"
 )
 
-// Metrics are requests, errors, and duration (RED) metrics for composition
+// PrometheusMetrics are requests, errors, and duration (RED) metrics for composition
 // function runs.
-type Metrics struct {
+type PrometheusMetrics struct {
 	requests  *prometheus.CounterVec
 	responses *prometheus.CounterVec
 	duration  *prometheus.HistogramVec
 }
 
-// NewMetrics creates metrics for composition function runs.
-func NewMetrics() *Metrics {
-	return &Metrics{
+// NewPrometheusMetrics creates metrics for composition function runs.
+func NewPrometheusMetrics() *PrometheusMetrics {
+	return &PrometheusMetrics{
 		requests: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Subsystem: "composition",
 			Name:      "run_function_request_total",
@@ -61,7 +61,7 @@ func NewMetrics() *Metrics {
 // Describe sends the super-set of all possible descriptors of metrics
 // collected by this Collector to the provided channel and returns once
 // the last descriptor has been sent.
-func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
+func (m *PrometheusMetrics) Describe(ch chan<- *prometheus.Desc) {
 	m.requests.Describe(ch)
 	m.responses.Describe(ch)
 	m.duration.Describe(ch)
@@ -70,15 +70,15 @@ func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
 // Collect is called by the Prometheus registry when collecting
 // metrics. The implementation sends each collected metric via the
 // provided channel and returns once the last metric has been sent.
-func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
+func (m *PrometheusMetrics) Collect(ch chan<- prometheus.Metric) {
 	m.requests.Collect(ch)
 	m.responses.Collect(ch)
 	m.duration.Collect(ch)
 }
 
 // CreateInterceptor returns a gRPC UnaryClientInterceptor for the named
 // function. The supplied package (pkg) should be the package's OCI reference.
-func (m *Metrics) CreateInterceptor(name, pkg string) grpc.UnaryClientInterceptor {
+func (m *PrometheusMetrics) CreateInterceptor(name, pkg string) grpc.UnaryClientInterceptor {
 	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
 		l := prometheus.Labels{"function_name": name, "function_package": pkg, "grpc_target": cc.Target(), "grpc_method": method}