audittools: add backing store to buffer events while rabbitmq is down

.github/workflows/ci.yaml

@@ -72,7 +72,11 @@ jobs:
           check-latest: true
           go-version: 1.27.2
       - 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@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7
         with:

.gitignore

@@ -1,3 +1,4 @@
 # test artifacts
 /build/
 *.actual
+/.testdb

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

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":"<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())

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")
+	}
+}

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
+}