add basic metrics for NRI framework

Introduces basic runtime metrics to improve observability and satisfy
v1.0 requirements. Exposes three primary metrics via a new `Metrics`
interface in pkg/adaptation:

- `nri_plugin_count` (Gauge): The number of active NRI plugins currently
  connected.
- `nri_plugin_invocations_total` (CounterVec): The total number of
  plugin invocations. Labeled by `plugin` name, `operation` (e.g.,
  "CreateContainer", "Synchronize"), and `status` ("success" or
  "failure").
- `nri_plugin_adjustments_total` (CounterVec): The total number of
  container/pod adjustments returned by a plugin. Labeled by `plugin`
  name, `operation`, and `adjustment_type` ("adjust", "update", or
  "evict").

To avoid burying state change events under a generic operation name,
`StateChange` metrics append the specific event type using human-readable
strings (e.g., `"StateChange/StartContainer"`).

These metrics are updated during plugin invocations and container
lifecycle events. A `WithMetrics` option is exposed to allow container
runtimes to attach these to their existing setups (e.g. Prometheus
registries).

Ref: #272
Signed-off-by: Chris Henzie <chrishenzie@gmail.com>
Assisted-by: gemini-cli

Author: chrishenzie

pkg/adaptation/adaptation.go

@@ -73,6 +73,7 @@ type Adaptation struct {
 	builtin     []*builtin.BuiltinPlugin
 	syncLock    sync.RWMutex
 	wasmService *api.PluginPlugin
+	metrics     Metrics
 }
 
 var (
@@ -135,6 +136,16 @@ func WithBuiltinPlugins(plugins ...*builtin.BuiltinPlugin) Option {
 	}
 }
 
