agent-events: fix more metrics (#20068)
* cleanuo * make all counters have analysis per status
Costa Tsaousis committed
Apr 6, 2025 at 17:03 UTC
0df2060dbff6818fb12d3a6f81b3eb7b9c5a1f60
2 files changed
+61
-32
packaging/tools/agent-events/server.go
+22
-21
@@ -107,16 +107,7 @@ func initMetrics() (*prometheus.Exporter, error) {
107
meter = meterProvider.Meter("agent-events")
108
109
// --- Create Metric Instruments ---
110
- // Helper to reduce repetition
111
- createCounter := func(name, desc string) (metric.Int64Counter, error) {
112
- counter, err := meter.Int64Counter(name, metric.WithDescription(desc))
113
- if err != nil {
114
- slog.Error("failed to create counter", "name", name, "error", err)
115
- }
116
- return counter, err
117
- }
118
-
119
- // Create a counter with a status label for consolidated metrics
110
+ // Helper to create counters with status labels for consolidated metrics
111
createLabeledCounter := func(name, desc string) (metric.Int64Counter, error) {
112
counter, err := meter.Int64Counter(
113
name,
@@ -146,9 +137,18 @@ func initMetrics() (*prometheus.Exporter, error) {
137
return hist, err
138
}
139
149
- // Create unified request counter with status label
150
- requestsCounter, _ = createLabeledCounter("agent_events_requests_ratio_total", "Total number of requests by status")
151
- bytesReceived, _ = createCounter("agent_events_bytes_received_total", "Total number of bytes received in request bodies")
140
+ // Create unified counters with status label
141
+ requestsCounter, _ = createLabeledCounter("agent_events_requests", "Number of requests by status")
142
+ bytesReceived, _ = createLabeledCounter("agent_events_received_bytes", "Number of bytes received in request bodies by status")
143
+
144
+ // Pre-initialize counters with all status labels set to zero
145
+ ctx := context.Background()
146
+ statusLabels := []string{"success", "duplicate", "invalid_json", "method_not_allowed", "body_too_large", "failed_to_read", "cant_marshal_output"}
147
+ for _, status := range statusLabels {
148
+ requestsCounter.Add(ctx, 0, metric.WithAttributes(attribute.String("status", status)))
149
+ bytesReceived.Add(ctx, 0, metric.WithAttributes(attribute.String("status", status)))
150
+ }
151
+
152
dedupCacheSize, _ = createGauge("agent_events_dedup_cache_entries", "Current number of entries in the deduplication cache")
153
activeConnectionsGauge, _ = createGauge("agent_events_active_connections", "Number of currently active connections")
154
uptimeGauge, _ = createGauge("agent_events_uptime_seconds", "How long the server has been running in seconds")
@@ -259,8 +259,6 @@ func handler(w http.ResponseWriter, r *http.Request) {
259
requestStartTime := time.Now()
260
ctx := r.Context()
261
262
- // Record metrics using OTEL API
263
- requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "total")))
262
defer func() {
263
requestDuration.Record(ctx, time.Since(requestStartTime).Seconds())
264
}()
@@ -281,15 +279,14 @@ func handler(w http.ResponseWriter, r *http.Request) {
279
if errors.As(err, &maxBytesErr) {
280
http.Error(w, fmt.Sprintf("Request body exceeds limit (%d bytes)", maxRequestBodySize), http.StatusRequestEntityTooLarge)
281
slog.Info("request discarded", "reason", "body_too_large", "limit", maxRequestBodySize, "remote_addr", r.RemoteAddr)
284
- requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "entity_too_large")))
282
+ requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "body_too_large")))
283
} else {
284
http.Error(w, "Error reading request", http.StatusInternalServerError)
285
slog.Error("request discarded", "reason", "error_reading_body", "error", err)
288
- requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "internal_error")))
286
+ requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "failed_to_read")))
287
}
288
return
289
}
292
- bytesReceived.Add(ctx, int64(len(body)))
290
291
// JSON Validation - Attempt to unmarshal directly to map (requires object)
292
var fullData map[string]interface{}
@@ -297,9 +294,12 @@ func handler(w http.ResponseWriter, r *http.Request) {
294
http.Error(w, "Invalid JSON", http.StatusBadRequest)
295
bodyDetail := ""; if slog.Default().Enabled(context.Background(), slog.LevelDebug) { bodyDetail = fmt.Sprintf(", Body: %s", string(body)) } else { bodyDetail = fmt.Sprintf(", Body snippet: %s", limitString(string(body), 100)) }
296
slog.Warn("request discarded", "reason", "invalid_json", "error", err.Error(), "body_detail", bodyDetail)
300
- requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "bad_request")))
297
+ bytesReceived.Add(ctx, int64(len(body)), metric.WithAttributes(attribute.String("status", "invalid_json")))
298
+ requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "invalid_json")))
299
return
300
}
301
+
302
+ bytesReceived.Add(ctx, int64(len(body)), metric.WithAttributes(attribute.String("status", "success")))
303
304
// Deduplication Logic
305
shouldProcess := true
@@ -316,6 +316,7 @@ func handler(w http.ResponseWriter, r *http.Request) {
316
317
if !checkAndRecordHash(dedupHash) {
318
shouldProcess = false
319
+ bytesReceived.Add(ctx, int64(len(body)), metric.WithAttributes(attribute.String("status", "duplicate")))
320
requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "duplicate")))
321
slog.Debug("discarded duplicate request", "hash", fmt.Sprintf("%x", dedupHash), "note", "timestamp refreshed")
322
if _, err := w.Write([]byte("OK")); err != nil { slog.Error("error writing response", "context", "after_duplicate_discard", "error", err) }
@@ -341,7 +342,7 @@ func handler(w http.ResponseWriter, r *http.Request) {
342
if err != nil {
343
http.Error(w, "Internal Server Error during output marshal", http.StatusInternalServerError)
344
slog.Error("request discarded", "reason", "json_marshal_failed", "error", err)
344
- requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "internal_error")))
345
+ requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "cant_marshal_output")))
346
return
347
}
348
@@ -479,4 +480,4 @@ func main() {
480
func limitString(s string, maxLen int) string {
481
if len(s) <= maxLen { return s }
482
return s[:maxLen] + "..."
482
-}
483
+}
\ No newline at end of file
packaging/tools/agent-events/server_test.go
+39
-11
@@ -242,19 +242,47 @@ func TestHandler(t *testing.T) {
242
metricsContent := string(metricsBodyBytes)
243
t.Logf("Metrics Output for Verification:\n%s", metricsContent) // Log for manual inspection if needed
244
245
- // Check for presence of key metrics (adjust names if needed)
246
- expectedMetrics := []string{
247
- "agent_events_requests_ratio_total", // Consolidated metric with status labels
248
- "agent_events_bytes_received_total",
249
- "agent_events_dedup_cache_entries",
250
- "agent_events_request_duration_seconds", // Check for histograms
251
- "go_goroutines",
245
+ // Check for presence of key metrics (handling both standard and suffixed names)
246
+ // OpenTelemetry may add suffixes like _ratio_total to counter metrics
247
+ metricChecks := []struct {
248
+ namePatterns []string
249
+ description string
250
+ }{
251
+ {
252
+ namePatterns: []string{"agent_events_requests", "agent_events_requests_ratio_total"},
253
+ description: "Requests counter",
254
+ },
255
+ {
256
+ namePatterns: []string{"agent_events_received_bytes", "agent_events_received_bytes_ratio_total"},
257
+ description: "Bytes received counter",
258
+ },
259
+ {
260
+ namePatterns: []string{"agent_events_dedup_cache_entries"},
261
+ description: "Dedup cache size gauge",
262
+ },
263
+ {
264
+ namePatterns: []string{"agent_events_request_duration_seconds"},
265
+ description: "Request duration histogram",
266
+ },
267
+ {
268
+ namePatterns: []string{"go_goroutines"},
269
+ description: "Go runtime metrics",
270
+ },
271
}
253
- for _, metricName := range expectedMetrics {
254
- if !strings.Contains(metricsContent, metricName) {
255
- t.Errorf("metrics response missing expected metric: %s", metricName)
272
+
273
+ for _, check := range metricChecks {
274
+ found := false
275
+ for _, pattern := range check.namePatterns {
276
+ if strings.Contains(metricsContent, pattern) {
277
+ found = true
278
+ break
279
+ }
280
+ }
281
+ if !found {
282
+ t.Errorf("metrics response missing expected metric: %s (patterns: %v)", check.description, check.namePatterns)
283
}
284
}
285
+
286
// OpenTelemetry histogram metrics have this pattern in the output:
287
// agent_events_request_duration_seconds_bucket{...
288
// agent_events_request_duration_seconds_sum{...
@@ -388,4 +416,4 @@ func TestHandler(t *testing.T) {
416
rr2 := httptest.NewRecorder(); stdout2, _ := captureOutput(t, func() { handler(rr2, req2) })
417
if rr2.Code != http.StatusOK || !strings.Contains(stdout2, "zero") { t.Errorf("Req 2 (duplicate) failed") }
418
})
391
-}
419
+}
\ No newline at end of file