Align multiple audit logs

Author: wzshiming

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

@@ -27,30 +27,64 @@ func init() {
 	pflag.Parse()
 }
 
-func main() {
-	go func() {
-		if delay > 0 {
-			time.Sleep(delay)
+func monitorAndStartExporters() {
+	paths := make([]string, 0, len(auditLogPath))
+	labels := make([]string, 0, len(auditLogPath))
+
+	for _, p := range auditLogPath {
+		path, label := getPathAndLabel(p)
+		paths = append(paths, path)
+		labels = append(labels, label)
+	}
+
+	for !validAuditLogs(paths) {
+		time.Sleep(time.Second)
+	}
+
+	if delay > 0 {
+		time.Sleep(delay)
+	}
+
+	for i, path := range paths {
+		e := exporter.NewExporter(
+			exporter.WithReplay(replay),
+			exporter.WithFile(path),
+			exporter.WithClusterLabel(labels[i]),
+		)
+		go e.Run()
+	}
+}
+
+func validAuditLogs(paths []string) bool {
+	for _, p := range paths {
+		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 getPathAndLabel(s string) (string, string) {
+	parts := strings.SplitN(s, ":", 2)
+	path := parts[0]
+	clusterLabel := cluster
+	if len(parts) > 1 {
+		clusterLabel = parts[1]
+	}
+	return path, clusterLabel
+}
+
+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

@@ -60,11 +60,6 @@ type Exporter struct {
 	batchJobCreationTimes map[target]*time.Time
 }
 
-func (p *Exporter) Start() {
-	// Process audit events
-	go p.processAuditEvents()
-}
-
 func ListenAndServe(addr string) error {
 	mux := http.NewServeMux()
 	handler := promhttp.HandlerFor(registry, promhttp.HandlerOpts{
@@ -76,8 +71,8 @@ func ListenAndServe(addr string) error {
 	return http.ListenAndServe(addr, mux)
 }
 
-// processAuditEvents handles audit log file changes
-func (p *Exporter) processAuditEvents() {
+// Run handles audit log file changes
+func (p *Exporter) Run() {
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 
@@ -90,7 +85,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 +97,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 +116,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

exporter/metrics.go

@@ -76,7 +76,14 @@ func (p *Exporter) updateMetrics(clusterLabel string, event auditv1.Event) {
 				target := buildTarget(event.ObjectRef)
 				createTime, exists := p.podCreationTimes[target]
 				if !exists {
-					slog.Warn("Pod not found", "target", target)
+					// Kueue's audit events may create pod/binding events before pod creation events
+					user := extractUserAgent(event.UserAgent)
+					podSchedulingLatency.WithLabelValues(
+						clusterLabel,
+						ns,
+						user,
+					).Observe(0)
+					p.podCreationTimes[target] = nil
 					return
 				}