feat(telemetry): per-controller metric views and span attributes, so controllers sharing a process stay attributable

pkg/controller/builder/builder.go

@@ -120,6 +120,7 @@ func Build(ctx context.Context, conf Configuration, optsFuncs ...FuncOption) (*c
 		GracefulShutdownTimeout: opts.gracefulShutdownTimeout,
 		TracerProvider:          opts.tracerProvider,
 		Ownership:               ownership,
+		SpanAttributes:          opts.spanAttributes,
 	})
 	if err != nil {
 		opts.logger.Error(err, "failed to create controller")

pkg/controller/builder/options.go

@@ -12,6 +12,7 @@ import (
 	metricsserver "github.com/krateo-platformops/unstructured-runtime/pkg/metrics/server"
 	"github.com/krateo-platformops/unstructured-runtime/pkg/pluralizer"
 	"github.com/krateo-platformops/unstructured-runtime/pkg/telemetry"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/time/rate"
 	"k8s.io/client-go/util/workqueue"
@@ -54,6 +55,9 @@ type options struct {
 	// tracerProvider is controller.Options.TracerProvider; nil => the process default.
 	tracerProvider trace.TracerProvider
 
+	// spanAttributes is controller.Options.SpanAttributes.
+	spanAttributes []attribute.KeyValue
+
 	// ownership is the Lease the controller must hold to reconcile; nil => none. Build fills in
 	// its Leases client from the Configuration's rest.Config.
 	ownership *controller.Ownership
@@ -106,6 +110,15 @@ func WithGracefulShutdownTimeout(d time.Duration) func(o *options) {
 	}
 }
 
+// WithSpanAttributes sets attributes on every reconcile root span the controller starts, naming the
+// controller (e.g. krateo.composition.gvr) so its traces stay attributable in a process that hosts
+// several. See controller.Options.SpanAttributes.
+func WithSpanAttributes(attrs ...attribute.KeyValue) func(o *options) {
+	return func(o *options) {
+		o.spanAttributes = append(o.spanAttributes, attrs...)
+	}
+}
+
 // WithOwnership makes the controller reconcile only while it holds the coordination.k8s.io Lease
 // namespace/name, as identity (the pod name), so that a second process for the same resource waits
 // instead of reconciling alongside. Every process that may reconcile the same resource must be given

pkg/controller/controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/krateo-platformops/unstructured-runtime/pkg/ratelimit"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sync"
@@ -201,6 +202,12 @@ type Options struct {
 	// krateo.io/traceparent-continued marker. Mirrors provider-runtime's WithTracerProvider.
 	TracerProvider trace.TracerProvider
 
+	// SpanAttributes are set on every reconcile root span this controller starts, so a process that
+	// hosts several controllers (core-provider's cdc pool) keeps each one's traces attributable. The
+	// span already names the object (k8s.object.kind/name/namespace); these name the controller, e.g.
+	// krateo.composition.gvr. Nil: none, as before.
+	SpanAttributes []attribute.KeyValue
+
 	// Ownership makes Run reconcile only while it holds a coordination.k8s.io Lease, so a second
 	// process for the same resource waits instead of reconciling alongside. Nil: no Lease. See
 	// Ownership.
@@ -255,6 +262,8 @@ type Controller struct {
 	traces traceMemo
 	// tracerProv is Options.TracerProvider; nil reads telemetry.TracerProvider on every pass.
 	tracerProv trace.TracerProvider
+	// spanAttrs is Options.SpanAttributes, set on every reconcile root.
+	spanAttrs []attribute.KeyValue
 	// runCtx is the context Run was given, cancelled at SIGTERM. A worker waiting for a local slot
 	// waits on it, because reconcileCtx stays live through a drain.
 	runCtx context.Context
@@ -656,6 +665,7 @@ func New(sid *shortid.Shortid, opts Options) (*Controller, error) {
 		maxRetries:              opts.MaxRetries,
 		governor:                opts.RateLimitGovernor,
 		tracerProv:              opts.TracerProvider,
+		spanAttrs:               opts.SpanAttributes,
 		objects:                 newObjectGate(),
 		gracefulShutdownTimeout: gracefulShutdownTimeout,
 		ownership:               opts.Ownership,

pkg/controller/traces_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/krateo-platformops/unstructured-runtime/pkg/meta"
 	"github.com/krateo-platformops/unstructured-runtime/pkg/telemetry"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 	"go.opentelemetry.io/otel/trace"
@@ -578,3 +579,50 @@ func TestTraceDeferredActMarksOnlyARealWrite(t *testing.T) {
 func TestMarkActedOutsideAPassIsANoop(t *testing.T) {
 	MarkActed(context.Background())
 }
+
+// Options.SpanAttributes name the controller on every reconcile root, so a process hosting several
+// controllers (core-provider's cdc pool, step 3) keeps each one's traces attributable. The control
+// is the same pass without the option: no such attribute.
+func TestTraceRootCarriesSpanAttributes(t *testing.T) {
+	gvr := attribute.String("krateo.composition.gvr", "test.example.org/v1/tests")
+	for _, tc := range []struct {
+		name  string
+		attrs []attribute.KeyValue
+		want  bool
+	}{{"with SpanAttributes", []attribute.KeyValue{gvr}, true}, {"without (control)", nil, false}} {
+		t.Run(tc.name, func(t *testing.T) {
+			rec := tracetest.NewSpanRecorder()
+			sdk := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec))
+			t.Cleanup(func() { _ = sdk.Shutdown(context.Background()) })
+			sid, err := shortid.New(1, shortid.DefaultABC, 2342)
+			require.NoError(t, err)
+			opts := createTestOptions()
+			opts.TracerProvider = sdk
+			opts.SpanAttributes = tc.attrs
+			ctrl, err := New(sid, opts)
+			require.NoError(t, err)
+			ctrl.externalClient = &actingClient{exists: true}
+			obj := createTestUnstructured("attrs", opts.Namespace)
+			_, err = opts.Client.Resource(opts.GVR).Namespace(opts.Namespace).Create(context.TODO(), obj, metav1.CreateOptions{})
+			require.NoError(t, err)
+			require.NoError(t, ctrl.processItem(context.TODO(), ctrlevent.Event{
+				EventType: ctrlevent.Observe,
+				ObjectRef: objectref.ObjectRef{APIVersion: "test.example.org/v1", Kind: "Test", Name: "attrs", Namespace: opts.Namespace},
+			}))
+			var root sdktrace.ReadOnlySpan
+			for _, s := range rec.Ended() {
+				if s.Name() == tracecontext.SpanReconcile {
+					root = s
+				}
+			}
+			require.NotNil(t, root, "control: the pass started a reconcile root")
+			got := false
+			for _, kv := range root.Attributes() {
+				if kv.Key == gvr.Key && kv.Value.AsString() == gvr.Value.AsString() {
+					got = true
+				}
+			}
+			require.Equal(t, tc.want, got, "root carries %s", gvr.Key)
+		})
+	}
+}

pkg/controller/worker.go

@@ -473,6 +473,9 @@ func (c *Controller) processItem(ctx context.Context, obj interface{}) (err erro
 	decision := c.traces.decide(tracer != nil, traceKey, el, time.Now(), tracecontext.DefaultContinueWindow)
 	traceObj := tracecontext.Object{Kind: el.GetKind(), Name: el.GetName(), Namespace: el.GetNamespace()}
 	ctx, span := tracecontext.StartReconcile(ctx, tracer, traceObj, decision, time.Now())
+	if len(c.spanAttrs) > 0 {
+		span.SetAttributes(c.spanAttrs...) // which controller (Options.SpanAttributes)
+	}
 	pass := offPass
 	if tracer != nil {
 		pass = &tracePass{tracer: tracer, span: span, d: decision, obj: traceObj, key: traceKey, uid: el.GetUID(), memo: &c.traces, c: c, target: el}