agent events No 7 (#20074)
* 1. Enhanced run.sh with multiple dedup keys and a dedicated log file 2. Added Cloudflare headers processing for all requests, not just successful ones 3. Implemented logging of duplicated requests to a separate file 4. Added health check endpoint improvements with more deduplication info 5. Improved test code structure with output capture * dont log duplicates by default
Costa Tsaousis committed
Apr 7, 2025 at 11:26 UTC
59cba7e98140266b308c2508f2827c955da2830c
3 files changed
+103
-31
packaging/tools/agent-events/run.sh
+10
-1
@@ -1,6 +1,15 @@
1
#!/usr/bin/env bash
2
3
-stdbuf -oL /opt/agent-events/server --port=30001 --dedup-key agent.id --dedup-window 1800 2>/opt/agent-events/log/agent-events.log \
3
+# --dedup-logfile=/opt/agent-events/log/dedup.log \
4
+
5
+stdbuf -oL /opt/agent-events/server \
6
+ --port=30001 \
7
+ --dedup-key=agent.id \
8
+ --dedup-key=host.id \
9
+ --dedup-key=host.boot.id \
10
+ --dedup-key=exit_cause \
11
+ --dedup-window=1800 \
12
+ 2>/opt/agent-events/log/stderr.log \
13
| stdbuf -oL log2journal json \
14
--prefix 'AE_' \
15
--inject 'SYSLOG_IDENTIFIER=agent-events' \
packaging/tools/agent-events/server.go
+84
-27
@@ -57,6 +57,8 @@ var (
57
keyPaths dedupPaths
58
dedupSeparator string
59
startTime time.Time
60
+ dedupLogger *os.File
61
+ dedupLogFile string // Store the deduplication log file path
62
63
// Track active connections for graceful shutdown
64
activeConnections int32
@@ -301,8 +303,17 @@ func handler(w http.ResponseWriter, r *http.Request) {
303
304
bytesReceived.Add(ctx, int64(len(body)), metric.WithAttributes(attribute.String("status", "success")))
305
306
+ // Add Cloudflare Headers to all requests
307
+ cfHeaders := make(map[string]string)
308
+ 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"}
309
+ for _, name := range cfHeaderPrefixes { if value := r.Header.Get(name); value != "" { key := strings.TrimPrefix(name, "CF-"); cfHeaders[key] = value } }
310
+ 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] } } }
311
+ if len(cfHeaders) > 0 { fullData["cf"] = cfHeaders; slog.Debug("added cloudflare headers", "count", len(cfHeaders)) }
312
+
313
// Deduplication Logic
314
shouldProcess := true
315
+ var finalKeyString string
316
+ var dedupHash [32]byte
317
if len(keyPaths) > 0 {
318
var keyBuilder strings.Builder
319
for i, path := range keyPaths {
@@ -310,50 +321,61 @@ func handler(w http.ResponseWriter, r *http.Request) {
321
keyBuilder.WriteString(result.String()) // gjson returns "" for non-existent paths
322
if i < len(keyPaths)-1 { keyBuilder.WriteString(dedupSeparator) }
323
}
313
- finalKeyString := keyBuilder.String()
314
- dedupHash := sha256.Sum256([]byte(finalKeyString))
324
+ finalKeyString = keyBuilder.String()
325
+ dedupHash = sha256.Sum256([]byte(finalKeyString))
326
slog.Debug("generated dedup key", "key_string", finalKeyString, "hash", fmt.Sprintf("%x", dedupHash))
327
+
328
+ // Add _dedup key to all requests (regardless of duplicate status)
329
+ fullData["_dedup"] = map[string]interface{}{
330
+ "key": finalKeyString,
331
+ "hash": fmt.Sprintf("%x", dedupHash),
332
+ }
333
334
+ // Check if this is a duplicate
335
if !checkAndRecordHash(dedupHash) {
336
shouldProcess = false
337
bytesReceived.Add(ctx, int64(len(body)), metric.WithAttributes(attribute.String("status", "duplicate")))
338
requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "duplicate")))
339
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) }
323
- return
340
}
341
} else {
342
slog.Debug("skipping deduplication", "reason", "no dedup keys provided")
343
}
344
329
- // Process Request if Not Duplicate
330
- if shouldProcess {
331
- // We've already parsed fullData in the initial check
332
-
333
- // Add Cloudflare Headers
334
- cfHeaders := make(map[string]string)
335
- 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"}
336
- for _, name := range cfHeaderPrefixes { if value := r.Header.Get(name); value != "" { key := strings.TrimPrefix(name, "CF-"); cfHeaders[key] = value } }
337
- 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] } } }
338
- if len(cfHeaders) > 0 { fullData["cf"] = cfHeaders; slog.Debug("added cloudflare headers", "count", len(cfHeaders)) }
339
-
340
- // Marshal Output
341
- outputBytes, err := json.Marshal(fullData)
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)
345
- requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "cant_marshal_output")))
346
- return
347
- }
345
+ // Marshal the fully prepared object for either stdout or dedup log
346
+ outputBytes, err := json.Marshal(fullData)
347
+ if err != nil {
348
+ http.Error(w, "Internal Server Error during output marshal", http.StatusInternalServerError)
349
+ slog.Error("request discarded", "reason", "json_marshal_failed", "error", err)
350
+ requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "cant_marshal_output")))
351
+ return
352
+ }
353
349
- // Write Output & Response
354
+ // Decide where to output based on deduplication status
355
+ if shouldProcess {
356
+ // Write to stdout for normal processing
357
fmt.Println(string(outputBytes))
351
- // Count successful processing
358
requestsCounter.Add(ctx, 1, metric.WithAttributes(attribute.String("status", "success")))
353
- if _, err := w.Write([]byte("OK")); err != nil {
359
+ } else if dedupLogger != nil {
360
+ // Write to dedup log file if it's a duplicate and logging is enabled
361
+ if _, err := fmt.Fprintln(dedupLogger, string(outputBytes)); err != nil {
362
+ slog.Error("failed to write to deduplication log file", "error", err)
363
+ }
364
+ }
365
+
366
+ // Send response
367
+ if _, err := w.Write([]byte("OK")); err != nil {
368
+ if shouldProcess {
369
slog.Error("error writing response", "context", "after_successful_processing", "error", err)
370
+ } else {
371
+ slog.Error("error writing response", "context", "after_duplicate_discard", "error", err)
372
}
373
}
374
+
375
+ // For duplicates, return early
376
+ if !shouldProcess {
377
+ return
378
+ }
379
}
380
381
// healthHandler provides a simple health check endpoint
@@ -367,7 +389,20 @@ func healthHandler(w http.ResponseWriter, r *http.Request) {
389
"version": "agent-events v1.0",
390
}
391
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}
370
- mapMutex.Lock(); mapSize := len(seenIDs); mapMutex.Unlock(); status["deduplication"] = map[string]interface{}{"enabled": len(keyPaths) > 0, "keys": keyPaths, "window": dedupWindow.String(), "map_size": mapSize}
392
+ mapMutex.Lock(); mapSize := len(seenIDs); mapMutex.Unlock();
393
+ dedupLogEnabled := dedupLogger != nil
394
+ dedupLogPath := ""
395
+ if dedupLogEnabled {
396
+ dedupLogPath = dedupLogFile
397
+ }
398
+ status["deduplication"] = map[string]interface{}{
399
+ "enabled": len(keyPaths) > 0,
400
+ "keys": keyPaths,
401
+ "window": dedupWindow.String(),
402
+ "map_size": mapSize,
403
+ "log_enabled": dedupLogEnabled,
404
+ "log_file": dedupLogPath,
405
+ }
406
w.Header().Set("Content-Type", "application/json"); w.WriteHeader(http.StatusOK);
407
if err := json.NewEncoder(w).Encode(status); err != nil { slog.Error("error encoding health check response", "error", err) }
408
}
@@ -389,6 +424,8 @@ func main() {
424
healthPath := flag.String("health-path", "/healthz", "Path for health check endpoint")
425
logFormat := flag.String("log-format", "json", "Log format: 'json' or 'text'")
426
logLevelFlag := flag.String("log-level", "info", "Log level: 'debug', 'info', 'warn', 'error'")
427
+ // Use the global dedupLogFile variable
428
+ flag.StringVar(&dedupLogFile, "dedup-logfile", "", "File to log deduplicated requests (empty to disable)")
429
flag.Var(&keyPaths, "dedup-key", "JSON path (dot-notation) for deduplication key (can be used multiple times)")
430
flag.StringVar(&dedupSeparator, "dedup-separator", "-", "Separator used between multi-key values")
431
flag.Parse()
@@ -411,6 +448,18 @@ func main() {
448
// Initialize core components
449
seenIDs = make(map[[32]byte]seenEntry)
450
dedupWindow = time.Duration(*dedupSeconds) * time.Second
451
+
452
+ // Open deduplication log file if specified
453
+ if dedupLogFile != "" {
454
+ var err error
455
+ dedupLogger, err = os.OpenFile(dedupLogFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
456
+ if err != nil {
457
+ slog.Error("failed to open deduplication log file", "file", dedupLogFile, "error", err)
458
+ os.Exit(1)
459
+ }
460
+ slog.Info("deduplication log file opened", "path", dedupLogFile)
461
+ }
462
+
463
if _, err := initMetrics(); err != nil { // Handle potential error from initMetrics
464
slog.Error("failed to initialize metrics", "error", err)
465
os.Exit(1)
@@ -471,6 +520,14 @@ func main() {
520
// Shutdown HTTP server
521
slog.Info("shutting down HTTP server")
522
if err := server.Shutdown(shutdownCtx); err != nil { slog.Error("server shutdown failed", "error", err) } else { slog.Info("server shutdown completed gracefully") }
523
+
524
+ // Close deduplication log file if open
525
+ if dedupLogger != nil {
526
+ slog.Info("closing deduplication log file")
527
+ if err := dedupLogger.Close(); err != nil {
528
+ slog.Error("failed to close deduplication log file", "error", err)
529
+ }
530
+ }
531
}
532
533
slog.Info("server exiting")
packaging/tools/agent-events/server_test.go
+9
-3
@@ -225,9 +225,15 @@ func TestHandler(t *testing.T) {
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
228
+ captureOutput(t, func() {
229
+ handler(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "m1"}`)))
230
+ })
231
+ captureOutput(t, func() {
232
+ handler(httptest.NewRecorder(), httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id": "m1"}`))) // duplicate
233
+ })
234
+ captureOutput(t, func() {
235
+ handler(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) // method not allowed
236
+ })
237
238
// Fetch metrics
239
resp, err := http.Get(metricsServer.URL)