Go CI: unit tests, race-enabled checks, and cmd/internal structure

.github/workflows/go-ci.yaml

@@ -65,8 +65,8 @@ jobs:
           set -euo pipefail
           go vet ./...
 
-      - name: go test
+      - name: go test (race)
         working-directory: ${{ matrix.module }}
         run: |
           set -euo pipefail
-          go test ./...
+          go test -race ./...

README.md

@@ -372,8 +372,8 @@ This pipeline exists to ensure the Helm chart remains:
 - Add CI checks for Go code quality:
   - `go test ./...` (with `-race` where feasible)
   - `golangci-lint` (or at least `go vet`)
-  - formatting checks (`gofmt`) and module tidiness (`go mod tidy` / `go mod verify`)
-- Keep these checks fast to run on every PR, and required before release
+  - formatting checks (`gofmt`) and module tidiness (`go mod tidy` / `go mod verify`) ✅
+- Keep these checks fast to run on every PR, and required before release ✅
 
 ### Streaming (SSE) – final validation (edge cases)
 - Client disconnect propagation (client → proxy → upstream)

collector/Dockerfile

@@ -1,20 +1,21 @@
-# -------- build stage --------
-FROM golang:1.22-alpine AS builder
-
-WORKDIR /src
-COPY go.mod ./
-RUN go mod download
-
-COPY . .
-RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath -ldflags="-s -w" -o /out/llm-collector .
-
-# -------- runtime stage --------
-FROM gcr.io/distroless/static:nonroot
-
-WORKDIR /
-COPY --from=builder /out/llm-collector /llm-collector
-
-EXPOSE 8081
-USER nonroot:nonroot
-
-ENTRYPOINT ["/llm-collector"]
+# -------- build stage --------
+FROM golang:1.22-alpine AS builder
+
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+
+COPY . .
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \
+    go build -trimpath -ldflags="-s -w" -o /out/llm-collector ./cmd/collector
+
+# -------- runtime stage --------
+FROM gcr.io/distroless/static:nonroot
+
+WORKDIR /
+COPY --from=builder /out/llm-collector /llm-collector
+
+EXPOSE 8081
+USER nonroot:nonroot
+
+ENTRYPOINT ["/llm-collector"]

collector/cmd/collector/main.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"log"
+	"net/http"
+	"time"
+
+	collector "llm-collector/internal/collector"
+)
+
+func main() {
+	addr := ":" + collector.Getenv("PORT", "8081")
+	outPath := collector.Getenv("EVENT_LOG_PATH", "")
+
+	s, err := collector.NewServer(outPath)
+	if err != nil {
+		log.Fatalf("open log file: %v", err)
+	}
+	defer s.Close()
+
+	log.Printf("collector listening on %s", addr)
+
+	srv := &http.Server{
+		Addr:              addr,
+		Handler:           s.Mux(),
+		ReadHeaderTimeout: 5 * time.Second,
+	}
+
+	log.Fatal(srv.ListenAndServe())
+}

collector/go.mod

@@ -1,3 +1,11 @@
 module llm-collector
 
 go 1.22
+
+require github.com/stretchr/testify v1.9.0
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

collector/go.sum

@@ -0,0 +1,10 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

collector/internal/collector/server.go

