From 0fa4a8c503941d94c60045dc7a4dbff4d93fca34 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 5 Nov 2026 02:20:06 +0800 Subject: [PATCH 1/2] fix(logger): keep reloaded webhook and audit targets alive after a peer reload A runtime change to audit_webhook, logger_webhook or audit_kafka is applied on every peer by serviceReloadDynamic, under a 30-second context that is cancelled as soon as the handler returns. applyDynamicConfigForSubSys passed that context to the logger target updates, and the targets tie their queue worker and, with queue_dir, their disk replay to the Init context. On every server except the one that served the admin request the new targets therefore stopped right after they were created: records filled the in-memory queue and were then dropped with "log buffer full" until the server restarted. - Build logger webhook, audit webhook and audit Kafka targets with GlobalContext, as startup and the serving node already do. The caller's context is still used for logging. - Open the HTTP disk queue store before starting the queue worker, and cancel the derived context when it cannot be opened. A worker started before a failed Open dereferenced the nil store on the first record and crashed the server; with the process context that would have happened on every server. - Let Cancel hand records still waiting in the channel to the shared buffer of the replacement target. A target whose queue store never opened has no worker to do this, so repairing queue_dir and applying the configuration again used to discard the records it had accepted. - TestDynamicWebhookOutlivesConfigContext applies each case through applyDynamicConfigForSubSys under a short context, cancels it as the peer handler does, and requires delivery: audit and logger webhooks with a memory queue, a disk queue and an unopenable disk queue (recovered by applying the configuration again), and audit Kafka with and without queue_dir against a sarama mock broker. Each case runs in a child process without logger and audit environment variables. This completes earlier uncommitted maintainer work on this issue, which covered the two HTTP call sites, the store-before-worker order and the first version of the test. Refs #259 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Feng Ruohang --- cmd/config-current.go | 9 +- cmd/config-current_test.go | 234 ++++++++++++++++++++++++++++ internal/logger/target/http/http.go | 25 ++- 3 files changed, 263 insertions(+), 5 deletions(-) diff --git a/cmd/config-current.go b/cmd/config-current.go index 64ca0673e..c0de6ebf3 100644 --- a/cmd/config-current.go +++ b/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: diff --git a/cmd/config-current_test.go b/cmd/config-current_test.go index 338cebbe8..b19246678 100644 --- a/cmd/config-current_test.go +++ b/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() diff --git a/internal/logger/target/http/http.go b/internal/logger/target/http/http.go index 1deb98666..be330b482 100644 --- a/internal/logger/target/http/http.go +++ b/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() From 14d49f14f7558e8eafb529b0e005b8d246ba73cf Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Thu, 5 Nov 2026 16:08:11 +0800 Subject: [PATCH 2/2] test: record webhook environment prefix compatibility Signed-off-by: Feng Ruohang --- buildscripts/rebrand-guard/compat-baseline.json | 2 ++ 1 file changed, 2 insertions(+) diff --git a/buildscripts/rebrand-guard/compat-baseline.json b/buildscripts/rebrand-guard/compat-baseline.json index 2ff232340..252c75de8 100644 --- a/buildscripts/rebrand-guard/compat-baseline.json +++ b/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",