fix: notification config not broadcast to swarm nodes (#4563)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

internal/agent/client_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/amir20/dozzle/internal/container"
 	"github.com/amir20/dozzle/internal/utils"
+	"github.com/amir20/dozzle/types"
 	"github.com/go-faker/faker/v4"
 	"github.com/go-faker/faker/v4/pkg/options"
 	"github.com/stretchr/testify/assert"
@@ -28,6 +29,16 @@ var lis *bufconn.Listener
 var certs tls.Certificate
 var mockService *MockedClientService
 
+type mockNotificationHandler struct{}
+
+func (m *mockNotificationHandler) HandleNotificationConfig(subscriptions []types.SubscriptionConfig, dispatchers []types.DispatcherConfig) error {
+	return nil
+}
+
+func (m *mockNotificationHandler) GetNotificationStats() []types.SubscriptionStats {
+	return nil
+}
+
 type MockedClientService struct {
 	mock.Mock
 }
@@ -134,7 +145,7 @@ func init() {
 
 	mockService.On("Client").Return(nil)
 
-	server, _ := NewServer(mockService, certs, "test", nil)
+	server, _ := NewServer(mockService, certs, "test", &mockNotificationHandler{})
 	go server.Serve(lis)
 }
 

internal/agent/pb/rpc.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.36.11
-// 	protoc        v7.34.0
+// 	protoc        v7.34.1
 // source: rpc.proto
 
 package pb

internal/agent/pb/rpc_grpc.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.6.0
-// - protoc             v7.34.0
+// - protoc             v7.34.1
 // source: rpc.proto
 
 package pb

internal/agent/pb/types.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.36.11
-// 	protoc        v7.34.0
+// 	protoc        v7.34.1
 // source: types.proto
 
 package pb

internal/agent/server.go

@@ -58,6 +58,10 @@ type server struct {
 }
 
 func newServer(service ClientService, dozzleVersion string, notificationHandler NotificationConfigHandler) pb.AgentServiceServer {
+	if notificationHandler == nil {
+		log.Fatal().Msg("No notification config handler registered")
+	}
+
 	return &server{
 		service:                   service,
 		version:                   dozzleVersion,
@@ -390,11 +394,6 @@ func (s *server) ContainerAttach(stream pb.AgentService_ContainerAttachServer) e
 }
 
 func (s *server) UpdateNotificationConfig(ctx context.Context, req *pb.UpdateNotificationConfigRequest) (*pb.UpdateNotificationConfigResponse, error) {
-	if s.notificationConfigHandler == nil {
-		log.Warn().Msg("No notification config handler registered, ignoring config update")
-		return &pb.UpdateNotificationConfigResponse{}, nil
-	}
-
 	// Validate request sizes to prevent memory exhaustion
 	const maxSubscriptions = 1000
 	const maxDispatchers = 100
@@ -446,17 +445,13 @@ func (s *server) UpdateNotificationConfig(ctx context.Context, req *pb.UpdateNot
 }
 
 func (s *server) GetNotificationStats(ctx context.Context, req *pb.GetNotificationStatsRequest) (*pb.GetNotificationStatsResponse, error) {
-	if s.notificationConfigHandler == nil {
-		return &pb.GetNotificationStatsResponse{}, nil
-	}
-
 	stats := s.notificationConfigHandler.GetNotificationStats()
 
 	pbStats := make([]*pb.NotificationSubscriptionStats, len(stats))
 	for i, s := range stats {
 		pbStat := &pb.NotificationSubscriptionStats{
-			SubscriptionId:       int32(s.SubscriptionID),
-			TriggerCount:         s.TriggerCount,
+			SubscriptionId:        int32(s.SubscriptionID),
+			TriggerCount:          s.TriggerCount,
 			TriggeredContainerIds: s.TriggeredContainerIDs,
 		}
 		if s.LastTriggeredAt != nil {

internal/support/docker/multi_host_service.go

@@ -253,6 +253,7 @@ func (m *MultiHostService) broadcastNotificationConfig() {
 			LogExpression:       sub.LogExpression,
 			ContainerExpression: sub.ContainerExpression,
 			MetricExpression:    sub.MetricExpression,
+			EventExpression:     sub.EventExpression,
 			Cooldown:            sub.Cooldown,
 			SampleWindow:        sub.SampleWindow,
 		}
@@ -267,6 +268,7 @@ func (m *MultiHostService) broadcastNotificationConfig() {
 			Type:      d.Type,
 			URL:       d.URL,
 			Template:  d.Template,
+			Headers:   d.Headers,
 			APIKey:    d.APIKey,
 			Prefix:    d.Prefix,
 			ExpiresAt: d.ExpiresAt,
@@ -291,6 +293,12 @@ func (m *MultiHostService) broadcastNotificationConfig() {
 	wg.Wait()
 }
 
+// NotificationHandler returns the notification manager as an agent.NotificationConfigHandler.
+// This is used in swarm mode to pass the handler to the local agent server.
+func (m *MultiHostService) NotificationHandler() *notification.Manager {
+	return m.notificationManager
+}
+
 // AddSubscription adds a subscription to local manager and broadcasts to agents
 func (m *MultiHostService) AddSubscription(sub *notification.Subscription) error {
 	if err := m.notificationManager.AddSubscription(sub); err != nil {

main.go

@@ -94,8 +94,7 @@ func main() {
 		}
 		// Create client service for agent server in swarm mode
 		clientService := docker_support.NewDockerClientService(localClient, args.Filter)
-		// TODO add notification for swarm mode
-		server, err := agent.NewServer(clientService, certs, args.Version(), nil)
+		server, err := agent.NewServer(clientService, certs, args.Version(), multiHostService.NotificationHandler())
 		if err != nil {
 			log.Fatal().Err(err).Msg("failed to create agent")
 		}