Add path latency metrics and race-detection CI job

Author: sidmohan0

.github/workflows/main-cicd.yml

@@ -59,3 +59,15 @@ jobs:
         with:
           name: benchmark-results
           path: /tmp/bench/benchmark.txt
+
+  race:
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: "1.24.13"
+      - name: Run race detector
+        run: go test -race ./...

README.md

@@ -40,9 +40,9 @@ The wrapper process is still named `datafog-shim` for compatibility, but we desc
 
 ## Positioning
 
-- **Developers and agent builders:** DataFog is a **privacy firewall for CLI tools and AI agents**. It sits in your PATH or runtime, inspects what is flowing through your commands, and enforces policy before data-sensitive actions execute.
-- **Security/compliance buyers:** DataFog is runtime policy-as-code enforcement at the process level with receipts for every decision.
-- **Broader view:** DataFog is the **data plane for agent governance** — detect, decide, enforce, and audit—not just “PII redaction.”
+- **Developers and agent builders:** DataFog is a **data-aware policy enforcement layer** for CLI tools and AI agents. It sits in your PATH or runtime, inspects data flowing through commands, and enforces policy before sensitive actions execute.
+- **Security/compliance buyers:** DataFog maps closely to runtime DLP for developer workstations, but without the legacy footprint: policy is programmable (OPA-style), decision-aware, and process-bound.
+- **Broader view:** DataFog is the **data plane for agent governance** — detect, decide, enforce, and audit.
 
 ## Repository layout
 
@@ -373,8 +373,9 @@ spec:
 ## If something fails, check these first
 
 1. `go test ./...` (build/runtime validation before changing policy)
-2. `/health` response for policy id/version mismatch
-3. Environment variables are set and files are writable
-4. API token/header if `DATAFOG_API_TOKEN` is configured
-5. Policy JSON is valid and rules match expected action fields
-6. Optional benchmark sweep: `scripts/run-benchmarks.sh`
+2. `go test -race ./...` (check race conditions on concurrency-sensitive paths)
+3. `/health` response for policy id/version mismatch
+4. Environment variables are set and files are writable
+5. API token/header if `DATAFOG_API_TOKEN` is configured
+6. Policy JSON is valid and rules match expected action fields
+7. Optional benchmark sweep: `scripts/run-benchmarks.sh`

docs/OBSERVABILITY.md

@@ -21,6 +21,8 @@ In-process counters are exposed at `GET /metrics`:
 - `by_status`
 - `by_path`
 - `by_method`
+- `avg_latency_ms`
+- `by_path_avg_latency_ms`
 - `uptime_seconds`
 - `started_at`
 

internal/server/server.go

@@ -27,24 +27,27 @@ import (
 )
 
 type Server struct {
-	policy      models.Policy
-	store       *receipts.ReceiptStore
-	eventReader shim.EventReader
-	apiToken    string
-	rateLimiter *tokenBucket
-	startedAt   time.Time
-	logger      *log.Logger
-	mu          sync.Mutex
-	statsMu     sync.Mutex
-	decisions   map[string]idempotentDecision
-	scans       map[string]idempotentCachedResponse
-	transforms  map[string]idempotentCachedResponse
-	anonymizes  map[string]idempotentCachedResponse
-	totalCount  int64
-	errorCount  int64
-	statusHits  map[int]int64
-	pathHits    map[string]int64
-	methodHits  map[string]int64
+	policy            models.Policy
+	store             *receipts.ReceiptStore
+	eventReader       shim.EventReader
+	apiToken          string
+	rateLimiter       *tokenBucket
+	startedAt         time.Time
+	logger            *log.Logger
+	mu                sync.Mutex
+	statsMu           sync.Mutex
+	decisions         map[string]idempotentDecision
+	scans             map[string]idempotentCachedResponse
+	transforms        map[string]idempotentCachedResponse
+	anonymizes        map[string]idempotentCachedResponse
+	totalCount        int64
+	errorCount        int64
+	statusHits        map[int]int64
+	pathHits          map[string]int64
+	methodHits        map[string]int64
+	totalLatencyNs    int64
+	pathLatencyNs     map[string]int64
+	pathLatencyCounts map[string]int64
 }
 
 type requestIDContextKey struct{}
@@ -82,13 +85,15 @@ type idempotentCachedResponse struct {
 }
 
 type metricsResponse struct {
-	TotalRequests int64            `json:"total_requests"`
-	ErrorRequests int64            `json:"error_requests"`
-	ByStatus      map[string]int64 `json:"by_status"`
-	ByPath        map[string]int64 `json:"by_path"`
-	ByMethod      map[string]int64 `json:"by_method"`
-	StartedAt     string           `json:"started_at"`
-	UptimeSeconds float64          `json:"uptime_seconds"`
+	TotalRequests int64              `json:"total_requests"`
+	ErrorRequests int64              `json:"error_requests"`
+	ByStatus      map[string]int64   `json:"by_status"`
+	ByPath        map[string]int64   `json:"by_path"`
+	ByMethod      map[string]int64   `json:"by_method"`
+	ByPathLatency map[string]float64 `json:"by_path_avg_latency_ms"`
+	AvgLatencyMs  float64            `json:"avg_latency_ms"`
+	StartedAt     string             `json:"started_at"`
+	UptimeSeconds float64            `json:"uptime_seconds"`
 }
 
 func New(policyData models.Policy, store *receipts.ReceiptStore, logger *log.Logger, apiToken string, rateLimitRPS int) *Server {
@@ -97,19 +102,22 @@ func New(policyData models.Policy, store *receipts.ReceiptStore, logger *log.Log
 	}
 	policyData = policy.NormalizeForEvaluation(policyData)
 	return &Server{
-		policy:      policyData,
-		store:       store,
-		apiToken:    apiToken,
-		rateLimiter: newTokenBucket(rateLimitRPS),
-		startedAt:   time.Now().UTC(),
-		logger:      logger,
-		decisions:   map[string]idempotentDecision{},
-		scans:       map[string]idempotentCachedResponse{},
-		transforms:  map[string]idempotentCachedResponse{},
-		anonymizes:  map[string]idempotentCachedResponse{},
-		statusHits:  map[int]int64{},
-		pathHits:    map[string]int64{},
-		methodHits:  map[string]int64{},
+		policy:            policyData,
+		store:             store,
+		apiToken:          apiToken,
+		rateLimiter:       newTokenBucket(rateLimitRPS),
+		startedAt:         time.Now().UTC(),
+		logger:            logger,
+		decisions:         map[string]idempotentDecision{},
+		scans:             map[string]idempotentCachedResponse{},
+		transforms:        map[string]idempotentCachedResponse{},
+		anonymizes:        map[string]idempotentCachedResponse{},
+		statusHits:        map[int]int64{},
+		pathHits:          map[string]int64{},
+		methodHits:        map[string]int64{},
+		totalLatencyNs:    0,
+		pathLatencyNs:     map[string]int64{},
+		pathLatencyCounts: map[string]int64{},
 	}
 }
 