@@ -0,0 +1,102 @@
+package collector
+
+import (
+	"bufio"
+	"encoding/json"
+	"log"
+	"net/http"
+	"os"
+	"sync"
+	"time"
+)
+
+type Server struct {
+	mu   sync.Mutex
+	file *os.File
+	w    *bufio.Writer
+}
+
+// NewServer creates a server. If outPath is empty, events are printed to stdout.
+func NewServer(outPath string) (*Server, error) {
+	s := &Server{}
+	if outPath == "" {
+		log.Printf("collector: EVENT_LOG_PATH not set; events will be printed to stdout")
+		return s, nil
+	}
+
+	f, err := os.OpenFile(outPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
+	if err != nil {
+		return nil, err
+	}
+
+	s.file = f
+	s.w = bufio.NewWriterSize(f, 1<<20)
+	log.Printf("collector: writing events to %s", outPath)
+	return s, nil
+}
+
+// Close flushes and closes the underlying file if configured.
+func (s *Server) Close() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.w != nil {
+		_ = s.w.Flush()
+	}
+	if s.file != nil {
+		_ = s.file.Close()
+	}
+}
+
+func (s *Server) Mux() *http.ServeMux {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok"))
+	})
+	mux.HandleFunc("/events", s.HandleEvents)
+	return mux
+}
+
+func (s *Server) HandleEvents(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	var ev MeteringEvent
+	dec := json.NewDecoder(r.Body)
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&ev); err != nil {
+		http.Error(w, "invalid json: "+err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	if ev.RequestID == "" {
+		http.Error(w, "missing request_id", http.StatusBadRequest)
+		return
+	}
+	if ev.At.IsZero() {
+		ev.At = time.Now().UTC()
+	}
+
+	b, err := json.Marshal(ev)
+	if err != nil {
+		http.Error(w, "failed to marshal", http.StatusInternalServerError)
+		return
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.w != nil {
+		_, _ = s.w.Write(b)
+		_, _ = s.w.WriteString("\n")
+		_ = s.w.Flush()
+	} else {
+		log.Printf("EVENT %s", string(b))
+	}
+
+	w.WriteHeader(http.StatusAccepted)
+	_, _ = w.Write([]byte("accepted"))
+}

collector/internal/collector/server_test.go

@@ -0,0 +1,61 @@
+package collector
+
+import (
+	"bytes"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHandleEvents_Accepted(t *testing.T) {
+	s, err := NewServer("") // stdout mode
+	require.NoError(t, err)
+
+	ev := MeteringEvent{
+		RequestID: "req_123",
+		Tenant:    "default",
+		Provider:  "openai",
+		Model:     "gpt-4",
+		At:        time.Now().UTC(),
+	}
+
+	body, _ := json.Marshal(ev)
+	req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body))
+	rec := httptest.NewRecorder()
+
+	s.HandleEvents(rec, req)
+
+	require.Equal(t, http.StatusAccepted, rec.Code)
+	require.Contains(t, rec.Body.String(), "accepted")
+}
+
+func TestHandleEvents_InvalidJSON(t *testing.T) {
+	s, err := NewServer("")
+	require.NoError(t, err)
+
+	req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewBufferString("{bad json"))
+	rec := httptest.NewRecorder()
+
+	s.HandleEvents(rec, req)
+
+	require.Equal(t, http.StatusBadRequest, rec.Code)
+}
+
+func TestHandleEvents_MissingRequestID(t *testing.T) {
+	s, err := NewServer("")
+	require.NoError(t, err)
+
+	ev := MeteringEvent{}
+	body, _ := json.Marshal(ev)
+
+	req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body))
+	rec := httptest.NewRecorder()
+
+	s.HandleEvents(rec, req)
+
+	require.Equal(t, http.StatusBadRequest, rec.Code)
+}

collector/internal/collector/types.go

@@ -0,0 +1,18 @@
+package collector
+
+import "time"
+
+type MeteringEvent struct {
+	RequestID        string    `json:"request_id"`
+	Tenant           string    `json:"tenant"`
+	AppKey           string    `json:"app_key"`
+	Provider         string    `json:"provider"`
+	Model            string    `json:"model"`
+	PromptTokens     int       `json:"prompt_tokens"`
+	CompletionTokens int       `json:"completion_tokens"`
+	TotalTokens      int       `json:"total_tokens"`
+	LatencyMs        int64     `json:"latency_ms"`
+	StatusCode       int       `json:"status_code"`
+	At               time.Time `json:"ts"`
+	Stream           bool      `json:"stream,omitempty"`
+}

collector/internal/collector/util.go

@@ -0,0 +1,11 @@
+package collector
+
+import "os"
+
+func Getenv(k, def string) string {
+	v := os.Getenv(k)
+	if v == "" {
+		return def
+	}
+	return v
+}