enhancement: Write audit logs asynchronously (#2104)

If requests are very large, serializing them to stdout or files can take
a while. This PR makes the audit log write asynchronous so that the
response can be returned without waiting for slow output sinks.

Signed-off-by: Charith Ellawala <charith@cerbos.dev>

Write audit log entries asynchronously

---------

Signed-off-by: Charith Ellawala <charith@cerbos.dev>

Author: charithe

go.mod

@@ -54,7 +54,6 @@ require (
 	github.com/microsoft/go-mssqldb v1.7.0
 	github.com/minio/minio-go/v7 v7.0.69
 	github.com/nlepage/go-tarfs v1.2.1
-	github.com/ohler55/ojg v1.21.4
 	github.com/oklog/ulid/v2 v2.1.0
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/ory/dockertest/v3 v3.10.0

go.sum

@@ -579,8 +579,6 @@ github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJm
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nlepage/go-tarfs v1.2.1 h1:o37+JPA+ajllGKSPfy5+YpsNHDjZnAoyfvf5GsUa+Ks=
 github.com/nlepage/go-tarfs v1.2.1/go.mod h1:rno18mpMy9aEH1IiJVftFsqPyIpwqSUiAOpJYjlV2NA=
-github.com/ohler55/ojg v1.21.4 h1:2iWyz/xExx0XySVIxR9kWFxIdsLNrpWLrKuAcs5aOZU=
-github.com/ohler55/ojg v1.21.4/go.mod h1:gQhDVpQLqrmnd2eqGAvJtn+NfKoYJbe/A4Sj3/Vro4o=
 github.com/oklog/ulid/v2 v2.1.0 h1:+9lhoxAP56we25tyYETBBY1YLA2SaoLvUFgrP2miPJU=
 github.com/oklog/ulid/v2 v2.1.0/go.mod h1:rcEKHmBBKfef9DhnvX7y1HZBYxjXb0cP5ExxNsTT1QQ=
 github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=

internal/audit/kafka/kafka_test.go

@@ -79,6 +79,7 @@ func TestProduceWithTLS(t *testing.T) {
 		}, nil
 	})
 	require.NoError(t, err)
+	require.NoError(t, log.Close())
 
 	// validate we see this entries in kafka
 	records, err := fetchKafkaTopic(t, uri, defaultIntegrationTopic, true)
@@ -118,6 +119,7 @@ func TestSyncProduce(t *testing.T) {
 		}, nil
 	})
 	require.NoError(t, err)
+	require.NoError(t, log.Close())
 
 	// validate we see this entries in kafka
 	records, err := fetchKafkaTopic(t, uri, defaultIntegrationTopic, false)
@@ -156,6 +158,7 @@ func TestCompression(t *testing.T) {
 			}, nil
 		})
 		require.NoError(t, err)
+		require.NoError(t, log.Close())
 	}
 
 	// validate we see these entries in kafka
@@ -196,6 +199,7 @@ func TestAsyncProduce(t *testing.T) {
 		}, nil
 	})
 	require.NoError(t, err)
+	require.NoError(t, log.Close())
 
 	// validate we see this entries in kafka, eventually
 	require.Eventually(t, func() bool {

internal/audit/log.go

@@ -8,11 +8,16 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"runtime"
 	"sync"
 	"time"
 
+	"github.com/sourcegraph/conc/pool"
+	"go.uber.org/zap"
+
 	auditv1 "github.com/cerbos/cerbos/api/genpb/cerbos/audit/v1"
 	"github.com/cerbos/cerbos/internal/config"
+	"github.com/cerbos/cerbos/internal/observability/logging"
 	"github.com/cerbos/cerbos/internal/observability/metrics"
 )
 
@@ -103,8 +108,7 @@ func NewLogFromConf(ctx context.Context, confW *config.Wrapper) (Log, error) {
 		return nil, fmt.Errorf("failed to create backend: %w", err)
 	}
 
-	lw := &logWrapper{conf: conf, backend: backend}
-
+	lw := newLogWrapper(conf, backend)
 	if q, ok := backend.(QueryableLog); ok {
 		return &queryableLogWrapper{logWrapper: lw, queryable: q}, nil
 	}
@@ -115,13 +119,24 @@ func NewLogFromConf(ctx context.Context, confW *config.Wrapper) (Log, error) {
 // NewNopLog returns an audit log that does nothing.
 func NewNopLog() Log {
 	conf := &Conf{confHolder: confHolder{Enabled: false, AccessLogsEnabled: false, DecisionLogsEnabled: false}}
-	return &logWrapper{conf: conf}
+	return newLogWrapper(conf, nil)
+}
+
+func newLogWrapper(conf *Conf, backend Log) *logWrapper {
+	lw := &logWrapper{conf: conf}
+	if backend != nil {
+		lw.backend = backend
+		lw.pool = pool.New().WithMaxGoroutines(runtime.NumCPU())
+	}
+
+	return lw
 }
 
 // logWrapper wraps the backends and enforces the config options.
 type logWrapper struct {
 	conf    *Conf
 	backend Log
+	pool    *pool.Pool
 }
 
 func (lw *logWrapper) Backend() string {
@@ -137,10 +152,12 @@ func (lw *logWrapper) WriteAccessLogEntry(ctx context.Context, entry AccessLogEn
 		return nil
 	}
 
-	if err := lw.backend.WriteAccessLogEntry(ctx, entry); err != nil {
-		metrics.Inc(ctx, metrics.AuditErrorCount(), metrics.KindKey(KindAccess))
-		return err
-	}
+	lw.pool.Go(func() {
+		if err := lw.backend.WriteAccessLogEntry(ctx, entry); err != nil {
+			metrics.Inc(ctx, metrics.AuditErrorCount(), metrics.KindKey(KindAccess))
+			logging.FromContext(ctx).Warn("Failed to write access log entry", zap.Error(err))
+		}
+	})
 
 	return nil
 }
@@ -150,16 +167,19 @@ func (lw *logWrapper) WriteDecisionLogEntry(ctx context.Context, entry DecisionL
 		return nil
 	}
 
-	if err := lw.backend.WriteDecisionLogEntry(ctx, entry); err != nil {
-		metrics.Inc(ctx, metrics.AuditErrorCount(), metrics.KindKey(KindDecision))
-		return err
-	}
+	lw.pool.Go(func() {
+		if err := lw.backend.WriteDecisionLogEntry(ctx, entry); err != nil {
+			metrics.Inc(ctx, metrics.AuditErrorCount(), metrics.KindKey(KindDecision))
+			logging.FromContext(ctx).Warn("Failed to write decision log entry", zap.Error(err))
+		}
+	})
 
 	return nil
 }
 
 func (lw *logWrapper) Close() error {
 	if lw.backend != nil {
+		lw.pool.Wait()
 		return lw.backend.Close()
 	}
 	return nil