feat(node): add per-request audit and accounting for FinOps telemetry

[aojea]

• Added wrapper to for raw streams.
• Added and to to capture per-request bytes on HTTP services, fulfilling FinOps requirement.
• Modified E2E datapath test to verify synchronous log flushing.
• Ensures all logs correlate the OIDC identity with bytes spent per service.

[gemini-code-assist]

Code Review

This pull request introduces stream accounting and audit traceability logging to track bandwidth usage and request context across libp2p streams and HTTP ingress. The reviewer feedback highlights several important improvements: logging pre-auth bytes under a default target to prevent untracked bandwidth during failures, implementing http.Flusher on the custom response writer wrapper to avoid breaking downstream streaming, wrapping HTTP requests earlier to capture early validation bytes, and fixing a potential out-of-bounds panic in the test helper when log files are small.

[gemini-code-assist]

If the auth handshake fails or is rate-limited, any bytes read/written before the failure are not logged because reqCtx.Target is empty. This can lead to untracked bandwidth usage (e.g., from malicious clients sending large payloads that fail auth). We should log these bytes under a default target like "system/auth" to ensure complete FinOps accounting.

		defer func() {
			target := reqCtx.Target
			if target == "" && (ts.bytesRead > 0 || ts.bytesWritten > 0) {
				target = "system/auth"
			}
			if target != "" {
				logger.Infow("Stream Accounting",
					"peer_id", remotePeer.String(),
					"target", target,
					"protocol", reqCtx.Protocol,
					"bytes_read", ts.bytesRead,
					"bytes_written", ts.bytesWritten,
				)
			}
			if err := ts.Close(); err != nil {
				logger.Debugf("[Auth] Failed to close stream: %v", err)
			}
		}()

[gemini-code-assist]

Wrapping http.ResponseWriter in responseWriterWithCount hides other interface implementations like http.Flusher. If downstream handlers or middleware attempt to assert w.(http.Flusher) to flush streaming data (e.g., for SSE or chunked transfers), the assertion will fail. Implementing the Flush method on responseWriterWithCount and forwarding it to the underlying writer if supported prevents this issue.

type responseWriterWithCount struct {
	http.ResponseWriter
	bytesWritten int64
	statusCode   int
}

func (w *responseWriterWithCount) Write(p []byte) (int, error) {
	n, err := w.ResponseWriter.Write(p)
	w.bytesWritten += int64(n)
	return n, err
}

func (w *responseWriterWithCount) WriteHeader(statusCode int) {
	w.statusCode = statusCode
	w.ResponseWriter.WriteHeader(statusCode)
}

func (w *responseWriterWithCount) Flush() {
	if f, ok := w.ResponseWriter.(http.Flusher); ok {
		f.Flush()
	}
}

[gemini-code-assist]

Wrapping the request body and response writer for accounting at this point misses all bytes read/written during the early validation and authorization phases (which happen before this block). If a request fails authorization or has an invalid path, those bytes are not accounted for in "Stream Accounting". To ensure accurate FinOps telemetry, consider wrapping r.Body and w at the very beginning of the HTTP handler.

[gemini-code-assist]

The assertLogInFile helper reads the log file and prints the last 1000 bytes on failure using b[len(b)-1000:]. If the log file is smaller than 1000 bytes, this slice operation will panic with an out-of-bounds error, causing the test suite to crash instead of failing gracefully. We should safely bound the slice index.

Suggested change

	func assertLogInFile(t *testing.T, logFile, expectedMsg string) {
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
	b, err := os.ReadFile(logFile)
	if err == nil && strings.Contains(string(b), expectedMsg) {
	return
	}
	time.Sleep(100 * time.Millisecond)
	}
	b, _ := os.ReadFile(logFile)
	t.Errorf("Expected log file %s to contain %q, but it didn't.\nLog end:\n%s", logFile, expectedMsg, string(b[len(b)-1000:])) // print last 1000 bytes
	}
	func assertLogInFile(t *testing.T, logFile, expectedMsg string) {
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
	b, err := os.ReadFile(logFile)
	if err == nil && strings.Contains(string(b), expectedMsg) {
	return
	}
	time.Sleep(100 * time.Millisecond)
	}
	b, _ := os.ReadFile(logFile)
	logEnd := string(b)
	if len(b) > 1000 {
	logEnd = string(b[len(b)-1000:])
	}
	t.Errorf("Expected log file %s to contain %q, but it didn't.\\nLog end:\\n%s", logFile, expectedMsg, logEnd)
	}

[aojea]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces stream accounting and audit traceability features by wrapping network streams and HTTP requests/responses to track bytes read and written, and extracting user metadata from biscuit tokens for audit logging. The feedback recommends using the safer and more idiomatic atomic.Int64 type instead of raw int64 with atomic.AddInt64 to prevent potential alignment panics on 32-bit architectures. Additionally, a duplicated assignment to r.URL.RawPath in internal/node/node.go should be removed.

[aojea]

/gemini review

[gemini-code-assist]

Code Review

This pull request updates several dependencies and simplifies the dht.New constructor by removing the context.Context parameter. It introduces stream accounting and audit logging for both libp2p streams and HTTP ingress requests by wrapping readers and writers to track data usage. Integration tests were updated to verify these logs. The review feedback highlights the need to handle errors when querying audit facts and suggests refactoring the duplicated query logic into a helper function to improve maintainability.

[gemini-code-assist]

The error returned from authorizer.Query is being ignored. This can mask problems like query timeouts or other evaluation errors. The error should be checked and logged. This applies to the subsequent queries for email and role as well.

	if facts, err := authorizer.Query(biscuit.Rule{
		Head: biscuit.Predicate{Name: "get_user", IDs: []biscuit.Term{biscuit.Variable("u")}},
		Body: []biscuit.Predicate{{Name: api.FactUser, IDs: []biscuit.Term{biscuit.Variable("u")}}},
	}); err != nil {
		logger.Warnf("Audit traceability: failed to query user fact: %v", err)
	} else if len(facts) > 0 && len(facts[0].IDs) > 0 {
		if s, ok := facts[0].IDs[0].(biscuit.String); ok {
			userStr = string(s)
		}
	}

[gemini-code-assist]

The logic for querying a fact (user, email, role) and extracting its string value is duplicated. Consider refactoring this into a helper function to improve readability and maintainability, especially since error handling should also be added to each query as noted in another comment. A single function could handle querying, error logging, and extracting the value.