add agentkit_runtime_operation_latency metric for agentkit

Author: shunjiazhu

observability/constant.go

@@ -68,6 +68,9 @@ const (
 	// APMPlus specific metrics
 	MetricNameAPMPlusSpanLatency    = "apmplus_span_latency"
 	MetricNameAPMPlusToolTokenUsage = "apmplus_tool_token_usage"
+
+	// AgentKit specific metrics
+	MetricNameAgentKitDuration = "agentkit_runtime_operation_latency"
 )
 
 // General attributes

observability/metrics.go

@@ -16,7 +16,7 @@ package observability
 
 import (
 	"context"
-
+	"fmt"
 	"sync"
 
 	"go.opentelemetry.io/otel"
@@ -46,6 +46,11 @@ var (
 	genAIServerTimePerOutputTokenBuckets = []float64{
 		0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75, 1.0, 2.5,
 	}
+
+	// Time duration buckets for agent_kit (seconds)
+	agentkitDurationSecondBuckets = []float64{
+		0.01, 0.02, 0.04, 0.08, 0.16, 0.32, 0.64, 1.28, 2.56, 5.12, 10.24, 20.48, 40.96, 81.92, 163.84,
+	}
 )
 
 var (
@@ -69,6 +74,9 @@ var (
 	// special metrics for APMPlus
 	apmPlusLatencyHistograms        []metric.Float64Histogram
 	apmPlusToolTokenUsageHistograms []metric.Float64Histogram
+
+	// special metrics for AgentKit
+	agentkitDurationHistograms []metric.Float64Histogram
 )
 
 // registerLocalMetrics initializes the metrics system with a local isolated MeterProvider.
@@ -200,6 +208,16 @@ func initializeInstruments(m metric.Meter) {
 	); err == nil {
 		apmPlusToolTokenUsageHistograms = append(apmPlusToolTokenUsageHistograms, h)
 	}
+
+	// AgentKit Duration
+	if h, err := m.Float64Histogram(
+		MetricNameAgentKitDuration,
+		metric.WithDescription("operation latency"),
+		metric.WithUnit("s"),
+		metric.WithExplicitBucketBoundaries(agentkitDurationSecondBuckets...),
+	); err == nil {
+		agentkitDurationHistograms = append(agentkitDurationHistograms, h)
+	}
 }
 
 // RecordTokenUsage records the number of tokens used.
@@ -271,3 +289,12 @@ func RecordAPMPlusToolTokenUsage(ctx context.Context, tokens int64, attrs ...att
 		histogram.Record(ctx, float64(tokens), metric.WithAttributes(attrs...))
 	}
 }
+
+func RecordAgentKitDuration(ctx context.Context, durationSeconds float64, err error, attrs ...attribute.KeyValue) {
+	if err != nil {
+		attrs = append(attrs, attribute.String("error_type", fmt.Sprintf("%T", err)))
+	}
+	for _, histogram := range agentkitDurationHistograms {
+		histogram.Record(ctx, durationSeconds, metric.WithAttributes(attrs...))
+	}
+}

observability/metrics_test.go

@@ -16,6 +16,7 @@ package observability
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -117,6 +118,34 @@ func TestMetricsRecording(t *testing.T) {
 		}
 		assert.True(t, found, "Streaming time to first token not found")
 	})
