Merge webhook and audit target reload fix (#259)

Reloaded logger webhook and audit targets stay alive on every node after a peer reload.

buildscripts/rebrand-guard/compat-baseline.json

@@ -152,6 +152,7 @@
     "MINIO_API_TRANSITION_WORKERS",
     "MINIO_API_TRUSTED_PROXIES",
     "MINIO_ARGS",
+    "MINIO_AUDIT_",
     "MINIO_AUDIT_KAFKA_BROKERS",
     "MINIO_AUDIT_KAFKA_CLIENT_TLS_CERT",
     "MINIO_AUDIT_KAFKA_CLIENT_TLS_KEY",
@@ -313,6 +314,7 @@
     "MINIO_LAMBDA_WEBHOOK_ENABLE",
     "MINIO_LAMBDA_WEBHOOK_ENDPOINT",
     "MINIO_LISTENERS",
+    "MINIO_LOGGER_",
     "MINIO_LOGGER_HTTP_ENDPOINT",
     "MINIO_LOGGER_WEBHOOK_AUTH_TOKEN",
     "MINIO_LOGGER_WEBHOOK_BATCH_SIZE",

cmd/config-current.go

@@ -609,7 +609,8 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
 			}
 			loggerCfg.HTTP[n] = l
 		}
-		if errs := logger.UpdateHTTPWebhooks(ctx, loggerCfg.HTTP); len(errs) > 0 {
+		// Webhook workers outlive the bounded configuration reload.
+		if errs := logger.UpdateHTTPWebhooks(GlobalContext, loggerCfg.HTTP); len(errs) > 0 {
 			configLogIf(ctx, fmt.Errorf("Unable to update logger webhook config: %v", errs))
 		}
 	case config.AuditWebhookSubSys:
@@ -627,7 +628,8 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
 			loggerCfg.AuditWebhook[n] = l
 		}
 
-		if errs := logger.UpdateAuditWebhooks(ctx, loggerCfg.AuditWebhook); len(errs) > 0 {
+		// Both the queue worker and disk replay follow the server lifetime.
+		if errs := logger.UpdateAuditWebhooks(GlobalContext, loggerCfg.AuditWebhook); len(errs) > 0 {
 			configLogIf(ctx, fmt.Errorf("Unable to update audit webhook targets: %v", errs))
 		}
 	case config.AuditKafkaSubSys:
@@ -644,7 +646,8 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
 				loggerCfg.AuditKafka[n] = l
 			}
 		}
-		if errs := logger.UpdateAuditKafkaTargets(ctx, loggerCfg); len(errs) > 0 {
+		// The Kafka disk replay follows the server lifetime as well.
+		if errs := logger.UpdateAuditKafkaTargets(GlobalContext, loggerCfg); len(errs) > 0 {
 			configLogIf(ctx, fmt.Errorf("Unable to update audit kafka targets: %v", errs))
 		}
 	case config.StorageClassSubSys:

cmd/config-current_test.go

@@ -19,12 +19,246 @@ package cmd
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
 	"testing"
+	"time"
 
+	"github.com/IBM/sarama"
 	"github.com/minio/minio/internal/config"
+	"github.com/minio/minio/internal/logger"
+	httpTarget "github.com/minio/minio/internal/logger/target/http"
+	types "github.com/minio/minio/internal/logger/target/loggertypes"
 )
 
