Align multiple audit logs

Author: wzshiming

cmd/kube-apiserver-audit-exporter/main.go

@@ -27,30 +27,56 @@ func init() {
 	pflag.Parse()
 }
 
-func main() {
-	go func() {
-		if delay > 0 {
-			time.Sleep(delay)
+func monitorAndStartExporters() {
+	for !validAuditLogs() {
+		time.Sleep(time.Second)
+	}
+
+	if delay > 0 {
+		time.Sleep(delay)
+	}
+
+	for _, path := range auditLogPath {
+		startExporterForPath(path)
+	}
+}
+
+func validAuditLogs() bool {
+	for _, p := range auditLogPath {
+		info, err := os.Stat(p)
+		if err != nil {
+			slog.Warn("Failed to stat audit log", "path", p, "err", err)
+			return false
 		}
-		for _, p := range auditLogPath {
-			ns := strings.SplitN(p, ":", 2)
-			path := ns[0]
-			clusterLabel := cluster
-			if len(ns) > 1 {
-				clusterLabel = ns[1]
-			}
-
-			e := exporter.NewExporter(
-				exporter.WithReplay(replay),
-				exporter.WithFile(path),
-				exporter.WithClusterLabel(clusterLabel),
-			)
-			e.Start()
+		if info.Size() == 0 {
+			slog.Info("Audit log is empty, waiting for content", "path", p)
+			return false
 		}
-	}()
+	}
+	return true
+}
+
+func startExporterForPath(pathWithLabel string) {
+	parts := strings.SplitN(pathWithLabel, ":", 2)
+	path := parts[0]
+	clusterLabel := cluster
+	if len(parts) > 1 {
+		clusterLabel = parts[1]
+	}
+
+	e := exporter.NewExporter(
+		exporter.WithReplay(replay),
+		exporter.WithFile(path),
+		exporter.WithClusterLabel(clusterLabel),
+	)
+	e.Start()
+}
+
+func main() {
+	go monitorAndStartExporters()
 
 	if err := exporter.ListenAndServe(address); err != nil {
-		slog.Error("Failed", "err", err)
+		slog.Error("Failed to start metrics server", "err", err)
 		os.Exit(1)
 	}
 }

exporter/exporter.go

@@ -90,7 +90,7 @@ func (p *Exporter) processAuditEvents() {
 // handleFileEvent processes filesystem events
 func (p *Exporter) handleFileEvent(path string) {
 	if err := p.processFileUpdate(path); err != nil {
-		slog.Error("Error processing file", "error", err)
+		slog.Error("Error processing file", "cluster", p.clusterLabel, "error", err)
 	}
 }
 
@@ -102,10 +102,10 @@ func (p *Exporter) processFileUpdate(path string) error {
 	}
 
 	if size := fileInfo.Size(); size < p.offset {
-		slog.Info("Log file truncated, resetting offset")
+		slog.Info("Log file truncated, resetting offset", "cluster", p.clusterLabel)
 		p.offset = 0
 	} else if size == p.offset {
-		slog.Info("No new updates in log file", "offset", p.offset)
+		slog.Info("No new updates in log file", "cluster", p.clusterLabel, "offset", p.offset)
 		return nil
 	}
 
@@ -121,7 +121,7 @@ func (p *Exporter) processFileUpdate(path string) error {
 
 	start := time.Now()
 	defer func() {
-		slog.Info("File processing complete", "new_offset", p.offset, "duration", time.Since(start))
+		slog.Info("File processing complete", "cluster", p.clusterLabel, "new_offset", p.offset, "duration", time.Since(start))
 	}()
 
 	reader := bufio.NewReaderSize(file, 1<<24) // 16MB buffer