diff --git a/weed/command/scaffold/notification.toml b/weed/command/scaffold/notification.toml index ebbd240f9f..7d327984e4 100644 --- a/weed/command/scaffold/notification.toml +++ b/weed/command/scaffold/notification.toml @@ -81,3 +81,16 @@ workers = 5 # optional: concurrent worker buffer_size = 10000 # optional: event buffer size (default: 10000, range: 100-1000000) # event_types = ["create", "update", "delete", "rename"] # optional: filter by event types (default: all) # path_prefixes = ["/important", "/data"] # optional: filter by path prefixes (default: all) +# +# Additional endpoints can be declared as named subsections, each with its own +# endpoint, filters, retry and worker settings, e.g.: +# [notification.webhook.file_ingress] +# enabled = true +# endpoint = "https://ingress.example.com/hook" +# event_types = ["create"] +# path_prefixes = ["/buckets/uploads/data"] +# +# [notification.webhook.metadata_extractor] +# enabled = true +# endpoint = "https://metadata.example.com/hook" +# path_prefixes = ["/buckets/repository"] diff --git a/weed/notification/configuration.go b/weed/notification/configuration.go index 1c620f2e6a..d271323149 100644 --- a/weed/notification/configuration.go +++ b/weed/notification/configuration.go @@ -26,10 +26,10 @@ func LoadConfiguration(config *util.ViperProxy, prefix string) { return } - validateOneEnabledQueue(config) + validateOneEnabledQueue(config, prefix) for _, queue := range MessageQueues { - if config.GetBool(prefix + queue.GetName() + ".enabled") { + if queueEnabled(config, prefix, queue) { if err := queue.Initialize(config, prefix+queue.GetName()+"."); err != nil { glog.Fatalf("Failed to initialize notification for %s: %+v", queue.GetName(), err) @@ -42,10 +42,26 @@ func LoadConfiguration(config *util.ViperProxy, prefix string) { } -func validateOneEnabledQueue(config *util.ViperProxy) { +// queueEnabled reports whether a queue is switched on — by the flat +// ".enabled" flag or, for queues accepting named sub-instances +// ([notification..]), by any enabled subconfig. +func queueEnabled(config util.Configuration, prefix string, queue MessageQueue) bool { + key := prefix + queue.GetName() + "." + if config.GetBool(key + "enabled") { + return true + } + if q, ok := queue.(interface { + HasEnabledSubconfig(util.Configuration, string) bool + }); ok { + return q.HasEnabledSubconfig(config, key) + } + return false +} + +func validateOneEnabledQueue(config *util.ViperProxy, prefix string) { enabledQueue := "" for _, queue := range MessageQueues { - if config.GetBool(queue.GetName() + ".enabled") { + if queueEnabled(config, prefix, queue) { if enabledQueue == "" { enabledQueue = queue.GetName() } else { diff --git a/weed/notification/webhook/types.go b/weed/notification/webhook/types.go index f177f635a4..1567e832f2 100644 --- a/weed/notification/webhook/types.go +++ b/weed/notification/webhook/types.go @@ -3,7 +3,6 @@ package webhook import ( "fmt" "net/url" - "strconv" "github.com/seaweedfs/seaweedfs/weed/notification" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -30,12 +29,6 @@ func (e eventType) valid() bool { return notification.ValidEventType(string(e)) } -var ( - pubSubHandlerNameTemplate = func(n int) string { - return "webhook_handler_" + strconv.Itoa(n) - } -) - type client interface { sendMessage(message *webhookMessage) error } diff --git a/weed/notification/webhook/webhook_queue.go b/weed/notification/webhook/webhook_queue.go index 2ac9db3ebb..afaa326245 100644 --- a/weed/notification/webhook/webhook_queue.go +++ b/weed/notification/webhook/webhook_queue.go @@ -4,6 +4,9 @@ import ( "context" "errors" "fmt" + "sort" + "strings" + "sync" "time" "github.com/ThreeDotsLabs/watermill" @@ -22,45 +25,176 @@ func init() { notification.MessageQueues = append(notification.MessageQueues, &Queue{}) } +// target is one webhook endpoint with its own filter, http client and +// concurrency budget. The flat [notification.webhook] section is the unnamed +// target; each enabled [notification.webhook.] subsection adds another. +type target struct { + name string + config *config + filter *filter + client client + sem chan struct{} + inbox chan *message.Message + pending chan struct{} +} + +func newTarget(name string, cfg *config) (*target, error) { + t := &target{ + name: name, + config: cfg, + filter: newFilter(cfg), + sem: make(chan struct{}, cfg.nWorkers), + inbox: make(chan *message.Message, cfg.bufferSize), + pending: make(chan struct{}, cfg.bufferSize), + } + hClient, err := newHTTPClient(cfg) + if err != nil { + return nil, fmt.Errorf("failed to create webhook http client: %w", err) + } + t.client = hClient + return t, nil +} + +func (t *target) topic() string { + if t.name == "" { + return pubSubTopicName + } + return pubSubTopicName + "_" + t.name +} + +func (t *target) handlerName() string { + if t.name == "" { + return "webhook_handler" + } + return "webhook_handler_" + t.name +} + type Queue struct { - router *message.Router + routers []*message.Router queueChannel *gochannel.GoChannel - config *config - client client - filter *filter + targets []*target ctx context.Context cancel context.CancelFunc - - // Semaphore for controlling concurrent webhook requests - sem chan struct{} } func (w *Queue) GetName() string { return queueName } +// HasEnabledSubconfig reports whether any [notification.webhook.] +// subsection is enabled, so the queue activates on named targets even when the +// flat [notification.webhook] section is absent. +func (w *Queue) HasEnabledSubconfig(configuration util.Configuration, prefix string) bool { + return len(namedTargetPrefixes(configuration, prefix)) > 0 +} + +// stringMapConfig is implemented by *util.ViperProxy; anything narrower simply +// gets no named-target discovery. +type stringMapConfig interface { + GetStringMap(key string) map[string]interface{} +} + +func namedTargetPrefixes(configuration util.Configuration, prefix string) []string { + mg, ok := configuration.(stringMapConfig) + if !ok { + return nil + } + var prefixes []string + for k, v := range mg.GetStringMap(strings.TrimSuffix(prefix, ".")) { + if _, ok := v.(map[string]interface{}); !ok { + continue + } + p := prefix + k + "." + if !configuration.GetBool(p + "enabled") { + continue + } + prefixes = append(prefixes, p) + } + sort.Strings(prefixes) + return prefixes +} + +func buildTargets(configuration util.Configuration, prefix string) ([]*target, error) { + var prefixes []string + if configuration.GetBool(prefix + "enabled") { + prefixes = append(prefixes, prefix) + } + prefixes = append(prefixes, namedTargetPrefixes(configuration, prefix)...) + + var targets []*target + for _, p := range prefixes { + cfg := newConfigWithDefaults(configuration, p) + if err := cfg.validate(); err != nil { + return nil, err + } + t, err := newTarget(strings.TrimSuffix(strings.TrimPrefix(p, prefix), "."), cfg) + if err != nil { + return nil, err + } + targets = append(targets, t) + } + return targets, nil +} + func (w *Queue) SendMessage(key string, msg proto.Message) error { eventNotification, ok := msg.(*filer_pb.EventNotification) if !ok { return nil } - if w.filter != nil && !w.filter.shouldPublish(key, eventNotification) { + var deferred []routedMessage + for _, t := range w.targets { + if t.filter != nil && !t.filter.shouldPublish(key, eventNotification) { + continue + } + m := newWebhookMessage(key, msg) + if m == nil { + continue + } + wMsg, err := m.toWaterMillMessage() + if err != nil { + return err + } + wMsg.Metadata.Set("target", t.name) + // a free reservation routes the event without waiting on slower + // targets; a saturated target defers to the blocking pass below so + // it applies backpressure instead of dropping or starving the rest + select { + case t.pending <- struct{}{}: + if err := t.enqueue(w.ctx, wMsg); err != nil { + return err + } + default: + deferred = append(deferred, routedMessage{t, wMsg}) + } + } + for _, r := range deferred { + select { + case r.target.pending <- struct{}{}: + case <-w.ctx.Done(): + return w.ctx.Err() + } + if err := r.target.enqueue(w.ctx, r.msg); err != nil { + return err + } + } + return nil +} + +type routedMessage struct { + target *target + msg *message.Message +} + +func (t *target) enqueue(ctx context.Context, wMsg *message.Message) error { + select { + case t.inbox <- wMsg: return nil + case <-ctx.Done(): + t.releasePending() + return ctx.Err() } - - m := newWebhookMessage(key, msg) - if m == nil { - return nil - } - - wMsg, err := m.toWaterMillMessage() - if err != nil { - return err - } - - return w.queueChannel.Publish(pubSubTopicName, wMsg) } func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) { @@ -78,105 +212,158 @@ func (w *webhookMessage) toWaterMillMessage() (*message.Message, error) { } func (w *Queue) Initialize(configuration util.Configuration, prefix string) error { - c := newConfigWithDefaults(configuration, prefix) - - if err := c.validate(); err != nil { + targets, err := buildTargets(configuration, prefix) + if err != nil { return err } + if len(targets) == 0 { + return fmt.Errorf("webhook endpoint is required") + } - return w.initialize(c) + return w.initialize(targets) } -func (w *Queue) initialize(cfg *config) error { +func (w *Queue) initialize(targets []*target) error { w.ctx, w.cancel = context.WithCancel(context.Background()) - w.config = cfg - w.filter = newFilter(cfg) + w.targets = targets - // Initialize semaphore for controlling concurrent webhook requests - w.sem = make(chan struct{}, cfg.nWorkers) - - hClient, err := newHTTPClient(cfg) - if err != nil { - return fmt.Errorf("failed to create webhook http client: %w", err) - } - w.client = hClient - - if err = w.setupWatermillQueue(cfg); err != nil { + if err := w.setupWatermillQueue(targets); err != nil { return fmt.Errorf("failed to setup watermill queue: %w", err) } - if err = w.logDeadLetterMessages(); err != nil { + if err := w.logDeadLetterMessages(); err != nil { return err } return nil } -func (w *Queue) setupWatermillQueue(cfg *config) error { +func (w *Queue) setupWatermillQueue(targets []*target) error { logger := watermill.NewStdLogger(false, false) + bufferSize := 0 + for _, t := range targets { + if t.config.bufferSize > bufferSize { + bufferSize = t.config.bufferSize + } + } pubSubConfig := gochannel.Config{ - OutputChannelBuffer: int64(cfg.bufferSize), + OutputChannelBuffer: int64(bufferSize), Persistent: false, } w.queueChannel = gochannel.NewGoChannel(pubSubConfig, logger) - router, err := message.NewRouter( - message.RouterConfig{ - CloseTimeout: 60 * time.Second, - }, - logger, - ) - if err != nil { - return fmt.Errorf("failed to create router: %w", err) - } - w.router = router + var wg sync.WaitGroup + for _, t := range targets { + // One router per target so each keeps its own retry/backoff settings. + // Handlers subscribe to the target's own topic, so a message is + // delivered to an endpoint exactly once per matching target. + router, err := message.NewRouter( + message.RouterConfig{ + CloseTimeout: 60 * time.Second, + }, + logger, + ) + if err != nil { + return fmt.Errorf("failed to create router: %w", err) + } + w.routers = append(w.routers, router) - retryMiddleware := middleware.Retry{ - MaxRetries: cfg.maxRetries, - InitialInterval: time.Duration(cfg.backoffSeconds) * time.Second, - MaxInterval: time.Duration(cfg.maxBackoffSeconds) * time.Second, - Multiplier: 2.0, - RandomizationFactor: 0.3, - Logger: logger, - }.Middleware + retryMiddleware := middleware.Retry{ + MaxRetries: t.config.maxRetries, + InitialInterval: time.Duration(t.config.backoffSeconds) * time.Second, + MaxInterval: time.Duration(t.config.maxBackoffSeconds) * time.Second, + Multiplier: 2.0, + RandomizationFactor: 0.3, + Logger: logger, + }.Middleware - poisonQueue, err := middleware.PoisonQueue(w.queueChannel, deadLetterTopic) - if err != nil { - return fmt.Errorf("failed to create poison queue: %w", err) - } - - router.AddPlugin(plugin.SignalsHandler) - router.AddMiddleware(retryMiddleware, poisonQueue) - - // Add a single handler to avoid duplicate message delivery. - // With gochannel's default behavior, each handler call creates - // a separate subscription, and all subscriptions receive their own copy of each message. - // Using a single handler ensures each webhook is sent only once. - // Concurrency is controlled via semaphore in handleWebhook based on nWorkers config. - router.AddConsumerHandler( - "webhook_handler", - pubSubTopicName, - w.queueChannel, - w.handleWebhook, - ) - - go func() { - // cancels the queue context so the dead letter logger exists in case context not canceled by the shutdown signal already - defer w.cancel() - - if err := router.Run(w.ctx); err != nil && !errors.Is(err, context.Canceled) { - glog.Errorf("webhook pubsub worker stopped with error: %v", err) + poisonQueue, err := middleware.PoisonQueue(w.queueChannel, deadLetterTopic) + if err != nil { + return fmt.Errorf("failed to create poison queue: %w", err) } - glog.Info("webhook pubsub worker stopped") + router.AddPlugin(plugin.SignalsHandler) + // outermost: the reservation SendMessage took releases once the whole + // chain (retries, dead letter) is done, bounding in-flight deliveries. + // poison wraps retry so the delivery is retried before dead-lettering. + releasePending := func(h message.HandlerFunc) message.HandlerFunc { + return func(msg *message.Message) ([]*message.Message, error) { + defer t.releasePending() + return h(msg) + } + } + router.AddMiddleware(releasePending, poisonQueue, retryMiddleware) + + t := t + router.AddConsumerHandler( + t.handlerName(), + t.topic(), + w.queueChannel, + t.handleWebhook, + ) + + routerStopped := make(chan struct{}) + wg.Add(1) + go func() { + defer wg.Done() + defer close(routerStopped) + if err := router.Run(w.ctx); err != nil && !errors.Is(err, context.Canceled) { + glog.Errorf("webhook pubsub worker stopped with error: %v", err) + } + glog.Info("webhook pubsub worker stopped") + }() + + go w.forwardTargetMessages(t) + + // without a subscriber GoChannel drops the publish as undeliverable + select { + case <-router.Running(): + case <-routerStopped: + return fmt.Errorf("webhook router for target %q stopped before it started", t.name) + case <-w.ctx.Done(): + return w.ctx.Err() + } + } + + // the queue context lives until every target's worker has stopped so the + // dead letter logger keeps running while any of them is still up + go func() { + wg.Wait() + w.cancel() }() return nil } -func (w *Queue) handleWebhook(msg *message.Message) error { +// forwardTargetMessages publishes each queued event to the target's topic in +// order. Publish detaches its delivery, so the pending reservation is what +// keeps a saturated target from queuing without bound. +func (w *Queue) forwardTargetMessages(t *target) { + for { + select { + case msg := <-t.inbox: + if err := w.queueChannel.Publish(t.topic(), msg); err != nil { + glog.Errorf("webhook target %q failed to publish message: %v", t.name, err) + t.releasePending() + } + case <-w.ctx.Done(): + return + } + } +} + +// releasePending frees one delivery reservation. It never blocks: a nacked +// message GoChannel resends reaches the release again after its token is gone. +func (t *target) releasePending() { + select { + case <-t.pending: + default: + } +} + +func (t *target) handleWebhook(msg *message.Message) error { // Acquire semaphore slot (blocks if at capacity) - w.sem <- struct{}{} - defer func() { <-w.sem }() + t.sem <- struct{}{} + defer func() { <-t.sem }() var n filer_pb.EventNotification if err := proto.Unmarshal(msg.Payload, &n); err != nil { @@ -191,8 +378,8 @@ func (w *Queue) handleWebhook(msg *message.Message) error { Notification: &n, } - if err := w.client.sendMessage(webhookMsg); err != nil { - glog.Errorf("failed to send message to webhook %s: %v", webhookMsg.Key, err) + if err := t.client.sendMessage(webhookMsg); err != nil { + glog.Errorf("webhook target %q failed to send message for %s: %v", t.name, webhookMsg.Key, err) return err } @@ -218,10 +405,12 @@ func (w *Queue) logDeadLetterMessages() error { continue } key := "unknown" + targetName := "" if msg.Metadata != nil { if keyValue, exists := msg.Metadata["key"]; exists { key = keyValue } + targetName = msg.Metadata["target"] } payload := "" if msg.Payload != nil { @@ -232,7 +421,7 @@ func (w *Queue) logDeadLetterMessages() error { payload = n.String() } } - glog.Errorf("received dead letter message: %s, key: %s", payload, key) + glog.Errorf("received dead letter message: %s, key: %s, target: %s", payload, key, targetName) case <-w.ctx.Done(): return } diff --git a/weed/notification/webhook/webhook_queue_test.go b/weed/notification/webhook/webhook_queue_test.go index a084ef975b..ba75ecb147 100644 --- a/weed/notification/webhook/webhook_queue_test.go +++ b/weed/notification/webhook/webhook_queue_test.go @@ -4,6 +4,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "time" @@ -170,6 +171,29 @@ func TestWebhookMessageSerialization(t *testing.T) { } } +func newQueueForTest(t *testing.T, cfg *config) *Queue { + t.Helper() + tgt, err := newTarget("", cfg) + if err != nil { + t.Fatalf("newTarget() error = %v", err) + } + q := &Queue{} + if err := q.initialize([]*target{tgt}); err != nil { + t.Fatalf("initialize() error = %v", err) + } + return q +} + +func closeQueueForTest(q *Queue) { + if q.cancel != nil { + q.cancel() + } + time.Sleep(100 * time.Millisecond) + for _, r := range q.routers { + r.Close() + } +} + func TestQueueInitialize(t *testing.T) { cfg := &config{ endpoint: "https://example.com/webhook", @@ -182,32 +206,19 @@ func TestQueueInitialize(t *testing.T) { bufferSize: 100, } - q := &Queue{} - err := q.initialize(cfg) - if err != nil { - t.Errorf("Initialize() error = %v", err) - } + q := newQueueForTest(t, cfg) + defer closeQueueForTest(q) - defer func() { - if q.cancel != nil { - q.cancel() - } - time.Sleep(100 * time.Millisecond) - if q.router != nil { - q.router.Close() - } - }() - - if q.router == nil { + if len(q.routers) != 1 { t.Error("Expected router to be initialized") } if q.queueChannel == nil { t.Error("Expected queueChannel to be initialized") } - if q.client == nil { + if q.targets[0].client == nil { t.Error("Expected client to be initialized") } - if q.config == nil { + if q.targets[0].config == nil { t.Error("Expected config to be initialized") } } @@ -225,21 +236,8 @@ func TestQueueSendMessage(t *testing.T) { bufferSize: 10, } - q := &Queue{} - err := q.initialize(cfg) - if err != nil { - t.Fatalf("Failed to initialize queue: %v", err) - } - - defer func() { - if q.cancel != nil { - q.cancel() - } - time.Sleep(100 * time.Millisecond) - if q.router != nil { - q.router.Close() - } - }() + q := newQueueForTest(t, cfg) + defer closeQueueForTest(q) msg := &filer_pb.EventNotification{ NewEntry: &filer_pb.Entry{ @@ -247,8 +245,7 @@ func TestQueueSendMessage(t *testing.T) { }, } - err = q.SendMessage("/test/path", msg) - if err != nil { + if err := q.SendMessage("/test/path", msg); err != nil { t.Errorf("SendMessage() error = %v", err) } } @@ -271,7 +268,7 @@ func TestQueueHandleWebhook(t *testing.T) { } client, _ := newHTTPClient(cfg) - q := &Queue{ + tgt := &target{ client: client, sem: make(chan struct{}, cfg.nWorkers), } @@ -287,8 +284,7 @@ func TestQueueHandleWebhook(t *testing.T) { t.Fatalf("Failed to create watermill message: %v", err) } - err = q.handleWebhook(wmMsg) - if err != nil { + if err := tgt.handleWebhook(wmMsg); err != nil { t.Errorf("handleWebhook() error = %v", err) } } @@ -307,21 +303,8 @@ func TestQueueEndToEnd(t *testing.T) { bufferSize: 10, } - q := &Queue{} - err := q.initialize(cfg) - if err != nil { - t.Fatalf("Failed to initialize queue: %v", err) - } - - defer func() { - if q.cancel != nil { - q.cancel() - } - time.Sleep(100 * time.Millisecond) - if q.router != nil { - q.router.Close() - } - }() + q := newQueueForTest(t, cfg) + defer closeQueueForTest(q) msg := &filer_pb.EventNotification{ NewEntry: &filer_pb.Entry{ @@ -329,8 +312,7 @@ func TestQueueEndToEnd(t *testing.T) { }, } - err = q.SendMessage("/test/path", msg) - if err != nil { + if err := q.SendMessage("/test/path", msg); err != nil { t.Errorf("SendMessage() error = %v", err) } } @@ -347,33 +329,20 @@ func TestQueueRetryMechanism(t *testing.T) { bufferSize: 10, } - q := &Queue{} - err := q.initialize(cfg) - if err != nil { - t.Fatalf("Failed to initialize queue: %v", err) - } - - defer func() { - if q.cancel != nil { - q.cancel() - } - time.Sleep(100 * time.Millisecond) - if q.router != nil { - q.router.Close() - } - }() + q := newQueueForTest(t, cfg) + defer closeQueueForTest(q) // Verify that the queue is properly configured for retries - if q.config.maxRetries != 3 { - t.Errorf("Expected maxRetries=3, got %d", q.config.maxRetries) + if q.targets[0].config.maxRetries != 3 { + t.Errorf("Expected maxRetries=3, got %d", q.targets[0].config.maxRetries) } - if q.config.backoffSeconds != 2 { - t.Errorf("Expected backoffSeconds=2, got %d", q.config.backoffSeconds) + if q.targets[0].config.backoffSeconds != 2 { + t.Errorf("Expected backoffSeconds=2, got %d", q.targets[0].config.backoffSeconds) } - if q.config.maxBackoffSeconds != 10 { - t.Errorf("Expected maxBackoffSeconds=10, got %d", q.config.maxBackoffSeconds) + if q.targets[0].config.maxBackoffSeconds != 10 { + t.Errorf("Expected maxBackoffSeconds=10, got %d", q.targets[0].config.maxBackoffSeconds) } // Test that we can send a message (retry behavior is handled by Watermill middleware) @@ -381,8 +350,7 @@ func TestQueueRetryMechanism(t *testing.T) { NewEntry: &filer_pb.Entry{Name: "test.txt"}, } - err = q.SendMessage("/test/retry", msg) - if err != nil { + if err := q.SendMessage("/test/retry", msg); err != nil { t.Errorf("SendMessage() error = %v", err) } } @@ -419,21 +387,8 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) { bufferSize: 10, } - q := &Queue{} - err := q.initialize(cfg) - if err != nil { - t.Fatalf("Failed to initialize queue: %v", err) - } - - defer func() { - if q.cancel != nil { - q.cancel() - } - time.Sleep(100 * time.Millisecond) - if q.router != nil { - q.router.Close() - } - }() + q := newQueueForTest(t, cfg) + defer closeQueueForTest(q) // Wait for router and subscriber to be fully ready time.Sleep(200 * time.Millisecond) @@ -444,8 +399,7 @@ func TestQueueNoDuplicateWebhooks(t *testing.T) { }, } - err = q.SendMessage("/test/path", msg) - if err != nil { + if err := q.SendMessage("/test/path", msg); err != nil { t.Errorf("SendMessage() error = %v", err) } @@ -607,3 +561,298 @@ func TestQueueSendMessageWithFilter(t *testing.T) { }) } } + +type mapConfig struct { + values map[string]interface{} +} + +func (m *mapConfig) GetString(key string) string { + v, _ := m.values[key].(string) + return v +} + +func (m *mapConfig) GetBool(key string) bool { + v, _ := m.values[key].(bool) + return v +} + +func (m *mapConfig) GetInt(key string) int { + v, _ := m.values[key].(int) + return v +} + +func (m *mapConfig) GetStringSlice(key string) []string { + v, _ := m.values[key].([]string) + return v +} + +func (m *mapConfig) SetDefault(key string, value interface{}) {} + +func (m *mapConfig) GetStringMap(key string) map[string]interface{} { + v, _ := m.values[key].(map[string]interface{}) + return v +} + +func TestNamedTargets(t *testing.T) { + var hitsA, hitsB atomic.Int32 + serverA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hitsA.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer serverA.Close() + serverB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hitsB.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer serverB.Close() + + cfg := &mapConfig{values: map[string]interface{}{ + "notification.webhook": map[string]interface{}{ + "file_ingress": map[string]interface{}{ + "enabled": true, + "endpoint": serverA.URL, + "path_prefixes": []string{"/data/"}, + }, + "metadata_extractor": map[string]interface{}{ + "enabled": true, + "endpoint": serverB.URL, + "path_prefixes": []string{"/repository/"}, + }, + "disabled_one": map[string]interface{}{ + "enabled": false, + "endpoint": "https://example.com/never", + }, + }, + "notification.webhook.file_ingress.enabled": true, + "notification.webhook.file_ingress.endpoint": serverA.URL, + "notification.webhook.file_ingress.path_prefixes": []string{"/data/"}, + "notification.webhook.metadata_extractor.enabled": true, + "notification.webhook.metadata_extractor.endpoint": serverB.URL, + "notification.webhook.metadata_extractor.path_prefixes": []string{"/repository/"}, + }} + + q := &Queue{} + if !q.HasEnabledSubconfig(cfg, "notification.webhook.") { + t.Fatal("expected named subconfig to be detected as enabled") + } + + targets, err := buildTargets(cfg, "notification.webhook.") + if err != nil { + t.Fatalf("buildTargets() error = %v", err) + } + if len(targets) != 2 { + t.Fatalf("expected 2 targets, got %d", len(targets)) + } + if targets[0].name != "file_ingress" || targets[1].name != "metadata_extractor" { + t.Fatalf("unexpected target names: %s, %s", targets[0].name, targets[1].name) + } + + if err := q.initialize(targets); err != nil { + t.Fatalf("initialize() error = %v", err) + } + defer closeQueueForTest(q) + + time.Sleep(200 * time.Millisecond) + + msg := &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{Name: "f.txt"}, + } + for _, key := range []string{"/data/f.txt", "/repository/f.txt", "/elsewhere/f.txt"} { + if err := q.SendMessage(key, msg); err != nil { + t.Errorf("SendMessage(%q) error = %v", key, err) + } + } + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) && (hitsA.Load() == 0 || hitsB.Load() == 0) { + time.Sleep(50 * time.Millisecond) + } + time.Sleep(200 * time.Millisecond) + + if got := hitsA.Load(); got != 1 { + t.Errorf("expected file_ingress to receive 1 event, got %d", got) + } + if got := hitsB.Load(); got != 1 { + t.Errorf("expected metadata_extractor to receive 1 event, got %d", got) + } +} + +func TestQueueBurstKeepsPendingEvents(t *testing.T) { + var hits atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + time.Sleep(20 * time.Millisecond) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + cfg := &config{ + endpoint: server.URL, + timeoutSeconds: 5, + maxRetries: 0, + backoffSeconds: 1, + maxBackoffSeconds: 1, + nWorkers: 1, + bufferSize: 2, + } + q := newQueueForTest(t, cfg) + defer closeQueueForTest(q) + + const total = 6 + msg := &filer_pb.EventNotification{NewEntry: &filer_pb.Entry{Name: "f.txt"}} + for i := 0; i < total; i++ { + if err := q.SendMessage("/test/path", msg); err != nil { + t.Fatalf("SendMessage() error = %v", err) + } + } + + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) && hits.Load() < total { + time.Sleep(50 * time.Millisecond) + } + if got := hits.Load(); got != total { + t.Errorf("expected %d deliveries, got %d", total, got) + } +} + +// A saturated target still applies backpressure to the sender, but it must +// not starve the targets after it: the event reaches a healthy endpoint +// while the slow one is still out of pending slots. +func TestSaturatedTargetDoesNotStarveOthers(t *testing.T) { + var hitsA, hitsB atomic.Int32 + serverA := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hitsA.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer serverA.Close() + serverB := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hitsB.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer serverB.Close() + + newTestTarget := func(name, endpoint string) *target { + tgt, err := newTarget(name, &config{ + endpoint: endpoint, + timeoutSeconds: 5, + maxRetries: 0, + backoffSeconds: 1, + maxBackoffSeconds: 1, + nWorkers: 1, + bufferSize: 2, + }) + if err != nil { + t.Fatalf("newTarget() error = %v", err) + } + return tgt + } + slow := newTestTarget("slow", serverA.URL) + fast := newTestTarget("fast", serverB.URL) + + q := &Queue{} + if err := q.initialize([]*target{slow, fast}); err != nil { + t.Fatalf("initialize() error = %v", err) + } + defer closeQueueForTest(q) + + for i := 0; i < cap(slow.pending); i++ { + slow.pending <- struct{}{} + } + + done := make(chan error, 1) + go func() { + done <- q.SendMessage("/test/path", &filer_pb.EventNotification{NewEntry: &filer_pb.Entry{Name: "f.txt"}}) + }() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) && hitsB.Load() == 0 { + time.Sleep(20 * time.Millisecond) + } + if got := hitsB.Load(); got != 1 { + t.Fatalf("expected fast target to receive the event while slow is saturated, got %d", got) + } + select { + case err := <-done: + t.Fatalf("SendMessage() returned %v while the saturated target had no free slot", err) + default: + } + + <-slow.pending + if err := <-done; err != nil { + t.Fatalf("SendMessage() error = %v", err) + } + + deadline = time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) && hitsA.Load() == 0 { + time.Sleep(20 * time.Millisecond) + } + if got := hitsA.Load(); got != 1 { + t.Errorf("expected the saturated target to deliver once a slot freed, got %d", got) + } +} + +// A disabled flat section keeps no target even when an endpoint is set — +// the scaffold ships exactly that combination, and a named target must not +// reactivate it. +func TestBuildTargetsDisabledFlat(t *testing.T) { + cfg := &mapConfig{values: map[string]interface{}{ + "notification.webhook": map[string]interface{}{ + "enabled": false, + "endpoint": "https://your-server.com/webhook", + "named": map[string]interface{}{ + "enabled": true, + "endpoint": "https://example.com/hook", + }, + }, + "notification.webhook.enabled": false, + "notification.webhook.endpoint": "https://your-server.com/webhook", + "notification.webhook.named.enabled": true, + "notification.webhook.named.endpoint": "https://example.com/hook", + }} + + targets, err := buildTargets(cfg, "notification.webhook.") + if err != nil { + t.Fatalf("buildTargets() error = %v", err) + } + if len(targets) != 1 || targets[0].name != "named" { + t.Fatalf("expected only the named target, got %v", targets) + } +} + +// A failing endpoint must be retried maxRetries times before the event is +// dead-lettered — the poison middleware has to sit outside the retry chain +// or it swallows the first error and retries never run. +func TestQueueRetriesBeforeDeadLetter(t *testing.T) { + var hits atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + w.WriteHeader(http.StatusInternalServerError) + })) + defer server.Close() + + cfg := &config{ + endpoint: server.URL, + timeoutSeconds: 5, + maxRetries: 2, + backoffSeconds: 1, + maxBackoffSeconds: 1, + nWorkers: 1, + bufferSize: 10, + } + q := newQueueForTest(t, cfg) + defer closeQueueForTest(q) + + msg := &filer_pb.EventNotification{NewEntry: &filer_pb.Entry{Name: "f.txt"}} + if err := q.SendMessage("/test/path", msg); err != nil { + t.Fatalf("SendMessage() error = %v", err) + } + + want := int32(cfg.maxRetries + 1) + deadline := time.Now().Add(15 * time.Second) + for time.Now().Before(deadline) && hits.Load() < want { + time.Sleep(50 * time.Millisecond) + } + if got := hits.Load(); got != want { + t.Errorf("expected %d delivery attempts (1 + %d retries), got %d", want, cfg.maxRetries, got) + } +}