From 924239d0070b04d30660fafb2d843de6db4dc753 Mon Sep 17 00:00:00 2001 From: Nathan Oyler Date: Fri, 2 Oct 2026 07:45:08 -0700 Subject: [PATCH] audittools: add backing store to buffer events while rabbitmq is down --- .github/workflows/ci.yaml | 6 +- .gitignore | 1 + Makefile | 2 +- audittools/auditor.go | 109 +++++++ audittools/auditor_backing_store_test.go | 109 +++++++ audittools/backing_store.go | 90 ++++++ audittools/backing_store_file.go | 320 ++++++++++++++++++++ audittools/backing_store_memory.go | 87 ++++++ audittools/backing_store_sql.go | 244 +++++++++++++++ audittools/backing_store_sql_test.go | 212 +++++++++++++ audittools/backing_store_test.go | 366 +++++++++++++++++++++++ audittools/rabbitmq.go | 45 ++- audittools/trail.go | 141 ++++++--- audittools/trail_test.go | 212 +++++++++++++ go.mod | 1 + go.sum | 2 + shell.nix | 1 + 17 files changed, 1905 insertions(+), 43 deletions(-) create mode 100644 audittools/auditor_backing_store_test.go create mode 100644 audittools/backing_store.go create mode 100644 audittools/backing_store_file.go create mode 100644 audittools/backing_store_memory.go create mode 100644 audittools/backing_store_sql.go create mode 100644 audittools/backing_store_sql_test.go create mode 100644 audittools/backing_store_test.go create mode 100644 audittools/trail_test.go diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 64b6a5ae..6f2eecf9 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -72,7 +72,11 @@ jobs: check-latest: true go-version: 1.27.1 - name: Run tests and generate coverage report - run: make build/cover.out + run: | + sudo /usr/share/postgresql-common/pgdg/apt.postgresql.org.sh -y + sudo apt-get install -y --no-install-recommends postgresql-18 + export PATH=/usr/lib/postgresql/18/bin:$PATH + make build/cover.out - name: Archive code coverage results uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 with: diff --git a/.gitignore b/.gitignore index 8cf79abf..1a6cc0da 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ # test artifacts /build/ *.actual +/.testdb diff --git a/Makefile b/Makefile index f71a5f8f..21b7d0aa 100644 --- a/Makefile +++ b/Makefile @@ -101,7 +101,7 @@ run-typos: FORCE install-typos build/cover.out: FORCE | build @printf "\e[1;36m>> Running tests\e[0m\n" - @env $(GO_TESTENV) go test -shuffle=on -coverprofile=build/coverprofile.out $(GO_BUILDFLAGS) -ldflags '-s -w -X github.com/sapcc/go-api-declarations/bininfo.binName=go-bits -X github.com/sapcc/go-api-declarations/bininfo.version=$(BININFO_VERSION) -X github.com/sapcc/go-api-declarations/bininfo.commit=$(BININFO_COMMIT_HASH) -X github.com/sapcc/go-api-declarations/bininfo.buildDate=$(BININFO_BUILD_DATE) $(GO_LDFLAGS)' -covermode=count -coverpkg=$(subst $(space),$(comma),$(GO_COVERPKGS)) $(GO_TESTFLAGS) $(GO_TESTPKGS) + @env $(GO_TESTENV) go test -shuffle=on -p 1 -coverprofile=build/coverprofile.out $(GO_BUILDFLAGS) -ldflags '-s -w -X github.com/sapcc/go-api-declarations/bininfo.binName=go-bits -X github.com/sapcc/go-api-declarations/bininfo.version=$(BININFO_VERSION) -X github.com/sapcc/go-api-declarations/bininfo.commit=$(BININFO_COMMIT_HASH) -X github.com/sapcc/go-api-declarations/bininfo.buildDate=$(BININFO_BUILD_DATE) $(GO_LDFLAGS)' -covermode=count -coverpkg=$(subst $(space),$(comma),$(GO_COVERPKGS)) $(GO_TESTFLAGS) $(GO_TESTPKGS) @awk < build/coverprofile.out '$$1 != "mode:" { is_filename[$$1] = true; counts1[$$1]+=$$2; counts2[$$1]+=$$3 } END { for (filename in is_filename) { printf "%s %d %d\n", filename, counts1[filename], counts2[filename]; } }' | sort | $(SED) '1s/^/mode: count\n/' > $@ build/cover.html: build/cover.out diff --git a/audittools/auditor.go b/audittools/auditor.go index 295c2837..ce218bac 100644 --- a/audittools/auditor.go +++ b/audittools/auditor.go @@ -7,6 +7,63 @@ a RabbitMQ server (with sane defaults) and publishing audit messages in the CADF Events are serialized into the subset of CADF that Hermes (https://github.com/sapcc/hermes) can consume. To use this library, build an Auditor object, then use its Record() function to push events. + +# Backing store + +Events that cannot be published right away (e.g. while RabbitMQ is unreachable) are put in a backing store, +from which publishing is retried once per minute. By default, the backing store keeps events in memory without limit, +so they are lost when the process exits. + +To keep events across restarts, the application can offer more types of backing store in AuditorOpts.BackingStoreFactories. +The operator then selects one through the environment variable "${PREFIX}_BACKING_STORE", +which contains a JSON payload like `{"type":"","params":{...}}`. +For example, an application with a PostgreSQL database could do: + + auditor, err := audittools.NewAuditor(ctx, audittools.AuditorOpts{ + EnvPrefix: "MYSERVICE_AUDIT_RABBITMQ", + Observer: observer, + BackingStoreFactories: map[string]audittools.BackingStoreFactory{ + "sql": audittools.SQLBackingStoreFactoryWithPostgresDB(db), + }, + }) + + # in the environment of the process + MYSERVICE_AUDIT_RABBITMQ_BACKING_STORE='{"type":"sql"}' + +The type "memory" (see NewInMemoryBackingStore) is always available. +Applications that do not use EnvPrefix can build a BackingStore themselves and put it in AuditorOpts.BackingStore. + +Besides the memory store, this package provides two types of backing store. +SQLBackingStoreFactoryWithPostgresDB is the right choice for applications that already have a PostgreSQL database. +NewFileBackingStore keeps events in files, and is only useful if each replica has its own persistent volume, +because files on the container filesystem are lost when the pod is replaced. + +If the backing store is full or cannot be written to, no further events are taken until that event can be published or stored, +so Record() blocks once its small buffer has filled up. +Events that Record() has accepted, but that were neither published nor written to the backing store yet +(at most 21: the 20 events in its buffer and the one being processed) only exist in memory, +so they are lost when the process exits, even with a durable backing store. +Events from the backing store may be published out of order, and may be published twice +if the process exits after publishing them, but before removing them from the backing store, +or if removing them from the backing store fails. + +# SQL backing store + +The SQL backing store keeps events in a table like the one below. +The application must create this table, e.g. in its schema migrations (see pgruntime.ConnectionBehavior). +On startup, the SQL backing store only checks that the table exists. + + CREATE TABLE audit_events ( + id BIGSERIAL NOT NULL PRIMARY KEY, + event_data JSONB NOT NULL + ); + +Several processes can share one table, e.g. multiple replicas of the same API. +Each stored event is then published by only one of them. +But processes that publish to different RabbitMQ queues must use different tables, +because the table does not record which queue an event belongs to. +The gauge "audittools_backing_store_events" counts all events in the table, +so every process that shares the table reports the same value. */ package audittools @@ -15,8 +72,10 @@ import ( "encoding/json" "errors" "fmt" + "maps" "net" "net/url" + "slices" "strconv" "strings" @@ -52,6 +111,7 @@ type AuditorOpts struct { // - "${PREFIX}_USERNAME" (defaults to "guest") // - "${PREFIX}_PASSWORD" (defaults to "guest") // - "${PREFIX}_QUEUE_NAME" (required) + // - "${PREFIX}_BACKING_STORE" (defaults to `{"type":"memory"}`, see "Backing store" in the package documentation) EnvPrefix string // Required if EnvPrefix is empty, ignored otherwise. @@ -63,7 +123,21 @@ type AuditorOpts struct { // The following metrics are registered: // - "audittools_successful_submissions" (counter, no labels) // - "audittools_failed_submissions" (counter, no labels) + // The backing stores in this package also register these metrics with this registry: + // - "audittools_backing_store_writes_total" (counter, no labels) + // - "audittools_backing_store_reads_total" (counter, no labels) + // - "audittools_backing_store_errors_total" (counter, label "operation") + // - "audittools_backing_store_events" (gauge, no labels; memory and SQL backing stores only) + // - "audittools_backing_store_size_bytes" and "audittools_backing_store_files" (gauges, no labels; file backing store only) Registry prometheus.Registerer + + // Optional. If given, events that cannot be published right away are put in this BackingStore. + // Otherwise, the backing store is chosen through "${PREFIX}_BACKING_STORE" (see "Backing store" in the package documentation). + BackingStore BackingStore + + // Optional. More types of backing store that can be chosen through "${PREFIX}_BACKING_STORE", keyed by type name. + // The type "memory" (NewInMemoryBackingStore) is always available. + BackingStoreFactories map[string]BackingStoreFactory } func (opts AuditorOpts) getConnectionOptions() (rabbitURL url.URL, queueName string, err error) { @@ -145,11 +219,19 @@ func NewAuditor(ctx context.Context, opts AuditorOpts) (Auditor, error) { if err != nil { return nil, err } + backingStore := opts.BackingStore + if backingStore == nil { + backingStore, err = opts.newBackingStoreFromEnv() + if err != nil { + return nil, err + } + } eventChan := make(chan cadf.Event, 20) go auditTrail{ EventSink: eventChan, OnSuccessfulPublish: func() { successCounter.Inc() }, OnFailedPublish: func() { failureCounter.Inc() }, + BackingStore: backingStore, }.Commit(ctx, rabbitURL, queueName) return &standardAuditor{ @@ -158,6 +240,33 @@ func NewAuditor(ctx context.Context, opts AuditorOpts) (Auditor, error) { }, nil } +func (opts AuditorOpts) newBackingStoreFromEnv() (BackingStore, error) { + configJSON := `{"type":"memory"}` + if opts.EnvPrefix != "" { + configJSON = osext.GetenvOrDefault(opts.EnvPrefix+"_BACKING_STORE", configJSON) + } + var cfg struct { + Type string `json:"type"` + Params json.RawMessage `json:"params"` + } + err := unmarshalJSONStrict([]byte(configJSON), &cfg) + if err != nil { + return nil, fmt.Errorf("invalid value for %s_BACKING_STORE: %w", opts.EnvPrefix, err) + } + if len(cfg.Params) == 0 { + cfg.Params = json.RawMessage(`{}`) + } + + factories := map[string]BackingStoreFactory{"memory": NewInMemoryBackingStore} + maps.Copy(factories, opts.BackingStoreFactories) + factory, ok := factories[cfg.Type] + if !ok { + return nil, fmt.Errorf("invalid value for %s_BACKING_STORE: unknown backing store type %q (available: %s)", + opts.EnvPrefix, cfg.Type, strings.Join(slices.Sorted(maps.Keys(factories)), ", ")) + } + return factory(cfg.Params, opts) +} + // Record implements the Auditor interface. func (a *standardAuditor) Record(event Event) { a.EventSink <- event.ToCADF(a.Observer.ToCADF()) diff --git a/audittools/auditor_backing_store_test.go b/audittools/auditor_backing_store_test.go new file mode 100644 index 00000000..3416aafe --- /dev/null +++ b/audittools/auditor_backing_store_test.go @@ -0,0 +1,109 @@ +// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "fmt" + "net/http" + "net/http/httptest" + "regexp" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/sapcc/go-api-declarations/cadf" + "go.xyrillian.de/gg/assert" + + "github.com/sapcc/go-bits/must" +) + +func TestBackingStoreFromEnv(t *testing.T) { + opts := AuditorOpts{ + EnvPrefix: "TEST_AUDIT", + Registry: prometheus.NewPedanticRegistry(), + BackingStoreFactories: map[string]BackingStoreFactory{ + "file": NewFileBackingStore, + }, + } + + // without configuration: unlimited memory store, same as before backing stores existed + store := must.ReturnT(opts.newBackingStoreFromEnv())(t) + memoryStore, ok := store.(*inMemoryBackingStore) + assert.Equal(t, ok, true) + assert.Equal(t, memoryStore.maxEvents, 0) + + // with configuration + dir := t.TempDir() + t.Setenv("TEST_AUDIT_BACKING_STORE", fmt.Sprintf(`{"type":"file","params":{"directory":%q}}`, dir)) + opts.Registry = prometheus.NewPedanticRegistry() + store = must.ReturnT(opts.newBackingStoreFromEnv())(t) + fileStore, ok := store.(*fileBackingStore) + assert.Equal(t, ok, true) + assert.Equal(t, fileStore.directory, dir) + + // with invalid configuration + t.Setenv("TEST_AUDIT_BACKING_STORE", `{"type":"sql","params":{}}`) + _, err := opts.newBackingStoreFromEnv() + assert.ErrEqual(t, err, `invalid value for TEST_AUDIT_BACKING_STORE: unknown backing store type "sql" (available: file, memory)`) + + t.Setenv("TEST_AUDIT_BACKING_STORE", `{"type":"memory","param":{}}`) + _, err = opts.newBackingStoreFromEnv() + assert.ErrEqual(t, err, regexp.MustCompile(`unknown field "param"`)) +} + +// recordingBackingStore is a BackingStore that reports each written event on a channel. +type recordingBackingStore struct { + written chan cadf.Event +} + +func (s recordingBackingStore) Write(event cadf.Event) error { + s.written <- event + return nil +} + +func (s recordingBackingStore) ReadBatch() ([]cadf.Event, func(int) error, error) { + return nil, nil, nil +} + +func (s recordingBackingStore) UpdateMetrics() error { + return nil +} + +type testUserInfo struct{} + +func (testUserInfo) AsInitiator(_ cadf.Host) cadf.Resource { return cadf.Resource{} } + +type testTarget struct{} + +func (testTarget) Render() cadf.Resource { + return cadf.Resource{TypeURI: "test/target", ID: "target-id"} +} + +func TestNewAuditorWithConnectionURLAndBackingStore(t *testing.T) { + // applications that set ConnectionURL instead of EnvPrefix can still choose a backing store + store := recordingBackingStore{written: make(chan cadf.Event, 1)} + auditor := must.ReturnT(NewAuditor(t.Context(), AuditorOpts{ + Observer: Observer{TypeURI: "service/test", Name: "test-service", ID: "test-id"}, + ConnectionURL: "amqp://127.0.0.1:1/", // nothing listens there, so publishing fails + QueueName: "test-queue", + Registry: prometheus.NewPedanticRegistry(), + BackingStore: store, + }))(t) + + auditor.Record(Event{ + Time: time.Now(), + Request: httptest.NewRequest(http.MethodGet, "/test", http.NoBody), + User: testUserInfo{}, + ReasonCode: 200, + Action: cadf.CreateAction, + Target: testTarget{}, + }) + + select { + case event := <-store.written: + assert.Equal(t, event.Target.ID, "target-id") + case <-time.After(30 * time.Second): + t.Fatal("event was not written into the backing store") + } +} diff --git a/audittools/backing_store.go b/audittools/backing_store.go new file mode 100644 index 00000000..71bda51e --- /dev/null +++ b/audittools/backing_store.go @@ -0,0 +1,90 @@ +// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "bytes" + "encoding/json" + "errors" + + "github.com/prometheus/client_golang/prometheus" + "github.com/sapcc/go-api-declarations/cadf" +) + +// BackingStore holds audit events that could not be published to RabbitMQ yet. +// See the package documentation for how to configure one. +// +// All methods are called from the single goroutine that publishes events, +// so implementations do not need to be safe for concurrent use. +type BackingStore interface { + // Write adds an event to the store. + // If this fails, the event is kept in memory and retried later. + // Until then, no new events are taken from Auditor.Record(), which will block once its small buffer is full. + Write(event cadf.Event) error + + // ReadBatch returns the oldest events in the store without removing them. + // If the store is empty, all return values are nil. + // + // The caller tries to publish the events in order, and then calls commit with the number of events that were published. + // commit removes those events from the store. The remaining events will be returned again by a later ReadBatch. + // commit is always called before the next call to ReadBatch. + // + // Stored entries that cannot be decoded are not included in events. + // They are removed by commit once all events in the batch were published. + // If a batch consists only of such entries, events is empty, but commit is not nil. + ReadBatch() (events []cadf.Event, commit func(published int) error, err error) + + // UpdateMetrics updates Prometheus metrics describing the state of the backing store (e.g. size, file count), if there are any. + // It is called once per minute. + UpdateMetrics() error +} + +// BackingStoreFactory builds a BackingStore from the "params" section of its configuration (see AuditorOpts.BackingStoreFactories). +// If the BackingStore has its own Prometheus metrics, it shall register them with opts.Registry, +// or with the default registry if opts.Registry is nil. +type BackingStoreFactory func(params json.RawMessage, opts AuditorOpts) (BackingStore, error) + +// errBackingStoreFull is returned by BackingStore.Write when the store has reached its configured size limit. +var errBackingStoreFull = errors.New("audittools: backing store is full") + +// unmarshalJSONStrict is like json.Unmarshal, but rejects unknown fields to catch typos in the configuration. +func unmarshalJSONStrict(buf []byte, target any) error { + dec := json.NewDecoder(bytes.NewReader(buf)) + dec.DisallowUnknownFields() + return dec.Decode(target) +} + +// backingStoreMetrics are the metrics that all BackingStore implementations in this package have in common. +type backingStoreMetrics struct { + Writes prometheus.Counter + Reads prometheus.Counter + Errors *prometheus.CounterVec +} + +// newBackingStoreMetrics builds and registers the common metrics, as well as any additional metrics given by the caller. +func newBackingStoreMetrics(registry prometheus.Registerer, additional ...prometheus.Collector) backingStoreMetrics { + m := backingStoreMetrics{ + Writes: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "audittools_backing_store_writes_total", + Help: "Number of audit events written to the backing store.", + }), + Reads: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "audittools_backing_store_reads_total", + Help: "Number of audit events that were published from the backing store and removed from it.", + }), + Errors: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "audittools_backing_store_errors_total", + Help: `Number of failed backing store operations. The "read" operation also counts stored entries that cannot be decoded.`, + }, []string{"operation"}), + } + for _, op := range []string{"write", "read", "commit"} { + m.Errors.WithLabelValues(op).Add(0) + } + + if registry == nil { + registry = prometheus.DefaultRegisterer + } + registry.MustRegister(append([]prometheus.Collector{m.Writes, m.Reads, m.Errors}, additional...)...) + return m +} diff --git a/audittools/backing_store_file.go b/audittools/backing_store_file.go new file mode 100644 index 00000000..ea373f8c --- /dev/null +++ b/audittools/backing_store_file.go @@ -0,0 +1,320 @@ +// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "slices" + "strings" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/sapcc/go-api-declarations/cadf" + "go.xyrillian.de/gg/option" + + "github.com/sapcc/go-bits/logg" +) + +// NewFileBackingStore is a BackingStoreFactory for a BackingStore that keeps events in files in a local directory. +// To keep events across restarts, the directory must be on a persistent volume. +// The directory must not be shared with other processes. Params: +// +// { +// "directory": "/var/cache/audit", // required +// "max_file_size": 10485760, // optional; once a file is this large, a new file is started (default: 10 MiB) +// "max_total_size": 1073741824 // optional; default: 0 (unlimited) +// } +// +// Each file is read and published as one batch. +// If max_total_size is set and reached, Auditor.Record() blocks until RabbitMQ is reachable again. +// Lines that cannot be decoded are moved into a file with the suffix ".corrupt" for manual inspection. +func NewFileBackingStore(params json.RawMessage, opts AuditorOpts) (BackingStore, error) { + var cfg struct { + Directory string `json:"directory"` + MaxFileSize option.Option[int64] `json:"max_file_size"` + MaxTotalSize option.Option[int64] `json:"max_total_size"` + } + err := unmarshalJSONStrict(params, &cfg) + if err != nil { + return nil, fmt.Errorf("audittools: cannot parse params for file backing store: %w", err) + } + if cfg.Directory == "" { + return nil, errors.New("audittools: missing required param for file backing store: directory") + } + + s := &fileBackingStore{ + directory: cfg.Directory, + maxFileSize: cfg.MaxFileSize.UnwrapOr(10 << 20), + maxTotalSize: cfg.MaxTotalSize.UnwrapOr(0), + sizeGauge: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "audittools_backing_store_size_bytes", + Help: "Total size of the files in the backing store.", + }), + filesGauge: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "audittools_backing_store_files", + Help: "Number of files in the backing store.", + }), + } + if s.maxFileSize <= 0 { + return nil, fmt.Errorf("audittools: invalid max_file_size for file backing store: %d", s.maxFileSize) + } + if s.maxTotalSize < 0 { + return nil, fmt.Errorf("audittools: invalid max_total_size for file backing store: %d", s.maxTotalSize) + } + + // audit events can contain sensitive data + err = os.MkdirAll(s.directory, 0700) + if err != nil { + return nil, fmt.Errorf("audittools: cannot create directory for file backing store: %w", err) + } + err = os.Chmod(s.directory, 0700) // in case it already existed + if err != nil { + return nil, fmt.Errorf("audittools: cannot chmod directory for file backing store: %w", err) + } + + err = s.UpdateMetrics() + if err != nil { + return nil, err + } + s.metrics = newBackingStoreMetrics(opts.Registry, s.sizeGauge, s.filesGauge) + return s, nil +} + +// fileBackingStore is the BackingStore built by NewFileBackingStore. +// Events are stored as one JSON document per line. +type fileBackingStore struct { + directory string + maxFileSize int64 + maxTotalSize int64 + + currentFile string // where Write() appends to (empty if a new file shall be started) + currentFileSize int64 + totalSize int64 + + metrics backingStoreMetrics + sizeGauge prometheus.Gauge + filesGauge prometheus.Gauge +} + +// Write implements the BackingStore interface. +func (s *fileBackingStore) Write(event cadf.Event) error { + if s.maxTotalSize > 0 && s.totalSize >= s.maxTotalSize { + s.metrics.Errors.WithLabelValues("write").Inc() + return fmt.Errorf("%w (max_total_size = %d)", errBackingStoreFull, s.maxTotalSize) + } + + buf, err := json.Marshal(event) + if err != nil { + s.metrics.Errors.WithLabelValues("write").Inc() + return fmt.Errorf("audittools: cannot serialize event for file backing store: %w", err) + } + buf = append(buf, '\n') + + isNewFile := s.currentFile == "" || s.currentFileSize >= s.maxFileSize + if isNewFile { + s.currentFile = filepath.Join(s.directory, fmt.Sprintf("audit-events-%d.jsonl", time.Now().UnixNano())) + s.currentFileSize = 0 + } + err = appendToFile(s.currentFile, buf) + if err == nil && isNewFile { + err = syncDir(s.directory) + } + if err != nil { + // the file may now end in a partial line, so do not append further events to it + s.currentFile = "" + s.metrics.Errors.WithLabelValues("write").Inc() + return fmt.Errorf("audittools: cannot write to file backing store: %w", err) + } + + s.currentFileSize += int64(len(buf)) + s.totalSize += int64(len(buf)) + s.metrics.Writes.Inc() + return nil +} + +// ReadBatch implements the BackingStore interface. +func (s *fileBackingStore) ReadBatch() ([]cadf.Event, func(int) error, error) { + files, _, err := s.listFiles() + if err != nil { + s.metrics.Errors.WithLabelValues("read").Inc() + return nil, nil, err + } + if len(files) == 0 { + return nil, nil, nil + } + path := files[0] + if path == s.currentFile { + // commit will remove or replace this file, so do not append to it anymore + s.currentFile = "" + } + + contents, err := os.ReadFile(path) + if err != nil { + s.metrics.Errors.WithLabelValues("read").Inc() + return nil, nil, fmt.Errorf("audittools: cannot read from file backing store: %w", err) + } + var ( + events []cadf.Event + eventLines [][]byte // the lines from which `events` were decoded, in the same order + corruptLines [][]byte + ) + for line := range bytes.Lines(contents) { + line = bytes.TrimSuffix(line, []byte("\n")) + if len(bytes.TrimSpace(line)) == 0 { + continue + } + var event cadf.Event + if json.Unmarshal(line, &event) != nil { + corruptLines = append(corruptLines, line) + continue + } + events = append(events, event) + eventLines = append(eventLines, line) + } + + commit := func(published int) error { + err := s.commit(path, published, eventLines, corruptLines) + if err != nil { + s.metrics.Errors.WithLabelValues("commit").Inc() + return err + } + if published > 0 || len(events) == 0 { + s.totalSize -= int64(len(contents)) + if published < len(events) { + s.totalSize += int64(len(joinLines(eventLines[published:], corruptLines))) + } + } + s.metrics.Reads.Add(float64(published)) + return nil + } + return events, commit, nil +} + +func (s *fileBackingStore) commit(path string, published int, eventLines, corruptLines [][]byte) error { + switch { + case published == 0 && len(eventLines) > 0: + return nil + case published < len(eventLines): + // keep what was not published yet for the next ReadBatch + return writeFileAtomically(path, joinLines(eventLines[published:], corruptLines)) + case len(corruptLines) > 0: + logg.Error("audittools: moving %d lines that cannot be decoded as audit events into %s.corrupt", len(corruptLines), path) + s.metrics.Errors.WithLabelValues("read").Add(float64(len(corruptLines))) + err := writeFileAtomically(path+".corrupt", joinLines(corruptLines)) + if err != nil { + return err + } + return os.Remove(path) + default: + return os.Remove(path) + } +} + +// UpdateMetrics implements the BackingStore interface. +func (s *fileBackingStore) UpdateMetrics() error { + files, totalSize, err := s.listFiles() + if err != nil { + return err + } + s.totalSize = totalSize // in case our own bookkeeping has drifted + s.sizeGauge.Set(float64(totalSize)) + s.filesGauge.Set(float64(len(files))) + return nil +} + +// listFiles returns the event files in the order in which they were written, as well as their total size. +func (s *fileBackingStore) listFiles() (files []string, totalSize int64, err error) { + entries, err := os.ReadDir(s.directory) + if err != nil { + return nil, 0, fmt.Errorf("audittools: cannot list files in file backing store: %w", err) + } + for _, entry := range entries { + name := entry.Name() + if entry.IsDir() || !strings.HasPrefix(name, "audit-events-") || !strings.HasSuffix(name, ".jsonl") { + continue + } + files = append(files, filepath.Join(s.directory, name)) + info, err := entry.Info() + if err == nil { + totalSize += info.Size() + } + } + slices.Sort(files) // file names contain a timestamp + return files, totalSize, nil +} + +// joinLines concatenates the given lines with a newline after each one. +func joinLines(lineSets ...[][]byte) []byte { + var buf []byte + for _, lines := range lineSets { + for _, line := range lines { + buf = append(buf, line...) + buf = append(buf, '\n') + } + } + return buf +} + +func appendToFile(path string, buf []byte) (err error) { + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600) + if err != nil { + return err + } + defer func() { + closeErr := f.Close() + if err == nil { + err = closeErr + } + }() + + _, err = f.Write(buf) + if err != nil { + return err + } + // fsync so that the event survives a node crash, not just a process crash + // (for a new file, Write also fsyncs the directory) + return f.Sync() +} + +func writeFileAtomically(path string, buf []byte) (err error) { + tmpPath := path + ".tmp" + f, err := os.OpenFile(tmpPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600) + if err != nil { + return err + } + defer func() { + closeErr := f.Close() + if err == nil { + err = closeErr + } + if err == nil { + err = os.Rename(tmpPath, path) + } + if err != nil { + _ = os.Remove(tmpPath) + return + } + err = syncDir(filepath.Dir(path)) + }() + + _, err = f.Write(buf) + if err != nil { + return err + } + return f.Sync() +} + +// syncDir fsyncs a directory, so that files created in it or renamed into it survive a node crash. +func syncDir(path string) error { + d, err := os.Open(path) + if err != nil { + return err + } + return errors.Join(d.Sync(), d.Close()) +} diff --git a/audittools/backing_store_memory.go b/audittools/backing_store_memory.go new file mode 100644 index 00000000..2560a83d --- /dev/null +++ b/audittools/backing_store_memory.go @@ -0,0 +1,87 @@ +// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "encoding/json" + "fmt" + + "github.com/prometheus/client_golang/prometheus" + "github.com/sapcc/go-api-declarations/cadf" +) + +// NewInMemoryBackingStore is a BackingStoreFactory for a BackingStore that keeps events in memory. +// Events in it are lost when the process exits. +// This is the backing store that NewAuditor uses when no other backing store is configured. +// Params (all optional): +// +// { +// "max_events": 1000 // default: 0 (unlimited) +// } +// +// If max_events is set and the store is full, Auditor.Record() blocks until RabbitMQ is reachable again. +func NewInMemoryBackingStore(params json.RawMessage, opts AuditorOpts) (BackingStore, error) { + var cfg struct { + MaxEvents int `json:"max_events"` + } + err := unmarshalJSONStrict(params, &cfg) + if err != nil { + return nil, fmt.Errorf("audittools: cannot parse params for memory backing store: %w", err) + } + if cfg.MaxEvents < 0 { + return nil, fmt.Errorf("audittools: invalid max_events for memory backing store: %d", cfg.MaxEvents) + } + + s := &inMemoryBackingStore{ + maxEvents: cfg.MaxEvents, + eventsGauge: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "audittools_backing_store_events", + Help: "Number of audit events in the backing store.", + }), + } + s.metrics = newBackingStoreMetrics(opts.Registry, s.eventsGauge) + return s, nil +} + +// inMemoryBackingStore is the BackingStore built by NewInMemoryBackingStore. +type inMemoryBackingStore struct { + maxEvents int + events []cadf.Event + metrics backingStoreMetrics + eventsGauge prometheus.Gauge +} + +// Write implements the BackingStore interface. +func (s *inMemoryBackingStore) Write(event cadf.Event) error { + if s.maxEvents > 0 && len(s.events) >= s.maxEvents { + s.metrics.Errors.WithLabelValues("write").Inc() + return fmt.Errorf("%w (max_events = %d)", errBackingStoreFull, s.maxEvents) + } + s.events = append(s.events, event) + s.metrics.Writes.Inc() + return nil +} + +// ReadBatch implements the BackingStore interface. +func (s *inMemoryBackingStore) ReadBatch() ([]cadf.Event, func(int) error, error) { + count := len(s.events) + if count == 0 { + return nil, nil, nil + } + commit := func(published int) error { + s.events = s.events[published:] + if len(s.events) == 0 { + s.events = nil // release the backing array + } + s.metrics.Reads.Add(float64(published)) + return nil + } + return s.events[:count:count], commit, nil +} + +// UpdateMetrics implements the BackingStore interface. +func (s *inMemoryBackingStore) UpdateMetrics() error { + s.eventsGauge.Set(float64(len(s.events))) + return nil +} diff --git a/audittools/backing_store_sql.go b/audittools/backing_store_sql.go new file mode 100644 index 00000000..551ba763 --- /dev/null +++ b/audittools/backing_store_sql.go @@ -0,0 +1,244 @@ +// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "regexp" + "strconv" + "strings" + + "github.com/prometheus/client_golang/prometheus" + "github.com/sapcc/go-api-declarations/cadf" + "go.xyrillian.de/gg/option" + + "github.com/sapcc/go-bits/logg" + "github.com/sapcc/go-bits/sqlext" +) + +var sqlIdentifierRx = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`) + +// SQLBackingStoreFactoryWithPostgresDB returns a BackingStoreFactory for a BackingStore that keeps events +// in a table in the given PostgreSQL database. Params (all optional): +// +// { +// "table_name": "audit_events", +// "batch_size": 100, // number of events read and published in one transaction +// "max_events": 0 // default: 0 (unlimited) +// } +// +// The application must create the table (see "SQL backing store" in the package documentation). +// If max_events is set and reached, or if the database is unreachable, +// Auditor.Record() blocks until RabbitMQ or the database is reachable again. +func SQLBackingStoreFactoryWithPostgresDB(db *sql.DB) BackingStoreFactory { + return func(params json.RawMessage, opts AuditorOpts) (BackingStore, error) { + var cfg struct { + TableName option.Option[string] `json:"table_name"` + BatchSize option.Option[int] `json:"batch_size"` + MaxEvents option.Option[int] `json:"max_events"` + } + err := unmarshalJSONStrict(params, &cfg) + if err != nil { + return nil, fmt.Errorf("audittools: cannot parse params for SQL backing store: %w", err) + } + + if db == nil { + return nil, errors.New("audittools: SQL backing store requires a database connection") + } + s := &sqlBackingStore{ + db: db, + tableName: cfg.TableName.UnwrapOr("audit_events"), + batchSize: cfg.BatchSize.UnwrapOr(100), + maxEvents: cfg.MaxEvents.UnwrapOr(0), + eventsGauge: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "audittools_backing_store_events", + Help: "Number of audit events in the backing store.", + }), + } + // the table name is concatenated into queries below, so it must be a plain identifier + if !sqlIdentifierRx.MatchString(s.tableName) { + return nil, fmt.Errorf("audittools: invalid table name for SQL backing store: %q", s.tableName) + } + if s.batchSize <= 0 { + return nil, fmt.Errorf("audittools: invalid batch_size for SQL backing store: %d", s.batchSize) + } + + if s.maxEvents < 0 { + return nil, fmt.Errorf("audittools: invalid max_events for SQL backing store: %d", s.maxEvents) + } + + query := `SELECT id, event_data FROM ` + s.tableName + ` LIMIT 0` //nolint:gosec // table name was validated above + _, err = db.Exec(query) + if err != nil { + return nil, fmt.Errorf("audittools: cannot use table %s for SQL backing store (see the package documentation for how to create it): %w", s.tableName, err) + } + + s.metrics = newBackingStoreMetrics(opts.Registry, s.eventsGauge) + return s, nil + } +} + +// sqlBackingStore is the BackingStore built by SQLBackingStoreFactoryWithPostgresDB. +type sqlBackingStore struct { + db *sql.DB + tableName string + batchSize int + maxEvents int + metrics backingStoreMetrics + eventsGauge prometheus.Gauge +} + +// Write implements the BackingStore interface. +func (s *sqlBackingStore) Write(event cadf.Event) error { + err := s.write(event) + if err != nil { + s.metrics.Errors.WithLabelValues("write").Inc() + return err + } + s.metrics.Writes.Inc() + return nil +} + +func (s *sqlBackingStore) write(event cadf.Event) error { + // The count and the insert are not atomic, so with several writers, max_events can be exceeded slightly. + if s.maxEvents > 0 { + count, err := s.countEvents() + if err != nil { + return err + } + if count >= int64(s.maxEvents) { + return fmt.Errorf("%w (max_events = %d)", errBackingStoreFull, s.maxEvents) + } + } + + buf, err := json.Marshal(event) + if err != nil { + return fmt.Errorf("audittools: cannot serialize event for SQL backing store: %w", err) + } + query := `INSERT INTO ` + s.tableName + ` (event_data) VALUES ($1)` //nolint:gosec // table name was validated in the factory + _, err = s.db.Exec(query, buf) + if err != nil { + return fmt.Errorf("audittools: cannot insert into %s: %w", s.tableName, err) + } + return nil +} + +// ReadBatch implements the BackingStore interface. +// +// The rows are locked until commit is called, so other processes sharing the same table will skip them. +func (s *sqlBackingStore) ReadBatch() (events []cadf.Event, commit func(int) error, err error) { + tx, err := s.db.Begin() + if err != nil { + s.metrics.Errors.WithLabelValues("read").Inc() + return nil, nil, fmt.Errorf("audittools: cannot begin transaction for SQL backing store: %w", err) + } + defer func() { + if commit == nil { + rollback(tx) + } + }() + + var ( + eventIDs []int64 // IDs of the rows in `events`, in the same order + corruptIDs []int64 // IDs of rows that cannot be decoded + ) + query := `SELECT id, event_data FROM ` + s.tableName + ` ORDER BY id LIMIT $1 FOR UPDATE SKIP LOCKED` + err = sqlext.ForeachRow(tx, query, []any{s.batchSize}, func(rows *sql.Rows) error { + var ( + id int64 + buf []byte + ) + err := rows.Scan(&id, &buf) + if err != nil { + return err + } + var event cadf.Event + if json.Unmarshal(buf, &event) == nil { + events = append(events, event) + eventIDs = append(eventIDs, id) + } else { + corruptIDs = append(corruptIDs, id) + } + return nil + }) + if err != nil { + s.metrics.Errors.WithLabelValues("read").Inc() + return nil, nil, fmt.Errorf("audittools: cannot read from %s: %w", s.tableName, err) + } + if len(eventIDs) == 0 && len(corruptIDs) == 0 { + return nil, nil, nil + } + + commit = func(published int) error { + idsToDelete := eventIDs[:published:published] + if published == len(events) { + idsToDelete = append(idsToDelete, corruptIDs...) + } + err := s.deleteAndCommit(tx, idsToDelete) + if err != nil { + s.metrics.Errors.WithLabelValues("commit").Inc() + return err + } + s.metrics.Reads.Add(float64(published)) + if published == len(events) && len(corruptIDs) > 0 { + logg.Error("audittools: deleted %d rows from %s that could not be decoded as audit events", len(corruptIDs), s.tableName) + s.metrics.Errors.WithLabelValues("read").Add(float64(len(corruptIDs))) + } + return nil + } + return events, commit, nil +} + +func (s *sqlBackingStore) deleteAndCommit(tx *sql.Tx, ids []int64) error { + if len(ids) > 0 { + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for idx, id := range ids { + placeholders[idx] = "$" + strconv.Itoa(idx+1) + args[idx] = id + } + query := `DELETE FROM ` + s.tableName + ` WHERE id IN (` + strings.Join(placeholders, ",") + `)` //nolint:gosec // table name was validated in the factory + _, err := tx.Exec(query, args...) + if err != nil { + rollback(tx) + return fmt.Errorf("audittools: cannot delete from %s: %w", s.tableName, err) + } + } + err := tx.Commit() + if err != nil { + return fmt.Errorf("audittools: cannot commit transaction for SQL backing store: %w", err) + } + return nil +} + +// UpdateMetrics implements the BackingStore interface. +func (s *sqlBackingStore) UpdateMetrics() error { + count, err := s.countEvents() + if err != nil { + return err + } + s.eventsGauge.Set(float64(count)) + return nil +} + +func (s *sqlBackingStore) countEvents() (int64, error) { + var count int64 + err := s.db.QueryRow(`SELECT COUNT(*) FROM ` + s.tableName).Scan(&count) + if err != nil { + return 0, fmt.Errorf("audittools: cannot count rows in %s: %w", s.tableName, err) + } + return count, nil +} + +// rollback is like sqlext.RollbackUnlessCommitted, but does not log anything on success. +// (Rolling back is the normal outcome of ReadBatch when the table is empty, and this happens once per minute.) +func rollback(tx *sql.Tx) { + err := tx.Rollback() + if err != nil && !errors.Is(err, sql.ErrTxDone) { + logg.Error("audittools: cannot roll back transaction for SQL backing store: %s", err.Error()) + } +} diff --git a/audittools/backing_store_sql_test.go b/audittools/backing_store_sql_test.go new file mode 100644 index 00000000..9e063067 --- /dev/null +++ b/audittools/backing_store_sql_test.go @@ -0,0 +1,212 @@ +// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "database/sql" + "encoding/json" + "fmt" + "regexp" + "sync" + "testing" + + _ "github.com/lib/pq" + "github.com/prometheus/client_golang/prometheus" + "github.com/sapcc/go-api-declarations/cadf" + "go.xyrillian.de/gg/assert" + "go.xyrillian.de/gg/pgruntime" + + "github.com/sapcc/go-bits/must" +) + +// TestMain spawns a PostgreSQL server for the duration of the test run. +func TestMain(m *testing.M) { + pgruntime.WithTestDB(m, m.Run) +} + +// createTableQuery creates a table with the schema from the package documentation. +func createTableQuery(tableName string) string { + return `CREATE TABLE ` + tableName + ` ( + id BIGSERIAL NOT NULL PRIMARY KEY, + event_data JSONB NOT NULL + )` +} + +// connectForTest opens a fresh test database with the table "audit_events", +// which is created by a migration in the same way as in an application. +func connectForTest(t *testing.T) *sql.DB { + t.Helper() + db, _ := pgruntime.StdConnector("postgres").ConnectForTest(t, pgruntime.ConnectionBehavior{ + Migrations: map[int64]string{1: createTableQuery("audit_events")}, + }) + return db.DB +} + +func newTestSQLBackingStore(t *testing.T, db *sql.DB, params string) *sqlBackingStore { + t.Helper() + store := must.ReturnT(SQLBackingStoreFactoryWithPostgresDB(db)(json.RawMessage(params), AuditorOpts{ + Registry: prometheus.NewPedanticRegistry(), + }))(t) + return store.(*sqlBackingStore) +} + +func TestSQLBackingStoreMaxEvents(t *testing.T) { + store := newTestSQLBackingStore(t, connectForTest(t), `{"max_events":3}`) + mustWrite(t, store, "event-1", "event-2", "event-3") + assert.ErrEqual(t, store.Write(testEvent("event-4")), errBackingStoreFull) + + mustReadBatch(t, store, 1) + mustWrite(t, store, "event-4") + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-2", "event-3", "event-4"}) +} + +func TestSQLBackingStoreIsUnlimitedByDefault(t *testing.T) { + store := newTestSQLBackingStore(t, connectForTest(t), `{}`) + assert.Equal(t, store.maxEvents, 0) +} + +func TestSQLBackingStoreBatchSize(t *testing.T) { + store := newTestSQLBackingStore(t, connectForTest(t), `{"batch_size":2}`) + mustWrite(t, store, "event-1", "event-2", "event-3") + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-1", "event-2"}) + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-3"}) +} + +func TestSQLBackingStoreSharedTable(t *testing.T) { + // two replicas of the same service, sharing one table + db := connectForTest(t) + storeA := newTestSQLBackingStore(t, db, `{"batch_size":2}`) + storeB := newTestSQLBackingStore(t, db, `{"batch_size":2}`) + mustWrite(t, storeA, "event-1", "event-2", "event-3") + + // while A holds a batch, B only sees the rows that are not in it + eventsA, commitA, err := storeA.ReadBatch() + must.SucceedT(t, err) + assert.Equal(t, len(eventsA), 2) + assert.Equal(t, mustReadBatch(t, storeB, -1), []string{"event-3"}) + assert.Equal(t, mustReadBatch(t, storeB, -1), []string{}) + + // if A fails to publish, the rows become available again + must.SucceedT(t, commitA(0)) + assert.Equal(t, mustReadBatch(t, storeB, -1), []string{"event-1", "event-2"}) + assert.Equal(t, mustReadBatch(t, storeA, -1), []string{}) +} + +func TestSQLBackingStoreUnreadableRows(t *testing.T) { + db := connectForTest(t) + store := newTestSQLBackingStore(t, db, `{"batch_size":2}`) + // "id" must be a string, so these rows cannot be decoded + for range 2 { + must.ReturnT(db.Exec(`INSERT INTO audit_events (event_data) VALUES ('{"id":42}')`))(t) + } + mustWrite(t, store, "event-1") + + // the first batch consists only of unreadable rows, which must not stop the drain + var published []string + trail := auditTrail{BackingStore: store} + trail.drainBackingStore(func(e *cadf.Event) bool { + published = append(published, e.ID) + return true + }) + assert.Equal(t, published, []string{"event-1"}) + assert.Equal(t, must.ReturnT(store.countEvents())(t), 0) +} + +func TestSQLBackingStoreRequiresTable(t *testing.T) { + db := connectForTest(t) + factory := SQLBackingStoreFactoryWithPostgresDB(db) + params := json.RawMessage(`{"table_name":"my_audit_events"}`) + must.ReturnT(db.Exec(`DROP TABLE IF EXISTS my_audit_events`))(t) // the test DB is reused between runs + + // the table is not created by the backing store + _, err := factory(params, AuditorOpts{Registry: prometheus.NewPedanticRegistry()}) + assert.ErrEqual(t, err, regexp.MustCompile(`cannot use table my_audit_events .* relation "my_audit_events" does not exist`)) + + must.ReturnT(db.Exec(createTableQuery("my_audit_events")))(t) + store := must.ReturnT(factory(params, AuditorOpts{Registry: prometheus.NewPedanticRegistry()}))(t) + mustWrite(t, store, "event-1") + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-1"}) +} + +func TestSQLBackingStoreConcurrentStartup(t *testing.T) { + // several replicas of the same service starting at the same time, sharing one table + db := connectForTest(t) + factory := SQLBackingStoreFactoryWithPostgresDB(db) + stores := make([]BackingStore, 8) + errs := make([]error, len(stores)) + var wg sync.WaitGroup + for idx := range stores { + wg.Go(func() { + stores[idx], errs[idx] = factory(json.RawMessage(`{}`), AuditorOpts{Registry: prometheus.NewPedanticRegistry()}) + }) + } + wg.Wait() + for _, err := range errs { + assert.ErrEqual(t, err, nil) + } + + // all of them work on the same table + for idx, store := range stores { + mustWrite(t, store, fmt.Sprintf("event-%d", idx)) + } + assert.Equal(t, len(mustReadBatch(t, stores[0], -1)), len(stores)) +} + +func TestSQLBackingStoreReopen(t *testing.T) { + // events written before a restart are published by the new process + db := connectForTest(t) + mustWrite(t, newTestSQLBackingStore(t, db, `{}`), "event-1", "event-2") + + registry := prometheus.NewPedanticRegistry() + store := must.ReturnT(SQLBackingStoreFactoryWithPostgresDB(db)(json.RawMessage(`{}`), AuditorOpts{Registry: registry}))(t) + must.SucceedT(t, store.UpdateMetrics()) + assert.Equal(t, gatherMetrics(t, registry)["audittools_backing_store_events"], 2) + + var published []string + auditTrail{BackingStore: store}.drainBackingStore(func(e *cadf.Event) bool { + published = append(published, e.ID) + return true + }) + assert.Equal(t, published, []string{"event-1", "event-2"}) + assert.Equal(t, mustReadBatch(t, store, -1), []string{}) +} + +func TestSQLBackingStoreMetrics(t *testing.T) { + registry := prometheus.NewPedanticRegistry() + store := must.ReturnT(SQLBackingStoreFactoryWithPostgresDB(connectForTest(t))(json.RawMessage(`{"max_events":3}`), AuditorOpts{Registry: registry}))(t) + + mustWrite(t, store, "event-1", "event-2", "event-3") + assert.ErrEqual(t, store.Write(testEvent("event-4")), errBackingStoreFull) + mustReadBatch(t, store, 2) + must.SucceedT(t, store.UpdateMetrics()) + + assert.Equal(t, gatherMetrics(t, registry), map[string]float64{ + "audittools_backing_store_writes_total": 3, + "audittools_backing_store_reads_total": 2, + "audittools_backing_store_errors_total": 1, + "audittools_backing_store_events": 1, + }) +} + +func TestSQLBackingStoreTableNameValidation(t *testing.T) { + db := connectForTest(t) + factory := SQLBackingStoreFactoryWithPostgresDB(db) + + for _, tableName := range []string{"audit_events; DROP TABLE users;", "audit-events", "audit.events", "123_events", ""} { + t.Run("invalid/"+tableName, func(t *testing.T) { + params := fmt.Sprintf(`{"table_name":%q}`, tableName) + _, err := factory(json.RawMessage(params), AuditorOpts{Registry: prometheus.NewPedanticRegistry()}) + assert.ErrEqual(t, err, regexp.MustCompile("invalid table name")) + }) + } + for _, tableName := range []string{"AuditEvents", "_audit_events", "audit_events_123"} { + t.Run("valid/"+tableName, func(t *testing.T) { + must.ReturnT(db.Exec(`DROP TABLE IF EXISTS ` + tableName))(t) // the test DB is reused between runs + must.ReturnT(db.Exec(createTableQuery(tableName)))(t) + params := fmt.Sprintf(`{"table_name":%q}`, tableName) + _, err := factory(json.RawMessage(params), AuditorOpts{Registry: prometheus.NewPedanticRegistry()}) + assert.ErrEqual(t, err, nil) + }) + } +} diff --git a/audittools/backing_store_test.go b/audittools/backing_store_test.go new file mode 100644 index 00000000..0bc6cc30 --- /dev/null +++ b/audittools/backing_store_test.go @@ -0,0 +1,366 @@ +// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "regexp" + "strings" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/sapcc/go-api-declarations/cadf" + "go.xyrillian.de/gg/assert" + + "github.com/sapcc/go-bits/must" +) + +//////////////////////////////////////////////////////////////////////////////// +// shared test infrastructure + +func testEvent(id string) cadf.Event { + return cadf.Event{ + ID: id, + EventType: "activity", + Action: "create", + Outcome: "success", + } +} + +func mustWrite(t *testing.T, store BackingStore, ids ...string) { + t.Helper() + for _, id := range ids { + must.SucceedT(t, store.Write(testEvent(id))) + } +} + +// mustReadBatch reads one batch, and then reports the first `published` events in it as published. +// If `published` is negative, all events are reported as published. +// Returns the IDs of all events in the batch. +func mustReadBatch(t *testing.T, store BackingStore, published int) []string { + t.Helper() + events, commit, err := store.ReadBatch() + must.SucceedT(t, err) + + ids := []string{} + for _, e := range events { + ids = append(ids, e.ID) + } + if commit != nil { + if published < 0 { + published = len(events) + } + must.SucceedT(t, commit(published)) + } + return ids +} + +func newTestMemoryBackingStore(t *testing.T, params string) *inMemoryBackingStore { + t.Helper() + store := must.ReturnT(NewInMemoryBackingStore(json.RawMessage(params), AuditorOpts{ + Registry: prometheus.NewPedanticRegistry(), + }))(t) + return store.(*inMemoryBackingStore) +} + +func newTestFileBackingStore(t *testing.T, params string) *fileBackingStore { + t.Helper() + params = strings.Replace(params, "{", fmt.Sprintf(`{"directory":%q,`, t.TempDir()), 1) + params = strings.Replace(params, ",}", "}", 1) + store := must.ReturnT(NewFileBackingStore(json.RawMessage(params), AuditorOpts{ + Registry: prometheus.NewPedanticRegistry(), + }))(t) + return store.(*fileBackingStore) +} + +func testWithEachTypeOfStore(t *testing.T, action func(*testing.T, BackingStore)) { + t.Run("with memory store", func(t *testing.T) { + action(t, newTestMemoryBackingStore(t, `{}`)) + }) + t.Run("with file store", func(t *testing.T) { + action(t, newTestFileBackingStore(t, `{}`)) + }) + t.Run("with PostgreSQL store", func(t *testing.T) { + action(t, newTestSQLBackingStore(t, connectForTest(t), `{}`)) + }) +} + +//////////////////////////////////////////////////////////////////////////////// +// tests for all types of store + +func TestBackingStoreWriteAndRead(t *testing.T) { + testWithEachTypeOfStore(t, func(t *testing.T, store BackingStore) { + events, commit, err := store.ReadBatch() + assert.ErrEqual(t, err, nil) + assert.Equal(t, len(events), 0) + assert.Equal(t, commit == nil, true) + + mustWrite(t, store, "event-1", "event-2", "event-3") + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-1", "event-2", "event-3"}) + assert.Equal(t, mustReadBatch(t, store, -1), []string{}) + }) +} + +func TestBackingStorePartialCommit(t *testing.T) { + testWithEachTypeOfStore(t, func(t *testing.T, store BackingStore) { + mustWrite(t, store, "event-1", "event-2", "event-3") + + // nothing published -> same batch again + assert.Equal(t, mustReadBatch(t, store, 0), []string{"event-1", "event-2", "event-3"}) + // first event published -> only that one is removed + assert.Equal(t, mustReadBatch(t, store, 1), []string{"event-1", "event-2", "event-3"}) + mustWrite(t, store, "event-4") + ids := mustReadBatch(t, store, -1) + ids = append(ids, mustReadBatch(t, store, -1)...) // the file store has event-4 in a separate file + assert.Equal(t, ids, []string{"event-2", "event-3", "event-4"}) + assert.Equal(t, mustReadBatch(t, store, -1), []string{}) + }) +} + +func TestBackingStoreRejectsUnknownParams(t *testing.T) { + db := connectForTest(t) + factories := map[string]BackingStoreFactory{ + "memory": NewInMemoryBackingStore, + "file": NewFileBackingStore, + "sql": SQLBackingStoreFactoryWithPostgresDB(db), + } + for name, factory := range factories { + t.Run(name, func(t *testing.T) { + params := fmt.Sprintf(`{"directory":%q,"max_event":10}`, t.TempDir()) + if name != "file" { + params = `{"max_event":10}` + } + _, err := factory(json.RawMessage(params), AuditorOpts{Registry: prometheus.NewPedanticRegistry()}) + assert.ErrEqual(t, err, regexp.MustCompile(`unknown field "max_event"`)) + }) + } +} + +func TestBackingStoreRejectsInvalidParams(t *testing.T) { + db := connectForTest(t) + testCases := []struct { + factory BackingStoreFactory + params string + err string + }{ + {NewInMemoryBackingStore, `{"max_events":-1}`, "invalid max_events for memory backing store: -1"}, + {NewFileBackingStore, `{"directory":"$DIR","max_file_size":0}`, "invalid max_file_size for file backing store: 0"}, + {NewFileBackingStore, `{"directory":"$DIR","max_total_size":-1}`, "invalid max_total_size for file backing store: -1"}, + {SQLBackingStoreFactoryWithPostgresDB(db), `{"batch_size":0}`, "invalid batch_size for SQL backing store: 0"}, + {SQLBackingStoreFactoryWithPostgresDB(db), `{"max_events":-1}`, "invalid max_events for SQL backing store: -1"}, + } + for _, tc := range testCases { + t.Run(tc.err, func(t *testing.T) { + params := strings.ReplaceAll(tc.params, "$DIR", t.TempDir()) + _, err := tc.factory(json.RawMessage(params), AuditorOpts{Registry: prometheus.NewPedanticRegistry()}) + assert.ErrEqual(t, err, "audittools: "+tc.err) + }) + } +} + +// gatherMetrics returns the values of all metrics in the registry, summed over all label values. +func gatherMetrics(t *testing.T, registry *prometheus.Registry) map[string]float64 { + t.Helper() + result := make(map[string]float64) + for _, mf := range must.ReturnT(registry.Gather())(t) { + for _, m := range mf.GetMetric() { + result[mf.GetName()] += m.GetCounter().GetValue() + m.GetGauge().GetValue() + } + } + return result +} + +//////////////////////////////////////////////////////////////////////////////// +// memory store + +func TestMemoryBackingStoreIsUnlimitedByDefault(t *testing.T) { + store := newTestMemoryBackingStore(t, `{}`) + for i := range 10000 { + must.SucceedT(t, store.Write(testEvent(fmt.Sprintf("event-%d", i)))) + } + assert.Equal(t, len(mustReadBatch(t, store, -1)), 10000) +} + +func TestMemoryBackingStoreMaxEvents(t *testing.T) { + store := newTestMemoryBackingStore(t, `{"max_events":3}`) + mustWrite(t, store, "event-1", "event-2", "event-3") + assert.ErrEqual(t, store.Write(testEvent("event-4")), errBackingStoreFull) + + mustReadBatch(t, store, 1) + mustWrite(t, store, "event-4") + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-2", "event-3", "event-4"}) +} + +func TestMemoryBackingStoreMetrics(t *testing.T) { + registry := prometheus.NewPedanticRegistry() + store := must.ReturnT(NewInMemoryBackingStore(json.RawMessage(`{"max_events":3}`), AuditorOpts{Registry: registry}))(t) + + mustWrite(t, store, "event-1", "event-2", "event-3") + assert.ErrEqual(t, store.Write(testEvent("event-4")), errBackingStoreFull) + mustReadBatch(t, store, 2) + must.SucceedT(t, store.UpdateMetrics()) + + assert.Equal(t, gatherMetrics(t, registry), map[string]float64{ + "audittools_backing_store_writes_total": 3, + "audittools_backing_store_reads_total": 2, + "audittools_backing_store_errors_total": 1, + "audittools_backing_store_events": 1, + }) +} + +//////////////////////////////////////////////////////////////////////////////// +// file store + +func TestFileBackingStoreLargeEvent(t *testing.T) { + // bufio.Scanner, which an earlier version used, cannot read lines longer than 64 KiB + store := newTestFileBackingStore(t, `{}`) + event := testEvent("large") + event.Attachments = []cadf.Attachment{{Name: "payload", TypeURI: "mime:text/plain", Content: strings.Repeat("x", 100<<10)}} + must.SucceedT(t, store.Write(event)) + mustWrite(t, store, "small") + + events, commit, err := store.ReadBatch() + must.SucceedT(t, err) + assert.Equal(t, len(events), 2) + assert.Equal(t, events[0], event) + must.SucceedT(t, commit(2)) + assert.Equal(t, mustReadBatch(t, store, -1), []string{}) +} + +func TestFileBackingStoreReopen(t *testing.T) { + // events written before a restart are published by the new process + store := newTestFileBackingStore(t, `{"max_file_size":1}`) + mustWrite(t, store, "event-1", "event-2") + + registry := prometheus.NewPedanticRegistry() + params := fmt.Sprintf(`{"directory":%q}`, store.directory) + reopened := must.ReturnT(NewFileBackingStore(json.RawMessage(params), AuditorOpts{Registry: registry}))(t) + assert.Equal(t, gatherMetrics(t, registry)["audittools_backing_store_files"], 2) + + var published []string + auditTrail{BackingStore: reopened}.drainBackingStore(func(e *cadf.Event) bool { + published = append(published, e.ID) + return true + }) + assert.Equal(t, published, []string{"event-1", "event-2"}) + assert.Equal(t, len(mustListFiles(t, store)), 0) +} + +func TestFileBackingStoreRotation(t *testing.T) { + store := newTestFileBackingStore(t, `{"max_file_size":1}`) + mustWrite(t, store, "event-1", "event-2") + assert.Equal(t, len(mustListFiles(t, store)), 2) + + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-1"}) + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-2"}) + assert.Equal(t, len(mustListFiles(t, store)), 0) +} + +func TestFileBackingStoreMaxTotalSize(t *testing.T) { + store := newTestFileBackingStore(t, `{"max_total_size":200}`) + mustWrite(t, store, "event-1", "event-2") // about 150 bytes each + assert.ErrEqual(t, store.Write(testEvent("event-3")), errBackingStoreFull) + + mustReadBatch(t, store, 1) + mustWrite(t, store, "event-3") + ids := mustReadBatch(t, store, -1) + ids = append(ids, mustReadBatch(t, store, -1)...) + assert.Equal(t, ids, []string{"event-2", "event-3"}) +} + +func TestFileBackingStorePermissions(t *testing.T) { + store := newTestFileBackingStore(t, `{}`) + mustWrite(t, store, "event-1") + + assert.Equal(t, must.ReturnT(os.Stat(store.directory))(t).Mode().Perm(), 0700) + files := mustListFiles(t, store) + assert.Equal(t, len(files), 1) + assert.Equal(t, must.ReturnT(os.Stat(files[0]))(t).Mode().Perm(), 0600) +} + +func TestFileBackingStoreCorruptLines(t *testing.T) { + store := newTestFileBackingStore(t, `{}`) + mustWrite(t, store, "event-1") + files := mustListFiles(t, store) + assert.Equal(t, len(files), 1) + appendLine(t, files[0], `{"id":42}`) // "id" must be a string + mustWrite(t, store, "event-2") + + // while events cannot be published, the corrupt line is left alone + assert.Equal(t, mustReadBatch(t, store, 0), []string{"event-1", "event-2"}) + assert.Equal(t, mustReadBatch(t, store, 1), []string{"event-1", "event-2"}) + assert.Equal(t, len(mustGlob(t, store, "*.corrupt")), 0) + + // once all events are published, the corrupt line is moved aside + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-2"}) + assert.Equal(t, len(mustListFiles(t, store)), 0) + corruptFiles := mustGlob(t, store, "*.corrupt") + assert.Equal(t, len(corruptFiles), 1) + assert.Equal(t, string(must.ReturnT(os.ReadFile(corruptFiles[0]))(t)), "{\"id\":42}\n") +} + +func TestFileBackingStoreOnlyCorruptLines(t *testing.T) { + store := newTestFileBackingStore(t, `{}`) + path := filepath.Join(store.directory, "audit-events-1.jsonl") + appendLine(t, path, `garbage`) + + events, commit, err := store.ReadBatch() + must.SucceedT(t, err) + assert.Equal(t, len(events), 0) + must.SucceedT(t, commit(0)) + assert.Equal(t, len(mustListFiles(t, store)), 0) + assert.Equal(t, len(mustGlob(t, store, "*.corrupt")), 1) +} + +func TestFileBackingStoreWriteErrorStartsNewFile(t *testing.T) { + if os.Getuid() == 0 { + t.Skip("cannot provoke write errors with file permissions when running as root") + } + store := newTestFileBackingStore(t, `{}`) + mustWrite(t, store, "event-1") + files := mustListFiles(t, store) + assert.Equal(t, len(files), 1) + + // after a failed write, the next event must not be appended to the same file, + // where it might be glued to a partially written line + must.SucceedT(t, os.Chmod(files[0], 0400)) + assert.ErrEqual(t, store.Write(testEvent("event-2")), regexp.MustCompile(`permission denied`)) + must.SucceedT(t, os.Chmod(files[0], 0600)) + mustWrite(t, store, "event-3") + assert.Equal(t, len(mustListFiles(t, store)), 2) + + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-1"}) + assert.Equal(t, mustReadBatch(t, store, -1), []string{"event-3"}) +} + +func TestWriteFileAtomicallyRemovesTempFile(t *testing.T) { + // renaming fails because the target is a non-empty directory + path := filepath.Join(t.TempDir(), "target") + must.SucceedT(t, os.MkdirAll(filepath.Join(path, "subdir"), 0700)) + assert.ErrEqual(t, writeFileAtomically(path, []byte("foo\n")), regexp.MustCompile(`rename`)) + _, err := os.Stat(path + ".tmp") + assert.Equal(t, errors.Is(err, fs.ErrNotExist), true) +} + +func mustListFiles(t *testing.T, store *fileBackingStore) []string { + t.Helper() + files, _, err := store.listFiles() + must.SucceedT(t, err) + return files +} + +func mustGlob(t *testing.T, store *fileBackingStore, pattern string) []string { + t.Helper() + return must.ReturnT(filepath.Glob(filepath.Join(store.directory, pattern)))(t) +} + +func appendLine(t *testing.T, path, line string) { + t.Helper() + must.SucceedT(t, appendToFile(path, []byte(line+"\n"))) +} diff --git a/audittools/rabbitmq.go b/audittools/rabbitmq.go index 98407bb0..5bfb727d 100644 --- a/audittools/rabbitmq.go +++ b/audittools/rabbitmq.go @@ -31,12 +31,18 @@ type rabbitConnection struct { // newRabbitConnection returns a new rabbitConnection using the specified amqp URI // and queue name. -func newRabbitConnection(uri url.URL, queueName string) (*rabbitConnection, error) { +func newRabbitConnection(uri url.URL, queueName string) (rc *rabbitConnection, err error) { // establish a connection with the RabbitMQ server - conn, err := amqp.Dial(uri.String()) + // (with a shorter timeout than the default of 30 seconds, since Record() blocks while we wait here) + conn, err := amqp.DialConfig(uri.String(), amqp.Config{Dial: amqp.DefaultDial(5 * time.Second)}) if err != nil { return nil, fmt.Errorf("audittools: rabbitmq: failed to establish a connection with the server: %w", err) } + defer func() { + if err != nil { + conn.CloseDeadline(time.Now().Add(time.Second)) //nolint:errcheck // we are already returning the more relevant error + } + }() // open a unique, concurrent server channel to process the bulk of AMQP messages ch, err := conn.Channel() @@ -44,6 +50,12 @@ func newRabbitConnection(uri url.URL, queueName string) (*rabbitConnection, erro return nil, fmt.Errorf("audittools: rabbitmq: failed to open a channel: %w", err) } + // with publisher confirms, PublishEvent can wait until the server has taken responsibility for the event + err = ch.Confirm(false) + if err != nil { + return nil, fmt.Errorf("audittools: rabbitmq: failed to enable publisher confirms: %w", err) + } + // declare a queue to hold and deliver messages to consumers _, err = ch.QueueDeclare( queueName, // name of the queue @@ -66,15 +78,16 @@ func newRabbitConnection(uri url.URL, queueName string) (*rabbitConnection, erro } // Disconnect is a helper function for closing a rabbitConnection. +// Closing the connection also closes the channel. +// We do not wait for the server for more than a second, since it may not be responding. func (c *rabbitConnection) Disconnect() { - c.Channel.Close() - c.Inner.Close() + c.Inner.CloseDeadline(time.Now().Add(time.Second)) //nolint:errcheck // the connection is not used anymore either way } // IsNilOrClosed is like (*amqp.Connection).IsClosed() but it also returns true -// if rabbitConnection or the underlying amqp.Connection are nil. +// if rabbitConnection or the underlying amqp.Connection are nil, or if the channel is closed. func (c *rabbitConnection) IsNilOrClosed() bool { - return c == nil || c.Inner == nil || c.Inner.IsClosed() + return c == nil || c.Inner == nil || c.Inner.IsClosed() || c.Channel.IsClosed() } // PublishEvent publishes a cadf.Event to a specific RabbitMQ Connection. @@ -93,7 +106,7 @@ func (c *rabbitConnection) PublishEvent(ctx context.Context, event *cadf.Event) return err } - return c.Channel.PublishWithContext( + confirmation, err := c.Channel.PublishWithDeferredConfirmWithContext( ctx, "", // exchange: publish to default c.QueueName, // routing key: same as queue name @@ -104,4 +117,22 @@ func (c *rabbitConnection) PublishEvent(ctx context.Context, event *cadf.Event) Body: b, }, ) + if err != nil { + return err + } + + // Without waiting for the confirm, events would be lost when the server stops responding, + // since publishing only puts them in the socket buffer. + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + acked, err := confirmation.WaitContext(ctx) + if err != nil { + // the server is not responding, so reconnect for the next event + c.Disconnect() + return fmt.Errorf("no publisher confirm from RabbitMQ: %w", err) + } + if !acked { + return errors.New("RabbitMQ did not accept the event") + } + return nil } diff --git a/audittools/trail.go b/audittools/trail.go index 2bdaac98..952bec77 100644 --- a/audittools/trail.go +++ b/audittools/trail.go @@ -5,10 +5,12 @@ package audittools import ( "context" + "encoding/json" "net/url" "time" "github.com/sapcc/go-api-declarations/cadf" + "go.xyrillian.de/gg/option" "github.com/sapcc/go-bits/logg" ) @@ -17,21 +19,31 @@ type auditTrail struct { EventSink <-chan cadf.Event OnSuccessfulPublish func() OnFailedPublish func() + BackingStore BackingStore } // Commit takes a AuditTrail that receives audit events from an event sink and publishes them to // a specific RabbitMQ Connection using the specified amqp URI and queue name. // The OnSuccessfulPublish and OnFailedPublish closures are executed as per their respective case. +// Events that cannot be published are put in the BackingStore, which is drained once per minute. // // This function blocks the current goroutine forever. It should be invoked with the "go" keyword. func (t auditTrail) Commit(ctx context.Context, rabbitmqURI url.URL, rabbitmqQueueName string) { + lastConnectAttempt := time.Now() rc, err := newRabbitConnection(rabbitmqURI, rabbitmqQueueName) if err != nil { logg.Error(err.Error()) } sendEvent := func(e *cadf.Event) bool { - rc = refreshConnectionIfClosedOrOld(rc, rabbitmqURI, rabbitmqQueueName) + // While RabbitMQ is unreachable, only try to reconnect every 10 seconds, and put events into the backing store in between. + // (Each attempt can take several seconds, and Record() blocks once its small buffer is full.) + if !rc.IsNilOrClosed() || time.Since(lastConnectAttempt) >= 10*time.Second { + if rc.IsNilOrClosed() { + lastConnectAttempt = time.Now() + } + rc = refreshConnectionIfClosedOrOld(rc, rabbitmqURI, rabbitmqQueueName) + } err := rc.PublishEvent(ctx, e) if err != nil { t.OnFailedPublish() @@ -42,52 +54,113 @@ func (t auditTrail) Commit(ctx context.Context, rabbitmqURI url.URL, rabbitmqQue return true } - var pendingEvents []cadf.Event + t.processEvents(sendEvent) +} + +func refreshConnectionIfClosedOrOld(rc *rabbitConnection, uri url.URL, queueName string) *rabbitConnection { + if !rc.IsNilOrClosed() && time.Since(rc.LastConnectedAt) < 5*time.Minute { + return rc + } + if rc != nil { + rc.Disconnect() + } + + connection, err := newRabbitConnection(uri, queueName) + if err != nil { + logg.Error(err.Error()) + return nil + } + + return connection +} + +// processEvents is the main loop of Commit. It is separate from Commit only so that tests can provide their own sendEvent. +// It returns when t.EventSink is closed, which only happens in tests. +func (t auditTrail) processEvents(sendEvent func(*cadf.Event) bool) { ticker := time.NewTicker(1 * time.Minute) defer ticker.Stop() + + // If an event can neither be published nor stored, we hold on to it and stop taking new events, + // so that Record() blocks instead of losing events. + var stalledEvent option.Option[cadf.Event] + for { + eventSink := t.EventSink + if stalledEvent.IsSome() { + eventSink = nil + } + select { - case e := <-t.EventSink: - if successful := sendEvent(&e); !successful { - pendingEvents = append(pendingEvents, e) + case e, ok := <-eventSink: + if !ok { + return + } + if !t.handleEvent(e, sendEvent) { + stalledEvent = option.Some(e) } case <-ticker.C: - for len(pendingEvents) > 0 { - successful := false // until proven otherwise - - nextEvent := pendingEvents[0] - if successful = sendEvent(&nextEvent); !successful { - // One more try before giving up. We simply set rc to nil - // and sendEvent() will take care of refreshing the - // connection. - time.Sleep(5 * time.Second) - rc = nil - successful = sendEvent(&nextEvent) - } - - if successful { - pendingEvents = pendingEvents[1:] - } else { - break - } + t.drainBackingStore(sendEvent) + if e, ok := stalledEvent.Unpack(); ok && t.handleEvent(e, sendEvent) { + stalledEvent = option.None[cadf.Event]() + } + err := t.BackingStore.UpdateMetrics() + if err != nil { + logg.Error("audittools: cannot update backing store metrics: %s", err.Error()) } } } } -func refreshConnectionIfClosedOrOld(rc *rabbitConnection, uri url.URL, queueName string) *rabbitConnection { - if !rc.IsNilOrClosed() { - if time.Since(rc.LastConnectedAt) < 5*time.Minute { - return rc - } - rc.Disconnect() +// handleEvent publishes the event, or puts it in the backing store if it cannot be published. +// Returns false if neither worked. +// Events that cannot be serialized are dropped, since publishing them can never succeed. +// +// Events from the backing store are only published on the next drain, +// so events can be published out of order while RabbitMQ is coming back. +func (t auditTrail) handleEvent(e cadf.Event, sendEvent func(*cadf.Event) bool) bool { + if sendEvent(&e) { + return true } - - connection, err := newRabbitConnection(uri, queueName) + _, err := json.Marshal(e) if err != nil { - logg.Error(err.Error()) - return nil + logg.Error("audittools: dropping audit event with ID %q that cannot be serialized: %s", e.ID, err.Error()) + return true + } + err = t.BackingStore.Write(e) + if err != nil { + logg.Error("audittools: cannot write audit event with ID %q to backing store: %s", e.ID, err.Error()) + return false } + return true +} - return connection +// drainBackingStore publishes events from the backing store until it is empty or publishing fails. +func (t auditTrail) drainBackingStore(sendEvent func(*cadf.Event) bool) { + for { + events, commit, err := t.BackingStore.ReadBatch() + if err != nil { + logg.Error("audittools: cannot read from backing store: %s", err.Error()) + return + } + if commit == nil { + return // backing store is empty + } + + published := 0 + for _, e := range events { + if !sendEvent(&e) { + break + } + published++ + } + + err = commit(published) + if err != nil { + logg.Error("audittools: cannot remove published events from backing store: %s", err.Error()) + return + } + if published < len(events) { + return + } + } } diff --git a/audittools/trail_test.go b/audittools/trail_test.go new file mode 100644 index 00000000..b478e5a8 --- /dev/null +++ b/audittools/trail_test.go @@ -0,0 +1,212 @@ +// SPDX-FileCopyrightText: 2026 SAP SE or an SAP affiliate company +// SPDX-License-Identifier: Apache-2.0 + +package audittools + +import ( + "encoding/json" + "fmt" + "slices" + "sync" + "testing" + "testing/synctest" + "time" + + "github.com/sapcc/go-api-declarations/cadf" + "go.xyrillian.de/gg/assert" +) + +// fakeRabbitMQ provides a sendEvent function for auditTrail that can be switched between working and failing. +type fakeRabbitMQ struct { + mutex sync.Mutex + isUp bool + hangFor time.Duration // while down, each publish fails only after this long (like a server that does not respond) + published []string +} + +func (f *fakeRabbitMQ) SendEvent(e *cadf.Event) bool { + // like rabbitConnection.PublishEvent + _, err := json.Marshal(e) + if err != nil { + return false + } + + f.mutex.Lock() + isUp, hangFor := f.isUp, f.hangFor + if isUp { + f.published = append(f.published, e.ID) + } + f.mutex.Unlock() + if !isUp { + time.Sleep(hangFor) + } + return isUp +} + +func (f *fakeRabbitMQ) SetUp(isUp bool) { + f.mutex.Lock() + defer f.mutex.Unlock() + f.isUp = isUp +} + +func (f *fakeRabbitMQ) Published() []string { + f.mutex.Lock() + defer f.mutex.Unlock() + return f.published +} + +func eventIDs(count int) []string { + ids := make([]string, count) + for idx := range ids { + ids[idx] = fmt.Sprintf("event-%d", idx) + } + return ids +} + +// runTrail runs auditTrail.processEvents in the background until the returned channel is closed. +func runTrail(store BackingStore, rabbit *fakeRabbitMQ) chan<- cadf.Event { + eventSink := make(chan cadf.Event, 20) // same size as in NewAuditor + go auditTrail{EventSink: eventSink, BackingStore: store}.processEvents(rabbit.SendEvent) + return eventSink +} + +func TestTrailWithDefaultBackingStoreDoesNotBlock(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + // without configuration, events are buffered in memory without limit (like before backing stores existed), + // so a RabbitMQ outage does not block Record() + rabbit := &fakeRabbitMQ{isUp: false} + eventSink := runTrail(newTestMemoryBackingStore(t, `{}`), rabbit) + ids := eventIDs(5000) + sent := make(chan struct{}) + go func() { + for _, id := range ids { + eventSink <- testEvent(id) + } + close(sent) + }() + synctest.Wait() + select { + case <-sent: + default: + t.Fatal("trail stopped accepting events, so Record() would block") + } + assert.Equal(t, len(rabbit.Published()), 0) + + // once RabbitMQ is back, everything is published in order on the next drain + rabbit.SetUp(true) + time.Sleep(time.Minute) + synctest.Wait() + assert.Equal(t, rabbit.Published(), ids) + + close(eventSink) + }) +} + +func TestTrailKeepsEventWhenBackingStoreIsFull(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + rabbit := &fakeRabbitMQ{isUp: false} + eventSink := runTrail(newTestMemoryBackingStore(t, `{"max_events":2}`), rabbit) + ids := eventIDs(5) + for _, id := range ids { + eventSink <- testEvent(id) + } + + // event-0 and event-1 go into the store, event-2 is held by the trail, + // and the trail does not take event-3 and event-4 until event-2 is dealt with + synctest.Wait() + assert.Equal(t, len(eventSink), 2) + + // as long as RabbitMQ is down, the drain does not change anything (in particular, nothing is lost) + time.Sleep(time.Minute) + synctest.Wait() + assert.Equal(t, len(eventSink), 2) + assert.Equal(t, len(rabbit.Published()), 0) + + // once RabbitMQ is back, nothing was lost + rabbit.SetUp(true) + time.Sleep(time.Minute) + synctest.Wait() + assert.Equal(t, rabbit.Published(), ids) + assert.Equal(t, len(eventSink), 0) + + close(eventSink) + }) +} + +func TestTrailWithHangingRabbitMQ(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + // RabbitMQ accepts connections, but does not respond + rabbit := &fakeRabbitMQ{isUp: false, hangFor: 5 * time.Second} + eventSink := runTrail(newTestMemoryBackingStore(t, `{}`), rabbit) + ids := eventIDs(30) + + // Record() blocks once the buffer is full, until the trail has given up on enough events: + // the trail takes event-0 right away, the buffer takes the next 20 events, + // and each of the remaining 9 events has to wait for one failed publish + start := time.Now() + for _, id := range ids { + eventSink <- testEvent(id) + } + assert.Equal(t, time.Since(start), 9*5*time.Second) + + // once RabbitMQ is back, nothing was lost + rabbit.SetUp(true) + time.Sleep(3 * time.Minute) + synctest.Wait() + assert.Equal(t, slices.Sorted(slices.Values(rabbit.Published())), slices.Sorted(slices.Values(ids))) + + close(eventSink) + }) +} + +func TestTrailDropsEventThatCannotBeSerialized(t *testing.T) { + stores := map[string]func(*testing.T) BackingStore{ + "memory": func(t *testing.T) BackingStore { return newTestMemoryBackingStore(t, `{}`) }, + "file": func(t *testing.T) BackingStore { return newTestFileBackingStore(t, `{}`) }, + } + for name, newStore := range stores { + t.Run(name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + rabbit := &fakeRabbitMQ{isUp: true} + store := newStore(t) + eventSink := runTrail(store, rabbit) + + // this event can never be published, so it must not block the events after it + event := testEvent("unserializable") + event.Attachments = []cadf.Attachment{{Name: "payload", Content: make(chan int)}} + eventSink <- event + ids := eventIDs(3) + for _, id := range ids { + eventSink <- testEvent(id) + } + + // the event is neither held by the trail nor put in the backing store + close(eventSink) + synctest.Wait() + assert.Equal(t, rabbit.Published(), ids) + assert.Equal(t, mustReadBatch(t, store, -1), []string{}) + }) + }) + } +} + +func TestDrainBackingStoreWithPartialFailure(t *testing.T) { + store := newTestMemoryBackingStore(t, `{}`) + mustWrite(t, store, "event-1", "event-2", "event-3") + trail := auditTrail{BackingStore: store} + + // RabbitMQ goes away after the first event + rabbit := &fakeRabbitMQ{isUp: true} + trail.drainBackingStore(func(e *cadf.Event) bool { + ok := rabbit.SendEvent(e) + rabbit.SetUp(false) + return ok + }) + assert.Equal(t, rabbit.Published(), []string{"event-1"}) + + // only the unpublished events are retried, so nothing is published twice + rabbit.SetUp(true) + trail.drainBackingStore(rabbit.SendEvent) + assert.Equal(t, rabbit.Published(), []string{"event-1", "event-2", "event-3"}) + assert.Equal(t, mustReadBatch(t, store, -1), []string{}) +} diff --git a/go.mod b/go.mod index 9a6c0d21..c2bbf129 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/hashicorp/vault/api v1.23.0 github.com/itchyny/gojq v0.12.19 + github.com/lib/pq v1.12.3 github.com/prometheus/client_golang v1.24.1 github.com/prometheus/common v0.70.1 github.com/rabbitmq/amqp091-go v1.14.0 diff --git a/go.sum b/go.sum index da4d8876..ee7f3ec2 100644 --- a/go.sum +++ b/go.sum @@ -64,6 +64,8 @@ github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJn github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ= +github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA= github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= diff --git a/shell.nix b/shell.nix index 0aef44ed..010abbde 100644 --- a/shell.nix +++ b/shell.nix @@ -12,6 +12,7 @@ mkShell { go_1_27 golangci-lint gotools # goimports + postgresql_18 renovate reuse typos