diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index 6141cb8b..e5539a94 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -31,7 +31,7 @@ RUN apt-get update \ && echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main" \ > /etc/apt/sources.list.d/github-cli.list \ && apt-get update \ - && apt-get -y install --no-install-recommends docker-ce-cli gh \ + && apt-get -y install --no-install-recommends docker-ce-cli docker-buildx-plugin gh \ && apt-get autoremove -y \ && apt-get clean -y \ && rm -rf /var/lib/apt/lists/* diff --git a/.env.example b/.env.example new file mode 100644 index 00000000..f381b2e6 --- /dev/null +++ b/.env.example @@ -0,0 +1,6 @@ +# Local-only environment overrides for the Taskfile (loaded via `dotenv`). +# Copy this file to `.env` and adjust the values. `.env` is gitignored, so it +# never gets committed. + +# Target image for `task docker-push-dev` (build + push to your own registry). +DEV_IMG=registry.example.com/gitops-reverser:latest diff --git a/.github/actions/docker-build/action.yml b/.github/actions/docker-build/action.yml index f93983c0..96676c0f 100644 --- a/.github/actions/docker-build/action.yml +++ b/.github/actions/docker-build/action.yml @@ -19,6 +19,9 @@ inputs: labels: description: 'Image labels' required: false + build-args: + description: 'Build args passed to the image build (newline-separated KEY=VALUE)' + required: false outputs: description: 'Build outputs (e.g., type=image,push=true or type=image,name=...,push-by-digest=true)' required: true @@ -51,6 +54,7 @@ runs: push: true tags: ${{ inputs.tags }} labels: ${{ inputs.labels }} + build-args: ${{ inputs.build-args }} cache-from: type=gha,scope=${{ inputs.cache-scope }} cache-to: type=gha,mode=max,scope=${{ inputs.cache-scope }} outputs: ${{ inputs.outputs }} \ No newline at end of file diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 99dae674..1f2735e4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -233,11 +233,13 @@ jobs: - name: Checkout code uses: actions/checkout@v6 - - name: Set full image name + - name: Set full image name and build metadata id: image run: | FULL_IMAGE="${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ env.IMAGE_TAG }}" echo "name=${FULL_IMAGE}" >> $GITHUB_OUTPUT + echo "git_commit=${GITHUB_SHA:0:7}" >> $GITHUB_OUTPUT + echo "build_date=$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> $GITHUB_OUTPUT echo "Building image: ${FULL_IMAGE}" - name: Build and push Docker image @@ -250,6 +252,11 @@ jobs: tags: ${{ steps.image.outputs.name }} outputs: type=image,push=true cache-scope: build-linux/amd64 + build-args: | + VERSION=${{ env.IMAGE_TAG }} + GIT_COMMIT=${{ steps.image.outputs.git_commit }} + GIT_DIRTY=0 + BUILD_DATE=${{ steps.image.outputs.build_date }} e2e: name: E2E (${{ matrix.name }}) @@ -352,6 +359,12 @@ jobs: type=semver,pattern={{major}},value=${{ needs.release-please.outputs.version }} type=raw,value=latest + - name: Compute build metadata + id: buildmeta + run: | + echo "git_commit=${GITHUB_SHA:0:7}" >> $GITHUB_OUTPUT + echo "build_date=$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> $GITHUB_OUTPUT + - name: Build and push Docker image by digest id: build uses: ./.github/actions/docker-build @@ -363,6 +376,11 @@ jobs: labels: ${{ steps.meta.outputs.labels }} outputs: type=image,name=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }},push-by-digest=true,name-canonical=true cache-scope: build-${{ matrix.platform }} + build-args: | + VERSION=${{ needs.release-please.outputs.version }} + GIT_COMMIT=${{ steps.buildmeta.outputs.git_commit }} + GIT_DIRTY=0 + BUILD_DATE=${{ steps.buildmeta.outputs.build_date }} - name: Export digest run: | diff --git a/.golangci.yml b/.golangci.yml index aed82e6a..13cb536f 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -346,6 +346,10 @@ linters: linters: [gochecknoglobals, gochecknoinits] - path: 'cmd/main\.go' linters: [gochecknoglobals, cyclop, gochecknoinits, gocognit, funlen] + # Build metadata vars in cmd/buildinfo.go are injected via -ldflags, which + # requires them to be package-level variables. + - path: 'cmd/buildinfo\.go' + linters: [gochecknoglobals] - path: 'internal/telemetry/.*\.go' linters: [gochecknoglobals] # Allow test suite global variables and patterns diff --git a/Dockerfile b/Dockerfile index 2c78836f..a02547f9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,6 +5,12 @@ FROM golang:1.26.3 AS builder ARG TARGETOS ARG TARGETARCH +# Build metadata, injected into the binary via -ldflags (see cmd/buildinfo.go). +ARG VERSION=dev +ARG GIT_COMMIT=unknown +ARG GIT_DIRTY=0 +ARG BUILD_DATE=unknown + WORKDIR /workspaces # Copy the Go Modules manifests @@ -19,7 +25,9 @@ COPY api/ api/ COPY internal/ internal/ # Build for the target platform -RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build -o manager cmd/main.go +RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build \ + -ldflags "-X main.version=${VERSION} -X main.gitCommit=${GIT_COMMIT} -X main.gitDirty=${GIT_DIRTY} -X main.buildDate=${BUILD_DATE}" \ + -o manager ./cmd FROM alpine:3.23 AS sops-downloader ARG TARGETARCH diff --git a/Taskfile-build.yml b/Taskfile-build.yml index e338e8ce..1957b374 100644 --- a/Taskfile-build.yml +++ b/Taskfile-build.yml @@ -145,7 +145,7 @@ tasks: - fmt - vet cmds: - - go build -o bin/manager cmd/main.go + - go build -o bin/manager ./cmd run: desc: Run a controller from your host @@ -154,12 +154,24 @@ tasks: - fmt - vet cmds: - - go run ./cmd/main.go + - go run ./cmd docker-build: desc: Build docker image with the manager cmds: - - '{{.CONTAINER_TOOL}} build -t {{.IMG}} .' + - | + GIT_COMMIT="$(git rev-parse --short HEAD 2>/dev/null || echo unknown)" + GIT_DIRTY=0 + [ -z "$(git status --porcelain 2>/dev/null)" ] || GIT_DIRTY=1 + BUILD_DATE="$(date -u +%Y-%m-%dT%H:%M:%SZ)" + VERSION="$(git describe --tags --always --dirty 2>/dev/null || echo dev)" + VERSION="${VERSION#gitops-reverser-v}" + {{.CONTAINER_TOOL}} build \ + --build-arg VERSION="${VERSION}" \ + --build-arg GIT_COMMIT="${GIT_COMMIT}" \ + --build-arg GIT_DIRTY="${GIT_DIRTY}" \ + --build-arg BUILD_DATE="${BUILD_DATE}" \ + -t {{.IMG}} . docker-push: desc: Push docker image with the manager @@ -172,14 +184,38 @@ tasks: PLATFORMS: '{{.PLATFORMS | default "linux/arm64,linux/amd64,linux/s390x,linux/ppc64le"}}' cmds: - | + GIT_COMMIT="$(git rev-parse --short HEAD 2>/dev/null || echo unknown)" + GIT_DIRTY=0 + [ -z "$(git status --porcelain 2>/dev/null)" ] || GIT_DIRTY=1 + BUILD_DATE="$(date -u +%Y-%m-%dT%H:%M:%SZ)" + VERSION="$(git describe --tags --always --dirty 2>/dev/null || echo dev)" + VERSION="${VERSION#gitops-reverser-v}" sed -e '1 s/\(^FROM\)/FROM --platform=\${BUILDPLATFORM}/; t' \ -e '1,// s//FROM --platform=\${BUILDPLATFORM}/' Dockerfile > Dockerfile.cross {{.CONTAINER_TOOL}} buildx create --name gitops-reverser-builder || true {{.CONTAINER_TOOL}} buildx use gitops-reverser-builder - {{.CONTAINER_TOOL}} buildx build --push --platform={{.PLATFORMS}} --tag {{.IMG}} -f Dockerfile.cross . + {{.CONTAINER_TOOL}} buildx build --push --platform={{.PLATFORMS}} \ + --build-arg VERSION="${VERSION}" \ + --build-arg GIT_COMMIT="${GIT_COMMIT}" \ + --build-arg GIT_DIRTY="${GIT_DIRTY}" \ + --build-arg BUILD_DATE="${BUILD_DATE}" \ + --tag {{.IMG}} -f Dockerfile.cross . {{.CONTAINER_TOOL}} buildx rm gitops-reverser-builder || true rm -f Dockerfile.cross + docker-push-dev: + desc: Build and push the image to your dev registry (DEV_IMG, set in .env) + requires: + vars: + - DEV_IMG + cmds: + - task: docker-build + vars: + IMG: '{{.DEV_IMG}}' + - task: docker-push + vars: + IMG: '{{.DEV_IMG}}' + dist-install: desc: Generate consolidated YAML from the Helm chart deps: diff --git a/Taskfile.yml b/Taskfile.yml index d2ed0ea0..4c73c303 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -1,6 +1,10 @@ # yaml-language-server: $schema=https://taskfile.dev/schema.json version: '3' +# Local-only overrides (e.g. DEV_IMG) are loaded from .env, which is gitignored. +# See .env.example for the available keys. +dotenv: ['.env'] + set: - errexit - nounset diff --git a/charts/gitops-reverser/README.md b/charts/gitops-reverser/README.md index 62a4caf1..6e2c6279 100644 --- a/charts/gitops-reverser/README.md +++ b/charts/gitops-reverser/README.md @@ -178,6 +178,9 @@ nodeSelector: | `queue.redis.stream` | Redis stream name for audit events | `gitopsreverser.audit.events.v1` | | `queue.redis.maxLen` | Approximate stream max length (`0` disables trim) | `0` | | `queue.redis.tls.enabled` | Enable TLS for Redis connection | `false` | +| `webhook.audit.debugStream.enabled` | Append every decoded audit event to the early Redis debug stream | `false` | +| `webhook.audit.debugStream.stream` | Redis stream name for early decoded audit event debugging | `gitopsreverser.audit.debug.events.v1` | +| `webhook.audit.debugStream.maxLen` | Approximate early debug stream max length (`0` disables trim) | `0` | | `auditEventJoin.bodyTTL` | TTL for parked additional audit bodies waiting for the matching official event | `5m` | | `auditEventJoin.decisionTTL` | TTL for audit decision dedupe keys | `1h` | | `auditEventJoin.bodyWait` | Grace period for a bodyless official audit event to wait for a matching additional body while preserving official event order | `500ms` | diff --git a/charts/gitops-reverser/templates/deployment.yaml b/charts/gitops-reverser/templates/deployment.yaml index fa3c3160..b34288e9 100644 --- a/charts/gitops-reverser/templates/deployment.yaml +++ b/charts/gitops-reverser/templates/deployment.yaml @@ -72,6 +72,10 @@ spec: - --audit-redis-db={{ .Values.queue.redis.db }} - --audit-redis-stream={{ .Values.queue.redis.stream }} - --audit-redis-max-len={{ .Values.queue.redis.maxLen }} + {{- if .Values.webhook.audit.debugStream.enabled }} + - --audit-debug-redis-stream={{ .Values.webhook.audit.debugStream.stream }} + - --audit-debug-redis-max-len={{ .Values.webhook.audit.debugStream.maxLen }} + {{- end }} {{- if .Values.queue.redis.tls.enabled }} - --audit-redis-tls {{- end }} @@ -93,9 +97,6 @@ spec: {{- if .Values.logging.stacktraceLevel }} - --zap-stacktrace-level={{ .Values.logging.stacktraceLevel }} {{- end }} - {{- if .Values.webhook.audit.debugDumps }} - - --audit-dump-path=/var/run/audit-dumps - {{- end }} ports: - name: audit containerPort: {{ .Values.servers.audit.port }} @@ -137,10 +138,6 @@ spec: resources: {{- toYaml .Values.resources | nindent 12 }} volumeMounts: - {{- if .Values.webhook.audit.debugDumps }} - - name: audit-dumps - mountPath: /var/run/audit-dumps - {{- end }} - name: tmp-dir mountPath: /tmp {{- if .Values.servers.metrics.tls.enabled }} @@ -160,10 +157,6 @@ spec: {{- toYaml . | nindent 12 }} {{- end }} volumes: - {{- if .Values.webhook.audit.debugDumps }} - - name: audit-dumps - emptyDir: {} - {{- end }} - name: tmp-dir emptyDir: {} {{- if .Values.servers.metrics.tls.enabled }} diff --git a/charts/gitops-reverser/values.yaml b/charts/gitops-reverser/values.yaml index ddc9d65c..e961d6ee 100644 --- a/charts/gitops-reverser/values.yaml +++ b/charts/gitops-reverser/values.yaml @@ -97,8 +97,13 @@ servers: # Webhook behavior webhook: audit: - # Set to true if you want to write events to /var/run/audit-dumps - debugDumps: false + # Set to true to append every decoded audit event to a separate Redis stream + # before normal audit processing can filter, join, or drop it. + debugStream: + enabled: false + stream: "gitopsreverser.audit.debug.events.v1" + # Approximate max stream length; 0 means no trimming. + maxLen: 0 # Durable audit queue configuration. queue: diff --git a/cmd/buildinfo.go b/cmd/buildinfo.go new file mode 100644 index 00000000..b3068024 --- /dev/null +++ b/cmd/buildinfo.go @@ -0,0 +1,77 @@ +// SPDX-License-Identifier: Apache-2.0 +// +// Copyright 2025 ConfigButler +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package main + +import ( + "encoding/json" + "net/http" + "runtime" +) + +// Build information, injected via -ldflags "-X main.=" at build +// time (see Dockerfile and the docker-build task). The defaults below are what +// a plain `go build`/`go run` produces, which is useful to spot a non-release +// binary at a glance. +var ( + version = "dev" + gitCommit = "unknown" + gitDirty = "0" + buildDate = "unknown" +) + +// buildInfo is the build metadata reported at startup and by the /build-info +// endpoint. It lets an operator confirm a running pod is the build they expect. +type buildInfo struct { + Version string `json:"version"` + GitCommit string `json:"gitCommit"` + IsDirty bool `json:"isDirty"` + CommitWithDirty string `json:"commitWithDirty"` + BuildDate string `json:"buildDate"` + GoVersion string `json:"goVersion"` +} + +// currentBuildInfo assembles the build metadata from the ldflags-injected vars. +func currentBuildInfo() buildInfo { + dirty := gitDirty == "1" + commitWithDirty := gitCommit + if dirty { + commitWithDirty = gitCommit + "-dirty" + } + return buildInfo{ + Version: version, + GitCommit: gitCommit, + IsDirty: dirty, + CommitWithDirty: commitWithDirty, + BuildDate: buildDate, + GoVersion: runtime.Version(), + } +} + +// buildInfoHandler serves the build metadata as JSON on GET requests. It is +// registered as an extra handler on the metrics server (see main). +func buildInfoHandler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(currentBuildInfo()) + }) +} diff --git a/cmd/buildinfo_test.go b/cmd/buildinfo_test.go new file mode 100644 index 00000000..dd961455 --- /dev/null +++ b/cmd/buildinfo_test.go @@ -0,0 +1,83 @@ +// SPDX-License-Identifier: Apache-2.0 +// +// Copyright 2025 ConfigButler +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "runtime" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// setBuildVars overrides the ldflags-injected build vars for a test and returns +// a function that restores their original values. +func setBuildVars(v, commit, dirty, date string) func() { + origV, origC, origD, origDate := version, gitCommit, gitDirty, buildDate + version, gitCommit, gitDirty, buildDate = v, commit, dirty, date + return func() { + version, gitCommit, gitDirty, buildDate = origV, origC, origD, origDate + } +} + +func TestCurrentBuildInfo_Clean(t *testing.T) { + defer setBuildVars("1.2.3", "abc123", "0", "2026-05-22T00:00:00Z")() + + bi := currentBuildInfo() + assert.Equal(t, "1.2.3", bi.Version) + assert.Equal(t, "abc123", bi.GitCommit) + assert.False(t, bi.IsDirty) + assert.Equal(t, "abc123", bi.CommitWithDirty) + assert.Equal(t, "2026-05-22T00:00:00Z", bi.BuildDate) + assert.Equal(t, runtime.Version(), bi.GoVersion) +} + +func TestCurrentBuildInfo_Dirty(t *testing.T) { + defer setBuildVars("dev", "abc123", "1", "unknown")() + + bi := currentBuildInfo() + assert.True(t, bi.IsDirty) + assert.Equal(t, "abc123-dirty", bi.CommitWithDirty) +} + +func TestBuildInfoHandler_Get(t *testing.T) { + defer setBuildVars("9.9.9", "deadbeef", "0", "2026-01-01T00:00:00Z")() + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/build-info", nil) + buildInfoHandler().ServeHTTP(rec, req) + + require.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, "application/json; charset=utf-8", rec.Header().Get("Content-Type")) + + var got buildInfo + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &got)) + assert.Equal(t, "9.9.9", got.Version) + assert.Equal(t, "deadbeef", got.CommitWithDirty) + assert.False(t, got.IsDirty) +} + +func TestBuildInfoHandler_RejectsNonGet(t *testing.T) { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/build-info", nil) + buildInfoHandler().ServeHTTP(rec, req) + + assert.Equal(t, http.StatusMethodNotAllowed, rec.Code) +} diff --git a/cmd/main.go b/cmd/main.go index 4a01a971..19c1001c 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -92,6 +92,13 @@ func main() { cfg := parseFlags() ctrl.SetLogger(zap.New(zap.UseFlagOptions(&cfg.zapOpts))) + bi := currentBuildInfo() + setupLog.Info("Starting gitops-reverser", + "version", bi.Version, + "gitCommit", bi.CommitWithDirty, + "buildDate", bi.BuildDate, + "goVersion", bi.GoVersion) + if cfg.auditRedisPassword == "" { setupLog.Info( "no Redis password configured — "+ @@ -125,6 +132,11 @@ func main() { // Manager mgr := newManager(metricsServerOptions, cfg.probeAddr) + // Expose build metadata on the metrics server so an operator can confirm a + // running pod is the build they expect (also logged at startup above). + fatalIfErr(mgr.AddMetricsServerExtraHandler("/build-info", buildInfoHandler()), + "unable to register build-info endpoint") + // Initialize rule store for watch rules ruleStore := rulestore.NewStore() @@ -194,6 +206,20 @@ func main() { }) fatalIfErr(err, "unable to initialize audit redis queue") + var auditDebugQueue webhookhandler.AuditDebugEventQueue + if cfg.auditDebugRedisStream != "" { + auditDebugQueue, err = queue.NewRedisAuditDebugQueue(queue.RedisAuditQueueConfig{ + Addr: cfg.auditRedisAddr, + Username: cfg.auditRedisUsername, + AuthValue: cfg.auditRedisPassword, + DB: cfg.auditRedisDB, + Stream: cfg.auditDebugRedisStream, + MaxLen: cfg.auditDebugRedisMaxLen, + TLSEnabled: cfg.auditRedisTLS, + }) + fatalIfErr(err, "unable to initialize audit debug redis queue") + } + auditJoiner, err := webhookhandler.NewRedisAuditEventJoiner(webhookhandler.RedisAuditJoinerConfig{ Addr: cfg.auditRedisAddr, Username: cfg.auditRedisUsername, @@ -208,6 +234,7 @@ func main() { setupLog.Info("Audit pipeline configured", "redisAddress", cfg.auditRedisAddr, "stream", cfg.auditRedisStream, + "debugStream", cfg.auditDebugRedisStream, "db", cfg.auditRedisDB, "tlsEnabled", cfg.auditRedisTLS, "bodyTTL", cfg.auditEventBodyTTL, @@ -242,9 +269,9 @@ func main() { setupLog.Info("Audit stream consumer registered", "consumerID", consumerID) auditHandler, err := webhookhandler.NewAuditHandler(webhookhandler.AuditHandlerConfig{ - DumpDir: cfg.auditDumpPath, MaxRequestBodyBytes: cfg.auditMaxRequestBodyBytes, Queue: auditQueue, + DebugQueue: auditDebugQueue, Joiner: auditJoiner, }) fatalIfErr(err, "unable to create audit handler") @@ -256,10 +283,6 @@ func main() { auditCertWatcher = watcher fatalIfErr(mgr.Add(auditRunnable), "unable to add audit ingress server runnable") - if cfg.auditDumpPath != "" { - setupLog.Info("Audit ingress dump enabled", "dumpPath", cfg.auditDumpPath) - } - // Setup watch manager (must be after controllers are set up) fatalIfErr(watchMgr.SetupWithManager(mgr), "unable to setup watch ingestion manager") fatalIfErr(mgr.Add(watchMgr), "unable to add watch ingestion manager") @@ -309,7 +332,6 @@ type appConfig struct { probeAddr string metricsInsecure bool enableHTTP2 bool - auditDumpPath string auditListenAddress string auditPort int auditCertPath string @@ -328,6 +350,8 @@ type appConfig struct { auditRedisDB int auditRedisStream string auditRedisMaxLen int64 + auditDebugRedisStream string + auditDebugRedisMaxLen int64 auditRedisTLS bool auditEventBodyTTL time.Duration auditEventDecisionTTL time.Duration @@ -365,8 +389,6 @@ func parseFlagsWithArgs(fs *flag.FlagSet, args []string) (appConfig, error) { ) fs.BoolVar(&cfg.enableHTTP2, "enable-http2", false, "If set, HTTP/2 will be enabled for the metrics server and audit ingress server") - fs.StringVar(&cfg.auditDumpPath, "audit-dump-path", "", - "Directory to write audit events for debugging. If empty, audit event file dumping is disabled.") fs.StringVar(&cfg.auditListenAddress, "audit-listen-address", "0.0.0.0", "IP address for the dedicated audit ingress HTTPS server.") fs.IntVar(&cfg.auditPort, "audit-port", defaultAuditPort, "Port for the dedicated audit ingress HTTPS server.") @@ -401,6 +423,10 @@ func parseFlagsWithArgs(fs *flag.FlagSet, args []string) (appConfig, error) { "Redis stream name for audit event queueing.") fs.Int64Var(&cfg.auditRedisMaxLen, "audit-redis-max-len", 0, "Approximate max stream length (0 disables trimming).") + fs.StringVar(&cfg.auditDebugRedisStream, "audit-debug-redis-stream", "", + "Optional Redis stream name for every decoded audit event before normal audit processing.") + fs.Int64Var(&cfg.auditDebugRedisMaxLen, "audit-debug-redis-max-len", 0, + "Approximate max debug stream length (0 disables trimming).") fs.BoolVar(&cfg.auditRedisTLS, "audit-redis-tls", false, "If set, Redis connection for audit queueing uses TLS.") fs.DurationVar(&cfg.auditEventBodyTTL, "audit-event-body-ttl", defaultAuditEventBodyTTL, @@ -497,6 +523,9 @@ func validateAuditConfig(cfg appConfig) error { if cfg.auditRedisMaxLen < 0 { return fmt.Errorf("audit-redis-max-len must be >= 0, got %d", cfg.auditRedisMaxLen) } + if cfg.auditDebugRedisMaxLen < 0 { + return fmt.Errorf("audit-debug-redis-max-len must be >= 0, got %d", cfg.auditDebugRedisMaxLen) + } if cfg.auditEventBodyTTL <= 0 { return fmt.Errorf("audit-event-body-ttl must be > 0, got %s", cfg.auditEventBodyTTL) } diff --git a/cmd/main_audit_server_test.go b/cmd/main_audit_server_test.go index 5b390e28..2877fea5 100644 --- a/cmd/main_audit_server_test.go +++ b/cmd/main_audit_server_test.go @@ -56,6 +56,8 @@ func TestParseFlagsWithArgs_Defaults(t *testing.T) { assert.Equal(t, "valkey:6379", cfg.auditRedisAddr) assert.Equal(t, "gitopsreverser.audit.events.v1", cfg.auditRedisStream) assert.Equal(t, int64(0), cfg.auditRedisMaxLen) + assert.Empty(t, cfg.auditDebugRedisStream) + assert.Equal(t, int64(0), cfg.auditDebugRedisMaxLen) assert.False(t, cfg.auditRedisTLS) assert.Equal(t, 5*time.Minute, cfg.auditEventBodyTTL) assert.Equal(t, time.Hour, cfg.auditEventDecisionTTL) @@ -95,6 +97,8 @@ func TestParseFlagsWithArgs_CustomAuditValues(t *testing.T) { "--audit-redis-db=2", "--audit-redis-stream=gitopsreverser.audit.custom", "--audit-redis-max-len=1000", + "--audit-debug-redis-stream=gitopsreverser.audit.debug.custom", + "--audit-debug-redis-max-len=2000", "--audit-redis-tls", "--audit-event-body-ttl=2m", "--audit-event-decision-ttl=30m", @@ -121,6 +125,8 @@ func TestParseFlagsWithArgs_CustomAuditValues(t *testing.T) { assert.Equal(t, 2, cfg.auditRedisDB) assert.Equal(t, "gitopsreverser.audit.custom", cfg.auditRedisStream) assert.Equal(t, int64(1000), cfg.auditRedisMaxLen) + assert.Equal(t, "gitopsreverser.audit.debug.custom", cfg.auditDebugRedisStream) + assert.Equal(t, int64(2000), cfg.auditDebugRedisMaxLen) assert.True(t, cfg.auditRedisTLS) assert.Equal(t, 2*time.Minute, cfg.auditEventBodyTTL) assert.Equal(t, 30*time.Minute, cfg.auditEventDecisionTTL) @@ -175,6 +181,10 @@ func TestParseFlagsWithArgs_InvalidAuditSettings(t *testing.T) { name: "invalid redis max len", args: []string{"--audit-redis-max-len=-1"}, }, + { + name: "invalid audit debug redis max len", + args: []string{"--audit-debug-redis-max-len=-1"}, + }, { name: "invalid audit event body ttl", args: []string{"--audit-event-body-ttl=0s"}, diff --git a/docs/architecture.md b/docs/architecture.md index 1f105f5e..cbb5e34d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -233,7 +233,7 @@ of namespace. All namespace-level filtering must happen inside the operator. 1. **Kubernetes API Server** sends audit events via webhook POST to `/audit-webhook` 2. Supplementary audit sources can send matching `EventList` payloads to `/audit-webhook-additional` -3. [AuditHandler](internal/webhook/audit_handler.go) deserializes each `EventList`, lets the audit joiner park or merge body contributions by `auditID`, and calls `RedisAuditQueue.Enqueue()` for canonical events +3. [AuditHandler](internal/webhook/audit_handler.go) deserializes each `EventList`, optionally appends every decoded event to a separate early debug Redis stream, lets the audit joiner park or merge body contributions by `auditID`, and calls `RedisAuditQueue.Enqueue()` for canonical events 4. [RedisAuditQueue](internal/queue/redis_audit_queue.go) writes to Redis stream `gitopsreverser.audit.events.v1` via `XADD` 5. [AuditConsumer](internal/queue/redis_audit_consumer.go) reads batches of 50 via `XREADGROUP` (consumer group for HA-readiness) 6. For each message: filter to `ResponseComplete` stage + mutating verbs only, then call [RuleStore](internal/rulestore/store.go) to find matching rules @@ -369,6 +369,12 @@ flowchart LR Today there is a single Redis stream for all audit events. The consumer is leader-elected, meaning only one pod processes events at a time. +When `--audit-debug-redis-stream` is set, the handler also appends each successfully decoded +audit event to that stream immediately after `EventList` decode. This tap runs before per-event +validation, stage/verb filtering, shallow-body joins, deduplication, and canonical enqueueing. +It uses the canonical stream fields plus `source` (`official` or `additional`) and retains the raw +event as `payload_json`. The stream is diagnostic only; no consumer routes it to Git. + **Desired future**: each GitTarget (or at minimum each Git destination) gets its own Redis queue. This way a GitHub outage only stalls commits for the affected destination — events keep accumulating in Redis and are committed once the remote is reachable again. Other Git destinations @@ -658,6 +664,11 @@ allowed to claim the dedupe key, the later `ResponseComplete` for the same audit silently dropped as a duplicate. The handler filters by stage at the boundary before the joiner sees the event. +Audit events with a non-empty `objectRef.subresource` are rejected before the joiner too. +Current WatchRule planning mirrors top-level resource state only. Subresource requests can +carry mutating-looking audit verbs without describing a Git-writable object; for example a +successful `pods/exec` stream is audited as `verb=create` with no request/response resource body. + ### Quality classification The classifier ([audit_joiner.go:507](internal/webhook/audit_joiner.go#L507)) decides what to do @@ -782,7 +793,7 @@ aggregate to group/version without `label_replace`. The event-action label is `v | `gitopsreverser_audit_eventlists_total` | `source`, `outcome` | EventList request-boundary counter: `processed`, `empty`, `decode_error`, `process_error` | | `gitopsreverser_audit_eventlist_events_total` | `source`, `outcome` | Decoded audit event items delivered in EventLists (no `decode_error` sample) | | `gitopsreverser_audit_eventlist_duration_seconds` | `source`, `outcome` | Histogram of webhook request time, including in-pod join wait work | -| `gitopsreverser_audit_events_received_total` | `source`, `group`, `version`, `resource`, `verb` | Receive counter (username is logged, not labelled) | +| `gitopsreverser_audit_events_received_total` | `source`, `group`, `version`, `resource`, `subresource`, `verb` | Receive counter (username is logged, not labelled). `subresource` is empty for top-level resources and a bounded value (`status`, `exec`, …) otherwise | | `gitopsreverser_audit_event_quality_total` | `source`, `quality`, `group`, `version`, `resource`, `verb` | First-class shape classification | | `gitopsreverser_audit_join_parked_total` | — | Parked additional bodies | | `gitopsreverser_audit_join_emitted_total` | `source`, `result` | Canonical emissions: `as_is`, `merged` | @@ -791,7 +802,7 @@ aggregate to group/version without `label_replace`. The event-action label is `v | `gitopsreverser_audit_join_body_late_total` | `group`, `version`, `resource`, `verb` | Additional body arrived after the decision was committed | | `gitopsreverser_audit_join_skew_seconds` | `arrival`, `outcome` | Histogram of official↔additional arrival skew: `arrival=body_first` is the proxy's lead time, `arrival=official_first` is how long the official waited on the canonical gate | | `gitopsreverser_audit_official_gate_wait_seconds` | — | Histogram of how long an official event waited to acquire the in-pod canonical gate (backpressure signal) | -| `gitopsreverser_audit_pipeline_events_total` | `group`, `version`, `resource`, `verb`, `outcome` | Consumer-stage counter: `unmatched`, `dropped_no_body`, `routed`, `route_failed` | +| `gitopsreverser_audit_pipeline_events_total` | `group`, `version`, `resource`, `verb`, `outcome` | Consumer-stage counter: `unmatched`, `dropped_no_body`, `dropped_partial_object`, `routed`, `route_failed` | | `gitopsreverser_audit_pipeline_route_targets_total` | `git_target_namespace`, `git_target`, `rule_kind`, `outcome` | Per-GitTarget route attempts: `routed`, `route_failed` | Useful alerts: `audit_shallow_dropped_total` non-zero (operator misconfiguration — install diff --git a/docs/design/partial-object-audit-event-handling.md b/docs/design/partial-object-audit-event-handling.md new file mode 100644 index 00000000..2f3bb880 --- /dev/null +++ b/docs/design/partial-object-audit-event-handling.md @@ -0,0 +1,366 @@ +# Partial-object audit events: classify the finalizer-patch fragment as a benign drop + +> Status: implemented. The end-to-end miniredis metric test below was +> intentionally skipped — the `handleExtractObjectError` unit test already +> covers the `dropped_partial_object` outcome mapping. +> Date: 2026-05-22 +> Observed on: **CozyStack** prod (`gitops-reverser-v0.26.0`), deleting `mongodb-simon2`. +> Related: [Shallow audit events](./shallow-audit-event-misclassification.md), +> [Audit Ingestion Pipeline](../architecture.md#audit-ingestion-pipeline). + +## Summary + +The audit consumer logs an `error`-level "poison-pill" line for an audit event +whose object body is a **finalizer-removal patch fragment** rather than a full +Kubernetes object: + +``` +{"level":"error","logger":"audit-consumer.audit-consumer", + "msg":"Failed to route audit event; ACKing to avoid poison-pill", + "msgID":"1779479358795-0", + "error":"extracting object for tenant-root/mongodb-simon2: failed to unmarshal + object JSON: Object 'Kind' is missing in '{\"metadata\":{\"finalizers\":null}}'"} +``` + +The body is `{"metadata":{"finalizers":null}}` — the merge-patch a controller +sends to drop the last finalizer while a resource is being deleted. It is valid +JSON but carries no `apiVersion`/`kind`, so +[`(*Unstructured).UnmarshalJSON`](https://pkg.go.dev/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured#Unstructured.UnmarshalJSON) +rejects it with `Object 'Kind' is missing`. + +The **outcome is already correct** — the event is ACK'd and dropped, no +poison-pill loop, no crash, and the resource's real `[DELETE]` commit lands +through its own delete audit event (in the captured run, the +`[DELETE] helm.toolkit.fluxcd.io/v2/helmreleases/mongodb-simon2` commit at +`19:49:02`, ~16s before this line). The **classification is wrong**: a routine, +expected event shape is funnelled through the generic error path instead of the +benign-drop path that `errAuditEventObjectMissing` and `errAuditEventObjectIsStatus` +already have. The cost is alert-grade log noise and no way to measure how often +it happens. + +This design adds a third recognised body shape — **partial object** — alongside +"missing body" and "Status error body", with its own benign-drop path, log +gating, and metric outcome. + +## Root cause + +[`extractObject`](../../internal/queue/redis_audit_consumer.go#L708) selects a +raw body and unmarshals it: + +```go +obj := &unstructured.Unstructured{} +if err := obj.UnmarshalJSON(raw); err != nil { + return nil, fmt.Errorf("failed to unmarshal object JSON: %w", err) // <-- generic +} +``` + +`(*Unstructured).UnmarshalJSON` requires a non-empty `kind`. A merge-patch body +has none, so the call fails. The returned error is a plain wrapped error — it +matches neither `errAuditEventObjectMissing` nor `errAuditEventObjectIsStatus`, +so [`handleExtractObjectError`](../../internal/queue/redis_audit_consumer.go#L459) +falls through its `switch` to `return false`, and +[`routeAuditEvent`](../../internal/queue/redis_audit_consumer.go#L408-L417) +surfaces it. [`processMessage`](../../internal/queue/redis_audit_consumer.go#L365) +then logs it at `error`. + +### Why the body is a fragment, not the full object + +[`selectAuditObjectRaw`](../../internal/queue/redis_audit_consumer.go#L754) +prefers `responseObject` for non-delete verbs and only falls back to +`requestObject`. For a `PATCH`, `requestObject` *is* the patch document — a +fragment by definition. The full merged object normally arrives as +`responseObject`, but when the patch removes the **last finalizer** of an object +that already has a `deletionTimestamp`, the object is deleted as part of that +same request and the apiserver records no resource `responseObject`. The +fallback then picks the only body present — the patch fragment. + +So this is **not** a malformed-event anomaly. It is the normal audit shape of +the finalizer-removal step of any finalizer-driven deletion, and CozyStack's +operator-managed resources (`mongodbs` → `helmreleases`) hit it on every delete. + +### How it differs from the shapes we already handle + +| Body shape | Example | Sentinel | Recognised today | +| --- | --- | --- | --- | +| No body at all | `requestObject`/`responseObject` both empty | `errAuditEventObjectMissing` | yes | +| `metav1.Status` error body | `{"apiVersion":"v1","kind":"Status",...}` | `errAuditEventObjectIsStatus` | yes | +| **Partial object** | `{"metadata":{"finalizers":null}}` | **`errAuditEventObjectPartial`** (new) | **no — falls to error path** | + +A partial object is **valid JSON with no `kind`**. That is precisely the +condition under which `UnmarshalJSON` fails while the bytes are still +well-formed — which makes it cheap and unambiguous to detect. + +## Design + +### 1. New sentinel error + +In [redis_audit_consumer.go](../../internal/queue/redis_audit_consumer.go), next +to the existing sentinels: + +```go +// errAuditEventObjectPartial marks an audit event whose body is valid JSON but +// lacks the apiVersion/kind identity of a full Kubernetes object — typically a +// merge-patch fragment such as {"metadata":{"finalizers":null}} recorded as the +// requestObject of a finalizer-removal PATCH. It carries no routable resource +// state, so it is dropped before git routing rather than treated as a decode +// failure. The resource's real mutation is mirrored from its own (delete or +// full-body) audit event. +var errAuditEventObjectPartial = errors.New( + "audit event object body is a partial object (no kind)") +``` + +### 2. Detect it in `extractObject` + +Distinguish "valid JSON, no kind" (partial) from "not JSON at all" (a genuine +decode failure worth an error) when `UnmarshalJSON` fails: + +```go +obj := &unstructured.Unstructured{} +if err := obj.UnmarshalJSON(raw); err != nil { + if isPartialObjectBody(raw) { + return nil, errAuditEventObjectPartial + } + return nil, fmt.Errorf("failed to unmarshal object JSON: %w", err) +} + +// isPartialObjectBody reports whether raw is well-formed JSON describing an +// object that lacks a "kind" — the condition that makes +// (*Unstructured).UnmarshalJSON fail on an otherwise valid body. A merge-patch +// fragment (e.g. {"metadata":{"finalizers":null}}) matches; malformed bytes do +// not, so a real decode failure still surfaces as an error. +func isPartialObjectBody(raw []byte) bool { + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + return false + } + kind, _ := m["kind"].(string) + return kind == "" +} +``` + +This adds no API discovery, no catalog lookup — it is a pure function of the +bytes, consistent with the recognition-rule principle in the +[shallow-event design](./shallow-audit-event-misclassification.md#the-recognition-rule). + +### 3. Benign-drop path with its own outcome + +[`handleExtractObjectError`](../../internal/queue/redis_audit_consumer.go#L459) +currently returns a single `bool` and the caller hard-codes +`pipelineOutcomeDroppedNoBody`. To measure the new case distinctly, change the +signature to return the metric outcome it classified: + +```go +func (c *AuditConsumer) handleExtractObjectError( + log logr.Logger, auditEvent auditv1.Event, err error, gvr, namespace, name string, +) (outcome string, handled bool) +``` + +- `errAuditEventObjectMissing` → `(pipelineOutcomeDroppedNoBody, true)` — unchanged. +- `errAuditEventObjectIsStatus` → `(pipelineOutcomeDroppedNoBody, true)` — unchanged. +- `errAuditEventObjectPartial` → `(pipelineOutcomeDroppedPartialObject, true)` — new. +- default → `("", false)`. + +New `case` in the `switch`, mirroring the existing two — first occurrence at +`Info` with actionable text, the rest at `V(1)`, gated by a new +`firstPartialDropped sync.Once`: + +```go +case errors.Is(err, errAuditEventObjectPartial): + c.firstPartialDropped.Do(func() { + c.log.Info( + "First audit event dropped before git routing — body is a partial "+ + "object (no kind), typically a finalizer-removal PATCH fragment. "+ + "The resource's real change is mirrored from its own audit event; "+ + "this fragment is not routable. Further drops will log at V(1) only.", + "auditID", auditEvent.AuditID, "gvr", gvr, "verb", auditEvent.Verb) + }) + log.V(1).Info( + "audit event dropped before git routing: partial object body (no kind)", + "auditID", auditEvent.AuditID, "gvr", gvr, "verb", auditEvent.Verb, + "namespace", namespace, "name", name) + return pipelineOutcomeDroppedPartialObject, true +``` + +The caller in `routeAuditEvent`: + +```go +sanitized, err := extractObject(auditEvent, op, fullAPIVersion, ref.Resource, namespace, name) +if err != nil { + if outcome, handled := c.handleExtractObjectError( + log, auditEvent, err, fullAPIVersion+"/"+ref.Resource, namespace, name, + ); handled { + recordPipelineEvent(ctx, gvr, auditEvent.Verb, outcome) + return nil + } + return fmt.Errorf("extracting object for %s/%s: %w", namespace, name, err) +} +``` + +The generic `error`-level "poison-pill" log is now reached only by a *genuine* +decode failure (not valid JSON) — which is the anomaly that line was written +for. + +## Metric + +Add one outcome value to the existing +**`gitopsreverser_audit_pipeline_events_total`** counter — no new metric, no new +label, so existing dashboards and `InitTestExporter` wiring are untouched: + +```go +const ( + pipelineOutcomeUnmatched = "unmatched" + pipelineOutcomeDroppedNoBody = "dropped_no_body" + pipelineOutcomeDroppedPartialObject = "dropped_partial_object" // new + pipelineOutcomeRouted = "routed" + pipelineOutcomeRouteFailed = "route_failed" +) +``` + +A sample is emitted as +`gitopsreverser_audit_pipeline_events_total{group=…,version=…,resource=…,verb=…,outcome="dropped_partial_object"}`. + +Operational use: +- **Expected, low and flat** — one count per finalizer-driven delete. A baseline + rate that tracks deletions of finalizer-bearing resources is healthy. +- **Alert on sustained growth on non-`delete`/`patch` verbs** — a + `dropped_partial_object` for a `create`/`update` would mean a full-body event + is being lost upstream, which is a real gap, not a benign finalizer fragment. + Suggested expression: + `sum by (resource, verb) (rate(gitopsreverser_audit_pipeline_events_total{outcome="dropped_partial_object", verb!~"delete|patch"}[15m])) > 0`. + +The [architecture.md metrics table](../architecture.md#metrics) row for +`gitopsreverser_audit_pipeline_events_total` is updated to list the new outcome: +`unmatched`, `dropped_no_body`, `dropped_partial_object`, `routed`, +`route_failed`. + +## Tests + +All in package `queue`. Two unit tests on `extractObject`, one decision test on +`handleExtractObjectError`, one end-to-end metric test. + +### Unit — `extractObject` classifies the fragment + +In [redis_audit_consumer_test.go](../../internal/queue/redis_audit_consumer_test.go), +beside `TestExtractObject_RejectsStatusErrorBody`: + +```go +// TestExtractObject_ClassifiesPartialFinalizerPatch reproduces the CozyStack +// prod occurrence: deleting mongodb-simon2 produced an audit event whose only +// body was the finalizer-removal patch fragment {"metadata":{"finalizers":null}}. +// extractObject must classify it as a partial object, not a decode failure. +func TestExtractObject_ClassifiesPartialFinalizerPatch(t *testing.T) { + ev := auditv1.Event{ + Verb: "patch", + RequestObject: &runtime.Unknown{Raw: []byte(`{"metadata":{"finalizers":null}}`)}, + // ResponseObject deliberately nil: the object was deleted by this same + // PATCH (last finalizer removed), so the apiserver recorded no body. + } + + _, err := extractObject( + ev, configv1alpha1.OperationUpdate, + "helm.toolkit.fluxcd.io/v2", "helmreleases", "tenant-root", "mongodb-simon2", + ) + require.ErrorIs(t, err, errAuditEventObjectPartial) +} + +// TestExtractObject_MalformedBodyStillErrors guards the boundary: bytes that are +// not valid JSON are a genuine decode failure and must NOT be reclassified as a +// benign partial object — they still deserve the error-level poison-pill log. +func TestExtractObject_MalformedBodyStillErrors(t *testing.T) { + ev := auditv1.Event{ + ResponseObject: &runtime.Unknown{Raw: []byte(`{"metadata":`)}, // truncated + } + + _, err := extractObject(ev, configv1alpha1.OperationCreate, "v1", "ConfigMap", "default", "cm") + require.Error(t, err) + require.NotErrorIs(t, err, errAuditEventObjectPartial) + require.NotErrorIs(t, err, errAuditEventObjectMissing) + require.NotErrorIs(t, err, errAuditEventObjectIsStatus) +} +``` + +### Unit — `handleExtractObjectError` maps it to the benign outcome + +```go +// TestHandleExtractObjectError_PartialObjectIsBenign confirms a partial-object +// error is handled (ACK, no poison-pill) and reported under the +// dropped_partial_object metric outcome. +func TestHandleExtractObjectError_PartialObjectIsBenign(t *testing.T) { + c := &AuditConsumer{log: logr.Discard()} + outcome, handled := c.handleExtractObjectError( + logr.Discard(), auditv1.Event{Verb: "patch"}, + errAuditEventObjectPartial, "helm.toolkit.fluxcd.io/v2/helmreleases", + "tenant-root", "mongodb-simon2", + ) + assert.True(t, handled) + assert.Equal(t, pipelineOutcomeDroppedPartialObject, outcome) +} +``` + +### End-to-end — the metric is emitted, the router is not called + +> Not implemented — intentionally skipped to keep the change small. The +> `handleExtractObjectError` unit test above already pins the +> `dropped_partial_object` outcome, and `recordPipelineEvent` records whatever +> outcome that function returns. The sketch below is kept for reference. + +In [audit_metrics_test.go](../../internal/queue/audit_metrics_test.go), modelled +on `TestAuditPipelineEventsMetric_DroppedNoBody`: + +```go +func TestAuditPipelineEventsMetric_DroppedPartialObject(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + + mr := miniredis.RunT(t) + er := &fakeEventRouter{} + c := newTestConsumer(t, mr, configmapRuleStore(), er) + require.NoError(t, c.ensureConsumerGroup(context.Background())) + + // A patch event whose only body is a finalizer-removal fragment. + ev := makeAuditEvent("patch", auditv1.StageResponseComplete, "configmaps", "default", "cm") + ev.RequestObject = &runtime.Unknown{Raw: []byte(`{"metadata":{"finalizers":null}}`)} + pushAuditMessage(t, mr, ev) + require.NoError(t, c.readAndProcessBatch(context.Background())) + + pipeline, ok := telemetry.CollectInt64Sum(reader, pipelineEventsMetric, map[string]string{ + "resource": "configmaps", "verb": "patch", "outcome": "dropped_partial_object", + }) + require.True(t, ok, "expected a dropped_partial_object audit_pipeline_events_total sample") + assert.Equal(t, int64(1), pipeline) + + // The fragment must never reach the git pipeline. + assert.Empty(t, er.calls, "partial-object event must not be routed") +} +``` + +(`readAndProcessBatch` returning `nil` is itself the assertion that the event +was ACK'd on the benign path — a poison-pill error would surface here. +`fakeEventRouter.calls` records every `RouteToGitTargetEventStream` call, so an +empty slice proves the fragment never reached the git pipeline.) + +## Out of scope + +- **Splitting the `Status` drop into its own outcome.** `errAuditEventObjectIsStatus` + still reports `dropped_no_body`, which is mildly inaccurate (there *is* a body). + Correcting it is a one-line follow-up but is not required here and would change + an existing metric series. +- **Recovering state from the fragment.** A finalizer patch has no full object; + there is nothing to mirror. The resource's real change is already covered by + its delete/full-body audit event. Dropping is the correct terminal action. +- **The CozyStack aggregated-apiserver watch instability** (`INTERNAL_ERROR` + HTTP/2 stream resets, `bookmark expired`) seen in the same logs — an upstream + environment issue on a different code path, tracked separately. + +## References + +- [internal/queue/redis_audit_consumer.go](../../internal/queue/redis_audit_consumer.go) + — `extractObject`, `handleExtractObjectError`, `routeAuditEvent`, the sentinel + errors, `pipelineOutcome*`, `recordPipelineEvent`. +- [internal/queue/redis_audit_consumer_test.go](../../internal/queue/redis_audit_consumer_test.go) + — existing `extractObject` test patterns. +- [internal/queue/audit_metrics_test.go](../../internal/queue/audit_metrics_test.go) + — existing `audit_pipeline_events_total` outcome tests. +- [docs/architecture.md](../architecture.md#metrics) — the metrics table to update. +- [shallow-audit-event-misclassification.md](./shallow-audit-event-misclassification.md) + — the related body-shape classification work. diff --git a/docs/design/shallow-audit-event-misclassification.md b/docs/design/shallow-audit-event-misclassification.md new file mode 100644 index 00000000..4338e939 --- /dev/null +++ b/docs/design/shallow-audit-event-misclassification.md @@ -0,0 +1,432 @@ +# Shallow audit events: definition, recognition, and the `/v1/pods` flood + +> Status: root cause confirmed — the flood is `pods/exec`, not pod object creation. +> Date: 2026-05-22 +> Cluster under investigation: **CozyStack** (`cluster="cozystack"`, `tenant="tenant-root"`). +> Related: [Audit Ingestion Pipeline](../architecture.md#audit-ingestion-pipeline), +> [apiservice-audit-proxy README](../../external-sources/apiservice-audit-proxy/README.md). + +## Summary + +After a recent upgrade, `audit-joiner` emits a high-volume flood of warnings: + +``` +WARNING: official shallow audit event timed out waiting for additional body; +the official event will be dropped. Install or repair apiservice-audit-proxy +so request/response bodies arrive within the wait budget. + auditID=... gvr=/v1/pods verb=create wait=0.5 +``` + +The volume is confirmed by metrics (see [Evidence](#evidence-from-the-cluster)): +thousands of `official` / `identity_shallow` / `pods` / `create` events. + +The early debug Redis stream captured the exact incoming event. It is a +successful streaming `pods/exec` request, not a pod object create: + +- `objectRef.resource: pods` +- `objectRef.subresource: exec` +- `verb: create` +- `responseStatus.code: 101` +- no `requestObject` or `responseObject` + +That resolves the apparent policy contradiction. Kubernetes audits POST-style +exec requests with verb `create`, while GitOps Reverser's quality metric labels +resource but not subresource. The metric therefore looked like a flood of +bodyless pod creates. The reported audit policy excludes top-level `pods`; a +`pods/exec` audit-policy match is separate and falls through to its later +`RequestResponse` rule. + +The fix for this flood is to reject all non-empty `objectRef.subresource` audit +events before the join pipeline. Current WatchRule planning rejects subresources +too; they are not the top-level object state this Git writer mirrors. The +previous audit ingress check rejected only `status` subresources, which was too +narrow. + +This document also keeps the shallow-event analysis because the aggregated-API +body gap is real, but that gap did not produce this `/v1/pods` flood. + +## Confirmed offending event + +The captured event has this load-bearing shape: + +```json +{ + "level": "RequestResponse", + "stage": "ResponseComplete", + "requestURI": "/api/v1/namespaces/cozy-kubeovn/pods/ovn-central-7955dc78d8-lvwh4/exec?...", + "verb": "create", + "objectRef": { + "resource": "pods", + "namespace": "cozy-kubeovn", + "name": "ovn-central-7955dc78d8-lvwh4", + "apiVersion": "v1", + "subresource": "exec" + }, + "responseStatus": { + "code": 101 + } +} +``` + +`pods/exec` upgrades into a command stream. It cannot provide a resource object +for Git mirroring and no additional-body proxy can repair it. The joiner used to +see `verb=create` plus empty object bodies, classify it `identity_shallow`, then +wait for a body that can never arrive. + +## Background: the two audit channels + +GitOps Reverser ingests audit events from two endpoints: + +| Endpoint | Sender (intended) | Role | +| --- | --- | --- | +| `/audit-webhook` | kube-apiserver | Canonical audit source — authority for identity, verb, status | +| `/audit-webhook-additional` | `apiservice-audit-proxy` | Supplementary **body** source | + +The endpoint the payload arrives on *is* the source role — there is no in-band +marker ([audit_handler.go](../../internal/webhook/audit_handler.go#L534)). The +metric label `source="official"` therefore means only "POSTed to +`/audit-webhook`" — **not** "proven to come from kube-apiserver." That +distinction matters below. + +`apiservice-audit-proxy` exists for exactly one reason. Per its +[README](../../external-sources/apiservice-audit-proxy/README.md): + +> It exists to recover audit fields that kube-apiserver does not populate for +> **aggregated API requests**: `objectRef.name`, `requestObject`, and +> `responseObject`. + +For a request to an **aggregated (`APIService`-backed) API**, kube-apiserver +proxies the call to a separate backend and never sees the request or response +body. Its native audit event for that request is *hollow*. The proxy sits in +front of the backend, observes the bodies, and posts an enriched event to +`/audit-webhook-additional` so the joiner can merge it. + +For **built-in resources** — `pods`, `configmaps`, `deployments`, every core and +standard API — kube-apiserver handles the request itself and its native audit +event is **already complete**. The proxy is never in that path and contributes +nothing. + +Note CozyStack specifically: the proxy README has a +[CozyStack case study](../../external-sources/apiservice-audit-proxy/README.md) +— the proxy redirects CozyStack's existing `APIService` objects (the +`cozystack-api` aggregated API) to itself. That covers CozyStack's aggregated +API groups; it does **not** put core `v1/pods` behind the proxy. + +## What a shallow event actually is + +A **shallow event** is the hollow native event kube-apiserver emits for an +**aggregated API request** — the case the additional-body channel was built to +repair. It is *not* simply "an event without a body." + +The distinction is visible in a real captured payload — the proxy's checked-in +`audit-lane-a-kube-apiserver-hollow.json`, kube-apiserver's native event for a +`create` on the aggregated `flunders` API: + +```jsonc +{ + "level": "RequestResponse", // policy asked for bodies... + "verb": "create", + "objectRef": { + "resource": "flunders", + "namespace": "default", + "apiGroup": "wardle.example.com", + "apiVersion": "v1alpha1" + // ...but NO "name" + } + // ...and NO "requestObject", NO "responseObject" +} +``` + +Even though the audit policy requested `RequestResponse`, kube-apiserver could +not fill in `objectRef.name`, `requestObject`, or `responseObject`, because the +real work happened in an aggregated backend it cannot see. **That** is a shallow +event — and note it still carries `level: RequestResponse`. + +Contrast the event shapes the joiner must tell apart: + +| Event shape | `level` | `objectRef.name` | `requestObject` / `responseObject` | What it is | Correct treatment | +| --- | --- | --- | --- | --- | --- | +| Complete | `Request` / `RequestResponse` | set | present | Built-in resource, or a merged event | Emit as-is | +| **Shallow** | **`RequestResponse`** | **empty** | **empty** | **Aggregated-API hollow event** — policy asked for bodies, apiserver could not supply them | **Wait for an additional body** | +| Bodyless by policy | `Metadata` / `None` | set | empty | Built-in resource the audit policy never captures bodies for | Reject at ingress — no Git write possible, nothing to wait for | + +(Bodyless *deletes* are a separate, documented carve-out — `body_shallow_deletable` — +and emit on `objectRef` identity alone; they are not covered by this table.) + +## The recognition rule + +A shallow event is **recognisable from the event alone**. No API discovery, no +`APIResourceCatalog` lookup, no `APIService` enumeration, no new field anywhere +is needed: + +> An audit event is a **genuine shallow event** when `objectRef.name`, +> `requestObject`, and `responseObject` are **all empty** — typically together +> with `level: RequestResponse` (the policy wanted bodies; the apiserver could +> not supply them). + +`objectRef.name` is the load-bearing field. It is metadata, not body, so the +audit *level* does not strip it: a built-in resource keeps its name even at +`level: Metadata`. Only the aggregated-API path drops it. So the presence or +absence of `objectRef.name` on the offending event is exactly what tells the two +remaining hypotheses apart. + +This matches the **documented** definition in the architecture quality table, +which already says `identity_shallow` means *"No body, **missing `objectRef` +identity**"*. The code drifted from it — see [The classifier +divergence](#the-classifier-divergence). + +## Evidence from the cluster + +### The audit policy reported as configured + +```yaml +apiVersion: audit.k8s.io/v1 +kind: Policy +omitStages: + - "RequestReceived" +rules: + - level: None + resources: + - group: "" + resources: ["events", "endpoints", "nodes", "pods", "secrets", "bindings", "componentstatuses", "*/status"] + - group: "authentication.k8s.io" + resources: ["tokenreviews"] + - group: "authorization.k8s.io" + resources: ["subjectaccessreviews", "selfsubjectaccessreviews", "localsubjectaccessreviews", "selfsubjectrulesreviews"] + - group: "coordination.k8s.io" + resources: ["leases"] + - group: "apps" + resources: ["*/status"] + - group: "networking.k8s.io" + resources: ["*/status"] + - level: None + users: ["system:serviceaccount:kube-system:horizontal-pod-autoscaler"] + verbs: ["update", "patch"] + resources: + - group: "apps" + resources: ["*/scale"] + - level: RequestResponse + omitManagedFields: true + verbs: + - create + - update + - patch + - delete + - deletecollection +``` + +Kubernetes audit policy is **first-match-wins**. For a core top-level `pods` +object `create`: + +- Rule 1 has a `GroupResources` entry `group: ""`, `resources: [... "pods" ...]`. + A core pod create matches it. Rule 1 has no `users`/`verbs`/`namespaces` + selectors, so those are wildcards — the match holds. +- First match wins → **`level: None`** → the event is **not generated at all.** + +It never reaches rule 3 (the `RequestResponse` catch-all). So a kube-apiserver +running this exact policy emits **zero** top-level `pods` object audit events of +any level. The captured event is `pods/exec`, which is a separate audit-policy +resource match. The same top-level-resource distinction applies to `secrets`, +`events`, `endpoints`, `nodes`, `bindings`, and the explicit `*/status` +patterns. + +### The metric + +Query: `gitopsreverser_audit_event_quality_total{verb="create", quality!="complete"}` + +``` +gitopsreverser_audit_event_quality_total{ + quality="identity_shallow", resource="pods", version="v1", + source="official", verb="create", + cluster="cozystack", tenant="tenant-root", + pod="gitops-reverser-56f948fdd8-ttrm6", ... +} + min: 34 median: 2127 max: 4116 +``` + +`addQualityMetric` is recorded immediately after classification, before any drop +([audit_handler.go](../../internal/webhook/audit_handler.go#L292)) — so this +counts every such event that actually arrived. Thousands of `official`, +`identity_shallow`, core `v1/pods`, `create` events are reaching the handler. + +## The resolved contradiction + +The policy says **zero top-level pod object events.** The metric says +**thousands of events labeled `resource="pods"` and `verb="create"`**. Both can +describe the same apiserver because the metric does not label +`objectRef.subresource`, while Kubernetes audit policy distinguishes `pods` from +`pods/exec`. + +My earlier draft asserted the cause was a `Metadata`-level policy for pods. That +assertion was unverified and is now retracted. The later H1/H2 split below is +kept as shallow-event analysis, not as the root-cause fork for this flood. + +## Earlier hypotheses + +Before the debug stream captured `objectRef.subresource: exec`, the analysis +split on `objectRef.name`. That split is no longer needed for this flood, but it +records the remaining shallow-event question for top-level resources. + +### H1 — genuine shallow events (the joiner is behaving correctly) + +If the offending `pods/create` events have **empty `objectRef.name`** and empty +bodies, they are *genuine shallow events* by the [recognition +rule](#the-recognition-rule). Something is routing core pod creates through an +aggregated/proxied path where the emitting apiserver cannot see the body. In +that case: + +- The joiner is **right** to wait for an additional body. +- The real gap is that no additional-body source is enriching these — the + `apiservice-audit-proxy` deployment does not cover whatever path this is. +- The warning's advice ("install/repair apiservice-audit-proxy") is *correct*. +- This is **not** a classification bug. + +### H2 — bodyless-but-identified events (a real misclassification) + +If the offending events have a **populated `objectRef.name`** and empty bodies, +they are built-in pod events whose bodies were dropped by an audit policy +(a broad policy auditing pods at `Metadata`, contradicting the pasted one). In +that case: + +- They are *not* shallow — they have full identity. +- The classifier still labels them `identity_shallow` and routes them into the + wait path (see below), so the joiner waits for a body that can never come. +- This **is** a classification bug, and the [proposed fix](#proposed-fix) + applies. + +A debug-level (`V(1)`) log already records `hasRequestObject` / +`hasResponseObject` in `dropShallowOfficial` +([audit_joiner.go](../../internal/webhook/audit_joiner.go#L319)); it does **not** +record `objectRef.name` or `level`. Adding both to that log line is the cheapest +way to settle H1 vs H2 in production. + +## The classifier divergence + +Independent of which hypothesis holds, the classifier does not match its own +documented contract. + +`classifyAuditEventQuality` in +[internal/webhook/audit_joiner.go](../../internal/webhook/audit_joiner.go#L681) +decides quality from body presence and never inspects `objectRef.name` on the +create/update/patch path: + +```go +if hasAuditV1ObjectBody(event) { + return AuditEventQualityComplete +} +if allowsBodylessAuditV1Delete(event) { // delete-only carve-out + return AuditEventQualityBodyShallowDeletable +} +return AuditEventQualityIdentityShallow // <-- everything else bodyless +``` + +`hasAuditV1ObjectBody` only checks `requestObject` / `responseObject`. The only +place `objectRef.name` is consulted is `allowsBodylessAuditV1Delete`, gated on +`verb == "delete"`. So a bodyless **`create`** / `update` / `patch` with a +complete `objectRef` falls through to `identity_shallow` even though it has full +identity — exactly the H2 failure. The architecture doc says `identity_shallow` +requires *missing* `objectRef` identity; the code assigns it on body absence +alone. + +### Contributing history + +Commit `605964a` ("gate audit body joins by rule relevance") added a gate that +dropped shallow officials for resources no `WatchRule` could match — but it +gated on the **wrong axis** (WatchRule relevance, not shallow-vs-identified). +Commit `a688620` then removed that gate entirely to fix an unrelated e2e race. +Neither corrected the classification. A later "louder logging" change then made +the pre-existing behaviour impossible to ignore — which is why the flood appears +to be new even though the misroute (under H2) is older. + +## Impact + +- **Log flood** — one WARN per offending event, on every occurrence (the WARN is + deliberately not gated by `sync.Once`). +- **Throughput cost** — each such event holds the official canonical gate for + the full `--audit-event-body-wait` budget (`500ms`), so later official events + queue behind a wait that is guaranteed to fail. +- **Data loss** — the event is dropped (`audit_shallow_dropped_total`). Under H1 + this means genuinely missing data; under H2 it means a built-in mutation is + silently not mirrored. +- **Possibly misleading advice** — under H2 the "install apiservice-audit-proxy" + WARN points operators at a fix that cannot help. Under H1 it is correct. + +## Proposed fix + +For the confirmed flood: + +1. **Gate every non-empty `objectRef.subresource` at audit ingress.** This + generalizes the previous `status`-only check to current product capability: + subresources are not top-level resource state and are not supported by + WatchRule planning. +2. **Keep a literal `pods/exec` regression event.** The regression must keep + the observed `verb: create`, `subresource: exec`, `responseStatus.code: 101`, + and empty object bodies so the joiner cannot start waiting on it again. + +For top-level bodyless resources, two changes are still worth considering +because they harden diagnosis: + +1. **Log `objectRef.name` and `level`** in `dropShallowOfficial` and in the + `waitForBody` timeout WARN. This settles H1 vs H2 in-cluster without a packet + capture, and is a one-line change. +2. **Revisit whether the classifier contract and docs should key on identity or + body repairability.** The earlier draft proposed tightening + `identity_shallow` + should require `objectRef.name` to be empty **in addition to** both bodies + being empty. That needs its own decision because URL-named aggregated + subresource-free requests can still need body repair. + +If **H2** is confirmed, additionally: + +3. **Gate the official ingress on `level`.** An official event at `level: None` + or `level: Metadata` cannot drive a Git write and has nothing to wait for; it + should be rejected in `classifyAuditIngress` + ([audit_handler.go](../../internal/webhook/audit_handler.go#L597)) before the + joiner, with a rate-limited log pointing at the audit policy — not the proxy. + +If **H1** is confirmed, the classifier is already correct; the work moves to the +deployment side — identifying the path these pod creates take and extending an +additional-body source to cover it. + +None of this needs an `APIResourceCatalog` attribute, `APIService` enumeration, +or extra discovery calls. + +## Deferred top-level shallow-event checks + +The captured `pods/exec` event is enough to explain and gate this flood. If a +separate top-level resource still arrives bodyless after that gate, these checks +would settle whether it is an aggregated-API hollow event or an audit-policy +body gap: + +1. **`objectRef.name` on the top-level event** — empty or populated. This field + decides H1 vs H2 for a subresource-free bodyless event. +2. **The audit configuration the *relevant* kube-apiserver is actually running** + on the CozyStack cluster — the live `--audit-policy-file` / + `--audit-webhook-config-file` flags of the process emitting those events, not + only the ConfigMap or Helm value the policy was pasted from. +3. **Which sender produced those events** — the CozyStack management control + plane, a tenant (Kamaji) control plane, or `apiservice-audit-proxy`. The + `auditID`, `sourceIPs`, and `userAgent` on a sample event will indicate this. +4. **Whether `apiservice-audit-proxy` is deployed** on this cluster and which + endpoint it posts to (`/audit-webhook` vs `/audit-webhook-additional`). + +## Open question + +If H2 is confirmed: what should happen to a bodyless-but-identified +`create` / `update` / `patch` — a built-in resource fully identified but with +bodies stripped by policy? Drop it (treat a thin policy as misconfiguration, +with a rate-limited WARN), or emit it as an identity-only event (the +`body_shallow_deletable` carve-out already does this for `delete`)? Left open +deliberately — it is a design decision, not a classification bug. + +## References + +- [internal/webhook/audit_joiner.go](../../internal/webhook/audit_joiner.go) — + `classifyAuditEventQuality`, `handleOfficial`, `waitForBody`, + `dropShallowOfficial`, `allowsBodylessAuditV1Delete`, `hasAuditV1ObjectBody`. +- [internal/webhook/audit_handler.go](../../internal/webhook/audit_handler.go) — + `classifyAuditIngress`, `auditSourceFromPath`, `addQualityMetric`. +- [docs/architecture.md](../architecture.md#audit-ingestion-pipeline) — the + pipeline and the quality table the code should match. +- [apiservice-audit-proxy README](../../external-sources/apiservice-audit-proxy/README.md) + — aggregated-API hollow vs complete events, and the CozyStack case study. diff --git a/docs/interpreting-metrics.md b/docs/interpreting-metrics.md index 24bc8236..009ab03d 100644 --- a/docs/interpreting-metrics.md +++ b/docs/interpreting-metrics.md @@ -121,7 +121,7 @@ additional body arriving first and parking; when the official wins the race it w | `audit_eventlists_total` | counter | `source`, `outcome` | ① ingress | | `audit_eventlist_events_total` | counter | `source`, `outcome` | ① ingress | | `audit_eventlist_duration_seconds` | histogram | `source`, `outcome` | ① ingress | -| `audit_events_received_total` | counter | `source`, `group`, `version`, `resource`, `verb` | ① ingress | +| `audit_events_received_total` | counter | `source`, `group`, `version`, `resource`, `subresource`, `verb` | ① ingress | | `audit_event_quality_total` | counter | `source`, `quality`, `group`, `version`, `resource`, `verb` | ① ingress | | `audit_join_parked_total` | counter | — | ② join | | `audit_join_emitted_total` | counter | `source`, `result` (`as_is`, `merged`) | ② join | @@ -140,6 +140,13 @@ counts the decoded event items inside them. `outcome` is bounded: `processed`, ` arriving?" before any join or rule logic runs. `audit_events_received_total` and `audit_event_quality_total` then describe individual decoded events. +`audit_events_received_total` carries a `subresource` label: empty for top-level resources, and +a bounded value (`exec`, `status`, `scale`, `log`, …) for subresource requests. Subresource +events are counted here but then dropped at ingress — they do not describe a top-level object +the Git pipeline can mirror — so any non-empty `subresource` row is "received then dropped". The +label exists so a `pods/exec` flood is visible as exactly that, rather than collapsing into +`resource="pods"` and looking like real pod mutations. + **Stage ③ — consumer output.** `audit_pipeline_events_total` is recorded once per canonical event in the consumer, after rule matching. `outcome` tells you which resource events reach the consumer but do not become Git work: `routed` (reached at least one BranchWorker), `unmatched` @@ -172,6 +179,26 @@ sum by (source, outcome) (rate(gitopsreverser_audit_eventlists_total[5m])) sum by (source) (rate(gitopsreverser_audit_eventlist_events_total[5m])) ``` +**What strange or high-volume traffic is the webhook receiving?** The top received event shapes — +this surfaces an unexpected resource flood at a glance, and because the result is split by +`subresource`, a streaming `pods/exec` storm shows up as its own row instead of hiding inside +`resource="pods"`: + +```promql +topk(15, sum by (resource, subresource, verb) ( + rate(gitopsreverser_audit_events_received_total[5m]))) +``` + +Any row with a non-empty `subresource` is received-then-dropped at ingress (subresources are not +mirrorable top-level objects). A large `pods`/`exec` row is the canonical example — see +[shallow-audit-event-misclassification.md](design/shallow-audit-event-misclassification.md). To +look only at the dropped subresource traffic: + +```promql +topk(10, sum by (resource, subresource, verb) ( + rate(gitopsreverser_audit_events_received_total{subresource!=""}[5m]))) +``` + **Are EventLists failing to decode?** Should be zero — non-zero means a sender is posting something that is not an `audit.k8s.io/v1 EventList`: diff --git a/internal/queue/redis_audit_consumer.go b/internal/queue/redis_audit_consumer.go index 76879f50..829f2cc1 100644 --- a/internal/queue/redis_audit_consumer.go +++ b/internal/queue/redis_audit_consumer.go @@ -80,6 +80,15 @@ var errAuditEventObjectMissing = errors.New("audit event has no requestObject or // proxy body) so the Status is never written to Git as if it were the resource. var errAuditEventObjectIsStatus = errors.New("audit event object is a metav1.Status error body") +// errAuditEventObjectPartial marks an audit event whose body is valid JSON but +// lacks the apiVersion/kind identity of a full Kubernetes object — typically a +// merge-patch fragment such as {"metadata":{"finalizers":null}} recorded as the +// requestObject of a finalizer-removal PATCH. It carries no routable resource +// state, so it is dropped before git routing rather than treated as a decode +// failure. The resource's real mutation is mirrored from its own (delete or +// full-body) audit event. +var errAuditEventObjectPartial = errors.New("audit event object body is a partial object (no kind)") + // AuditEventRouter is the subset of watch.EventRouter used by the consumer. // watch.EventRouter satisfies this interface without modification. type AuditEventRouter interface { @@ -130,6 +139,7 @@ type AuditConsumer struct { firstRouted sync.Once firstShallowDropped sync.Once firstStatusDropped sync.Once + firstPartialDropped sync.Once } // NewAuditConsumer creates a new AuditConsumer. It does not start consuming; @@ -407,10 +417,10 @@ func (c *AuditConsumer) routeAuditEvent( sanitized, err := extractObject(auditEvent, op, fullAPIVersion, ref.Resource, namespace, name) if err != nil { - if c.handleExtractObjectError( + if outcome, handled := c.handleExtractObjectError( log, auditEvent, err, fullAPIVersion+"/"+ref.Resource, namespace, name, - ) { - recordPipelineEvent(ctx, gvr, auditEvent.Verb, pipelineOutcomeDroppedNoBody) + ); handled { + recordPipelineEvent(ctx, gvr, auditEvent.Verb, outcome) return nil } return fmt.Errorf("extracting object for %s/%s: %w", namespace, name, err) @@ -452,16 +462,18 @@ func (c *AuditConsumer) routeAuditEvent( } // handleExtractObjectError classifies an extractObject failure. For a benign -// drop — a shallow event with no body, or a metav1.Status error body from a -// failed API request — it logs and returns true so the caller acks the event -// without routing it. For any other error it returns false, leaving the caller -// to surface it. +// drop — a shallow event with no body, a metav1.Status error body from a failed +// API request, or a partial object such as a finalizer-removal patch fragment — +// it logs and returns (outcome, true): the audit_pipeline_events_total outcome +// the caller should record, and a handled flag so the event is ACK'd without +// routing. For any other error it returns ("", false), leaving the caller to +// surface it. func (c *AuditConsumer) handleExtractObjectError( log logr.Logger, auditEvent auditv1.Event, err error, gvr, namespace, name string, -) bool { +) (string, bool) { switch { case errors.Is(err, errAuditEventObjectMissing): c.firstShallowDropped.Do(func() { @@ -484,7 +496,7 @@ func (c *AuditConsumer) handleExtractObjectError( "hasRequestObject", hasAuditObjectRaw(auditEvent.RequestObject), "hasResponseObject", hasAuditObjectRaw(auditEvent.ResponseObject), ) - return true + return pipelineOutcomeDroppedNoBody, true case errors.Is(err, errAuditEventObjectIsStatus): c.firstStatusDropped.Do(func() { c.log.Info( @@ -505,9 +517,30 @@ func (c *AuditConsumer) handleExtractObjectError( "namespace", namespace, "name", name, ) - return true + return pipelineOutcomeDroppedNoBody, true + case errors.Is(err, errAuditEventObjectPartial): + c.firstPartialDropped.Do(func() { + c.log.Info( + "First audit event dropped before git routing — body is a partial object "+ + "(no kind), typically a finalizer-removal PATCH fragment. The resource's "+ + "real change is mirrored from its own audit event; this fragment is not "+ + "routable. Further drops will log at V(1) only.", + "auditID", auditEvent.AuditID, + "gvr", gvr, + "verb", auditEvent.Verb, + ) + }) + log.V(1).Info( + "audit event dropped before git routing: partial object body (no kind)", + "auditID", auditEvent.AuditID, + "gvr", gvr, + "verb", auditEvent.Verb, + "namespace", namespace, + "name", name, + ) + return pipelineOutcomeDroppedPartialObject, true default: - return false + return "", false } } @@ -574,10 +607,11 @@ type pipelineGVR struct { // Audit pipeline consumer-stage outcome and rule-kind label values. const ( - pipelineOutcomeUnmatched = "unmatched" - pipelineOutcomeDroppedNoBody = "dropped_no_body" - pipelineOutcomeRouted = "routed" - pipelineOutcomeRouteFailed = "route_failed" + pipelineOutcomeUnmatched = "unmatched" + pipelineOutcomeDroppedNoBody = "dropped_no_body" + pipelineOutcomeDroppedPartialObject = "dropped_partial_object" + pipelineOutcomeRouted = "routed" + pipelineOutcomeRouteFailed = "route_failed" ruleKindWatchRule = "watchrule" ruleKindClusterWatchRule = "clusterwatchrule" @@ -726,6 +760,9 @@ func extractObject( obj := &unstructured.Unstructured{} if err := obj.UnmarshalJSON(raw); err != nil { + if isPartialObjectBody(raw) { + return nil, errAuditEventObjectPartial + } return nil, fmt.Errorf("failed to unmarshal object JSON: %w", err) } @@ -736,6 +773,20 @@ func extractObject( return backfillSanitizedIdentity(sanitize.Sanitize(obj), apiVersion, resource, namespace, name), nil } +// isPartialObjectBody reports whether raw is well-formed JSON describing an +// object that lacks a "kind" — the condition that makes +// (*Unstructured).UnmarshalJSON fail on an otherwise valid body. A merge-patch +// fragment such as {"metadata":{"finalizers":null}} matches; malformed bytes do +// not, so a genuine decode failure still surfaces as an error. +func isPartialObjectBody(raw []byte) bool { + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + return false + } + kind, _ := m["kind"].(string) + return kind == "" +} + // isStatusObject reports whether obj is a core metav1.Status error response // (apiVersion: v1, kind: Status) rather than a real Kubernetes resource. The // API server returns such a body when a request fails — for example the diff --git a/internal/queue/redis_audit_consumer_test.go b/internal/queue/redis_audit_consumer_test.go index bad9f383..9500b8ce 100644 --- a/internal/queue/redis_audit_consumer_test.go +++ b/internal/queue/redis_audit_consumer_test.go @@ -351,6 +351,54 @@ func TestExtractObject_InvalidJSON(t *testing.T) { require.Error(t, err) } +// TestExtractObject_ClassifiesPartialFinalizerPatch reproduces the CozyStack +// prod occurrence: deleting mongodb-simon2 produced an audit event whose only +// body was the finalizer-removal patch fragment {"metadata":{"finalizers":null}}. +// extractObject must classify it as a partial object, not a decode failure. +func TestExtractObject_ClassifiesPartialFinalizerPatch(t *testing.T) { + ev := auditv1.Event{ + Verb: "patch", + RequestObject: &runtime.Unknown{Raw: []byte(`{"metadata":{"finalizers":null}}`)}, + // ResponseObject deliberately nil: the object was deleted by this same + // PATCH (last finalizer removed), so the apiserver recorded no body. + } + + _, err := extractObject( + ev, configv1alpha1.OperationUpdate, + "helm.toolkit.fluxcd.io/v2", "helmreleases", "tenant-root", "mongodb-simon2", + ) + require.ErrorIs(t, err, errAuditEventObjectPartial) +} + +// TestExtractObject_MalformedBodyStillErrors guards the boundary: bytes that are +// not valid JSON are a genuine decode failure and must NOT be reclassified as a +// benign partial object — they still deserve the error-level poison-pill log. +func TestExtractObject_MalformedBodyStillErrors(t *testing.T) { + ev := auditv1.Event{ + ResponseObject: &runtime.Unknown{Raw: []byte(`{"metadata":`)}, // truncated + } + + _, err := extractObject(ev, configv1alpha1.OperationCreate, "v1", "ConfigMap", "default", "cm") + require.Error(t, err) + require.NotErrorIs(t, err, errAuditEventObjectPartial) + require.NotErrorIs(t, err, errAuditEventObjectMissing) + require.NotErrorIs(t, err, errAuditEventObjectIsStatus) +} + +// TestHandleExtractObjectError_PartialObjectIsBenign confirms a partial-object +// error is handled (ACK, no poison-pill) and reported under the +// dropped_partial_object metric outcome. +func TestHandleExtractObjectError_PartialObjectIsBenign(t *testing.T) { + c := &AuditConsumer{log: logr.Discard()} + outcome, handled := c.handleExtractObjectError( + logr.Discard(), auditv1.Event{Verb: "patch"}, + errAuditEventObjectPartial, "helm.toolkit.fluxcd.io/v2/helmreleases", + "tenant-root", "mongodb-simon2", + ) + assert.True(t, handled) + assert.Equal(t, pipelineOutcomeDroppedPartialObject, outcome) +} + func TestEffectiveAuditOperation_TerminatingUpdateBecomesDelete(t *testing.T) { obj := &unstructured.Unstructured{} obj.SetAPIVersion("apiextensions.k8s.io/v1") diff --git a/internal/queue/redis_audit_queue.go b/internal/queue/redis_audit_queue.go index ab16b199..a4fb1a4b 100644 --- a/internal/queue/redis_audit_queue.go +++ b/internal/queue/redis_audit_queue.go @@ -36,6 +36,8 @@ import ( const ( // DefaultRedisAuditStream is the default stream used for audit ingestion events. DefaultRedisAuditStream = "gitopsreverser.audit.events.v1" + // DefaultRedisAuditDebugStream is the default stream used for early audit debug events. + DefaultRedisAuditDebugStream = "gitopsreverser.audit.debug.events.v1" ) // RedisAuditQueueConfig configures a Redis-backed audit queue. @@ -56,6 +58,11 @@ type RedisAuditQueue struct { maxLen int64 } +// RedisAuditDebugQueue enqueues early audit debug events into a Redis stream. +type RedisAuditDebugQueue struct { + queue *RedisAuditQueue +} + // NewRedisAuditQueue creates a Redis stream-backed audit queue. func NewRedisAuditQueue(cfg RedisAuditQueueConfig) (*RedisAuditQueue, error) { if strings.TrimSpace(cfg.Addr) == "" { @@ -87,6 +94,28 @@ func NewRedisAuditQueue(cfg RedisAuditQueueConfig) (*RedisAuditQueue, error) { // Enqueue writes one audit event to Redis stream storage. func (q *RedisAuditQueue) Enqueue(ctx context.Context, event auditv1.Event) error { + return q.enqueue(ctx, event, nil) +} + +// NewRedisAuditDebugQueue creates a Redis stream-backed early audit debug queue. +func NewRedisAuditDebugQueue(cfg RedisAuditQueueConfig) (*RedisAuditDebugQueue, error) { + if strings.TrimSpace(cfg.Stream) == "" { + cfg.Stream = DefaultRedisAuditDebugStream + } + + redisQueue, err := NewRedisAuditQueue(cfg) + if err != nil { + return nil, err + } + return &RedisAuditDebugQueue{queue: redisQueue}, nil +} + +// Enqueue writes one early audit debug event to Redis stream storage. +func (q *RedisAuditDebugQueue) Enqueue(ctx context.Context, source string, event auditv1.Event) error { + return q.queue.enqueue(ctx, event, map[string]any{"source": source}) +} + +func (q *RedisAuditQueue) enqueue(ctx context.Context, event auditv1.Event, extraValues map[string]any) error { payload, err := json.Marshal(event) if err != nil { return fmt.Errorf("failed to marshal audit event payload: %w", err) @@ -119,6 +148,9 @@ func (q *RedisAuditQueue) Enqueue(ctx context.Context, event auditv1.Event) erro "stage_timestamp": formatStageTimestamp(event.StageTimestamp.Time), "payload_json": string(payload), } + for key, value := range extraValues { + values[key] = value + } args := &redis.XAddArgs{ Stream: q.stream, diff --git a/internal/queue/redis_audit_queue_test.go b/internal/queue/redis_audit_queue_test.go index eb7380dc..21326934 100644 --- a/internal/queue/redis_audit_queue_test.go +++ b/internal/queue/redis_audit_queue_test.go @@ -128,3 +128,28 @@ func TestRedisAuditQueue_EnqueueCustomResourceStoresAPIGroup(t *testing.T) { assert.Equal(t, "default", entry["namespace"]) assert.Equal(t, "order-1", entry["name"]) } + +func TestRedisAuditDebugQueue_EnqueueStoresSource(t *testing.T) { + mr := miniredis.RunT(t) + + debugQueue, err := NewRedisAuditDebugQueue(RedisAuditQueueConfig{ + Addr: mr.Addr(), + Stream: "audit.debug.test", + MaxLen: 100, + }) + require.NoError(t, err) + + event := auditv1.Event{AuditID: "audit-debug-123", Verb: "create"} + err = debugQueue.Enqueue(context.Background(), "official", event) + require.NoError(t, err) + + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + entries, err := client.XRange(context.Background(), "audit.debug.test", "-", "+").Result() + require.NoError(t, err) + require.Len(t, entries, 1) + + entry := entries[0].Values + assert.Equal(t, "official", entry["source"]) + assert.Equal(t, "audit-debug-123", entry["audit_id"]) + assert.NotEmpty(t, entry["payload_json"]) +} diff --git a/internal/webhook/audit_handler.go b/internal/webhook/audit_handler.go index 9afcb846..81c985f3 100644 --- a/internal/webhook/audit_handler.go +++ b/internal/webhook/audit_handler.go @@ -24,8 +24,6 @@ import ( "fmt" "io" "net/http" - "os" - "path/filepath" "strings" "sync" "time" @@ -39,8 +37,6 @@ import ( "k8s.io/apimachinery/pkg/types" "k8s.io/apiserver/pkg/apis/audit" auditv1 "k8s.io/apiserver/pkg/apis/audit/v1" - "sigs.k8s.io/yaml" - logf "sigs.k8s.io/controller-runtime/pkg/log" "github.com/ConfigButler/gitops-reverser/internal/auditutil" @@ -59,23 +55,19 @@ type auditHandlerFirsts struct { impersonatedEvent sync.Once } -const ( - // DefaultAuditDumpDir is the default directory for audit event dumps. - DefaultAuditDumpDir = "/tmp/audit-events" - // DefaultAuditMaxRequestBodyBytes limits incoming audit payload size. - DefaultAuditMaxRequestBodyBytes = int64(10 * 1024 * 1024) -) +// DefaultAuditMaxRequestBodyBytes limits incoming audit payload size. +const DefaultAuditMaxRequestBodyBytes = int64(10 * 1024 * 1024) // AuditHandlerConfig contains configuration for the audit handler. type AuditHandlerConfig struct { - // DumpDir is the directory where audit events are written for debugging. - // If empty, defaults to DefaultAuditDumpDir. - DumpDir string // MaxRequestBodyBytes is the maximum accepted HTTP request body size. MaxRequestBodyBytes int64 // Queue enqueues accepted audit events to a durable backend. // If nil, queueing is disabled. Queue AuditEventQueue + // DebugQueue enqueues every decoded event before audit processing begins. + // If nil, early debug stream queueing is disabled. + DebugQueue AuditDebugEventQueue // Joiner optionally parks additional-source bodies and deduplicates canonical audit events. Joiner AuditEventJoiner } @@ -85,6 +77,11 @@ type AuditEventQueue interface { Enqueue(ctx context.Context, event auditv1.Event) error } +// AuditDebugEventQueue persists decoded events for early audit debugging. +type AuditDebugEventQueue interface { + Enqueue(ctx context.Context, source string, event auditv1.Event) error +} + // auditIngressDecision is the intrinsic accept/reject verdict for an audit event // before it enters the join pipeline. The verdict is derived purely from the // event — stage, verb, and body shape — and carries no knowledge of WatchRules; @@ -105,7 +102,6 @@ type AuditHandler struct { } // NewAuditHandler creates a new audit handler with the given configuration. -// If config.DumpDir is empty, file dumping is disabled. func NewAuditHandler(config AuditHandlerConfig) (*AuditHandler, error) { if config.MaxRequestBodyBytes <= 0 { config.MaxRequestBodyBytes = DefaultAuditMaxRequestBodyBytes @@ -182,6 +178,11 @@ func (h *AuditHandler) serveEventListRequest( } eventCount := len(eventListV1.Items) + if err := h.enqueueDebugEvents(ctx, source, eventListV1.Items); err != nil { + reqLog.Error(err, "Failed to enqueue early audit debug events") + http.Error(w, err.Error(), http.StatusInternalServerError) + return outcomeProcessError, eventCount + } if eventCount == 0 { reqLog.Info("Received empty audit event list", "eventCount", 0, "processingOutcome", "empty") h.writeResponse(w, reqLog, "Empty event list processed") @@ -255,6 +256,19 @@ func (h *AuditHandler) decodeEventList(r *http.Request) (*auditv1.EventList, err return &eventListV1, nil } +// enqueueDebugEvents preserves each decoded event before normal audit processing can filter it. +func (h *AuditHandler) enqueueDebugEvents(ctx context.Context, source AuditSource, events []auditv1.Event) error { + if h.config.DebugQueue == nil { + return nil + } + for _, event := range events { + if err := h.config.DebugQueue.Enqueue(ctx, string(source), event); err != nil { + return fmt.Errorf("failed to enqueue early audit debug event %q: %w", event.AuditID, err) + } + } + return nil +} + // processEvents processes a list of audit events. func (h *AuditHandler) processEvents(ctx context.Context, source AuditSource, events []auditv1.Event) error { for _, auditEventV1 := range events { @@ -325,7 +339,7 @@ func (h *AuditHandler) processEvent(ctx context.Context, source AuditSource, aud "ips", auditEvent.SourceIPs, "userAgent", auditEvent.UserAgent) - return h.writeCanonicalAuditEvent(eventToWrite, auditEvent) + return nil } // logFirstAuditRequest emits an Info banner the first time we accept an @@ -403,11 +417,13 @@ func (h *AuditHandler) recordReceivedMetric( // The username is intentionally not a metric label (cardinality bomb); it // stays in the structured logs only. group, version, resource := gvrParts(&auditEvent) + subresource := subresourcePart(&auditEvent) handlerLog.V(1).Info("Audit event received", "source", source, "group", group, "version", version, "resource", resource, + "subresource", subresource, "verb", auditEvent.Verb, "user", effectiveAuditUsername(auditEvent)) telemetry.AuditEventsReceivedTotal.Add(ctx, 1, metric.WithAttributes( @@ -415,6 +431,7 @@ func (h *AuditHandler) recordReceivedMetric( attribute.String("group", group), attribute.String("version", version), attribute.String("resource", resource), + attribute.String("subresource", subresource), attribute.String("verb", auditEvent.Verb), )) } @@ -505,18 +522,6 @@ func (h *AuditHandler) releaseJoinDecision(ctx context.Context, decision AuditJo } } -func (h *AuditHandler) writeCanonicalAuditEvent(eventToWrite *auditv1.Event, fallback audit.Event) error { - var emitted audit.Event - if err := h.scheme.Convert(eventToWrite, &emitted, nil); err != nil { - return fmt.Errorf("failed to convert emitted audit event: %w", err) - } - if emitted.AuditID == "" { - emitted = fallback - } - h.writeAuditEventToFile(&emitted) - return nil -} - func effectiveAuditUsername(event audit.Event) string { if event.ImpersonatedUser != nil && event.ImpersonatedUser.Username != "" { return event.ImpersonatedUser.Username @@ -561,6 +566,17 @@ func gvrParts(event *audit.Event) (string, string, string) { return gv.Group, gv.Version, orUnknownResource(event.ObjectRef.Resource) } +// subresourcePart returns the audit event's objectRef.subresource, or "" when +// the event has no objectRef or targets a top-level resource. Kubernetes +// subresources are a bounded, closed set (status, scale, exec, log, ...), so +// this is safe to use as a metric label. +func subresourcePart(event *audit.Event) string { + if event.ObjectRef == nil { + return "" + } + return event.ObjectRef.Subresource +} + func orUnknownResource(resource string) string { if resource == "" { return "unknown" @@ -630,7 +646,10 @@ func isFailedAuditRequest(event *auditv1.Event) bool { // checkEvent validates an audit event before processing. func (h *AuditHandler) checkEvent(event *audit.Event) (bool, error) { - process := event.ObjectRef == nil || event.ObjectRef.Subresource != "status" + // Subresource audit verbs do not describe top-level object mutations that + // the current Git routing path can mirror. Some streaming subresources such + // as pods/exec are audited as verb=create with no resource body at all. + process := event.ObjectRef == nil || event.ObjectRef.Subresource == "" if string(event.AuditID) == "" { return process, errors.New("invalid audit event: auditID cannot be empty") } @@ -645,42 +664,3 @@ func hasAuditV1ObjectBody(event *auditv1.Event) bool { func hasRuntimeUnknownBody(object *runtime.Unknown) bool { return object != nil && len(object.Raw) > 0 } - -// writeAuditEventToFile writes an audit event to a YAML file for debugging and testing. -// Assumes event has been validated by validateEvent() - auditID is guaranteed to be non-empty. -// Skips file writing if DumpDir is empty (disabled). -func (h *AuditHandler) writeAuditEventToFile(event *audit.Event) { - if h.config.DumpDir == "" { - return - } - - if err := os.MkdirAll(h.config.DumpDir, 0750); err != nil { - logf.Log.Error(err, "Failed to create audit events dump directory", "directory", h.config.DumpDir) - return - } - - filename := fmt.Sprintf("%s.yaml", event.AuditID) - filePath := filepath.Join(h.config.DumpDir, filename) - - var eventV1 auditv1.Event - if err := h.scheme.Convert(event, &eventV1, nil); err != nil { - logf.Log.Error(err, "Failed to convert event to v1", "auditID", event.AuditID) - return - } - - eventV1.Kind = "Event" - eventV1.APIVersion = "audit.k8s.io/v1" - - eventYAML, err := yaml.Marshal(&eventV1) - if err != nil { - logf.Log.Error(err, "Failed to marshal audit event to YAML", "auditID", event.AuditID) - return - } - - if err := os.WriteFile(filePath, eventYAML, 0600); err != nil { - logf.Log.Error(err, "Failed to write audit event to file", "file", filePath, "auditID", event.AuditID) - return - } - - logf.Log.V(1).Info("Audit event written to file", "file", filePath, "auditID", event.AuditID) -} diff --git a/internal/webhook/audit_handler_test.go b/internal/webhook/audit_handler_test.go index 19ae748f..a8a98529 100644 --- a/internal/webhook/audit_handler_test.go +++ b/internal/webhook/audit_handler_test.go @@ -27,7 +27,6 @@ import ( "net/http" "net/http/httptest" "os" - "path/filepath" "sync" "testing" "time" @@ -73,6 +72,31 @@ func (q *recordingAuditEventQueue) auditIDs() []string { return ids } +type errorAuditDebugQueue struct{ err error } + +func (q errorAuditDebugQueue) Enqueue(_ context.Context, _ string, _ auditv1.Event) error { + return q.err +} + +type recordingAuditDebugQueue struct { + sources []string + events []auditv1.Event +} + +func (q *recordingAuditDebugQueue) Enqueue(_ context.Context, source string, event auditv1.Event) error { + q.sources = append(q.sources, source) + q.events = append(q.events, event) + return nil +} + +func (q *recordingAuditDebugQueue) auditIDs() []string { + ids := make([]string, 0, len(q.events)) + for _, event := range q.events { + ids = append(ids, string(event.AuditID)) + } + return ids +} + type fakeAuditJoiner struct { decision AuditJoinDecision err error @@ -249,9 +273,7 @@ func TestAuditHandler_ServeHTTP(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: "/tmp/audit-events", - }) + handler, err := NewAuditHandler(AuditHandlerConfig{}) require.NoError(t, err) // Create request @@ -267,12 +289,52 @@ func TestAuditHandler_ServeHTTP(t *testing.T) { } } -func TestAuditHandler_extractGVR(t *testing.T) { +func TestAuditHandler_DebugQueueCapturesAllDecodedEventsBeforeProcessing(t *testing.T) { + debugQueue := &recordingAuditDebugQueue{} + handler, err := NewAuditHandler(AuditHandlerConfig{DebugQueue: debugQueue}) + require.NoError(t, err) + + body := `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[` + + `{"kind":"Event","auditID":"request-received","verb":"get","stage":"RequestReceived",` + + `"objectRef":{"resource":"pods","apiVersion":"v1"}},` + + `{"kind":"Event","auditID":"bodyless-update","verb":"update","stage":"ResponseComplete",` + + `"objectRef":{"resource":"configmaps","apiVersion":"v1","namespace":"default","name":"cm-a"}}]}` + req := httptest.NewRequest(http.MethodPost, "/audit-webhook-additional", bytes.NewReader([]byte(body))) + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, []string{"request-received", "bodyless-update"}, debugQueue.auditIDs()) + assert.Equal(t, []string{"additional", "additional"}, debugQueue.sources) +} + +func TestAuditHandler_DebugQueueFailureStopsEventProcessing(t *testing.T) { + queue := &recordingAuditEventQueue{} handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: "/tmp/audit-events", + Queue: queue, + DebugQueue: errorAuditDebugQueue{err: errors.New("debug stream down")}, }) require.NoError(t, err) + body := `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[` + + `{"kind":"Event","auditID":"canonical-event","verb":"create","stage":"ResponseComplete",` + + `"objectRef":{"resource":"configmaps","apiVersion":"v1","namespace":"default","name":"cm-a"},` + + `"responseObject":{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"cm-a"}}}]}` + req := httptest.NewRequest(http.MethodPost, "/audit-webhook", bytes.NewReader([]byte(body))) + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + assert.Equal(t, http.StatusInternalServerError, w.Code) + assert.Contains(t, w.Body.String(), "debug stream down") + assert.Empty(t, queue.events) +} + +func TestAuditHandler_extractGVR(t *testing.T) { + handler, err := NewAuditHandler(AuditHandlerConfig{}) + require.NoError(t, err) + tests := []struct { name string eventJSON string @@ -318,9 +380,7 @@ func TestAuditHandler_extractGVR(t *testing.T) { } func TestAuditHandler_InvalidJSON(t *testing.T) { - handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: "/tmp/audit-events", - }) + handler, err := NewAuditHandler(AuditHandlerConfig{}) require.NoError(t, err) req := httptest.NewRequest(http.MethodPost, "/audit-webhook", bytes.NewReader([]byte("invalid json"))) @@ -332,122 +392,8 @@ func TestAuditHandler_InvalidJSON(t *testing.T) { assert.Contains(t, w.Body.String(), "invalid audit event list") } -func TestAuditHandler_FileDump(t *testing.T) { - handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: "/tmp/audit-events", - }) - require.NoError(t, err) - - // 1. Read the YAML file - yamlContent, err := os.ReadFile("testdata/audit-events/config-update.yaml") - require.NoError(t, err) - - // 2. Unmarshal into the v1 Event struct - // The YAML file includes proper TypeMeta (kind/apiVersion) for Kubernetes consistency - var event auditv1.Event - err = yaml.Unmarshal(yamlContent, &event) - require.NoError(t, err) - - // 3. Create the EventList struct - eventList := auditv1.EventList{ - TypeMeta: metav1.TypeMeta{ - Kind: "EventList", - APIVersion: "audit.k8s.io/v1", - }, - Items: []auditv1.Event{event}, - } - - // 4. Marshal the whole thing to JSON - // This guarantees perfect K8s JSON structure - body, err := json.Marshal(eventList) - require.NoError(t, err) - - req := httptest.NewRequest(http.MethodPost, "/audit-webhook", bytes.NewReader(body)) - w := httptest.NewRecorder() - - // Call handler - handler.ServeHTTP(w, req) - - // Verify successful processing - assert.Equal(t, http.StatusOK, w.Code) - - // Check that the file was created and contains valid YAML - filePath := "/tmp/audit-events/89e50d9e-7963-4836-87ab-a18685930369.yaml" - fileContent, err := os.ReadFile(filePath) - require.NoError(t, err, "File should be created successfully") - - // Verify the file contains valid YAML that can be unmarshaled back to audit.Event - var dumpedEvent audit.Event - err = yaml.Unmarshal(fileContent, &dumpedEvent) - require.NoError(t, err, "File content should be valid audit.Event YAML") - - // Verify the auditID matches the actual value from the YAML file - assert.Equal(t, "89e50d9e-7963-4836-87ab-a18685930369", string(dumpedEvent.AuditID), "AuditID should match") - - // Verify key fields are preserved (from the actual YAML file) - assert.Equal(t, "patch", dumpedEvent.Verb) - assert.Equal(t, "system:admin", dumpedEvent.User.Username) - assert.Equal(t, "configmaps", dumpedEvent.ObjectRef.Resource) - - // Clean up file - err = os.Remove(filePath) - require.NoError(t, err, "File cleanup should succeed") - - // Test that events with empty auditID are properly rejected - t.Run("empty auditID should not create file", func(t *testing.T) { - os.RemoveAll("/tmp/audit-events") - handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: "/tmp/audit-events", - }) - require.NoError(t, err) - - // Create proper event with empty auditID - event := auditv1.Event{ - TypeMeta: metav1.TypeMeta{ - Kind: "Event", - APIVersion: "audit.k8s.io/v1", - }, - AuditID: "", - Verb: "create", - } - event.User.Username = "test-user" - event.ObjectRef = &auditv1.ObjectReference{ - Resource: "configmaps", - APIVersion: "v1", - } - - eventList := auditv1.EventList{ - TypeMeta: metav1.TypeMeta{ - Kind: "EventList", - APIVersion: "audit.k8s.io/v1", - }, - Items: []auditv1.Event{event}, - } - - eventJSON, err := json.Marshal(eventList) - require.NoError(t, err) - - req := httptest.NewRequest(http.MethodPost, "/audit-webhook", bytes.NewReader(eventJSON)) - w := httptest.NewRecorder() - - // Call handler - handler.ServeHTTP(w, req) - - // Verify that empty auditID returns 500 error (from processEvents) - assert.Equal(t, http.StatusInternalServerError, w.Code) - assert.Contains(t, w.Body.String(), "invalid audit event: auditID cannot be empty") - - // Verify that no file was created for empty auditID - emptyAuditIDFile := "/tmp/audit-events/.yaml" - _, statErr := os.Stat(emptyAuditIDFile) - assert.True(t, os.IsNotExist(statErr), "File should not be created for empty auditID") - }) -} - func TestAuditHandler_validateEvent(t *testing.T) { - handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: "/tmp/audit-events", - }) + handler, err := NewAuditHandler(AuditHandlerConfig{}) require.NoError(t, err) tests := []struct { @@ -476,7 +422,7 @@ func TestAuditHandler_validateEvent(t *testing.T) { expectedProcessed: true, }, { - name: "valid status event", + name: "status subresource event", event: audit.Event{ AuditID: "some-status", Verb: "update", @@ -494,6 +440,19 @@ func TestAuditHandler_validateEvent(t *testing.T) { expectedErr: "", expectedProcessed: false, }, + { + name: "exec subresource event", + event: audit.Event{ + AuditID: "some-exec", + Verb: "create", + ObjectRef: &audit.ObjectReference{ + Resource: "pods", + Subresource: "exec", + }, + }, + expectedErr: "", + expectedProcessed: false, + }, { name: "empty auditID", event: audit.Event{ @@ -575,10 +534,7 @@ func TestAuditHandler_ReadYAMLToJSON(t *testing.T) { } func TestAuditHandler_RejectsOversizedBody(t *testing.T) { - handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: "/tmp/audit-events", - MaxRequestBodyBytes: 32, - }) + handler, err := NewAuditHandler(AuditHandlerConfig{MaxRequestBodyBytes: 32}) require.NoError(t, err) oversizedBody := `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[]}` @@ -591,60 +547,6 @@ func TestAuditHandler_RejectsOversizedBody(t *testing.T) { assert.Contains(t, w.Body.String(), "request body too large") } -func TestAuditHandler_BodyPresenceControlsDumping(t *testing.T) { - tests := []struct { - name string - body string - expectedStatus int - expectedDumped []string - }{ - { - name: "events with object bodies are dumped", - body: `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[{"kind":"Event","auditID":"bodyful-1","verb":"update","stage":"ResponseComplete","user":{"username":"test-user"},"objectRef":{"resource":"configmaps","namespace":"default","name":"cm-a","apiVersion":"v1"},"responseObject":{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"cm-a","namespace":"default"}}}]}`, - expectedStatus: http.StatusOK, - expectedDumped: []string{"bodyful-1.yaml"}, - }, - { - name: "bodyless non-delete events are ignored", - body: `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[{"kind":"Event","auditID":"bodyless-update-1","verb":"update","stage":"ResponseComplete","user":{"username":"test-user"},"objectRef":{"resource":"configmaps","namespace":"default","name":"cm-a","apiVersion":"v1"}}]}`, - expectedStatus: http.StatusOK, - expectedDumped: nil, - }, - { - name: "bodyless delete events are still dumped", - body: `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[{"kind":"Event","auditID":"bodyless-delete-1","verb":"delete","stage":"ResponseComplete","user":{"username":"test-user"},"objectRef":{"resource":"flunders","namespace":"default","name":"flunder-a","apiVersion":"wardle.example.com/v1alpha1"}}]}`, - expectedStatus: http.StatusOK, - expectedDumped: []string{"bodyless-delete-1.yaml"}, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - dumpDir := t.TempDir() - - handler, err := NewAuditHandler(AuditHandlerConfig{ - DumpDir: dumpDir, - }) - require.NoError(t, err) - - req := httptest.NewRequest(http.MethodPost, "/audit-webhook", bytes.NewReader([]byte(tt.body))) - w := httptest.NewRecorder() - - handler.ServeHTTP(w, req) - - assert.Equal(t, tt.expectedStatus, w.Code) - - entries, err := os.ReadDir(dumpDir) - require.NoError(t, err) - require.Len(t, entries, len(tt.expectedDumped)) - for i, expectedFile := range tt.expectedDumped { - assert.Equal(t, expectedFile, entries[i].Name()) - assert.FileExists(t, filepath.Join(dumpDir, expectedFile)) - } - }) - } -} - func TestAuditHandler_EnqueueFailureReturnsInternalServerError(t *testing.T) { handler, err := NewAuditHandler(AuditHandlerConfig{ Queue: errorAuditEventQueue{err: errors.New("queue down")}, @@ -1015,6 +917,78 @@ func TestAuditHandler_SuccessfulUpdateStillReachesGit(t *testing.T) { assert.Equal(t, []string{"helmrelease-ok-1"}, queue.auditIDs()) } +func TestAuditHandler_PodExecCreateDoesNotEnterJoinPipeline(t *testing.T) { + queue := &recordingAuditEventQueue{} + joiner := &fakeAuditJoiner{decision: AuditJoinDecision{Action: AuditJoinActionEmit}} + handler, err := NewAuditHandler(AuditHandlerConfig{Queue: queue, Joiner: joiner}) + require.NoError(t, err) + + body := `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[ +{ + "level": "RequestResponse", + "auditID": "3df193b2-b83e-4375-a0c2-67ee0c045404", + "stage": "ResponseComplete", + "requestURI": "/api/v1/namespaces/cozy-kubeovn/pods/ovn-central-7955dc78d8-lvwh4/exec?command=ovsdb-client&command=query&command=unix%3A%2Fvar%2Frun%2Fovn%2Fovnsb_db.sock&command=%5B%22_Server%22%2C%7B%22op%22%3A%22select%22%2C%22table%22%3A%22Database%22%2C%22where%22%3A%5B%5B%22name%22%2C%22%3D%3D%22%2C%22OVN_Southbound%22%5D%5D%2C%22columns%22%3A%5B%22leader%22%2C%22connected%22%2C%22cid%22%2C%22sid%22%2C%22index%22%5D%7D%5D&container=ovn-central&stderr=true&stdout=true", + "verb": "create", + "user": { + "username": "system:serviceaccount:cozy-kubeovn:kube-ovn-plunger", + "uid": "c0ad9728-52c2-426b-817b-01c5f9e49eb7", + "groups": [ + "system:serviceaccounts", + "system:serviceaccounts:cozy-kubeovn", + "system:authenticated" + ], + "extra": { + "authentication.kubernetes.io/credential-id": [ + "JTI=46b362c9-c8e0-4295-aa6c-431298f08e6b" + ], + "authentication.kubernetes.io/node-name": [ + "talos-c1194" + ], + "authentication.kubernetes.io/node-uid": [ + "98394f25-77d1-42a4-be30-2b189366cf26" + ], + "authentication.kubernetes.io/pod-name": [ + "kube-ovn-plunger-9b759b798-x58r8" + ], + "authentication.kubernetes.io/pod-uid": [ + "6a021464-f96a-42be-8249-71b8ad212ece" + ] + } + }, + "sourceIPs": [ + "10.244.0.215" + ], + "userAgent": "Go-http-client/1.1", + "objectRef": { + "resource": "pods", + "namespace": "cozy-kubeovn", + "name": "ovn-central-7955dc78d8-lvwh4", + "apiVersion": "v1", + "subresource": "exec" + }, + "responseStatus": { + "metadata": {}, + "code": 101 + }, + "requestReceivedTimestamp": "2026-05-22T19:29:23.082656Z", + "stageTimestamp": "2026-05-22T19:29:23.093528Z", + "annotations": { + "authorization.k8s.io/decision": "allow", + "authorization.k8s.io/reason": "RBAC: allowed by RoleBinding \"kube-ovn-plunger/cozy-kubeovn\" of Role \"kube-ovn-plunger\" to ServiceAccount \"kube-ovn-plunger/cozy-kubeovn\"" + } +} +]}` + req := httptest.NewRequest(http.MethodPost, "/audit-webhook", bytes.NewReader([]byte(body))) + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Zero(t, joiner.calls, "pods/exec is a streaming subresource, not a pod create") + assert.Empty(t, queue.events, "pods/exec must not enter the canonical Git audit stream") +} + // TestClassifyAuditIngress_RejectsFailedRequests pins the intrinsic gate's // verdict on responseStatus.code: any non-success code (>= 300) is rejected as // a failed request, while a missing/zero code and 2xx codes pass. diff --git a/internal/webhook/audit_metrics_test.go b/internal/webhook/audit_metrics_test.go index 27cd4eb2..823509d2 100644 --- a/internal/webhook/audit_metrics_test.go +++ b/internal/webhook/audit_metrics_test.go @@ -48,6 +48,15 @@ const validCreateEventList = `{"kind":"EventList","apiVersion":"audit.k8s.io/v1" // emptyEventList decodes to zero items. const emptyEventList = `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[]}` +// subresourceExecEventList is a one-item official EventList for a pods/exec +// streaming request — audited as verb=create with a non-empty objectRef +// subresource and no resource body. +const subresourceExecEventList = `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[` + + `{"kind":"Event","level":"RequestResponse","auditID":"subres-exec-1","stage":"ResponseComplete",` + + `"verb":"create","user":{"username":"test-user"},` + + `"objectRef":{"resource":"pods","namespace":"default","name":"p","apiVersion":"v1","subresource":"exec"},` + + `"responseStatus":{"code":101}}]}` + // processErrorEventList decodes but fails processing: an event with an empty // auditID is rejected by checkEvent. const processErrorEventList = `{"kind":"EventList","apiVersion":"audit.k8s.io/v1","items":[` + @@ -149,3 +158,37 @@ func TestServeHTTP_EventListIngressMetrics(t *testing.T) { }) } } + +// TestServeHTTP_ReceivedMetricCarriesSubresource confirms audit_events_received_total +// labels objectRef.subresource, so a pods/exec flood is distinguishable from +// real pod mutations rather than collapsing into resource="pods". +func TestServeHTTP_ReceivedMetricCarriesSubresource(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + + handler, err := NewAuditHandler(AuditHandlerConfig{}) + require.NoError(t, err) + + // One top-level configmap create, one pods/exec streaming subresource. + for _, body := range []string{validCreateEventList, subresourceExecEventList} { + req := httptest.NewRequest(http.MethodPost, "/audit-webhook", bytes.NewReader([]byte(body))) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + require.Equal(t, http.StatusOK, w.Code) + } + + const receivedMetric = "gitopsreverser_audit_events_received_total" + + exec, ok := telemetry.CollectInt64Sum(reader, receivedMetric, map[string]string{ + "resource": "pods", "subresource": "exec", "verb": "create", + }) + require.True(t, ok, "expected a subresource=exec received sample") + assert.Equal(t, int64(1), exec) + + // A top-level resource carries an empty subresource label. + top, ok := telemetry.CollectInt64Sum(reader, receivedMetric, map[string]string{ + "resource": "configmaps", "subresource": "", "verb": "create", + }) + require.True(t, ok, `expected a subresource="" received sample for a top-level resource`) + assert.Equal(t, int64(1), top) +} diff --git a/test/e2e/Taskfile.yml b/test/e2e/Taskfile.yml index 7fed8f7f..1852151c 100644 --- a/test/e2e/Taskfile.yml +++ b/test/e2e/Taskfile.yml @@ -671,9 +671,18 @@ tasks: cmds: - mkdir -p "{{.IS}}" - | + GIT_COMMIT="$(git rev-parse --short HEAD 2>/dev/null || echo unknown)" + GIT_DIRTY=0 + [ -z "$(git status --porcelain 2>/dev/null)" ] || GIT_DIRTY=1 + BUILD_DATE="$(date -u +%Y-%m-%dT%H:%M:%SZ)" + VERSION="$(git describe --tags --always --dirty 2>/dev/null || echo dev)" {{.CONTAINER_TOOL}} build \ --build-arg TARGETOS="$(go env GOOS)" \ --build-arg TARGETARCH="$(go env GOARCH)" \ + --build-arg VERSION="${VERSION}" \ + --build-arg GIT_COMMIT="${GIT_COMMIT}" \ + --build-arg GIT_DIRTY="${GIT_DIRTY}" \ + --build-arg BUILD_DATE="${BUILD_DATE}" \ -t "{{.E2E_LOCAL_IMAGE}}" \ . {{.CONTAINER_TOOL}} inspect --format='{{"{{"}}.Id{{"}}"}}' "{{.E2E_LOCAL_IMAGE}}" > "{{.IS}}/controller.id"