plugin: Move reconcile queue callbacks to metrics

This removes a layer of indirection via the plugin methods, which means
we no longer have to require that the plugin is initialized before any
of the methods are called.

Author: sharnoff

pkg/plugin/entrypoint.go

@@ -54,6 +54,8 @@ func NewAutoscaleEnforcerPlugin(
 	promReg := prometheus.NewRegistry()
 	metrics.RegisterDefaultCollectors(promReg)
 
+	pluginMetrics := metrics.BuildPluginMetrics(promReg, config.NodeMetricLabels)
+
 	// pre-define this so that we can reference it in the handlers, knowing that it won't be used
 	// until we start the workers (which we do *after* we've set this value).
 	var pluginState *PluginState
@@ -78,21 +80,13 @@ func NewAutoscaleEnforcerPlugin(
 		},
 		reconcile.WithBaseContext(ctx),
 		reconcile.WithMiddleware(initEvents),
-		// Note: we need one layer of indirection for callbacks referencing pluginState, because
-		// it's initialized later, so directly referencing the methods at this point will use the
-		// nil pluginState and panic on use.
-		reconcile.WithQueueWaitDurationCallback(func(duration time.Duration) {
-			pluginState.reconcileQueueWaitCallback(duration)
-		}),
-		reconcile.WithResultCallback(func(params reconcile.ObjectParams, duration time.Duration, err error) {
-			pluginState.reconcileResultCallback(params, duration, err)
-		}),
+		reconcile.WithQueueWaitDurationCallback(pluginMetrics.Reconcile.QueueWaitDurationCallback),
+		reconcile.WithResultCallback(pluginMetrics.Reconcile.ResultCallback),
 		reconcile.WithErrorStatsCallback(func(params reconcile.ObjectParams, stats reconcile.ErrorStats) {
-			pluginState.reconcileErrorStatsCallback(logger, params, stats)
-		}),
-		reconcile.WithPanicCallback(func(params reconcile.ObjectParams) {
-			pluginState.reconcilePanicCallback(params)
+			pluginMetrics.Reconcile.ErrorStatsCallback(params, stats)
+			logSuccessiveReconcileFailure(logger, params, stats, config.LogSuccessiveFailuresThreshold)
 		}),
+		reconcile.WithPanicCallback(pluginMetrics.Reconcile.PanicCallback),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("could not setup reconcile queue: %w", err)
@@ -123,7 +117,7 @@ func NewAutoscaleEnforcerPlugin(
 		return nil, fmt.Errorf("could not start watch on VirtualMachineMigration events: %w", err)
 	}
 
-	pluginState = NewPluginState(*config, vmClient, promReg, podStore, nodeStore)
+	pluginState = NewPluginState(*config, vmClient, pluginMetrics, podStore, nodeStore)
 
 	// Start the workers for the queue. We can't do these earlier because our handlers depend on the
 	// PluginState that only exists now.

pkg/plugin/globalstate.go

@@ -10,7 +10,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 
 	corev1 "k8s.io/api/core/v1"
@@ -53,7 +52,7 @@ type PluginState struct {
 	// We use this when scoring pod placements.
 	maxNodeMem api.Bytes
 
-	metrics metrics.Plugin
+	metrics *metrics.Plugin
 
 	requeuePod      func(uid types.UID) error
 	requeueNode     func(nodeName string) error
@@ -81,16 +80,14 @@ type nodeState struct {
 func NewPluginState(
 	config Config,
 	vmClient vmclient.Interface,
-	reg prometheus.Registerer,
+	metrics *metrics.Plugin,
 	podWatchStore *watch.Store[corev1.Pod],
 	nodeWatchStore *watch.Store[corev1.Node],
 ) *PluginState {
 	crudTimeout := time.Second * time.Duration(config.K8sCRUDTimeoutSeconds)
 
 	indexedNodeStore := watch.NewIndexedStore(nodeWatchStore, watch.NewFlatNameIndex[corev1.Node]())
 
-	metrics := metrics.BuildPluginMetrics(reg, config.NodeMetricLabels)
-
 	return &PluginState{
 		mu: sync.Mutex{},
 

pkg/plugin/metrics/metrics.go

@@ -29,10 +29,10 @@ type Plugin struct {
 	K8sOps *prometheus.CounterVec
 }
 
-func BuildPluginMetrics(reg prometheus.Registerer, nodeMetricLabels map[string]string) Plugin {
+func BuildPluginMetrics(reg prometheus.Registerer, nodeMetricLabels map[string]string) *Plugin {
 	nodeLabels := buildNodeLabels(nodeMetricLabels)
 
-	return Plugin{
+	return &Plugin{
 		nodeLabels: nodeLabels,
 		Framework:  buildSchedFrameworkMetrics(reg, nodeLabels),
 		Nodes:      buildNodeMetrics(reg, nodeLabels),

pkg/plugin/metrics/reconcile.go

@@ -1,21 +1,24 @@
 package metrics
 
 import (
+	"time"
+
 	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/neondatabase/autoscaling/pkg/plugin/reconcile"
 	"github.com/neondatabase/autoscaling/pkg/util"
 )
 
 type Reconcile struct {
-	WaitDurations    prometheus.Histogram
-	ProcessDurations *prometheus.HistogramVec
-	Failing          *prometheus.GaugeVec
-	Panics           *prometheus.CounterVec
+	waitDurations    prometheus.Histogram
+	processDurations *prometheus.HistogramVec
+	failing          *prometheus.GaugeVec
+	panics           *prometheus.CounterVec
 }
 
 func buildReconcileMetrics(reg prometheus.Registerer) Reconcile {
 	return Reconcile{
-		WaitDurations: util.RegisterMetric(reg, prometheus.NewHistogram(
+		waitDurations: util.RegisterMetric(reg, prometheus.NewHistogram(
 			prometheus.HistogramOpts{
 				Name: "autoscaling_plugin_reconcile_queue_wait_durations",
 				Help: "Duration that items in the reconcile queue are waiting to be picked up",
@@ -29,7 +32,7 @@ func buildReconcileMetrics(reg prometheus.Registerer) Reconcile {
 				},
 			},
 		)),
-		ProcessDurations: util.RegisterMetric(reg, prometheus.NewHistogramVec(
+		processDurations: util.RegisterMetric(reg, prometheus.NewHistogramVec(
 			prometheus.HistogramOpts{
 				Name: "autoscaling_plugin_reconcile_duration_seconds",
 				Help: "Duration that items take to be reconciled",
@@ -44,14 +47,14 @@ func buildReconcileMetrics(reg prometheus.Registerer) Reconcile {
 			},
 			[]string{"kind", "outcome"},
 		)),
-		Failing: util.RegisterMetric(reg, prometheus.NewGaugeVec(
+		failing: util.RegisterMetric(reg, prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
 				Name: "autoscaling_plugin_reconcile_failing_objects",
 				Help: "Number of objects currently failing to be reconciled",
 			},
 			[]string{"kind"},
 		)),
-		Panics: util.RegisterMetric(reg, prometheus.NewCounterVec(
+		panics: util.RegisterMetric(reg, prometheus.NewCounterVec(
 			prometheus.CounterOpts{
 				Name: "autoscaling_plugin_reconcile_panics_count",
 				Help: "Number of times reconcile operations have panicked",
@@ -60,3 +63,24 @@ func buildReconcileMetrics(reg prometheus.Registerer) Reconcile {
 		)),
 	}
 }
+
+func (r Reconcile) QueueWaitDurationCallback(duration time.Duration) {
+	r.waitDurations.Observe(duration.Seconds())
+}
+
+func (r Reconcile) ResultCallback(params reconcile.ObjectParams, duration time.Duration, err error) {
+	outcome := "success"
+	if err != nil {
+		outcome = "failure"
+	}
+	r.processDurations.WithLabelValues(params.GVK.Kind, outcome).Observe(duration.Seconds())
+}
+
+func (r Reconcile) ErrorStatsCallback(params reconcile.ObjectParams, stats reconcile.ErrorStats) {
+	// update count of current failing objects
+	r.failing.WithLabelValues(params.GVK.Kind).Set(float64(stats.TypedCount))
+}
+
+func (r Reconcile) PanicCallback(params reconcile.ObjectParams) {
+	r.panics.WithLabelValues(params.GVK.Kind).Inc()
+}

pkg/plugin/reconcile.go

@@ -3,48 +3,29 @@ package plugin
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"go.uber.org/zap"
 
 	"github.com/neondatabase/autoscaling/pkg/plugin/reconcile"
 )
 
-func (s *PluginState) reconcileQueueWaitCallback(duration time.Duration) {
-	s.metrics.Reconcile.WaitDurations.Observe(duration.Seconds())
-}
-
-func (s *PluginState) reconcileResultCallback(params reconcile.ObjectParams, duration time.Duration, err error) {
-	outcome := "success"
-	if err != nil {
-		outcome = "failure"
-	}
-	s.metrics.Reconcile.ProcessDurations.
-		WithLabelValues(params.GVK.Kind, outcome).
-		Observe(duration.Seconds())
-}
-
-func (s *PluginState) reconcileErrorStatsCallback(logger *zap.Logger, params reconcile.ObjectParams, stats reconcile.ErrorStats) {
-	// update count of current failing objects
-	s.metrics.Reconcile.Failing.
-		WithLabelValues(params.GVK.Kind).
-		Set(float64(stats.TypedCount))
-
+func logSuccessiveReconcileFailure(
+	logger *zap.Logger,
+	params reconcile.ObjectParams,
+	stats reconcile.ErrorStats,
+	threshold int,
+) {
 	// Make sure that repeatedly failing objects are sufficiently noisy
-	if stats.SuccessiveFailures >= s.config.LogSuccessiveFailuresThreshold {
+	if stats.SuccessiveFailures >= threshold {
 		logger.Warn(
-			fmt.Sprintf("%s has failed to reconcile >%d times in a row", params.GVK.Kind, s.config.LogSuccessiveFailuresThreshold),
+			fmt.Sprintf("%s has failed to reconcile >%d times in a row", params.GVK.Kind, threshold),
 			zap.Int("SuccessiveFailures", stats.SuccessiveFailures),
 			zap.String("EventKind", string(params.EventKind)),
 			reconcile.ObjectMetaLogField(params.GVK.Kind, params.Obj),
 		)
 	}
 }
 
-func (s *PluginState) reconcilePanicCallback(params reconcile.ObjectParams) {
-	s.metrics.Reconcile.Panics.WithLabelValues(params.GVK.Kind).Inc()
-}
-
 func reconcileWorker(ctx context.Context, logger *zap.Logger, queue *reconcile.Queue) {
 	wait := queue.WaitChan()
 	for {