feat: add backing store for disk buffering of events

Author: notque

audittools/README.md

@@ -0,0 +1,142 @@
+<!--  SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# audittools
+
+`audittools` provides a standard interface for generating and sending CADF (Cloud Auditing Data Federation) audit events to a RabbitMQ message broker.
+
+## Certification Requirements (PCI DSS, SOC 2, and more)
+
+As a cloud provider subject to strict audits (including PCI DSS and more), we must ensure the **completeness** and **integrity** of audit logs while maintaining service **availability**.
+
+### Standard Production Configuration
+
+**You MUST configure a Backing Store with Persistent Storage (PVC).**
+
+*   **Configuration**: Set `BackingStorePath` to a mount point backed by a PVC.
+*   **Requirement**: This ensures that audit events are preserved even in double-failure scenarios (RabbitMQ outage + Pod crash/reschedule).
+*   **Compliance**: Satisfies requirements for guaranteed event delivery and audit trail completeness.
+
+### Non-Compliant Configurations
+
+The following configurations are available for development or specific edge cases but are **NOT** recommended for production services subject to audit:
+
+1.  **Ephemeral Storage (emptyDir)**:
+    *   *Risk*: Data loss if the Pod is rescheduled during a RabbitMQ outage.
+    *   *Status*: **Development / Testing Only**.
+
+2.  **No Backing Store**:
+    *   *Behavior*: The service will **block** (stop processing requests) if the RabbitMQ broker is down and the memory buffer fills up.
+    *   *Risk*: Service downtime (Violation of Availability targets).
+    *   *Status*: **Not Recommended**. Only acceptable if service downtime is preferred over *any* storage complexity.
+
+## Usage
+
+### Basic Setup
+
+To use `audittools`, you typically initialize an `Auditor` with your RabbitMQ connection details.
+
+```go
+import "github.com/sapcc/go-bits/audittools"
+
+func main() {
+    // ...
+    auditor, err := audittools.NewAuditor(audittools.AuditorOpts{
+        EnvPrefix: "MYSERVICE_AUDIT", // Configures env vars like MYSERVICE_AUDIT_RABBITMQ_URL
+    })
+    if err != nil {
+        log.Fatal(err)
+    }
+    // ...
+}
+```
+
+### Sending Events
+
+```go
+event := cadf.Event{
+    // ... fill in event details ...
+}
+auditor.Record(event)
+```
+
+## Disk-Based Buffering
+
+`audittools` includes buffering to ensure audit events are not lost if the RabbitMQ broker becomes unavailable. Events are temporarily written to disk and replayed once the connection is restored.
+
+### Configuration
+
+Buffering is enabled by providing a `BackingStorePath`.
+
+```go
+auditor, err := audittools.NewAuditor(audittools.AuditorOpts{
+    EnvPrefix:        "MYSERVICE_AUDIT",
+    BackingStorePath: "/var/lib/myservice/audit-buffer",
+})
+```
+
+Or via environment variables:
+
+*   `MYSERVICE_AUDIT_BACKING_STORE_PATH`: Directory to store buffered events.
+*   `MYSERVICE_AUDIT_BACKING_STORE_MAX_TOTAL_SIZE`: (Optional) Max total size of the buffer in bytes.
+
+### Kubernetes Deployment
+
+If running in Kubernetes, you have two main options for the backing store:
+
+1.  **Persistent Storage (PVC) - Recommended for Audit Compliance**:
+    *   Mount a Persistent Volume Claim (PVC) at the `BackingStorePath`.
+    *   **Pros**: Data survives Pod deletion, rescheduling, and rolling updates.
+    *   **Cons**: Adds complexity (volume management, access modes).
+    *   **Use Case**: **Required** for strict audit compliance to ensure no data is lost even if the service instance fails during a broker outage.
+
+2.  **Ephemeral Storage (emptyDir)**:
+    *   Mount an `emptyDir` volume at the `BackingStorePath`.
+    *   **Pros**: Simple, fast, no persistent volume management.
+    *   **Cons**: Data is lost if the *Pod* is deleted or rescheduled. However, it survives container restarts within the same Pod.
+    *   **Use Case**: Suitable for non-critical environments or where occasional data loss during complex failure scenarios (simultaneous broker outage + pod rescheduling) is acceptable.
+
+### Behavior
+
+The system transitions through the following states to ensure zero data loss:
+
+1.  **Normal Operation**: Events are sent directly to RabbitMQ.
+2.  **RabbitMQ Outage**: Events are written to the disk backing store. The application continues without blocking.
+3.  **Disk Full**: If the backing store reaches `BackingStoreMaxTotalSize`, writes fail. Events are buffered in memory (up to 20).
+4.  **Fail-Closed**: If the memory buffer fills up, `auditor.Record()` **blocks**. This pauses the application to prevent data loss.
+5.  **Recovery**: A background routine continuously drains the backing store to RabbitMQ once it becomes available. New events are persisted to disk during draining to prevent blocking.
+
+**Additional Details**:
+
+*   **Security**: The directory is created with `0700` permissions, and files with `0600`, ensuring only the service user can access the sensitive audit data.
+*   **Capacity**: If `BackingStoreMaxTotalSize` is configured, the store will reject new writes when full. The limit is approximate and may be exceeded by up to one event's size (typically a few KB) due to the check-then-write sequence. Set the limit with appropriate headroom for your filesystem.
+*   **Corrupted Event Handling**:
+    *   Corrupted events encountered during reads are written to dead-letter files (`audit-events-deadletter-*.jsonl`)
+    *   Dead-letter files contain metadata (timestamp, source file) and the raw corrupted data for investigation
+    *   The `corrupted_event` metric is incremented for monitoring
+    *   Source files are deleted after processing, even if all events were corrupted (after moving to dead-letter)
+    *   Dead-letter files should be monitored and investigated to identify data corruption issues
+
+### Metrics
+
+The backing store exports the following Prometheus metrics:
+
+*   `audittools_backing_store_writes_total`: Total number of audit events written to disk.
+*   `audittools_backing_store_reads_total`: Total number of audit events read from disk.
+*   `audittools_backing_store_errors_total`: Total number of errors, labeled by operation:
+    *   `write_stat`: Failed to stat file during rotation check
+    *   `write_full`: Backing store is full (exceeds `MaxTotalSize`)
+    *   `write_open`: Failed to open backing store file for writing
+    *   `write_marshal`: Failed to marshal event to JSON
+    *   `write_io`: Failed to write event to disk
+    *   `write_sync`: Failed to sync (flush) event to disk
+    *   `write_close`: Failed to close backing store file
+    *   `read_open`: Failed to open backing store file for reading
+    *   `read_scan`: Failed to scan backing store file
+    *   `corrupted_event`: Encountered corrupted event during read (written to dead-letter)
+    *   `deadletter_write`: Successfully wrote corrupted event to dead-letter file
+    *   `deadletter_write_failed`: Failed to write corrupted event to dead-letter file
+    *   `commit_remove`: Failed to remove file after successful processing
+*   `audittools_backing_store_size_bytes`: Current total size of the backing store in bytes.
+*   `audittools_backing_store_files_count`: Current number of files in the backing store.

