Merge branch 'main' into gm/flipt-validate-build-snapshot

config/flipt.schema.cue

@@ -223,16 +223,22 @@ import "strings"
 
 	#audit: {
 		sinks?: {
-			events?: [...string] | *["*:*"]
 			log?: {
 				enabled?: bool | *false
 				file?:    string | *""
 			}
+			webhook?: {
+				enabled?:              bool | *false
+				url?:                  string | *""
+				max_backoff_duration?: =~#duration | *"15s"
+				signing_secret?:       string | *""
+			}
 		}
 		buffer?: {
 			capacity?:     int | *2
 			flush_period?: string | *"2m"
 		}
+		events?: [...string] | *["*:*"]
 	}
 
 	#experimental: {}

config/flipt.schema.json

@@ -652,10 +652,6 @@
           "type": "object",
           "additionalProperties": false,
           "properties": {
-            "events": {
-              "type": "array",
-              "default": ["*:*"]
-            },
             "log": {
               "type": "object",
               "additionalProperties": false,
@@ -670,6 +666,36 @@
                 }
               },
               "title": "Log File"
+            },
+            "webhook": {
+              "type": "object",
+              "additionalProperties": false,
+              "properties": {
+                "enabled": {
+                  "type": "boolean",
+                  "default": false
+                },
+                "url": {
+                  "type": "string",
+                  "default": ""
+                },
+                "max_backoff_duration": {
+                  "oneOf": [
+                    {
+                      "type": "string",
+                      "pattern": "^([0-9]+(ns|us|µs|ms|s|m|h))+$"
+                    },
+                    {
+                      "type": "integer"
+                    }
+                  ],
+                  "default": "15s"
+                },
+                "signing_secret": {
+                  "type": "string",
+                  "default": ""
+                }
+              }
             }
           }
         },
@@ -686,6 +712,10 @@
               "default": "2m"
             }
           }
+        },
+        "events": {
+          "type": "array",
+          "default": ["*:*"]
         }
       },
       "title": "Audit"

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.18.39
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.38.5
 	github.com/blang/semver/v4 v4.0.0
+	github.com/cenkalti/backoff/v4 v4.2.1
 	github.com/coreos/go-oidc/v3 v3.6.0
 	github.com/docker/go-connections v0.4.0
 	github.com/fatih/color v1.15.0