@@ -164,6 +172,7 @@ func (s *Server) wrapMiddleware(mux *http.ServeMux) http.Handler {
 		startedAt := time.Now()
 		handler, pattern := mux.Handler(r)
 		defer func() {
+			latency := time.Since(startedAt)
 			if rec := recover(); rec != nil {
 				responseWriter.status = http.StatusInternalServerError
 				s.logger.Printf("request panic request_id=%s method=%s path=%s err=%v", reqID, r.Method, r.URL.Path, rec)
@@ -173,11 +182,11 @@ func (s *Server) wrapMiddleware(mux *http.ServeMux) http.Handler {
 				responseWriter.status = http.StatusOK
 			}
 			if pattern == "" {
-				s.recordRequestMetrics(r.Method, "/_not_found", responseWriter.status)
+				s.recordRequestMetrics(r.Method, "/_not_found", responseWriter.status, latency)
 			} else {
-				s.recordRequestMetrics(r.Method, canonicalizedRoute(pattern, r.URL.Path), responseWriter.status)
+				s.recordRequestMetrics(r.Method, canonicalizedRoute(pattern, r.URL.Path), responseWriter.status, latency)
 			}
-			s.logger.Printf("request complete request_id=%s method=%s path=%s status=%d latency_ms=%d", reqID, r.Method, r.URL.Path, responseWriter.status, time.Since(startedAt).Milliseconds())
+			s.logger.Printf("request complete request_id=%s method=%s path=%s status=%d latency_ms=%d", reqID, r.Method, r.URL.Path, responseWriter.status, latency.Milliseconds())
 		}()
 
 		if !s.authorized(r) {
@@ -275,7 +284,7 @@ func canonicalizedRoute(pattern string, path string) string {
 	return pattern
 }
 
-func (s *Server) recordRequestMetrics(method string, route string, status int) {
+func (s *Server) recordRequestMetrics(method string, route string, status int, latency time.Duration) {
 	s.statsMu.Lock()
 	defer s.statsMu.Unlock()
 	s.totalCount++
@@ -285,6 +294,11 @@ func (s *Server) recordRequestMetrics(method string, route string, status int) {
 	if status >= 400 {
 		s.errorCount++
 	}
+
+	latencyNs := latency.Nanoseconds()
+	s.totalLatencyNs += latencyNs
+	s.pathLatencyNs[route] += latencyNs
+	s.pathLatencyCounts[route]++
 }
 
 func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
@@ -740,12 +754,27 @@ func (s *Server) snapshotMetrics() metricsResponse {
 		byMethod[method] = count
 	}
 
+	byPathLatency := map[string]float64{}
+	for path, count := range s.pathLatencyCounts {
+		if count == 0 {
+			continue
+		}
+		byPathLatency[path] = float64(s.pathLatencyNs[path]) / float64(count) / float64(time.Millisecond)
+	}
+
+	avgLatencyMs := 0.0
+	if s.totalCount > 0 {
+		avgLatencyMs = float64(s.totalLatencyNs) / float64(s.totalCount) / float64(time.Millisecond)
+	}
+
 	return metricsResponse{
 		TotalRequests: s.totalCount,
 		ErrorRequests: s.errorCount,
 		ByStatus:      byStatus,
 		ByPath:        byPath,
 		ByMethod:      byMethod,
+		ByPathLatency: byPathLatency,
+		AvgLatencyMs:  avgLatencyMs,
 		StartedAt:     s.startedAt.Format(time.RFC3339),
 		UptimeSeconds: time.Since(s.startedAt).Seconds(),
 	}

internal/server/server_test.go

@@ -887,6 +887,12 @@ func TestMetricsEndpoint(t *testing.T) {
 	if got.ByPath["/_not_found"] != 1 {
 		t.Fatalf("expected /_not_found to be tracked once, got %d", got.ByPath["/_not_found"])
 	}
+	if got.AvgLatencyMs <= 0 {
+		t.Fatalf("expected positive average latency, got %f", got.AvgLatencyMs)
+	}
+	if got.ByPathLatency["/health"] <= 0 {
+		t.Fatalf("expected health average latency to be tracked, got %f", got.ByPathLatency["/health"])
+	}
 	if _, err := time.Parse(time.RFC3339, got.StartedAt); err != nil {
 		t.Fatalf("expected started_at to be RFC3339, got %q", got.StartedAt)
 	}