audittools/auditor.go

@@ -17,10 +17,12 @@ import (
 	"fmt"
 	"net"
 	"net/url"
+	"os"
 	"strconv"
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/sapcc/go-api-declarations/cadf"
 
 	"github.com/sapcc/go-bits/assert"
@@ -29,6 +31,13 @@ import (
 	"github.com/sapcc/go-bits/osext"
 )
 
+const (
+	// eventBufferSize is the maximum number of events that can be buffered in memory
+	// when the RabbitMQ connection is unavailable and the backing store is full or unavailable.
+	// When this limit is reached, Record() will block to prevent data loss.
+	eventBufferSize = 20
+)
+
 // Auditor is a high-level interface for audit event acceptors.
 // In a real process, use NewAuditor() or NewNullAuditor() depending on whether you have RabbitMQ client credentials.
 // In a test scenario, use NewMockAuditor() to get an assertable mock implementation.
@@ -63,6 +72,14 @@ type AuditorOpts struct {
 	//   - "audittools_successful_submissions" (counter, no labels)
 	//   - "audittools_failed_submissions" (counter, no labels)
 	Registry prometheus.Registerer
+
+	// Optional. If given, the Auditor will buffer events in this directory when the RabbitMQ server is unreachable.
+	// If EnvPrefix is given, this can also be set via the environment variable "${PREFIX}_BACKING_STORE_PATH".
+	BackingStorePath string
+
+	// Optional. If given, limits the total size of the backing store in bytes.
+	// If EnvPrefix is given, this can also be set via the environment variable "${PREFIX}_BACKING_STORE_MAX_TOTAL_SIZE".
+	BackingStoreMaxTotalSize int64
 }
 
 func (opts AuditorOpts) getConnectionOptions() (rabbitURL url.URL, queueName string, err error) {
@@ -99,6 +116,7 @@ func (opts AuditorOpts) getConnectionOptions() (rabbitURL url.URL, queueName str
 		User:   url.UserPassword(username, pass),
 		Path:   "/",
 	}
+
 	return rabbitURL, queueName, nil
 }
 
@@ -139,16 +157,50 @@ func NewAuditor(ctx context.Context, opts AuditorOpts) (Auditor, error) {
 		opts.Registry.MustRegister(failureCounter)
 	}
 
+	// read backing store options from environment if EnvPrefix is set
+	if opts.EnvPrefix != "" {
+		if opts.BackingStorePath == "" {
+			opts.BackingStorePath = os.Getenv(opts.EnvPrefix + "_BACKING_STORE_PATH")
+		}
+		if opts.BackingStoreMaxTotalSize == 0 {
+			if val := os.Getenv(opts.EnvPrefix + "_BACKING_STORE_MAX_TOTAL_SIZE"); val != "" {
+				size, err := strconv.ParseInt(val, 10, 64)
+				if err != nil {
+					return nil, fmt.Errorf("audittools: invalid value for %s_BACKING_STORE_MAX_TOTAL_SIZE: %w", opts.EnvPrefix, err)
+				}
+				opts.BackingStoreMaxTotalSize = size
+			}
+		}
+	}
+
 	// spawn event delivery goroutine
 	rabbitURL, queueName, err := opts.getConnectionOptions()
 	if err != nil {
 		return nil, err
 	}
-	eventChan := make(chan cadf.Event, 20)
+
+	var backingStore BackingStore
+	if opts.BackingStorePath != "" {
+		bsRegistry := opts.Registry
+		if bsRegistry == nil {
+			bsRegistry = prometheus.DefaultRegisterer
+		}
+		backingStore, err = NewFileBackingStore(FileBackingStoreOpts{
+			Directory:    opts.BackingStorePath,
+			MaxTotalSize: opts.BackingStoreMaxTotalSize,
+			Registry:     bsRegistry,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to initialize backing store: %w", err)
+		}
+	}
+
+	eventChan := make(chan cadf.Event, eventBufferSize)
 	go auditTrail{
 		EventSink:           eventChan,
 		OnSuccessfulPublish: func() { successCounter.Inc() },
 		OnFailedPublish:     func() { failureCounter.Inc() },
+		BackingStore:        backingStore,
 	}.Commit(ctx, rabbitURL, queueName)
 
 	return &standardAuditor{

audittools/auditor_test.go

@@ -0,0 +1,61 @@
+// SPDX-FileCopyrightText: 2025 SAP SE or an SAP affiliate company
+// SPDX-License-Identifier: Apache-2.0
+
+package audittools
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+func TestNewAuditorInvalidMaxTotalSize(t *testing.T) {
+	// Test that invalid BACKING_STORE_MAX_TOTAL_SIZE value returns an error
+	t.Setenv("TEST_AUDIT_BACKING_STORE_MAX_TOTAL_SIZE", "10GB")
+
+	_, err := NewAuditor(context.Background(), AuditorOpts{
+		EnvPrefix: "TEST_AUDIT",
+		Observer: Observer{
+			TypeURI: "service/test",
+			Name:    "test-service",
+			ID:      "test-id",
+		},
+		Registry: prometheus.NewRegistry(),
+	})
+
+	if err == nil {
+		t.Fatal("expected error for invalid BACKING_STORE_MAX_TOTAL_SIZE, got nil")
+	}
+	if !strings.Contains(err.Error(), "invalid value for TEST_AUDIT_BACKING_STORE_MAX_TOTAL_SIZE") {
+		t.Fatalf("expected error message to contain 'invalid value for TEST_AUDIT_BACKING_STORE_MAX_TOTAL_SIZE', got: %v", err)
+	}
+}
+
+func TestNewAuditorValidMaxTotalSize(t *testing.T) {
+	// Test that valid BACKING_STORE_MAX_TOTAL_SIZE value works
+	t.Setenv("TEST_AUDIT_BACKING_STORE_MAX_TOTAL_SIZE", "1073741824") // 1GB in bytes
+	t.Setenv("TEST_AUDIT_QUEUE_NAME", "test-queue")
+
+	tmpDir := t.TempDir()
+
+	auditor, err := NewAuditor(context.Background(), AuditorOpts{
+		EnvPrefix:                "TEST_AUDIT",
+		BackingStorePath:         tmpDir,
+		BackingStoreMaxTotalSize: 0, // Will be read from env var
+		Observer: Observer{
+			TypeURI: "service/test",
+			Name:    "test-service",
+			ID:      "test-id",
+		},
+		Registry: prometheus.NewRegistry(),
+	})
+
+	if err != nil {
+		t.Fatalf("expected no error, got: %v", err)
+	}
+	if auditor == nil {
+		t.Fatal("expected auditor to be created, got nil")
+	}
+}