+// WithMetrics allows consumers to register an implementation of the Metrics interface.
+func WithMetrics(m Metrics) Option {
+	return func(r *Adaptation) error {
+		if m != nil {
+			r.metrics = m
+		}
+		return nil
+	}
+}
+
 // WithDefaultValidator sets up builtin validator plugin if it is configured.
 func WithDefaultValidator(cfg *validator.DefaultValidatorConfig) Option {
 	return func(r *Adaptation) error {
@@ -174,6 +185,7 @@ func New(name, version string, syncFn SyncFn, updateFn UpdateFn, opts ...Option)
 		socketPath:  DefaultSocketPath,
 		syncLock:    sync.RWMutex{},
 		wasmService: wasmService,
+		metrics:     &noopMetrics{},
 	}
 
 	for _, o := range opts {
@@ -537,6 +549,7 @@ func (r *Adaptation) removeClosedPlugins() {
 
 	r.plugins = active
 	r.validators = validators
+	r.metrics.UpdatePluginCount(len(r.plugins))
 }
 
 func (r *Adaptation) startListener() error {
@@ -604,6 +617,7 @@ func (r *Adaptation) acceptPluginConnections(l net.Listener) error {
 					r.validators = append(r.validators, p)
 				}
 				r.sortPlugins()
+				r.metrics.UpdatePluginCount(len(r.plugins))
 				r.Unlock()
 				log.Infof(ctx, "plugin %q connected and synchronized", p.name())
 			}

pkg/adaptation/metrics.go

@@ -0,0 +1,39 @@
+/*
+   Copyright The containerd 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 adaptation
+
+// Metrics defines the interface that a consumer can implement to collect
+// and emit metrics regarding NRI plugin activity.
+type Metrics interface {
+	// RecordPluginInvocation records the invocation of a plugin for a specific operation.
+	RecordPluginInvocation(pluginName, operation string, err error)
+
+	// RecordPluginAdjustments records the adjustments returned by a plugin.
+	RecordPluginAdjustments(pluginName, operation string, adjust *ContainerAdjustment, updates, evicts int)
+
+	// UpdatePluginCount sets the number of currently active plugins.
+	UpdatePluginCount(count int)
+}
+
+// noopMetrics provides a default, no-operation implementation of the Metrics interface.
+type noopMetrics struct{}
+
+var _ Metrics = (*noopMetrics)(nil)
+
+func (n *noopMetrics) RecordPluginInvocation(_, _ string, _ error)                           {}
+func (n *noopMetrics) RecordPluginAdjustments(_, _ string, _ *ContainerAdjustment, _, _ int) {}
+func (n *noopMetrics) UpdatePluginCount(_ int)                                               {}

pkg/adaptation/metrics_test.go

@@ -0,0 +1,127 @@
+/*
+   Copyright The containerd 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 adaptation
+
+import (
+	"context"
+	"testing"
+
+	"github.com/containerd/nri/pkg/api"
+	"github.com/stretchr/testify/assert"
+)
+
+// mockMetrics provides a simple implementation of the Metrics interface for testing.
+type mockMetrics struct {
+	pluginCount int
+	invocations []mockInvocation
+	adjustments []mockAdjustment
+}
+
+type mockInvocation struct {
+	pluginName string
+	operation  string
+	err        error
+}
+
+type mockAdjustment struct {
+	pluginName string
+	operation  string
+	adjust     *ContainerAdjustment
+	updates    int
+	evicts     int
+}
+
+func (m *mockMetrics) RecordPluginInvocation(pluginName, operation string, err error) {
+	m.invocations = append(m.invocations, mockInvocation{
+		pluginName: pluginName,
+		operation:  operation,
+		err:        err,
+	})
+}
+
+func (m *mockMetrics) RecordPluginAdjustments(pluginName, operation string, adjust *ContainerAdjustment, updates, evicts int) {
+	m.adjustments = append(m.adjustments, mockAdjustment{
+		pluginName: pluginName,
+		operation:  operation,
+		adjust:     adjust,
+		updates:    updates,
+		evicts:     evicts,
+	})
+}
+
+func (m *mockMetrics) UpdatePluginCount(count int) {
+	m.pluginCount = count
+}
+
+func TestMockMetrics(t *testing.T) {
+	m := &mockMetrics{}
+
+	m.RecordPluginInvocation("test-plugin", "CreateContainer", nil)
+	assert.Len(t, m.invocations, 1)
+	assert.Equal(t, "test-plugin", m.invocations[0].pluginName)
+	assert.Equal(t, "CreateContainer", m.invocations[0].operation)
+	assert.Nil(t, m.invocations[0].err)
+
+	m.RecordPluginAdjustments("test-plugin", "CreateContainer", nil, 2, 3)
+	assert.Len(t, m.adjustments, 1)
+	assert.Equal(t, "test-plugin", m.adjustments[0].pluginName)
+	assert.Equal(t, "CreateContainer", m.adjustments[0].operation)
+	assert.Nil(t, m.adjustments[0].adjust)
+	assert.Equal(t, 2, m.adjustments[0].updates)
+	assert.Equal(t, 3, m.adjustments[0].evicts)
+
+	m.UpdatePluginCount(42)
+	assert.Equal(t, 42, m.pluginCount)
+}
+
+type dummyPlugin struct {
+	api.PluginService
+}
+
+func (d *dummyPlugin) StateChange(_ context.Context, _ *api.StateChangeEvent) (*api.Empty, error) {
+	return &api.Empty{}, nil
+}
+
+func TestPluginStateChangeMetrics(t *testing.T) {
+	m := &mockMetrics{}
+	adapt := &Adaptation{metrics: m}
+
+	impl := &pluginType{builtinImpl: &dummyPlugin{}}
+
+	var events api.EventMask
+	events.Set(api.Event_START_CONTAINER)
+
+	p := &plugin{
+		r:      adapt,
+		events: events,
+		impl:   impl,
+		idx:    "00",
+		base:   "test-plugin",
+	}
+
+	evt := &api.StateChangeEvent{
+		Event: api.Event_START_CONTAINER,
+	}
+
+	err := p.StateChange(context.Background(), evt)
+	assert.Nil(t, err)
+
+	assert.Len(t, m.invocations, 1)
+	assert.Equal(t, "00-test-plugin", m.invocations[0].pluginName)
+	assert.Equal(t, "StateChange/StartContainer", m.invocations[0].operation)
+	assert.Nil(t, m.invocations[0].err)
+}

pkg/adaptation/plugin.go

@@ -522,7 +522,11 @@ func (p *plugin) synchronize(ctx context.Context, pods []*PodSandbox, containers
 			len(req.Pods), len(podsToSend), len(req.Containers), len(ctrsToSend), req.More)
 
 		rpl, err = p.impl.Synchronize(ctx, req)
+		p.r.metrics.RecordPluginInvocation(p.name(), "Synchronize", err)
 		if err == nil {
+			if rpl != nil {
+				p.r.metrics.RecordPluginAdjustments(p.name(), "Synchronize", nil, len(rpl.Update), 0)
+			}
 			if !req.More {
 				break
 			}
@@ -606,6 +610,7 @@ func (p *plugin) createContainer(ctx context.Context, req *CreateContainerReques
 	defer cancel()
 
 	rpl, err := p.impl.CreateContainer(ctx, req)
+	p.r.metrics.RecordPluginInvocation(p.name(), "CreateContainer", err)
 	if err != nil {
 		if isFatalError(err) {
 			log.Errorf(ctx, "closing plugin %s, failed to handle CreateContainer request: %v",
@@ -615,6 +620,9 @@ func (p *plugin) createContainer(ctx context.Context, req *CreateContainerReques
 		}
 		return nil, err
 	}
+	if rpl != nil {
+		p.r.metrics.RecordPluginAdjustments(p.name(), "CreateContainer", rpl.Adjust, len(rpl.Update), len(rpl.Evict))
+	}
 
 	return rpl, nil
 }
@@ -629,6 +637,7 @@ func (p *plugin) updateContainer(ctx context.Context, req *UpdateContainerReques
 	defer cancel()
 
 	rpl, err := p.impl.UpdateContainer(ctx, req)
+	p.r.metrics.RecordPluginInvocation(p.name(), "UpdateContainer", err)
 	if err != nil {
 		if isFatalError(err) {
 			log.Errorf(ctx, "closing plugin %s, failed to handle UpdateContainer request: %v",
@@ -638,6 +647,9 @@ func (p *plugin) updateContainer(ctx context.Context, req *UpdateContainerReques
 		}
 		return nil, err
 	}
+	if rpl != nil {
+		p.r.metrics.RecordPluginAdjustments(p.name(), "UpdateContainer", nil, len(rpl.Update), len(rpl.Evict))
+	}
 
 	return rpl, nil
 }
@@ -652,6 +664,7 @@ func (p *plugin) stopContainer(ctx context.Context, req *StopContainerRequest) (
 	defer cancel()
 
 	rpl, err = p.impl.StopContainer(ctx, req)
+	p.r.metrics.RecordPluginInvocation(p.name(), "StopContainer", err)
 	if err != nil {
 		if isFatalError(err) {
 			log.Errorf(ctx, "closing plugin %s, failed to handle StopContainer request: %v",
@@ -661,6 +674,9 @@ func (p *plugin) stopContainer(ctx context.Context, req *StopContainerRequest) (
 		}
 		return nil, err
 	}
+	if rpl != nil {
+		p.r.metrics.RecordPluginAdjustments(p.name(), "StopContainer", nil, len(rpl.Update), 0)
+	}
 
 	return rpl, nil
 }
@@ -673,7 +689,9 @@ func (p *plugin) updatePodSandbox(ctx context.Context, req *UpdatePodSandboxRequ
 	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
-	if _, err := p.impl.UpdatePodSandbox(ctx, req); err != nil {
+	_, err := p.impl.UpdatePodSandbox(ctx, req)
+	p.r.metrics.RecordPluginInvocation(p.name(), "UpdatePodSandbox", err)
+	if err != nil {
 		if isFatalError(err) {
 			log.Errorf(ctx, "closing plugin %s, failed to handle event %d: %v",
 				p.name(), Event_UPDATE_POD_SANDBOX, err)
@@ -695,7 +713,14 @@ func (p *plugin) StateChange(ctx context.Context, evt *StateChangeEvent) (err er
 	ctx, cancel := context.WithTimeout(ctx, getPluginRequestTimeout())
 	defer cancel()
 
-	if err = p.impl.StateChange(ctx, evt); err != nil {
+	err = p.impl.StateChange(ctx, evt)
+
+	var mask api.EventMask
+	mask.Set(evt.Event)
+	operation := fmt.Sprintf("StateChange/%s", mask.PrettyString())
+	p.r.metrics.RecordPluginInvocation(p.name(), operation, err)
+
+	if err != nil {
 		if isFatalError(err) {
 			log.Errorf(ctx, "closing plugin %s, failed to handle event %d: %v",
 				p.name(), evt.Event, err)
@@ -717,6 +742,7 @@ func (p *plugin) ValidateContainerAdjustment(ctx context.Context, req *ValidateC
 	defer cancel()
 
 	rpl, err := p.impl.ValidateContainerAdjustment(ctx, req)
+	p.r.metrics.RecordPluginInvocation(p.name(), "ValidateContainerAdjustment", err)
 	if err != nil {
 		if isFatalError(err) {
 			log.Errorf(ctx, "closing plugin %s, failed to validate request: %v", p.name(), err)