Improve agent-events web server (#20063)
- Add strict JSON validation to correctly reject malformed input - Modernize logging using structured logging (slog) - Add comprehensive OTEL & Prometheus metrics - Extract and include Cloudflare headers in output - Implement graceful shutdown handling - Add health endpoint with server status - Improve error handling and status reporting - Update tests to cover various scenarios and edge cases - Update .gitignore to exclude agent-events binary and test files
Costa Tsaousis committed
Apr 6, 2025 at 14:50 UTC
b68d6cc47fea687cca7465d326b5c746341f633d
5 files changed
+626
-265
.gitignore
+9
@@ -186,3 +186,12 @@ src/go/plugin/go.d/vendor
186
187
# ignore files used with msi installer
188
packaging/windows/*.msi
189
+
190
+# agent-events build artifacts and test files
191
+packaging/tools/agent-events/agent-events
192
+packaging/tools/agent-events/expected_metrics.txt
193
+packaging/tools/agent-events/cf_patch.txt
194
+packaging/tools/agent-events/parseBehaviorTest.go
195
+packaging/tools/agent-events/server
196
+packaging/tools/agent-events/go.mod
197
+packaging/tools/agent-events/go.sum
packaging/tools/agent-events/.gitignore
deleted
-3
@@ -1,3 +0,0 @@
1
-server
2
-go.mod
3
-go.sum
\ No newline at end of file
packaging/tools/agent-events/run.sh
+1
-1
@@ -1,6 +1,6 @@
1
#!/usr/bin/env bash
2
3
-stdbuf -oL /opt/agent-events/server --port=30001 --dedup-key agent.ephemeral_id --dedup-window 1800 2>/dev/null \
3
+stdbuf -oL /opt/agent-events/server --port=30001 --dedup-key agent.id --dedup-window 1800 2>/var/log/agent-events.log \
4
| stdbuf -oL log2journal json \
5
--prefix 'AE_' \
6
--inject 'SYSLOG_IDENTIFIER=agent-events' \
packaging/tools/agent-events/server.go
+320
-97
@@ -1,20 +1,33 @@
1
package main
2
3
import (
4
+ "context"
5
"crypto/sha256"
6
"encoding/json"
7
"errors"
8
+ "expvar"
9
"flag"
10
"fmt"
11
"io"
10
- "log"
12
+ "log/slog"
13
"net/http"
14
+ _ "net/http/pprof" // Import for side effects - registers HTTP handlers
15
"os"
16
+ "os/signal"
17
+ "runtime"
18
"strings"
19
"sync"
20
+ "sync/atomic"
21
+ "syscall"
22
"time"
23
24
+ // Import promhttp for the handler
25
+ "github.com/prometheus/client_golang/prometheus/promhttp"
26
"github.com/tidwall/gjson"
27
+ "go.opentelemetry.io/otel"
28
+ "go.opentelemetry.io/otel/exporters/prometheus"
29
+ "go.opentelemetry.io/otel/metric"
30
+ sdkmetric "go.opentelemetry.io/otel/sdk/metric"
31
)
32
33
// --- Constants ---
@@ -36,12 +49,40 @@ func (d *dedupPaths) Set(value string) error {
49
50
// --- Global variables ---
51
var (
52
+ // Core functionality variables
53
seenIDs map[[32]byte]seenEntry
54
mapMutex = &sync.Mutex{}
55
dedupWindow time.Duration
42
- debugMode bool
56
keyPaths dedupPaths
57
dedupSeparator string
58
+ startTime time.Time
59
+
60
+ // Track active connections for graceful shutdown
61
+ activeConnections int32
62
+
63
+ // OpenTelemetry meter provider and metrics
64
+ meterProvider *sdkmetric.MeterProvider
65
+ meter metric.Meter
66
+
67
+ // Counters
68
+ requestsTotal metric.Int64Counter
69
+ duplicateRequests metric.Int64Counter
70
+ methodNotAllowedRequests metric.Int64Counter
71
+ badRequestsTotal metric.Int64Counter
72
+ entityTooLargeRequests metric.Int64Counter
73
+ internalErrorRequests metric.Int64Counter
74
+ bytesReceived metric.Int64Counter
75
+
76
+ // Gauges
77
+ dedupCacheSize metric.Int64ObservableGauge
78
+ activeConnectionsGauge metric.Int64ObservableGauge
79
+ uptimeGauge metric.Int64ObservableGauge
80
+
81
+ // Histograms
82
+ requestDuration metric.Float64Histogram
83
+
84
+ // Keep expvar for backward compatibility
85
+ eventMetrics = expvar.NewMap("agent_events")
86
)
87
88
// --- Data Structures ---
@@ -51,6 +92,99 @@ type seenEntry struct {
92
93
// --- Core Logic Functions ---
94
95
+// initMetrics initializes OpenTelemetry metrics and expvar metrics for backward compatibility
96
+func initMetrics() (*prometheus.Exporter, error) {
97
+ // Set up the OpenTelemetry Prometheus exporter
98
+ exporter, err := prometheus.New()
99
+ if err != nil {
100
+ // Return error instead of exiting directly
101
+ return nil, fmt.Errorf("failed to create Prometheus exporter: %w", err)
102
+ }
103
+
104
+ // Create a new meter provider with the Prometheus exporter as a Reader
105
+ meterProvider = sdkmetric.NewMeterProvider(
106
+ sdkmetric.WithReader(exporter),
107
+ )
108
+ otel.SetMeterProvider(meterProvider)
109
+
110
+ // Create a new meter
111
+ meter = meterProvider.Meter("agent-events")
112
+
113
+ // --- Create Metric Instruments ---
114
+ // Helper to reduce repetition
115
+ createCounter := func(name, desc string) (metric.Int64Counter, error) {
116
+ counter, err := meter.Int64Counter(name, metric.WithDescription(desc))
117
+ if err != nil {
118
+ slog.Error("failed to create counter", "name", name, "error", err)
119
+ }
120
+ return counter, err
121
+ }
122
+ createGauge := func(name, desc string) (metric.Int64ObservableGauge, error) {
123
+ gauge, err := meter.Int64ObservableGauge(name, metric.WithDescription(desc))
124
+ if err != nil {
125
+ slog.Error("failed to create gauge", "name", name, "error", err)
126
+ }
127
+ return gauge, err
128
+ }
129
+ createHistogram := func(name, desc string, buckets []float64) (metric.Float64Histogram, error) {
130
+ hist, err := meter.Float64Histogram(name,
131
+ metric.WithDescription(desc),
132
+ metric.WithExplicitBucketBoundaries(buckets...),
133
+ )
134
+ if err != nil {
135
+ slog.Error("failed to create histogram", "name", name, "error", err)
136
+ }
137
+ return hist, err
138
+ }
139
+
140
+ requestsTotal, _ = createCounter("agent_events_requests_total", "Total number of requests received")
141
+ duplicateRequests, _ = createCounter("agent_events_requests_duplicate_total", "Total number of duplicate requests detected")
142
+ methodNotAllowedRequests, _ = createCounter("agent_events_requests_method_not_allowed_total", "Total number of requests with incorrect HTTP method")
143
+ badRequestsTotal, _ = createCounter("agent_events_requests_bad_request_total", "Total number of requests with invalid JSON")
144
+ entityTooLargeRequests, _ = createCounter("agent_events_requests_entity_too_large_total", "Total number of requests exceeding size limits")
145
+ internalErrorRequests, _ = createCounter("agent_events_requests_internal_error_total", "Total number of internal server errors")
146
+ bytesReceived, _ = createCounter("agent_events_bytes_received_total", "Total number of bytes received in request bodies")
147
+ dedupCacheSize, _ = createGauge("agent_events_dedup_cache_size", "Current number of entries in the deduplication cache")
148
+ activeConnectionsGauge, _ = createGauge("agent_events_active_connections", "Number of currently active connections")
149
+ uptimeGauge, _ = createGauge("agent_events_uptime_seconds", "How long the server has been running in seconds")
150
+ requestDuration, _ = createHistogram("agent_events_request_duration_seconds",
151
+ "Histogram of request processing times in seconds",
152
+ []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
153
+ )
154
+ // Basic check if any metric failed (optional, depends on how critical individual metrics are)
155
+ // if requestsTotal == nil || ... { return exporter, fmt.Errorf("one or more metrics failed to initialize") }
156
+
157
+
158
+ // Register callbacks for observable metrics
159
+ _, err = meter.RegisterCallback(
160
+ func(_ context.Context, observer metric.Observer) error {
161
+ mapMutex.Lock()
162
+ observer.ObserveInt64(dedupCacheSize, int64(len(seenIDs)))
163
+ mapMutex.Unlock()
164
+ observer.ObserveInt64(activeConnectionsGauge, int64(atomic.LoadInt32(&activeConnections)))
165
+ observer.ObserveInt64(uptimeGauge, int64(time.Since(startTime).Seconds()))
166
+ return nil
167
+ },
168
+ dedupCacheSize, activeConnectionsGauge, uptimeGauge,
169
+ )
170
+ if err != nil {
171
+ // Log but don't necessarily exit, maybe gauges won't update
172
+ slog.Error("failed to register callback for observable metrics", "error", err)
173
+ }
174
+
175
+ // Create expvar metrics for backward compatibility
176
+ eventMetrics.Set("active_connections", expvar.Func(func() interface{} { return atomic.LoadInt32(&activeConnections) }))
177
+ eventMetrics.Set("uptime_seconds", expvar.Func(func() interface{} { return time.Since(startTime).Seconds() }))
178
+ eventMetrics.Set("start_time_seconds", expvar.Func(func() interface{} { return startTime.Unix() }))
179
+ eventMetrics.Set("runtime_stats", expvar.Func(func() interface{} {
180
+ memStats := &runtime.MemStats{}
181
+ runtime.ReadMemStats(memStats)
182
+ return *memStats
183
+ }))
184
+
185
+ return exporter, nil // Return the exporter (even though we use promhttp.Handler) and nil error
186
+}
187
+
188
// checkAndRecordHash accepts the SHA256 hash ([32]byte) for checking.
189
// It now REFRESHES the timestamp whenever a hash is found,
190
// effectively creating a sliding deduplication window.
@@ -58,38 +192,20 @@ func checkAndRecordHash(hash [32]byte) bool {
192
now := time.Now()
193
mapMutex.Lock()
194
defer mapMutex.Unlock()
61
-
195
var zeroHash [32]byte
196
if hash == zeroHash {
64
- log.Println("Warning: checkAndRecordHash received potentially zero hash.")
65
- // Decide if zero hash should always be discarded, e.g. return false
197
+ slog.Warn("potentially zero hash received", "hash", fmt.Sprintf("%x", hash))
198
}
67
-
68
- // Check if the hash exists in the map
199
entry, found := seenIDs[hash]
70
-
200
if found {
72
- // --- Hash Found ---
73
- // Check if it was a duplicate based on the *previous* timestamp
201
isRecentDuplicate := now.Sub(entry.timestamp) < dedupWindow
75
-
76
- // *** Always update the timestamp to 'now' to refresh the window ***
202
seenIDs[hash] = seenEntry{timestamp: now}
78
-
79
- // Return 'false' if it was a recent duplicate (suppress processing),
80
- // return 'true' if it was found but expired (allow processing).
203
return !isRecentDuplicate
82
-
83
- } else {
84
- // --- Hash Not Found ---
85
- // Record the new hash with the current timestamp
86
- seenIDs[hash] = seenEntry{timestamp: now}
87
- // Return 'true' as this is the first time (or first time after expiry)
88
- return true
204
}
205
+ seenIDs[hash] = seenEntry{timestamp: now}
206
+ return true
207
}
208
92
-
209
// cleanupExpiredEntries uses the hash ([32]byte) as the key type.
210
func cleanupExpiredEntries(interval time.Duration) {
211
ticker := time.NewTicker(interval)
@@ -99,154 +215,261 @@ func cleanupExpiredEntries(interval time.Duration) {
215
for range ticker.C {
216
mapMutex.Lock()
217
now := time.Now()
218
+ currentMapSize := len(seenIDs)
219
+ deletedInCycle := 0 // Track deletes per cycle for more granular debug
220
for h, entry := range seenIDs {
221
if now.Sub(entry.timestamp) >= dedupWindow {
222
delete(seenIDs, h)
223
cleanedCount++
224
+ deletedInCycle++
225
}
226
}
227
mapMutex.Unlock()
109
- // Simplified periodic logging for cleanup
110
- if cleanedCount > 0 && time.Since(lastCleanupLogTime) > time.Hour {
111
- if debugMode {
112
- log.Printf("Debug: Cleaned up %d expired entries in the past hour.", cleanedCount)
113
- }
114
- cleanedCount = 0 // Reset count after logging
228
+
229
+ // Log hourly summary if any were cleaned in the last hour
230
+ if cleanedCount > 0 && time.Since(lastCleanupLogTime) >= time.Hour {
231
+ slog.Debug("cleaned up expired entries",
232
+ "count_past_hour", cleanedCount,
233
+ "remaining_entries", currentMapSize-deletedInCycle) // Use size before delete for consistency
234
+ cleanedCount = 0 // Reset hourly count
235
lastCleanupLogTime = time.Now()
116
- }
236
+ } else if deletedInCycle > 0 {
237
+ // Optional: Log every cycle if debugging cleanup
238
+ // slog.Debug("cleanup cycle completed", "deleted", deletedInCycle, "remaining", currentMapSize-deletedInCycle)
239
+ }
240
}
241
}
242
243
+// connectionTracker is middleware that wraps an http.Handler to track active connections
244
+func connectionTracker(next http.Handler) http.Handler {
245
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
246
+ atomic.AddInt32(&activeConnections, 1)
247
+ defer atomic.AddInt32(&activeConnections, -1)
248
+ next.ServeHTTP(w, r)
249
+ })
250
+}
251
+
252
// --- HTTP Handler ---
253
func handler(w http.ResponseWriter, r *http.Request) {
254
+ requestStartTime := time.Now()
255
+ ctx := r.Context()
256
+
257
+ // Record metrics using OTEL API
258
+ requestsTotal.Add(ctx, 1)
259
+ defer func() {
260
+ requestDuration.Record(ctx, time.Since(requestStartTime).Seconds())
261
+ }()
262
+
263
+ // Method Check
264
if r.Method != http.MethodPost {
265
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
124
- log.Printf("Discarded: Method not allowed (%s) from %s", r.Method, r.RemoteAddr)
266
+ slog.Info("request discarded", "reason", "method_not_allowed", "method", r.Method, "remote_addr", r.RemoteAddr)
267
+ methodNotAllowedRequests.Add(ctx, 1)
268
return
269
}
270
+
271
+ // Read Body & Size Check
272
r.Body = http.MaxBytesReader(w, r.Body, maxRequestBodySize)
273
body, err := io.ReadAll(r.Body)
274
if err != nil {
275
var maxBytesErr *http.MaxBytesError
276
if errors.As(err, &maxBytesErr) {
277
http.Error(w, fmt.Sprintf("Request body exceeds limit (%d bytes)", maxRequestBodySize), http.StatusRequestEntityTooLarge)
133
- log.Printf("Discarded: Request body too large (limit %d bytes) from %s", maxRequestBodySize, r.RemoteAddr)
278
+ slog.Info("request discarded", "reason", "body_too_large", "limit", maxRequestBodySize, "remote_addr", r.RemoteAddr)
279
+ entityTooLargeRequests.Add(ctx, 1)
280
} else {
281
http.Error(w, "Error reading request", http.StatusInternalServerError)
136
- log.Printf("Discarded: Error reading request body: %v", err)
282
+ slog.Error("request discarded", "reason", "error_reading_body", "error", err)
283
+ internalErrorRequests.Add(ctx, 1)
284
}
285
return
286
}
287
+ bytesReceived.Add(ctx, int64(len(body)))
288
289
+ // JSON Validation - Attempt to unmarshal directly to map (requires object)
290
+ var fullData map[string]interface{}
291
+ if err := json.Unmarshal(body, &fullData); err != nil {
292
+ http.Error(w, "Invalid JSON", http.StatusBadRequest)
293
+ 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)) }
294
+ slog.Warn("request discarded", "reason", "invalid_json", "error", err.Error(), "body_detail", bodyDetail)
295
+ badRequestsTotal.Add(ctx, 1)
296
+ return
297
+ }
298
+
299
+ // Deduplication Logic
300
shouldProcess := true
301
if len(keyPaths) > 0 {
302
var keyBuilder strings.Builder
303
for i, path := range keyPaths {
304
result := gjson.GetBytes(body, path)
146
- var valueStr string
147
- if result.Exists() { valueStr = result.String() } else { valueStr = "" }
148
- keyBuilder.WriteString(valueStr)
305
+ keyBuilder.WriteString(result.String()) // gjson returns "" for non-existent paths
306
if i < len(keyPaths)-1 { keyBuilder.WriteString(dedupSeparator) }
307
}
308
finalKeyString := keyBuilder.String()
309
dedupHash := sha256.Sum256([]byte(finalKeyString))
153
- if debugMode {
154
- log.Printf("Debug: Generated dedup key string: \"%s\"", finalKeyString)
155
- log.Printf("Debug: Generated dedup hash: %x", dedupHash)
156
- }
310
+ slog.Debug("generated dedup key", "key_string", finalKeyString, "hash", fmt.Sprintf("%x", dedupHash))
311
158
- // Call the updated checkAndRecordHash function
312
if !checkAndRecordHash(dedupHash) {
160
- // It was determined to be a duplicate (based on previous timestamp)
313
shouldProcess = false
162
- if debugMode { log.Printf("Debug: Discarded duplicate hash: %x (timestamp refreshed)", dedupHash) } // Updated log message
163
- // Respond OK for duplicate and stop processing
164
- if _, err := w.Write([]byte("OK")); err != nil { log.Printf("Error writing response after duplicate discard: %v", err) }
165
- return // Exit handler early for duplicates
314
+ duplicateRequests.Add(ctx, 1)
315
+ slog.Debug("discarded duplicate request", "hash", fmt.Sprintf("%x", dedupHash), "note", "timestamp refreshed")
316
+ if _, err := w.Write([]byte("OK")); err != nil { slog.Error("error writing response", "context", "after_duplicate_discard", "error", err) }
317
+ return
318
}
167
- // If we reach here, it was not a recent duplicate (new or expired)
319
} else {
169
- if debugMode { log.Println("Debug: No --dedup-key flags provided, skipping deduplication.") }
320
+ slog.Debug("skipping deduplication", "reason", "no dedup keys provided")
321
}
322
323
+ // Process Request if Not Duplicate
324
if shouldProcess {
173
- // Always create a fresh map to prevent field reuse between requests
174
- fullData := make(map[string]interface{})
175
-
176
- if err := json.Unmarshal(body, &fullData); err != nil {
177
- http.Error(w, "Invalid JSON for full parsing", http.StatusBadRequest)
178
- bodyDetail := ""
179
- if debugMode { bodyDetail = fmt.Sprintf(", Body: %s", string(body)) } else { bodyDetail = fmt.Sprintf(", Body snippet: %s", limitString(string(body), 100)) }
180
- log.Printf("Discarded: Failed to fully parse JSON (post-dedup): %v%s", err, bodyDetail)
181
- return
182
- }
183
-
184
- // Marshal the map back to JSON
325
+ // We've already parsed fullData in the initial check
326
+
327
+ // Add Cloudflare Headers
328
+ cfHeaders := make(map[string]string)
329
+ cfHeaderPrefixes := []string{"CF-IPCountry", "CF-Ray", "CF-Connecting-IP", "CF-IPCity", "CF-IPContinent", "CF-IPLatitude", "CF-IPLongitude", "CF-IPRegion", "CF-IPTimeZone", "CF-Visitor", "CF-IPCOLO"}
330
+ for _, name := range cfHeaderPrefixes { if value := r.Header.Get(name); value != "" { key := strings.TrimPrefix(name, "CF-"); cfHeaders[key] = value } }
331
+ for name, values := range r.Header { if strings.HasPrefix(name, "CF-") && len(values) > 0 { key := strings.TrimPrefix(name, "CF-"); if _, exists := cfHeaders[key]; !exists { cfHeaders[key] = values[0] } } }
332
+ if len(cfHeaders) > 0 { fullData["cf"] = cfHeaders; slog.Debug("added cloudflare headers", "count", len(cfHeaders)) }
333
+
334
+ // Marshal Output
335
outputBytes, err := json.Marshal(fullData)
336
if err != nil {
337
http.Error(w, "Internal Server Error during output marshal", http.StatusInternalServerError)
188
- log.Printf("Discarded: Failed to marshal JSON for output: %v", err)
338
+ slog.Error("request discarded", "reason", "json_marshal_failed", "error", err)
339
+ internalErrorRequests.Add(ctx, 1)
340
return
341
}
191
-
342
+
343
+ // Write Output & Response
344
fmt.Println(string(outputBytes))
193
- if _, err := w.Write([]byte("OK")); err != nil { log.Printf("Error writing OK response: %v", err) }
345
+ if _, err := w.Write([]byte("OK")); err != nil {
346
+ slog.Error("error writing response", "context", "after_successful_processing", "error", err)
347
+ }
348
}
349
}
350
197
-// --- Main Function ---
351
+// healthHandler provides a simple health check endpoint
352
+func healthHandler(w http.ResponseWriter, r *http.Request) {
353
+ status := map[string]interface{}{
354
+ "status": "ok",
355
+ "timestamp": time.Now().Format(time.RFC3339),
356
+ "uptime": time.Since(startTime).String(),
357
+ "goroutines": runtime.NumGoroutine(),
358
+ "connections": atomic.LoadInt32(&activeConnections),
359
+ "version": "agent-events v1.0",
360
+ }
361
+ memStats := &runtime.MemStats{}; runtime.ReadMemStats(memStats); status["memory"] = map[string]interface{}{"alloc": memStats.Alloc, "total_alloc": memStats.TotalAlloc, "sys": memStats.Sys, "heap_alloc": memStats.HeapAlloc, "gc_cycles": memStats.NumGC}
362
+ mapMutex.Lock(); mapSize := len(seenIDs); mapMutex.Unlock(); status["deduplication"] = map[string]interface{}{"enabled": len(keyPaths) > 0, "keys": keyPaths, "window": dedupWindow.String(), "map_size": mapSize}
363
+ w.Header().Set("Content-Type", "application/json"); w.WriteHeader(http.StatusOK);
364
+ if err := json.NewEncoder(w).Encode(status); err != nil { slog.Error("error encoding health check response", "error", err) }
365
+}
366
+
367
+// main is the entry point of the application
368
func main() {
199
- log.SetOutput(os.Stderr)
200
- log.SetFlags(log.LstdFlags | log.Lshortfile)
369
+ // Setup initial logger before flag parsing
370
+ initialLogLevel := slog.LevelInfo
371
+ initialLogHandler := slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{ Level: initialLogLevel, AddSource: true })
372
+ slog.SetDefault(slog.New(initialLogHandler))
373
202
- // --- Command Line Flags ---
374
+ startTime = time.Now()
375
+
376
+ // Define flags
377
port := flag.Int("port", 8080, "Port to listen on")
204
- dedupSeconds := flag.Int("dedup-window", 1800, "Deduplication window in seconds (e.g., 1800 for 30 minutes)")
205
- flag.BoolVar(&debugMode, "debug", false, "Enable debug mode for verbose logging")
378
+ dedupSeconds := flag.Int("dedup-window", 1800, "Deduplication window in seconds")
379
+ metricsPath := flag.String("metrics-path", "/metrics", "Path for OpenTelemetry Prometheus metrics endpoint")
380
+ expvarPath := flag.String("expvar-path", "/debug/vars", "Path for expvar metrics endpoint (empty to disable)")
381
+ healthPath := flag.String("health-path", "/healthz", "Path for health check endpoint")
382
+ logFormat := flag.String("log-format", "json", "Log format: 'json' or 'text'")
383
+ logLevelFlag := flag.String("log-level", "info", "Log level: 'debug', 'info', 'warn', 'error'")
384
flag.Var(&keyPaths, "dedup-key", "JSON path (dot-notation) for deduplication key (can be used multiple times)")
207
- flag.StringVar(&dedupSeparator, "dedup-separator", "-", "Separator used between values from multiple --dedup-key paths")
385
+ flag.StringVar(&dedupSeparator, "dedup-separator", "-", "Separator used between multi-key values")
386
flag.Parse()
387
388
+ // Configure final logger based on flags
389
+ var level slog.Level
390
+ switch strings.ToLower(*logLevelFlag) {
391
+ case "debug": level = slog.LevelDebug
392
+ case "info": level = slog.LevelInfo
393
+ case "warn": level = slog.LevelWarn
394
+ case "error": level = slog.LevelError
395
+ default:
396
+ slog.Warn("invalid log level specified, defaulting to info", "value", *logLevelFlag)
397
+ level = slog.LevelInfo
398
+ }
399
+ var logHandler slog.Handler
400
+ if *logFormat == "text" { logHandler = slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{ Level: level, AddSource: true }) } else { logHandler = slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{ Level: level, AddSource: true }) }
401
+ slog.SetDefault(slog.New(logHandler))
402
+
403
+ // Initialize core components
404
seenIDs = make(map[[32]byte]seenEntry)
405
dedupWindow = time.Duration(*dedupSeconds) * time.Second
406
+ if _, err := initMetrics(); err != nil { // Handle potential error from initMetrics
407
+ slog.Error("failed to initialize metrics", "error", err)
408
+ os.Exit(1)
409
+ }
410
+
411
412
+ // Start background tasks
413
if dedupWindow > 0 && len(keyPaths) > 0 {
214
- cleanupInterval := dedupWindow / 10
215
- if cleanupInterval < 1*time.Minute { cleanupInterval = 1 * time.Minute } else if cleanupInterval > 15*time.Minute { cleanupInterval = 15 * time.Minute }
216
- log.Printf("Cleanup goroutine started. Interval: %v", cleanupInterval)
217
- go cleanupExpiredEntries(cleanupInterval)
218
- } else if dedupWindow <= 0 && len(keyPaths) > 0 {
219
- log.Println("Warning: Deduplication keys provided, but window is zero or negative. Deduplication effectively disabled.")
220
- }
414
+ cleanupInterval := dedupWindow / 10; if cleanupInterval < 1*time.Minute { cleanupInterval = 1 * time.Minute } else if cleanupInterval > 15*time.Minute { cleanupInterval = 15 * time.Minute }
415
+ slog.Info("cleanup goroutine started", "interval", cleanupInterval); go cleanupExpiredEntries(cleanupInterval)
416
+ } else if dedupWindow <= 0 && len(keyPaths) > 0 { slog.Warn("deduplication keys provided, but window is zero or negative", "keys", keyPaths, "window", dedupWindow) }
417
222
- // --- Configure HTTP Server ---
223
- readTimeout := 10 * time.Second
224
- writeTimeout := 10 * time.Second
225
- idleTimeout := 60 * time.Second
418
+ // Configure HTTP server
419
server := &http.Server{
420
Addr: fmt.Sprintf(":%d", *port),
228
- Handler: http.DefaultServeMux,
229
- ReadTimeout: readTimeout,
230
- WriteTimeout: writeTimeout,
231
- IdleTimeout: idleTimeout,
421
+ ReadTimeout: 10 * time.Second,
422
+ WriteTimeout: 10 * time.Second,
423
+ IdleTimeout: 60 * time.Second,
424
}
233
- http.HandleFunc("/", handler)
425
+ mux := http.NewServeMux()
426
+ mux.HandleFunc("/", handler)
427
+ mux.HandleFunc(*healthPath, healthHandler)
428
+ if *expvarPath != "" { mux.Handle(*expvarPath, expvar.Handler()) } // Register expvar if path not empty
429
+ mux.Handle(*metricsPath, promhttp.Handler()) // Use promhttp handler for OTEL metrics
430
+ // Add pprof handlers to custom mux
431
+ mux.HandleFunc("/debug/pprof/", http.DefaultServeMux.ServeHTTP)
432
+ mux.HandleFunc("/debug/pprof/cmdline", http.DefaultServeMux.ServeHTTP)
433
+ mux.HandleFunc("/debug/pprof/profile", http.DefaultServeMux.ServeHTTP)
434
+ mux.HandleFunc("/debug/pprof/symbol", http.DefaultServeMux.ServeHTTP)
435
+ mux.HandleFunc("/debug/pprof/trace", http.DefaultServeMux.ServeHTTP)
436
+ server.Handler = connectionTracker(mux) // Apply middleware
437
235
- // --- Start Server ---
236
- log.Printf("Server listening on port %d", *port)
237
- log.Printf("Maximum request body size: %d bytes", maxRequestBodySize)
238
- if len(keyPaths) > 0 {
239
- log.Printf("Deduplication enabled: Keys=%v, Separator='%s', Window=%v (Sliding window: timestamp refreshed on duplicate)", keyPaths, dedupSeparator, dedupWindow) // Updated log message
240
- } else {
241
- log.Println("Deduplication disabled (no --dedup-key specified).")
438
+ // Start server and handle shutdown
439
+ stop := make(chan os.Signal, 1)
440
+ signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
441
+ serverErrors := make(chan error, 1)
442
+
443
+ slog.Info("server starting", "port", *port, "metrics_path", *metricsPath, "expvar_path", *expvarPath, "health_path", *healthPath, "log_level", level.String()) // Simplified startup log
444
+
445
+ go func() {
446
+ slog.Info("server listening", "addr", server.Addr)
447
+ serverErrors <- server.ListenAndServe()
448
+ }()
449
+
450
+ select {
451
+ case err := <-serverErrors:
452
+ if err != nil && !errors.Is(err, http.ErrServerClosed) { slog.Error("server error", "error", err) }
453
+ case sig := <-stop:
454
+ slog.Info("shutdown initiated", "signal", sig.String())
455
+ shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second); defer cancel()
456
+
457
+ // Shutdown meter provider first
458
+ if meterProvider != nil {
459
+ slog.Info("shutting down OpenTelemetry meter provider")
460
+ if err := meterProvider.Shutdown(shutdownCtx); err != nil { slog.Error("meter provider shutdown failed", "error", err) }
461
+ }
462
+
463
+ // Shutdown HTTP server
464
+ slog.Info("shutting down HTTP server")
465
+ if err := server.Shutdown(shutdownCtx); err != nil { slog.Error("server shutdown failed", "error", err) } else { slog.Info("server shutdown completed gracefully") }
466
}
243
- log.Printf("Debug mode enabled: %t", debugMode)
244
- log.Printf("Server timeouts -> Read: %v, Write: %v, Idle: %v", readTimeout, writeTimeout, idleTimeout)
245
- log.Fatal(server.ListenAndServe())
467
+
468
+ slog.Info("server exiting")
469
}
470
471
// --- Helper Functions ---
472
func limitString(s string, maxLen int) string {
473
if len(s) <= maxLen { return s }
474
return s[:maxLen] + "..."
252
-}
\ No newline at end of file
475
+}
packaging/tools/agent-events/server_test.go
+296
-164
@@ -2,14 +2,21 @@ package main
2
3
import (
4
"bytes"
5
+ "context" // Import context
6
+ "encoding/json"
7
+ "fmt"
8
"io"
6
- "log"
9
+ "log/slog"
10
"net/http"
11
"net/http/httptest"
12
"os"
13
"strings"
14
+ "sync"
15
"testing"
16
"time"
17
+
18
+ // Import promhttp for testing the metrics endpoint handler
19
+ "github.com/prometheus/client_golang/prometheus/promhttp"
20
)
21
22
// Helper function to capture stdout/stderr during a test run
@@ -18,243 +25,368 @@ func captureOutput(t *testing.T, f func()) (stdout, stderr string) {
25
26
originalStdout := os.Stdout
27
originalStderr := os.Stderr
21
- originalLogOutput := log.Writer() // Get current log output writer
22
-
23
- // Create pipes to capture output
28
+ oldLogger := slog.Default()
29
rOut, wOut, _ := os.Pipe()
30
rErr, wErr, _ := os.Pipe()
26
-
27
- // Redirect stdout and stderr
31
os.Stdout = wOut
32
os.Stderr = wErr
30
- log.SetOutput(wErr) // Redirect default logger to stderr pipe
33
32
- // Use t.Cleanup to ensure restoration even if the test panics
34
t.Cleanup(func() {
35
os.Stdout = originalStdout
36
os.Stderr = originalStderr
36
- log.SetOutput(originalLogOutput) // Restore original log output
37
+ slog.SetDefault(oldLogger)
38
})
39
39
- // Channels to signal when reading is done
40
outCh := make(chan string)
41
errCh := make(chan string)
42
43
- // Goroutine to read stdout
43
go func() {
44
var buf bytes.Buffer
45
_, _ = io.Copy(&buf, rOut)
46
outCh <- buf.String()
47
}()
49
-
50
- // Goroutine to read stderr
48
go func() {
49
var buf bytes.Buffer
50
_, _ = io.Copy(&buf, rErr)
51
errCh <- buf.String()
52
}()
53
57
- // --- Execute the function under test ---
58
- f()
59
- // --- ---
54
+ f() // Execute the function
55
61
- // Close the writers to signal EOF to the readers
56
_ = wOut.Close()
57
_ = wErr.Close()
64
-
65
- // Read captured output
58
stdout = <-outCh
59
stderr = <-errCh
68
-
69
- // Optional: Print captured output via test logger if needed for debugging
70
- // t.Logf("Captured Stdout:\n%s", stdout)
71
- // t.Logf("Captured Stderr:\n%s", stderr)
72
-
60
return stdout, stderr
61
}
62
63
// --- Test Suite ---
64
78
-func TestHandler(t *testing.T) {
79
- // --- Test Setup ---
80
- // Configure global variables for the tests
81
- keyPaths = []string{"id"} // Simple dedup key for testing
65
+// setupTest initializes necessary components for tests
66
+func setupTest(t *testing.T) {
67
+ t.Helper()
68
+ keyPaths = []string{"id"}
69
dedupSeparator = "-"
83
- dedupWindow = 30 * time.Second // Use a reasonable window for tests
84
- debugMode = false // Start with debug off, can enable per test case
70
+ dedupWindow = 30 * time.Second
71
+ startTime = time.Now()
72
+ noopHandler := slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelDebug})
73
+ slog.SetDefault(slog.New(noopHandler))
74
+
75
+ initMetrics() // Initializes OTEL which feeds default registry
76
+
77
+ mapMutex.Lock()
78
+ seenIDs = make(map[[32]byte]seenEntry)
79
+ mapMutex.Unlock()
80
+
81
+ t.Cleanup(func() {
82
+ if meterProvider != nil {
83
+ ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
84
+ defer cancel()
85
+ if err := meterProvider.Shutdown(ctx); err != nil {
86
+ t.Logf("Warning: error shutting down meter provider in test cleanup: %v", err)
87
+ }
88
+ }
89
+ })
90
+}
91
86
- // Ensure the map is initialized and clean before starting tests
92
+// Helper to reset state between sub-tests if needed beyond setupTest
93
+func resetDedupState() {
94
mapMutex.Lock()
95
seenIDs = make(map[[32]byte]seenEntry)
96
mapMutex.Unlock()
97
+ keyPaths = []string{"id"}
98
+ dedupSeparator = "-"
99
+}
100
91
- // Helper to reset state between sub-tests
92
- resetState := func() {
93
- mapMutex.Lock()
94
- seenIDs = make(map[[32]byte]seenEntry) // Clear the map
95
- mapMutex.Unlock()
96
- keyPaths = []string{"id"} // Reset paths just in case
97
- debugMode = false
98
- // Reset other globals if they were modified
99
- }
101
+func TestHandler(t *testing.T) {
102
+ setupTest(t) // Setup once for all sub-tests
103
104
// --- Test Cases ---
105
106
t.Run("FirstValidRequest", func(t *testing.T) {
104
- t.Cleanup(resetState) // Ensure state is reset after this sub-test
105
-
106
- // Prepare request
107
+ t.Cleanup(resetDedupState) // Reset map for isolation
108
jsonBody := `{"id": "uuid-1", "data": "value1"}`
109
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
109
- rr := httptest.NewRecorder() // Records the HTTP response
110
-
111
- // Execute handler and capture output
112
- stdout, stderr := captureOutput(t, func() {
113
- handler(rr, req)
114
- })
115
-
116
- // Assertions
117
- if status := rr.Code; status != http.StatusOK {
118
- t.Errorf("handler returned wrong status code: got %v want %v", status, http.StatusOK)
119
- }
120
- expectedResponse := `OK`
121
- if rr.Body.String() != expectedResponse {
122
- t.Errorf("handler returned unexpected body: got %v want %v", rr.Body.String(), expectedResponse)
123
- }
124
- // Check stdout contains the *exact* JSON (json.Marshal might reorder fields)
125
- // A simpler check is that it's not empty and maybe contains key parts.
126
- // For exact match, we'd need to unmarshal stdout and compare.
127
- if !strings.Contains(stdout, `"id":"uuid-1"`) || !strings.Contains(stdout, `"data":"value1"`) {
128
- t.Errorf("handler produced unexpected stdout:\ngot: %q\nwant it to contain parts of: %q", stdout, jsonBody)
129
- }
130
- if stderr != "" {
131
- t.Errorf("handler produced unexpected stderr: got %q want empty", stderr)
132
- }
133
-
134
- // Check internal state (optional, needs mutex)
135
- mapMutex.Lock()
136
- if len(seenIDs) != 1 {
137
- t.Errorf("expected 1 entry in seenIDs map, got %d", len(seenIDs))
138
- }
139
- mapMutex.Unlock()
110
+ rr := httptest.NewRecorder()
111
+ stdout, _ := captureOutput(t, func() { handler(rr, req) })
112
+ if status := rr.Code; status != http.StatusOK { t.Errorf("status: got %v want %v", status, http.StatusOK) }
113
+ if rr.Body.String() != "OK" { t.Errorf("body: got %v want %v", rr.Body.String(), "OK") }
114
+ if !strings.Contains(stdout, `"id":"uuid-1"`) { t.Errorf("stdout missing id: %q", stdout) }
115
+ mapMutex.Lock(); mapLen := len(seenIDs); mapMutex.Unlock()
116
+ if mapLen != 1 { t.Errorf("map size: got %d want 1", mapLen) }
117
})
118
119
t.Run("DuplicateRequestWithinWindow", func(t *testing.T) {
143
- t.Cleanup(resetState)
144
-
145
- // --- Setup: Simulate the first request having happened ---
120
+ t.Cleanup(resetDedupState)
121
firstJsonBody := `{"id": "uuid-2", "data": "value2"}`
122
firstReq := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(firstJsonBody))
148
- firstRr := httptest.NewRecorder()
149
- // Run handler once, ignore output for this setup run
150
- captureOutput(t, func() { handler(firstRr, firstReq) })
151
- if firstRr.Code != http.StatusOK {
152
- t.Fatalf("Setup failed: first request did not return OK")
153
- }
154
- // Verify setup placed item in map
155
- mapMutex.Lock()
156
- if len(seenIDs) != 1 {
157
- t.Fatalf("Setup failed: map size not 1 after first request")
158
- }
159
- mapMutex.Unlock()
160
- // --- End Setup ---
161
-
162
-
163
- // Prepare the duplicate request
164
- // Note: Using the *same* body string as the first request
123
+ firstRr := httptest.NewRecorder(); captureOutput(t, func() { handler(firstRr, firstReq) })
124
+ if firstRr.Code != http.StatusOK { t.Fatalf("Setup failed") }
125
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(firstJsonBody))
126
+ rr := httptest.NewRecorder(); stdout, _ := captureOutput(t, func() { handler(rr, req) })
127
+ if status := rr.Code; status != http.StatusOK { t.Errorf("status: got %v want %v", status, http.StatusOK) }
128
+ if rr.Body.String() != "OK" { t.Errorf("body: got %v want %v", rr.Body.String(), "OK") }
129
+ if stdout != "" { t.Errorf("stdout not empty: %q", stdout) }
130
+ mapMutex.Lock(); mapLen := len(seenIDs); mapMutex.Unlock()
131
+ if mapLen != 1 { t.Errorf("map size: got %d want 1", mapLen) }
132
+ })
133
+
134
+ t.Run("InvalidJSON", func(t *testing.T) {
135
+ t.Cleanup(resetDedupState)
136
+ jsonBody := `{"id": "uuid-3", "data":`
137
+ req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
138
rr := httptest.NewRecorder()
139
+ stdout, _ := captureOutput(t, func() { handler(rr, req) })
140
+ if status := rr.Code; status != http.StatusBadRequest { t.Errorf("status: got %v want %v", status, http.StatusBadRequest) }
141
+ if stdout != "" { t.Errorf("stdout not empty: %q", stdout) }
142
+ })
143
168
- // Execute handler and capture output
169
- stdout, stderr := captureOutput(t, func() {
170
- handler(rr, req)
171
- })
144
+ t.Run("MissingDedupKey", func(t *testing.T) {
145
+ t.Cleanup(resetDedupState)
146
+ jsonBody := `{"other_id": "uuid-4", "data": "value4"}`
147
+ req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
148
+ rr := httptest.NewRecorder()
149
+ stdout, _ := captureOutput(t, func() { handler(rr, req) })
150
+ if status := rr.Code; status != http.StatusOK { t.Errorf("status: got %v want %v", status, http.StatusOK) }
151
+ if !strings.Contains(stdout, `"other_id":"uuid-4"`) { t.Errorf("stdout missing other_id: %q", stdout) }
152
+ mapMutex.Lock(); mapLen := len(seenIDs); mapMutex.Unlock()
153
+ if mapLen != 1 { t.Errorf("map size: got %d want 1", mapLen) }
154
+ })
155
173
- // Assertions
174
- if status := rr.Code; status != http.StatusOK {
175
- t.Errorf("handler returned wrong status code for duplicate: got %v want %v", status, http.StatusOK)
176
- }
177
- expectedResponse := `OK`
178
- if rr.Body.String() != expectedResponse {
179
- t.Errorf("handler returned unexpected body for duplicate: got %v want %v", rr.Body.String(), expectedResponse)
180
- }
181
- // Stdout should be empty for a duplicate
182
- if stdout != "" {
183
- t.Errorf("handler produced unexpected stdout for duplicate: got %q want empty", stdout)
184
- }
185
- // Stderr should be empty (unless debug mode logs duplicates)
186
- if stderr != "" {
187
- t.Errorf("handler produced unexpected stderr for duplicate: got %q want empty", stderr)
188
- }
189
- // Map size should remain 1
190
- mapMutex.Lock()
191
- if len(seenIDs) != 1 {
192
- t.Errorf("expected 1 entry in seenIDs map after duplicate, got %d", len(seenIDs))
193
- }
194
- mapMutex.Unlock()
156
+ t.Run("CloudflareHeaders", func(t *testing.T) {
157
+ t.Cleanup(resetDedupState)
158
+ jsonBody := `{"id": "uuid-cf", "data": "value-cf"}`
159
+ req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
160
+ req.Header.Set("CF-IPCountry", "US"); req.Header.Set("CF-Ray", "123"); req.Header.Set("CF-Connecting-IP", "1.2.3.4"); req.Header.Set("CF-IPCity", "Testville")
161
+ rr := httptest.NewRecorder(); stdout, _ := captureOutput(t, func() { handler(rr, req) })
162
+ if status := rr.Code; status != http.StatusOK { t.Errorf("status: got %v want %v", status, http.StatusOK) }
163
+ if !strings.Contains(stdout, `"cf":`) { t.Errorf("stdout missing cf object") }
164
+ if !strings.Contains(stdout, `"IPCountry":"US"`) { t.Errorf("stdout missing cf header IPCountry") }
165
+ if !strings.Contains(stdout, `"Connecting-IP":"1.2.3.4"`) { t.Errorf("stdout missing cf header Connecting-IP") }
166
})
167
197
- t.Run("InvalidJSON", func(t *testing.T) {
198
- t.Cleanup(resetState)
168
+ t.Run("MethodNotAllowed", func(t *testing.T) {
169
+ t.Cleanup(resetDedupState)
170
+ req := httptest.NewRequest(http.MethodGet, "/", nil)
171
+ rr := httptest.NewRecorder()
172
+ stdout, _ := captureOutput(t, func() { handler(rr, req) })
173
+ if status := rr.Code; status != http.StatusMethodNotAllowed { t.Errorf("status: got %v want %v", status, http.StatusMethodNotAllowed) }
174
+ if stdout != "" { t.Errorf("stdout not empty: %q", stdout) }
175
+ })
176
200
- // Prepare request
201
- jsonBody := `{"id": "uuid-3", "data":` // Invalid JSON
202
- req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
177
+ t.Run("RequestEntityTooLarge", func(t *testing.T) {
178
+ t.Cleanup(resetDedupState)
179
+ largeBody := make([]byte, maxRequestBodySize+1)
180
+ req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(largeBody))
181
rr := httptest.NewRecorder()
182
+ stdout, _ := captureOutput(t, func() { handler(rr, req) })
183
+ if status := rr.Code; status != http.StatusRequestEntityTooLarge { t.Errorf("status: got %v want %v", status, http.StatusRequestEntityTooLarge) }
184
+ if stdout != "" { t.Errorf("stdout not empty: %q", stdout) }
185
+ })
186
205
- // Execute handler and capture output
206
- stdout, stderr := captureOutput(t, func() {
207
- handler(rr, req)
208
- })
187
+ t.Run("MultiKeyDeduplication", func(t *testing.T) {
188
+ t.Cleanup(resetDedupState)
189
+ keyPaths = []string{"id", "source"}; dedupSeparator = "|"
190
+ req1 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "1", "source": "A", "data": "v1"}`))
191
+ rr1 := httptest.NewRecorder(); stdout1, _ := captureOutput(t, func() { handler(rr1, req1) })
192
+ if rr1.Code != http.StatusOK || !strings.Contains(stdout1, `"id":"1"`) { t.Errorf("Request 1 failed") }
193
+ req2 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "1", "source": "B", "data": "v2"}`))
194
+ rr2 := httptest.NewRecorder(); stdout2, _ := captureOutput(t, func() { handler(rr2, req2) })
195
+ if rr2.Code != http.StatusOK || !strings.Contains(stdout2, `"source":"B"`) { t.Errorf("Request 2 failed") }
196
+ req3 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "1", "source": "A", "data": "v3"}`))
197
+ rr3 := httptest.NewRecorder(); stdout3, _ := captureOutput(t, func() { handler(rr3, req3) })
198
+ if rr3.Code != http.StatusOK { t.Errorf("Request 3 status wrong") }
199
+ if stdout3 != "" { t.Errorf("Request 3 produced output") }
200
+ mapMutex.Lock(); mapLen := len(seenIDs); mapMutex.Unlock()
201
+ if mapLen != 2 { t.Errorf("map size: got %d want 2", mapLen) }
202
+ })
203
210
- // Assertions
211
- if status := rr.Code; status != http.StatusBadRequest {
212
- t.Errorf("handler returned wrong status code for invalid JSON: got %v want %v", status, http.StatusBadRequest)
204
+ t.Run("MetricsDelta", func(t *testing.T) {
205
+ t.Skip("Skipping MetricsDelta test: Verifying exact metric deltas with OTEL in unit tests is complex.")
206
+ })
207
+
208
+ t.Run("HealthEndpoint", func(t *testing.T) {
209
+ t.Cleanup(resetDedupState)
210
+ req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
211
+ rr := httptest.NewRecorder(); healthHandler(rr, req)
212
+ if status := rr.Code; status != http.StatusOK { t.Errorf("status: got %v want %v", status, http.StatusOK) }
213
+ if contentType := rr.Header().Get("Content-Type"); contentType != "application/json" { t.Errorf("content type: got %v want %v", contentType, "application/json") }
214
+ var result map[string]interface{}; if err := json.Unmarshal(rr.Body.Bytes(), &result); err != nil { t.Fatalf("invalid JSON: %v", err) }
215
+ requiredFields := []string{"status", "timestamp", "uptime", "goroutines", "memory", "deduplication"}; for _, field := range requiredFields { if _, ok := result[field]; !ok { t.Errorf("missing field: %s", field) } }
216
+ if status, ok := result["status"].(string); !ok || status != "ok" { t.Errorf("status field: got %v want ok", result["status"]) }
217
+ })
218
+
219
+ t.Run("MetricsEndpoint", func(t *testing.T) {
220
+ // Re-enabled: Uses promhttp.Handler which reads from default registry
221
+ t.Cleanup(resetDedupState)
222
+
223
+ // Create test server using the standard promhttp handler
224
+ metricsServer := httptest.NewServer(promhttp.Handler())
225
+ t.Cleanup(metricsServer.Close)
226
+
227
+ // Make requests to main handler to generate metrics
228
+ handler(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "m1"}`)))
229
+ handler(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "m1"}`))) // duplicate
230
+ handler(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) // method not allowed
231
+
232
+ // Fetch metrics
233
+ resp, err := http.Get(metricsServer.URL)
234
+ if err != nil { t.Fatalf("failed to get metrics: %v", err) }
235
+ defer resp.Body.Close()
236
+
237
+ if status := resp.StatusCode; status != http.StatusOK { t.Errorf("status: got %v want %v", status, http.StatusOK) }
238
+ if contentType := resp.Header.Get("Content-Type"); !strings.HasPrefix(contentType, "text/plain") { t.Errorf("content type: got %q want prefix text/plain", contentType) }
239
+
240
+ metricsBodyBytes, err := io.ReadAll(resp.Body)
241
+ if err != nil { t.Fatalf("failed to read metrics body: %v", err) }
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_total",
248
+ "agent_events_requests_duplicate_total",
249
+ "agent_events_requests_method_not_allowed_total",
250
+ "agent_events_dedup_cache_size",
251
+ "agent_events_request_duration_seconds_count", // Check for count specifically
252
+ "go_goroutines",
253
}
214
- // Stdout should be empty
215
- if stdout != "" {
216
- t.Errorf("handler produced unexpected stdout for invalid JSON: got %q want empty", stdout)
254
+ for _, metricName := range expectedMetrics {
255
+ if !strings.Contains(metricsContent, metricName) {
256
+ t.Errorf("metrics response missing expected metric: %s", metricName)
257
+ }
258
}
218
- // Stderr should contain the parsing error log message
219
- if !strings.Contains(stderr, "Failed to fully parse JSON") {
220
- t.Errorf("handler did not produce expected stderr log for invalid JSON: got %q", stderr)
259
+ // OpenTelemetry histogram metrics have this pattern in the output:
260
+ // agent_events_request_duration_seconds_bucket{...
261
+ // agent_events_request_duration_seconds_sum{...
262
+ // agent_events_request_duration_seconds_count{...
263
+ // So check for existence, not value line which can be variable
264
+ if !strings.Contains(metricsContent, "agent_events_request_duration_seconds_count{") {
265
+ t.Errorf("metrics response missing count line for histogram metric")
266
}
267
})
268
224
- t.Run("MissingDedupKey", func(t *testing.T) {
225
- t.Cleanup(resetState)
269
+ t.Run("DedupWindowExpiration", func(t *testing.T) {
270
+ t.Cleanup(resetDedupState); oldWindow := dedupWindow; dedupWindow = 50 * time.Millisecond; t.Cleanup(func() { dedupWindow = oldWindow })
271
+ req1 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "exp1"}`))
272
+ rr1 := httptest.NewRecorder(); stdout1, _ := captureOutput(t, func() { handler(rr1, req1) })
273
+ if rr1.Code != http.StatusOK || !strings.Contains(stdout1, "exp1") { t.Errorf("Req 1 failed") }
274
+ req2 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "exp1"}`))
275
+ rr2 := httptest.NewRecorder(); stdout2, _ := captureOutput(t, func() { handler(rr2, req2) })
276
+ if stdout2 != "" { t.Errorf("Immediate duplicate not suppressed") }
277
+ time.Sleep(100 * time.Millisecond)
278
+ req3 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "exp1"}`))
279
+ rr3 := httptest.NewRecorder(); stdout3, _ := captureOutput(t, func() { handler(rr3, req3) })
280
+ if rr3.Code != http.StatusOK || !strings.Contains(stdout3, "exp1") { t.Errorf("Req 3 after expiry failed") }
281
+ })
282
227
- // Prepare request - JSON is valid but missing the 'id' field used by keyPaths
228
- jsonBody := `{"other_id": "uuid-4", "data": "value4"}`
283
+ t.Run("LongKeyValues", func(t *testing.T) {
284
+ t.Cleanup(resetDedupState); longId := strings.Repeat("a", 500)
285
+ jsonBody := fmt.Sprintf(`{"id": "%s", "data": "long"}`, longId)
286
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
230
- rr := httptest.NewRecorder()
231
-
232
- // Execute handler and capture output
233
- stdout, stderr := captureOutput(t, func() {
234
- handler(rr, req)
235
- })
287
+ rr := httptest.NewRecorder(); stdout, _ := captureOutput(t, func() { handler(rr, req) })
288
+ if rr.Code != http.StatusOK || !strings.Contains(stdout, "long") { t.Errorf("Long key req failed") }
289
+ reqDup := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
290
+ rrDup := httptest.NewRecorder(); stdoutDup, _ := captureOutput(t, func() { handler(rrDup, reqDup) })
291
+ if stdoutDup != "" { t.Errorf("Long key duplicate not suppressed") }
292
+ })
293
237
- // Assertions for missing key (results in empty string "" for the key part)
238
- // This *should* be processed correctly, as "" is a valid key string before hashing
239
- // The hash of "" will be deduplicated like any other hash.
294
+ t.Run("OptionsMethod", func(t *testing.T) {
295
+ t.Cleanup(resetDedupState)
296
+ req := httptest.NewRequest(http.MethodOptions, "/", nil)
297
+ rr := httptest.NewRecorder(); stdout, _ := captureOutput(t, func() { handler(rr, req) })
298
+ if status := rr.Code; status != http.StatusMethodNotAllowed { t.Errorf("status: got %v want %v", status, http.StatusMethodNotAllowed) }
299
+ if stdout != "" { t.Errorf("stdout not empty: %q", stdout) }
300
+ })
301
241
- if status := rr.Code; status != http.StatusOK {
242
- t.Errorf("handler returned wrong status code for missing key: got %v want %v", status, http.StatusOK)
302
+ t.Run("VariousJSONFormats", func(t *testing.T) {
303
+ t.Cleanup(resetDedupState)
304
+ testCases := []struct{ name string; body string; expectStatus int; expectOutput bool }{
305
+ {"EmptyObject", `{}`, http.StatusOK, true}, {"ValidJSON", `{"id": "valid"}`, http.StatusOK, true},
306
+ {"SingleQuotes", `{'id': 'invalid'}`, http.StatusBadRequest, false}, {"TrailingComma", `{"id": "comma",}`, http.StatusBadRequest, false},
307
+ {"UnquotedKey", `{id: "unquoted"}`, http.StatusBadRequest, false},
308
}
244
- // Stdout should contain the JSON
245
- if !strings.Contains(stdout, `"other_id":"uuid-4"`) || !strings.Contains(stdout, `"data":"value4"`) {
246
- t.Errorf("handler produced unexpected stdout for missing key:\ngot: %q\nwant it to contain parts of: %q", stdout, jsonBody)
309
+ for _, tc := range testCases {
310
+ t.Run(tc.name, func(t *testing.T) {
311
+ req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(tc.body))
312
+ rr := httptest.NewRecorder(); stdout, _ := captureOutput(t, func() { handler(rr, req) })
313
+ if status := rr.Code; status != tc.expectStatus { t.Errorf("status: got %v want %v", status, tc.expectStatus) }
314
+ hasOutput := stdout != ""; if hasOutput != tc.expectOutput { t.Errorf("Output mismatch: expected %t, got %t (stdout: %q)", tc.expectOutput, hasOutput, stdout) }
315
+ })
316
}
248
- if stderr != "" {
249
- t.Errorf("handler produced unexpected stderr for missing key: got %q want empty", stderr)
317
+ })
318
+
319
+ t.Run("ConcurrentRequests", func(t *testing.T) {
320
+ t.Cleanup(resetDedupState); numRequests := 50; var wg sync.WaitGroup; wg.Add(numRequests)
321
+ process := func(id int) {
322
+ defer wg.Done(); jsonBody := fmt.Sprintf(`{"id": "conc-%d"}`, id); req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
323
+ rr := httptest.NewRecorder(); captureOutput(t, func() { handler(rr, req) })
324
+ if status := rr.Code; status != http.StatusOK { t.Logf("conc req %d status: got %v want %v", id, status, http.StatusOK); t.Fail() }
325
}
251
- // Check map (hash of "" should be present)
252
- mapMutex.Lock()
253
- if len(seenIDs) != 1 {
254
- t.Errorf("expected 1 entry in seenIDs map for missing key, got %d", len(seenIDs))
326
+ for i := 0; i < numRequests; i++ { go process(i) }; wg.Wait()
327
+ mapMutex.Lock(); mapLen := len(seenIDs); mapMutex.Unlock()
328
+ if mapLen != numRequests { t.Errorf("map size: got %d want %d", mapLen, numRequests) }
329
+ })
330
+
331
+ t.Run("CleanupExpiredEntries", func(t *testing.T) {
332
+ t.Cleanup(resetDedupState); oldWindow := dedupWindow; dedupWindow = 50 * time.Millisecond; t.Cleanup(func() { dedupWindow = oldWindow }); numEntries := 5
333
+ for i := 0; i < numEntries; i++ { req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(fmt.Sprintf(`{"id": "clean-%d"}`, i))); rr := httptest.NewRecorder(); captureOutput(t, func() { handler(rr, req) }); if rr.Code != http.StatusOK { t.Fatalf("Setup failed entry %d", i)} }
334
+ mapMutex.Lock(); if got := len(seenIDs); got != numEntries { t.Fatalf("Entries after add: %d != %d", got, numEntries) }; mapMutex.Unlock()
335
+ time.Sleep(100 * time.Millisecond); now := time.Now(); mapMutex.Lock()
336
+ for h, entry := range seenIDs { if now.Sub(entry.timestamp) >= dedupWindow { delete(seenIDs, h) } }
337
+ count := len(seenIDs); mapMutex.Unlock(); if count != 0 { t.Errorf("Entries after cleanup: %d != 0", count) }
338
+ })
339
+
340
+ t.Run("MixedKeyTypes", func(t *testing.T) {
341
+ t.Cleanup(resetDedupState); keyPaths = []string{"id", "count", "enabled"}; dedupSeparator = "|"
342
+ req1 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "mix", "count": 1, "enabled": true}`))
343
+ rr1 := httptest.NewRecorder(); stdout1, _ := captureOutput(t, func() { handler(rr1, req1) })
344
+ if rr1.Code != http.StatusOK || !strings.Contains(stdout1, `"id":"mix"`) { t.Errorf("Req 1 failed") }
345
+ req2 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"enabled": true, "count": 1.0, "id": "mix"}`))
346
+ rr2 := httptest.NewRecorder(); stdout2, _ := captureOutput(t, func() { handler(rr2, req2) })
347
+ if rr2.Code != http.StatusOK { t.Errorf("Req 2 status wrong") }; if stdout2 != "" { t.Errorf("Req 2 produced output") }
348
+ req3 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "mix", "count": 1, "enabled": false}`))
349
+ rr3 := httptest.NewRecorder(); stdout3, _ := captureOutput(t, func() { handler(rr3, req3) })
350
+ if rr3.Code != http.StatusOK || !strings.Contains(stdout3, `"enabled":false`) { t.Errorf("Req 3 failed") }
351
+ mapMutex.Lock(); mapLen := len(seenIDs); mapMutex.Unlock(); if mapLen != 2 { t.Errorf("map size: got %d want 2", mapLen) }
352
+ })
353
+
354
+ t.Run("JsonFormatTests", func(t *testing.T) {
355
+ t.Cleanup(resetDedupState)
356
+ testCases := []struct{ name string; body string; expectStatus int; expectOutput bool }{
357
+ {"EmptyObject", `{}`, http.StatusOK, true}, {"ValidJSON", `{"id": "valid"}`, http.StatusOK, true},
358
+ {"CompletelyInvalid", `not json`, http.StatusBadRequest, false}, {"IncompleteJSON", `{"id": "inc`, http.StatusBadRequest, false},
359
+ {"ArrayAsRoot", `[1, 2]`, http.StatusBadRequest, false}, // Expect 400 now
360
}
256
- mapMutex.Unlock()
361
+ for _, tc := range testCases {
362
+ t.Run(tc.name, func(t *testing.T) {
363
+ req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(tc.body))
364
+ rr := httptest.NewRecorder(); stdout, _ := captureOutput(t, func() { handler(rr, req) })
365
+ if status := rr.Code; status != tc.expectStatus { t.Errorf("status: got %v want %v", status, tc.expectStatus) }
366
+ hasOutput := stdout != ""; if hasOutput != tc.expectOutput { t.Errorf("Output mismatch: expected %t, got %t", tc.expectOutput, hasOutput) }
367
+ })
368
+ }
369
+ })
370
+
371
+ t.Run("MalformedJsonHandling", func(t *testing.T) {
372
+ // Verifies server returns BadRequest for invalid JSON
373
+ t.Cleanup(resetDedupState)
374
+ req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`not json`))
375
+ rr := httptest.NewRecorder(); captureOutput(t, func() { handler(rr, req) })
376
+ if status := rr.Code; status != http.StatusBadRequest { t.Errorf("status invalid: got %v want %v", status, http.StatusBadRequest) }
377
+ req2 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"open":`))
378
+ rr2 := httptest.NewRecorder(); captureOutput(t, func() { handler(rr2, req2) })
379
+ if status := rr2.Code; status != http.StatusBadRequest { t.Errorf("status incomplete: got %v want %v", status, http.StatusBadRequest) }
380
})
381
259
- // Add more test cases: Wrong method (GET), expired duplicate, multiple dedup keys, etc.
382
+ t.Run("ZeroLengthDedupWindow", func(t *testing.T) {
383
+ t.Cleanup(resetDedupState); oldWindow := dedupWindow; dedupWindow = 0; t.Cleanup(func() { dedupWindow = oldWindow })
384
+ jsonBody := `{"id": "zero"}`
385
+ req1 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
386
+ rr1 := httptest.NewRecorder(); stdout1, _ := captureOutput(t, func() { handler(rr1, req1) })
387
+ if rr1.Code != http.StatusOK || !strings.Contains(stdout1, "zero") { t.Errorf("Req 1 failed") }
388
+ req2 := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(jsonBody))
389
+ rr2 := httptest.NewRecorder(); stdout2, _ := captureOutput(t, func() { handler(rr2, req2) })
390
+ if rr2.Code != http.StatusOK || !strings.Contains(stdout2, "zero") { t.Errorf("Req 2 (duplicate) failed") }
391
+ })
392
}