+type webhookConfigObjectLayer struct {
+	ObjectLayer
+}
+
+func (webhookConfigObjectLayer) SetDriveCounts() []int { return []int{1} }
+
+// TestDynamicWebhookOutlivesConfigContext applies logger targets the way a
+// peer reload does: under a bounded context that ends when the handler returns.
+func TestDynamicWebhookOutlivesConfigContext(t *testing.T) {
+	for _, subSys := range []string{config.AuditWebhookSubSys, config.LoggerWebhookSubSys, config.AuditKafkaSubSys} {
+		for _, queue := range []struct {
+			name     string
+			disk     bool
+			failInit bool
+		}{
+			{name: "disk=false"},
+			{name: "disk=true", disk: true},
+			{name: "disk=unavailable", disk: true, failInit: true},
+		} {
+			if subSys == config.AuditKafkaSubSys && queue.failInit {
+				// An unopenable Kafka queue starts no goroutine.
+				continue
+			}
+			t.Run(subSys+"/"+queue.name, func(t *testing.T) {
+				// Isolate global target registries and wait for all workers to exit
+				// before the parent removes the disk queue.
+				if os.Getenv("_SILO_WEBHOOK_CONTEXT_TEST") != t.Name() {
+					executable, err := os.Executable()
+					if err != nil {
+						t.Fatal(err)
+					}
+					queueDir := ""
+					if queue.disk {
+						queueDir = t.TempDir()
+					}
+					ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+					defer cancel()
+					child := exec.CommandContext(ctx, executable,
+						"-test.run=^"+strings.ReplaceAll(t.Name(), "/", "$/^")+"$",
+						"-test.v", "-test.timeout=25s")
+					// Logger targets from the environment would join the registries.
+					for _, kv := range os.Environ() {
+						if !strings.HasPrefix(kv, "MINIO_AUDIT_") && !strings.HasPrefix(kv, "MINIO_LOGGER_") {
+							child.Env = append(child.Env, kv)
+						}
+					}
+					child.Env = append(child.Env,
+						"_SILO_WEBHOOK_CONTEXT_TEST="+t.Name(),
+						"_SILO_WEBHOOK_QUEUE_DIR="+queueDir)
+					output, err := child.CombinedOutput()
+					if err != nil {
+						t.Fatalf("isolated webhook test: %v\n%s", err, output)
+					}
+					t.Logf("%s", output)
+					return
+				}
+
+				if subSys == config.AuditKafkaSubSys {
+					testDynamicKafkaOutlivesConfigContext(t, os.Getenv("_SILO_WEBHOOK_QUEUE_DIR"))
+					return
+				}
+
+				targets := logger.AuditTargets
+				if subSys == config.LoggerWebhookSubSys {
+					targets = logger.SystemTargets
+				}
+				received := make(chan string, 8)
+				receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					var entry struct{ Marker string }
+					if err := json.NewDecoder(r.Body).Decode(&entry); err != nil {
+						t.Errorf("decode webhook: %v", err)
+					}
+					received <- entry.Marker
+					w.WriteHeader(http.StatusOK)
+				}))
+				t.Cleanup(receiver.Close)
+
+				queueDir := os.Getenv("_SILO_WEBHOOK_QUEUE_DIR")
+				if queue.failInit {
+					queueDir = filepath.Join(queueDir, "queue")
+					if err := os.WriteFile(queueDir, []byte("not a directory"), 0o600); err != nil {
+						t.Fatal(err)
+					}
+				}
+				cfg := newServerConfig()
+				_, err := cfg.SetKVS(fmt.Sprintf("%s:context-test enable=on endpoint=%s queue_size=16 batch_size=1 queue_dir=%s http_timeout=1s retry_interval=100ms",
+					subSys, receiver.URL, queueDir), config.DefaultKVS)
+				if err != nil {
+					t.Fatal(err)
+				}
+				// A reload cancels the target it replaces; clean up only the last one.
+				var current *httpTarget.Target
+				t.Cleanup(func() {
+					if current != nil {
+						current.Cancel()
+					}
+				})
+				// reload applies the subsystem the way a peer does and returns the
+				// registered target once the bounded reload context has ended.
+				reload := func() *httpTarget.Target {
+					t.Helper()
+					ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+					defer cancel()
+					if err := applyDynamicConfigForSubSys(ctx, webhookConfigObjectLayer{}, cfg, subSys); err != nil {
+						t.Fatal(err)
+					}
+					var target *httpTarget.Target
+					for _, candidate := range targets() {
+						if candidate.Endpoint() == receiver.URL {
+							if target != nil {
+								t.Fatal("configured webhook was registered twice")
+							}
+							target = candidate.(*httpTarget.Target)
+						}
+					}
+					if target == nil {
+						t.Fatal("configured webhook was not registered")
+					}
+					current = target
+					return target
+				}
+
+				target := reload()
+				if err := target.Send(t.Context(), struct{ Marker string }{"after-reload"}); err != nil {
+					t.Fatal(err)
+				}
+				if queue.failInit {
+					// A failed initialization starts no worker: the record stays queued.
+					select {
+					case got := <-received:
+						t.Fatalf("received %q before the queue was initialized", got)
+					case <-time.After(250 * time.Millisecond):
+					}
+					if got := target.Stats().QueueLength; got != 1 {
+						t.Fatalf("failed queue consumed an event: queue length %d", got)
+					}
+					// After the operator repairs queue_dir, applying the configuration
+					// again replaces the target, which must take over the record.
+					if err := os.Remove(queueDir); err != nil {
+						t.Fatal(err)
+					}
+					target = reload()
+				}
+				select {
+				case got := <-received:
+					if got != "after-reload" {
+						t.Fatalf("received marker %q, want after-reload", got)
+					}
+				case <-time.After(10 * time.Second):
+					t.Fatalf("webhook stopped after reload returned; stats: %+v", target.Stats())
+				}
+			})
+		}
+	}
+}
+
+// testDynamicKafkaOutlivesConfigContext runs in the isolated child process.
+// With a queue_dir, Kafka records reach the broker only through the disk
+// replay, which polls the queue every few seconds.
+func testDynamicKafkaOutlivesConfigContext(t *testing.T, queueDir string) {
+	const topic = "audit-context-test"
+	broker := sarama.NewMockBroker(t, 1)
+	t.Cleanup(broker.Close)
+	broker.SetHandlerByMap(map[string]sarama.MockResponse{
+		"ApiVersionsRequest": sarama.NewMockApiVersionsResponse(t).SetApiKeys([]sarama.ApiVersionsResponseKey{
+			{ApiKey: 0, MinVersion: 0, MaxVersion: 8},  // Produce
+			{ApiKey: 3, MinVersion: 0, MaxVersion: 11}, // Metadata
+			{ApiKey: 18, MinVersion: 0, MaxVersion: 3}, // ApiVersions
+		}),
+		"MetadataRequest": sarama.NewMockMetadataResponse(t).
+			SetBroker(broker.Addr(), broker.BrokerID()).
+			SetLeader(topic, 0, broker.BrokerID()),
+		"ProduceRequest": sarama.NewMockProduceResponse(t),
+	})
+	produced := func() (n int) {
+		for _, exchange := range broker.History() {
+			if _, ok := exchange.Request.(*sarama.ProduceRequest); ok {
+				n++
+			}
+		}
+		return n
+	}
+
+	cfg := newServerConfig()
+	_, err := cfg.SetKVS(fmt.Sprintf("%s:context-test enable=on brokers=%s topic=%s queue_size=16 queue_dir=%s",
+		config.AuditKafkaSubSys, broker.Addr(), topic, queueDir), config.DefaultKVS)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+	defer cancel()
+	if err := applyDynamicConfigForSubSys(ctx, webhookConfigObjectLayer{}, cfg, config.AuditKafkaSubSys); err != nil {
+		t.Fatal(err)
+	}
+	var target logger.Target
+	for _, candidate := range logger.AuditTargets() {
+		if candidate.Type() == types.TargetKafka {
+			if target != nil {
+				t.Fatal("more than one Kafka target was registered")
+			}
+			target = candidate
+		}
+	}
+	if target == nil {
+		t.Fatal("configured Kafka target was not registered")
+	}
+	t.Cleanup(target.Cancel)
+	// Returning from the peer reload cancels its bounded context.
+	cancel()
+
+	before := produced()
+	if err := target.Send(t.Context(), struct{ Marker string }{"after-reload"}); err != nil {
+		t.Fatal(err)
+	}
+	for deadline := time.Now().Add(10 * time.Second); produced() == before; time.Sleep(50 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatalf("Kafka target stopped after reload returned; stats: %+v", target.Stats())
+		}
+	}
+}
+
 func TestServerConfig(t *testing.T) {
 	ctx, cancel := context.WithCancel(t.Context())
 	defer cancel()

internal/logger/target/http/http.go

@@ -196,8 +196,6 @@ func (h *Target) Init(ctx context.Context) (err error) {
 func (h *Target) initDiskStore(ctx context.Context) (err error) {
 	ctx, cancel := context.WithCancel(ctx)
 	h.storeCtxCancel = cancel
-	h.lastStarted = time.Now()
-	go h.startQueueProcessor(ctx, true)
 
 	queueStore := store.NewQueueStore[any](
 		filepath.Join(h.config.QueueDir, h.Name()),
@@ -206,10 +204,14 @@ func (h *Target) initDiskStore(ctx context.Context) (err error) {
 	)
 
 	if err := queueStore.Open(); err != nil {
+		cancel()
 		return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
 	}
 
 	h.store = queueStore
+	// The writer may immediately consume entries migrated from an old target.
+	h.lastStarted = time.Now()
+	go h.startQueueProcessor(ctx, true)
 	store.StreamItems(h.store, h, ctx.Done(), h.config.LogOnceIf)
 
 	return nil
@@ -627,6 +629,25 @@ func (h *Target) Cancel() {
 	// and finish the existing ones.
 	// All future ones will be discarded.
 	h.logChMu.Lock()
+	// Hand entries that no worker drained to the replacement, for example
+	// when the queue store never opened and no worker was started.
+	logChLock.Lock()
+	if migrateBuffer, ok := logChBuffers[h.Name()]; ok {
+	drain:
+		for {
+			select {
+			case v := <-h.logCh:
+				select {
+				case migrateBuffer <- v:
+				default:
+					break drain
+				}
+			default:
+				break drain
+			}
+		}
+	}
+	logChLock.Unlock()
 	xioutil.SafeClose(h.logCh)
 	h.logCh = nil
 	h.logChMu.Unlock()