@@ -49,7 +50,7 @@ require (
 	go.flipt.io/flipt/errors v1.19.3
 	go.flipt.io/flipt/rpc/flipt v1.25.0
 	go.flipt.io/flipt/sdk/go v0.3.0
-	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.42.0
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.43.0
 	go.opentelemetry.io/otel v1.17.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.17.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.17.0
@@ -97,7 +98,6 @@ require (
 	github.com/benbjohnson/clock v1.3.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
-	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/cloudflare/circl v1.3.3 // indirect
 	github.com/cockroachdb/apd/v3 v3.2.0 // indirect

go.sum

@@ -636,8 +636,8 @@ go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.42.0 h1:ZOLJc06r4CB42laIXg/7udr0pbZyuAihN10A/XuiQRY=
-go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.42.0/go.mod h1:5z+/ZWJQKXa9YT34fQNx5K8Hd1EoIhvtUygUQPqEOgQ=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.43.0 h1:7XZai4VhA473clBrOqqHdjHBImGfyEtv0qW4nnn/kAo=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.43.0/go.mod h1:1WpsUwjQrUJSNugfMlPn0rPRJ9Do7wwBgTBPK7MLiS4=
 go.opentelemetry.io/otel v1.17.0 h1:MW+phZ6WZ5/uk2nd93ANk/6yJ+dVrvNWUjGhnnFU5jM=
 go.opentelemetry.io/otel v1.17.0/go.mod h1:I2vmBGtFaODIVMBSTPVDlJSzBDNf93k60E6Ft0nyjo0=
 go.opentelemetry.io/otel/exporters/jaeger v1.17.0 h1:D7UpUy2Xc2wsi1Ras6V40q806WM07rqoCWzXu7Sqy+4=

internal/cmd/grpc.go

@@ -21,6 +21,7 @@ import (
 	fliptserver "go.flipt.io/flipt/internal/server"
 	"go.flipt.io/flipt/internal/server/audit"
 	"go.flipt.io/flipt/internal/server/audit/logfile"
+	"go.flipt.io/flipt/internal/server/audit/webhook"
 	"go.flipt.io/flipt/internal/server/auth"
 	"go.flipt.io/flipt/internal/server/evaluation"
 	"go.flipt.io/flipt/internal/server/metadata"
@@ -330,10 +331,21 @@ func NewGRPCServer(
 		sinks = append(sinks, logFileSink)
 	}
 
+	if cfg.Audit.Sinks.Webhook.Enabled {
+		opts := []webhook.ClientOption{}
+		if cfg.Audit.Sinks.Webhook.MaxBackoffDuration != 0 {
+			opts = append(opts, webhook.WithMaxBackoffDuration(cfg.Audit.Sinks.Webhook.MaxBackoffDuration))
+		}
+
+		webhookSink := webhook.NewSink(logger, webhook.NewHTTPClient(logger, cfg.Audit.Sinks.Webhook.URL, cfg.Audit.Sinks.Webhook.SigningSecret, opts...))
+
+		sinks = append(sinks, webhookSink)
+	}
+
 	// based on audit sink configuration from the user, provision the audit sinks and add them to a slice,
 	// and if the slice has a non-zero length, add the audit sink interceptor
 	if len(sinks) > 0 {
-		checker, err := audit.NewChecker(cfg.Audit.Sinks.Events)
+		checker, err := audit.NewChecker(cfg.Audit.Events)
 		if err != nil {
 			return nil, err
 		}

internal/config/audit.go

@@ -15,26 +15,30 @@ var _ defaulter = (*AuditConfig)(nil)
 type AuditConfig struct {
 	Sinks  SinksConfig  `json:"sinks,omitempty" mapstructure:"sinks"`
 	Buffer BufferConfig `json:"buffer,omitempty" mapstructure:"buffer"`
+	Events []string     `json:"events,omitempty" mapstructure:"events"`
 }
 
 // Enabled returns true if any nested sink is enabled
 func (c *AuditConfig) Enabled() bool {
-	return c.Sinks.LogFile.Enabled
+	return c.Sinks.LogFile.Enabled || c.Sinks.Webhook.Enabled
 }
 
 func (c *AuditConfig) setDefaults(v *viper.Viper) error {
 	v.SetDefault("audit", map[string]any{
 		"sinks": map[string]any{
-			"events": []string{"*:*"},
 			"log": map[string]any{
 				"enabled": "false",
 				"file":    "",
 			},
+			"webhook": map[string]any{
+				"enabled": "false",
+			},
 		},
 		"buffer": map[string]any{
 			"capacity":     2,
 			"flush_period": "2m",
 		},
+		"events": []string{"*:*"},
 	})
 
 	return nil
@@ -45,6 +49,10 @@ func (c *AuditConfig) validate() error {
 		return errors.New("file not specified")
 	}
 
+	if c.Sinks.Webhook.Enabled && c.Sinks.Webhook.URL == "" {
+		return errors.New("url not provided")
+	}
+
 	if c.Buffer.Capacity < 2 || c.Buffer.Capacity > 10 {
 		return errors.New("buffer capacity below 2 or above 10")
 	}
@@ -59,8 +67,17 @@ func (c *AuditConfig) validate() error {
 // SinksConfig contains configuration held in structures for the different sinks
 // that we will send audits to.
 type SinksConfig struct {
-	Events  []string          `json:"events,omitempty" mapstructure:"events"`
 	LogFile LogFileSinkConfig `json:"log,omitempty" mapstructure:"log"`
+	Webhook WebhookSinkConfig `json:"webhook,omitempty" mapstructure:"webhook"`
+}
+
+// WebhookSinkConfig contains configuration for sending POST requests to specific
+// URL as its configured.
+type WebhookSinkConfig struct {
+	Enabled            bool          `json:"enabled,omitempty" mapstructure:"enabled"`
+	URL                string        `json:"url,omitempty" mapstructure:"url"`
+	MaxBackoffDuration time.Duration `json:"maxBackoffDuration,omitempty" mapstructure:"max_backoff_duration"`
+	SigningSecret      string        `json:"signingSecret,omitempty" mapstructure:"signing_secret"`
 }
 
 // LogFileSinkConfig contains fields that hold configuration for sending audits

internal/config/config.go

@@ -511,7 +511,6 @@ func Default() *Config {
 
 		Audit: AuditConfig{
 			Sinks: SinksConfig{
-				Events: []string{"*:*"},
 				LogFile: LogFileSinkConfig{
 					Enabled: false,
 					File:    "",
@@ -521,6 +520,7 @@ func Default() *Config {
 				Capacity:    2,
 				FlushPeriod: 2 * time.Minute,
 			},
+			Events: []string{"*:*"},
 		},
 	}
 }

internal/config/config_test.go

@@ -449,7 +449,6 @@ func TestLoad(t *testing.T) {
 
 				cfg.Audit = AuditConfig{
 					Sinks: SinksConfig{
-						Events: []string{"*:*"},
 						LogFile: LogFileSinkConfig{
 							Enabled: true,
 							File:    "/path/to/logs.txt",
@@ -459,6 +458,7 @@ func TestLoad(t *testing.T) {
 						Capacity:    10,
 						FlushPeriod: 3 * time.Minute,
 					},
+					Events: []string{"*:*"},
 				}
 
 				cfg.Log = LogConfig{
@@ -619,6 +619,11 @@ func TestLoad(t *testing.T) {
 			path:    "./testdata/audit/invalid_enable_without_file.yml",
 			wantErr: errors.New("file not specified"),
 		},
+		{
+			name:    "url not specified",
+			path:    "./testdata/audit/invalid_webhook_url_not_provided.yml",
+			wantErr: errors.New("url not provided"),
+		},
 		{
 			name: "local config provided",
 			path: "./testdata/storage/local_provided.yml",

internal/config/testdata/audit/invalid_webhook_url_not_provided.yml

@@ -0,0 +1,4 @@
+audit:
+  sinks:
+    webhook:
+      enabled: true

internal/server/audit/audit.go

@@ -180,7 +180,7 @@ type Metadata struct {
 // Sink is the abstraction for various audit sink configurations
 // that Flipt will support.
 type Sink interface {
-	SendAudits([]Event) error
+	SendAudits(context.Context, []Event) error
 	Close() error
 	fmt.Stringer
 }
@@ -195,7 +195,7 @@ type SinkSpanExporter struct {
 type EventExporter interface {
 	ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error
 	Shutdown(ctx context.Context) error
-	SendAudits(es []Event) error
+	SendAudits(ctx context.Context, es []Event) error
 }
 
 // NewSinkSpanExporter is the constructor for a SinkSpanExporter.
@@ -224,7 +224,7 @@ func (s *SinkSpanExporter) ExportSpans(ctx context.Context, spans []sdktrace.Rea
 		}
 	}
 
-	return s.SendAudits(es)
+	return s.SendAudits(ctx, es)
 }
 
 // Shutdown will close all the registered sinks.
@@ -242,14 +242,14 @@ func (s *SinkSpanExporter) Shutdown(ctx context.Context) error {
 }
 
 // SendAudits wraps the methods of sending audits to various sinks.
-func (s *SinkSpanExporter) SendAudits(es []Event) error {
+func (s *SinkSpanExporter) SendAudits(ctx context.Context, es []Event) error {
 	if len(es) < 1 {
 		return nil
 	}
 
 	for _, sink := range s.sinks {
 		s.logger.Debug("performing batched sending of audit events", zap.Stringer("sink", sink), zap.Int("batch size", len(es)))
-		err := sink.SendAudits(es)
+		err := sink.SendAudits(ctx, es)
 		if err != nil {
 			s.logger.Debug("failed to send audits to sink", zap.Stringer("sink", sink))
 		}

internal/server/audit/audit_test.go

@@ -18,7 +18,7 @@ type sampleSink struct {
 	fmt.Stringer
 }
 
-func (s *sampleSink) SendAudits(es []Event) error {
+func (s *sampleSink) SendAudits(ctx context.Context, es []Event) error {
 	go func() {
 		s.ch <- es[0]
 	}()

internal/server/audit/logfile/logfile.go

@@ -1,6 +1,7 @@
 package logfile
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -35,7 +36,7 @@ func NewSink(logger *zap.Logger, path string) (audit.Sink, error) {
 	}, nil
 }
 
-func (l *Sink) SendAudits(events []audit.Event) error {
+func (l *Sink) SendAudits(ctx context.Context, events []audit.Event) error {
 	l.mtx.Lock()
 	defer l.mtx.Unlock()
 	var result error