+
+	t.Run("RecordAgentKitDurationWithError", func(t *testing.T) {
+		testErr := fmt.Errorf("test error")
+		RecordAgentKitDuration(ctx, 2.5, testErr, attrs...)
+
+		var rm metricdata.ResourceMetrics
+		err := reader.Collect(ctx, &rm)
+		assert.NoError(t, err)
+
+		var found bool
+		for _, sm := range rm.ScopeMetrics {
+			for _, m := range sm.Metrics {
+				if m.Name == MetricNameAgentKitDuration {
+					data := m.Data.(metricdata.Histogram[float64])
+					for _, dp := range data.DataPoints {
+						if dp.Count > 0 {
+							assert.Equal(t, uint64(1), dp.Count)
+							assert.Equal(t, 2.5, dp.Sum)
+							errType, _ := dp.Attributes.Value("error_type")
+							assert.Equal(t, "*errors.errorString", errType.AsString())
+							found = true
+						}
+					}
+				}
+			}
+		}
+		assert.True(t, found, "AgentKit duration not found")
+	})
 }
 
 func TestRegisterLocalMetrics(t *testing.T) {

observability/plugin.go

@@ -179,16 +179,28 @@ func (p *adkObservabilityPlugin) AfterRun(ctx agent.InvocationContext) {
 
 			// Record final metrics for invocation
 			if !meta.StartTime.IsZero() {
-				elapsed := time.Since(meta.StartTime).Seconds()
-				metricAttrs := []attribute.KeyValue{
-					attribute.String("gen_ai_operation_name", "chain"),
-					attribute.String("gen_ai_operation_type", "workflow"),
-					attribute.String("gen_ai.system", GetModelProvider(context.Context(ctx))),
-				}
 				if p.isMetricsEnabled() {
+					elapsed := time.Since(meta.StartTime).Seconds()
+					metricAttrs := []attribute.KeyValue{
+						attribute.String("gen_ai_operation_name", "chain"),
+						attribute.String("gen_ai_operation_type", "workflow"),
+						attribute.String("gen_ai.system", GetModelProvider(context.Context(ctx))),
+					}
 					RecordOperationDuration(context.Background(), elapsed, metricAttrs...)
 					RecordAPMPlusSpanLatency(context.Background(), elapsed, metricAttrs...)
+
+					agentKitsAttrs := []attribute.KeyValue{
+						attribute.String("gen_ai_operation_name", "chain"),
+						attribute.String("gen_ai_operation_type", "workflow"),
+					}
+
+					var lastErr error
+					if val, _ := ctx.Session().State().Get(stateKeyLastError); val != nil {
+						lastErr = val.(error)
+					}
+					RecordAgentKitDuration(context.Background(), elapsed, lastErr, agentKitsAttrs...)
 				}
+
 			}
 
 			// Clean up from global map with delay to allow children to be exported.
@@ -346,6 +358,7 @@ func (p *adkObservabilityPlugin) AfterModel(ctx agent.CallbackContext, resp *mod
 
 	if err != nil {
 		span.SetStatus(codes.Error, err.Error())
+		_ = ctx.State().Set(stateKeyLastError, err)
 		// Record Exceptions metric
 		if p.isMetricsEnabled() {
 			meta := p.getSpanMetadata(ctx.State())
@@ -357,6 +370,7 @@ func (p *adkObservabilityPlugin) AfterModel(ctx agent.CallbackContext, resp *mod
 				attribute.String("error_type", "error"), // Simple error type
 			}
 			RecordExceptions(context.Context(ctx), 1, metricAttrs...)
+			p.recordFinalResponseMetrics(ctx, meta, meta.ModelName, err)
 		}
 		return nil, nil
 	}
@@ -450,7 +464,7 @@ func (p *adkObservabilityPlugin) AfterModel(ctx agent.CallbackContext, resp *mod
 
 	if !resp.Partial {
 		// Record Operation Duration and Latency
-		p.recordFinalResponseMetrics(ctx, meta, finalModelName)
+		p.recordFinalResponseMetrics(ctx, meta, finalModelName, nil)
 	}
 
 	return nil, nil
@@ -563,7 +577,7 @@ func (p *adkObservabilityPlugin) recordStreamingGenerationMetrics(ctx agent.Call
 	}
 }
 
-func (p *adkObservabilityPlugin) recordFinalResponseMetrics(ctx agent.CallbackContext, meta *spanMetadata, finalModelName string) {
+func (p *adkObservabilityPlugin) recordFinalResponseMetrics(ctx agent.CallbackContext, meta *spanMetadata, finalModelName string, err error) {
 	if !meta.StartTime.IsZero() {
 		duration := time.Since(meta.StartTime).Seconds()
 		metricAttrs := []attribute.KeyValue{
@@ -575,6 +589,12 @@ func (p *adkObservabilityPlugin) recordFinalResponseMetrics(ctx agent.CallbackCo
 		if p.isMetricsEnabled() {
 			RecordOperationDuration(context.Context(ctx), duration, metricAttrs...)
 			RecordAPMPlusSpanLatency(context.Context(ctx), duration, metricAttrs...)
+
+			agentKitsAttrs := []attribute.KeyValue{
+				attribute.String("gen_ai_operation_name", "chat"),
+				attribute.String("gen_ai_operation_type", "llm"),
+			}
+			RecordAgentKitDuration(context.Context(ctx), duration, err, agentKitsAttrs...)
 		}
 	}
 }
@@ -901,6 +921,9 @@ func (p *adkObservabilityPlugin) BeforeTool(ctx tool.Context, tool tool.Tool, ar
 
 // AfterTool is called after a tool is executed.
 func (p *adkObservabilityPlugin) AfterTool(ctx tool.Context, tool tool.Tool, args, result map[string]any, err error) (map[string]any, error) {
+	if err != nil {
+		_ = ctx.State().Set(stateKeyLastError, err)
+	}
 	// Metrics recording only
 	meta := p.getSpanMetadata(ctx.State())
 	if !meta.StartTime.IsZero() {
@@ -913,6 +936,12 @@ func (p *adkObservabilityPlugin) AfterTool(ctx tool.Context, tool tool.Tool, arg
 		if p.isMetricsEnabled() {
 			RecordOperationDuration(context.Background(), duration, metricAttrs...)
 			RecordAPMPlusSpanLatency(context.Background(), duration, metricAttrs...)
+
+			agentKitsAttrs := []attribute.KeyValue{
+				attribute.String("gen_ai_operation_name", tool.Name()),
+				attribute.String("gen_ai_operation_type", "tool"),
+			}
+			RecordAgentKitDuration(context.Background(), duration, err, agentKitsAttrs...)
 		}
 
 		if p.isMetricsEnabled() {
@@ -1052,6 +1081,7 @@ const (
 	stateKeyMetadata        = "veadk.observability.metadata"
 	stateKeyStreamingOutput = "veadk.observability.streaming_output"
 	stateKeyStreamingSpan   = "veadk.observability.streaming_span"
+	stateKeyLastError       = "veadk.observability.last_error"
 )
 
 // spanMetadata groups various observational data points in a single structure