fix: refactoring to remove pubsub flags to improve experience

Makefile

@@ -224,8 +224,6 @@ ifeq ($(ENABLE_PUBSUB),true)
 		--set disabledBuiltins={http.send} \
 		--set logMutations=true \
 		--set audit.enablePubsub=${ENABLE_PUBSUB} \
-		--set audit.connection=${AUDIT_CONNECTION} \
-		--set audit.channel=${AUDIT_CHANNEL} \
 		--set-string auditPodAnnotations.dapr\\.io/enabled=true \
 		--set-string auditPodAnnotations.dapr\\.io/app-id=audit \
 		--set-string auditPodAnnotations.dapr\\.io/metrics-port=9999 \

cmd/build/helmify/replacements.go

@@ -95,8 +95,6 @@ var replacements = map[string]string{
 
 	"- HELMSUBST_PUBSUB_ARGS": `{{ if .Values.audit.enablePubsub}}
         - --enable-pub-sub={{ .Values.audit.enablePubsub }}
-        - --audit-connection={{ .Values.audit.connection }}
-        - --audit-channel={{ .Values.audit.channel }}
         {{- end }}`,
 
 	"HELMSUBST_MUTATING_WEBHOOK_FAILURE_POLICY": `{{ .Values.mutatingWebhookFailurePolicy }}`,

cmd/build/helmify/static/values.yaml

@@ -214,9 +214,7 @@ controllerManager:
       #   - ipBlock:
       #       cidr: 0.0.0.0/0
 audit:
-  enablePubsub: false
-  connection: audit-connection
-  channel: audit-channel
+  enablePubSub: false
   hostNetwork: false
   dnsPolicy: ClusterFirst
   metricsPort: 8888

manifest_staging/charts/gatekeeper/templates/gatekeeper-audit-deployment.yaml

@@ -68,8 +68,6 @@ spec:
         - --operation=status
         {{ if .Values.audit.enablePubsub}}
         - --enable-pub-sub={{ .Values.audit.enablePubsub }}
-        - --audit-connection={{ .Values.audit.connection }}
-        - --audit-channel={{ .Values.audit.channel }}
         {{- end }}
         {{ if not .Values.disableMutation}}- --operation=mutation-status{{- end }}
         - --logtostderr

manifest_staging/charts/gatekeeper/values.yaml

@@ -214,9 +214,7 @@ controllerManager:
       #   - ipBlock:
       #       cidr: 0.0.0.0/0
 audit:
-  enablePubsub: false
-  connection: audit-connection
-  channel: audit-channel
+  enablePubSub: false
   hostNetwork: false
   dnsPolicy: ClusterFirst
   metricsPort: 8888

pkg/audit/manager.go

@@ -67,8 +67,6 @@ var (
 	auditEventsInvolvedNamespace = flag.Bool("audit-events-involved-namespace", false, "emit audit events for each violation in the involved objects namespace, the default (false) generates events in the namespace Gatekeeper is installed in. Audit events from cluster-scoped resources will still follow the default behavior")
 	auditMatchKindOnly           = flag.Bool("audit-match-kind-only", false, "only use kinds specified in all constraints for auditing cluster resources. if kind is not specified in any of the constraints, it will audit all resources (same as setting this flag to false)")
 	apiCacheDir                  = flag.String("api-cache-dir", defaultAPICacheDir, "The directory where audit from api server cache are stored, defaults to /tmp/audit")
-	auditConnection              = flag.String("audit-connection", defaultConnection, "Connection name for publishing audit violation messages. Defaults to audit-connection")
-	auditChannel                 = flag.String("audit-channel", defaultChannel, "Channel name for publishing audit violation messages. Defaults to audit-channel")
 	emptyAuditResults            = newLimitQueue(0)
 	logStatsAudit                = flag.Bool("log-stats-audit", false, "(alpha) log stats metrics for the audit run")
 )
@@ -901,7 +899,7 @@ func (am *Manager) addAuditResponsesToUpdateLists(
 		labels := r.obj.GetLabels()
 		logViolation(am.log, constraint, ea, gvk, namespace, name, msg, details, labels)
 		if *pubsubController.PubsubEnabled {
-			err := am.pubsubSystem.Publish(context.Background(), *auditConnection, *auditChannel, violationMsg(constraint, ea, gvk, namespace, name, msg, details, labels, timestamp))
+			err := am.pubsubSystem.Publish(context.Background(), violationMsg(constraint, ea, gvk, namespace, name, msg, details, labels, timestamp))
 			if err != nil {
 				am.log.Error(err, "pubsub audit Publishing")
 			}

pkg/pubsub/connection/connection.go

@@ -7,7 +7,7 @@ import (
 // PubSub is the interface that wraps pubsub methods.
 type Connection interface {
 	// Publish single message over a specific topic/channel
-	Publish(ctx context.Context, data interface{}, topic string) error
+	Publish(ctx context.Context, data interface{}) error
 
 	// Close connections
 	CloseConnection() error

pkg/pubsub/dapr/dapr.go

@@ -12,6 +12,9 @@ import (
 type ClientConfig struct {
 	// Name of the component to be used for pub sub messaging
 	Component string `json:"component"`
+
+	// Topic where the messages would be published for the connection
+	Topic string `json:"topic"`
 }
 
 // Dapr represents driver for interacting with pub sub using dapr.
@@ -21,19 +24,22 @@ type Dapr struct {
 
 	// Name of the pubsub component
 	pubSubComponent string
+
+	// Topic where the messages would be published for the connection
+	topic string
 }
 
 const (
 	Name = "dapr"
 )
 
-func (r *Dapr) Publish(_ context.Context, data interface{}, topic string) error {
+func (r *Dapr) Publish(_ context.Context, data interface{}) error {
 	jsonData, err := json.Marshal(data)
 	if err != nil {
 		return fmt.Errorf("error marshaling data: %w", err)
 	}
 
-	err = r.client.PublishEvent(context.Background(), r.pubSubComponent, topic, jsonData)
+	err = r.client.PublishEvent(context.Background(), r.pubSubComponent, r.topic, jsonData)
 	if err != nil {
 		return fmt.Errorf("error publishing message to dapr: %w", err)
 	}
@@ -56,6 +62,11 @@ func (r *Dapr) UpdateConnection(_ context.Context, config interface{}) error {
 		return fmt.Errorf("failed to get value of component")
 	}
 	r.pubSubComponent = cfg.Component
+	cfg.Topic, ok = m["topic"].(string)
+	if !ok {
+		return fmt.Errorf("failed to get value of topic")
+	}
+	r.topic = cfg.Topic
 	return nil
 }
 
@@ -70,6 +81,10 @@ func NewConnection(_ context.Context, config interface{}) (connection.Connection
 	if !ok {
 		return nil, fmt.Errorf("failed to get value of component")
 	}
+	cfg.Topic, ok = m["topic"].(string)
+	if !ok {
+		return nil, fmt.Errorf("failed to get value of topic")
+	}
 
 	tmp, err := daprClient.NewClient()
 	if err != nil {
@@ -79,5 +94,6 @@ func NewConnection(_ context.Context, config interface{}) (connection.Connection
 	return &Dapr{
 		client:          tmp,
 		pubSubComponent: cfg.Component,
+		topic:           cfg.Topic,
 	}, nil
 }

pkg/pubsub/dapr/dapr_test.go

@@ -55,9 +55,8 @@ func TestDapr_Publish(t *testing.T) {
 	ctx := context.Background()
 
 	type args struct {
-		ctx   context.Context
-		data  interface{}
-		topic string
+		ctx  context.Context
+		data interface{}
 	}
 
 	tests := []struct {
@@ -72,35 +71,22 @@ func TestDapr_Publish(t *testing.T) {
 				data: map[string]interface{}{
 					"test": "test",
 				},
-				topic: "test",
 			},
 			wantErr: false,
 		},
 		{
 			name: "test publish without data",
 			args: args{
-				ctx:   ctx,
-				data:  nil,
-				topic: "test",
+				ctx:  ctx,
+				data: nil,
 			},
 			wantErr: false,
 		},
-		{
-			name: "test publish without topic",
-			args: args{
-				ctx: ctx,
-				data: map[string]interface{}{
-					"test": "test",
-				},
-				topic: "",
-			},
-			wantErr: true,
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			r := testClient
-			if err := r.Publish(tt.args.ctx, tt.args.data, tt.args.topic); (err != nil) != tt.wantErr {
+			if err := r.Publish(tt.args.ctx, tt.args.data); (err != nil) != tt.wantErr {
 				t.Errorf("Dapr.Publish() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})
@@ -117,6 +103,7 @@ func TestDapr_UpdateConnection(t *testing.T) {
 			name: "test update connection",
 			config: map[string]interface{}{
 				"component": "foo",
+				"topic":     "bar",
 			},
 			wantErr: false,
 		},

pkg/pubsub/dapr/fake_dapr_client.go

@@ -328,6 +328,7 @@ func FakeConnection() (connection.Connection, func()) {
 	return &Dapr{
 		client:          c,
 		pubSubComponent: "test",
+		topic:           "test",
 	}, f
 }
 
@@ -338,11 +339,14 @@ type FakeDapr struct {
 	// Name of the pubsub component
 	pubSubComponent string
 
+	// Name of the topic
+	topic string
+
 	// closing function
 	f func()
 }
 
-func (r *FakeDapr) Publish(_ context.Context, _ interface{}, _ string) error {
+func (r *FakeDapr) Publish(_ context.Context, _ interface{}) error {
 	return nil
 }
 
@@ -376,12 +380,17 @@ func FakeNewConnection(ctx context.Context, config interface{}) (connection.Conn
 	if !ok {
 		return nil, fmt.Errorf("failed to get value of component")
 	}
+	cfg.Topic, ok = m["topic"].(string)
+	if !ok {
+		return nil, fmt.Errorf("failed to get value of topic")
+	}
 
 	c, f := getTestClient(ctx)
 
 	return &FakeDapr{
 		client:          c,
 		pubSubComponent: cfg.Component,
 		f:               f,
+		topic:           cfg.Topic,
 	}, nil
 }

pkg/pubsub/system.go

@@ -2,6 +2,7 @@ package pubsub
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -19,16 +20,19 @@ func NewSystem() *System {
 	return &System{}
 }
 
-func (s *System) Publish(_ context.Context, connection string, topic string, msg interface{}) error {
+func (s *System) Publish(ctx context.Context, msg interface{}) error {
 	s.mux.RLock()
 	defer s.mux.RUnlock()
-	if len(s.connections) > 0 {
-		if c, ok := s.connections[connection]; ok {
-			return c.Publish(context.Background(), msg, topic)
-		}
-		return fmt.Errorf("connection is not initialized, name: %s ", connection)
+	var errs error
+
+	if len(s.connections) == 0 {
+		return fmt.Errorf("no connections are established")
+	}
+
+	for _, c := range s.connections {
+		errs = errors.Join(errs, c.Publish(ctx, msg))
 	}
-	return fmt.Errorf("No connections are established")
+	return errs
 }
 
 func (s *System) UpsertConnection(ctx context.Context, config interface{}, name string, provider string) error {

pkg/pubsub/system_test.go

@@ -24,6 +24,7 @@ func TestMain(m *testing.M) {
 	cfg := map[string]interface{}{
 		dapr.Name: map[string]interface{}{
 			"component": "pubsub",
+			"topic":     "audit",
 		},
 	}
 	for name, fakeConn := range tmp {
@@ -90,6 +91,7 @@ func TestSystem_UpsertConnection(t *testing.T) {
 				ctx: context.Background(),
 				config: map[string]interface{}{
 					"component": "pubsub",
+					"topic":     "test",
 				},
 				name:     "dapr",
 				provider: "dapr",
@@ -111,6 +113,7 @@ func TestSystem_UpsertConnection(t *testing.T) {
 				ctx: context.Background(),
 				config: map[string]interface{}{
 					"component": "pubsub",
+					"topic":     "test",
 				},
 				name:     "audit",
 				provider: "test",
@@ -133,6 +136,7 @@ func TestSystem_UpsertConnection(t *testing.T) {
 				ctx: context.Background(),
 				config: map[string]interface{}{
 					"component": "test",
+					"topic":     "audit",
 				},
 				name:     "audit",
 				provider: "dapr",
@@ -222,15 +226,6 @@ func TestSystem_Publish(t *testing.T) {
 			args:    args{ctx: context.Background(), connection: "audit", topic: "test", msg: nil},
 			wantErr: true,
 		},
-		{
-			name: "Publishing to a connection that does not exist",
-			fields: fields{
-				connections: map[string]connection.Connection{"audit": &dapr.Dapr{}},
-				providers:   map[string]string{"audit": "dapr"},
-			},
-			args:    args{ctx: context.Background(), connection: "test", topic: "test", msg: nil},
-			wantErr: true,
-		},
 		{
 			name: "Publishing to a connection that does exist",
 			fields: fields{
@@ -248,7 +243,7 @@ func TestSystem_Publish(t *testing.T) {
 				connections: tt.fields.connections,
 				providers:   tt.fields.providers,
 			}
-			if err := s.Publish(tt.args.ctx, tt.args.connection, tt.args.topic, tt.args.msg); (err != nil) != tt.wantErr {
+			if err := s.Publish(tt.args.ctx, tt.args.msg); (err != nil) != tt.wantErr {
 				t.Errorf("System.Publish() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})

test/pubsub/publish-components.yaml

@@ -24,5 +24,6 @@ data:
   provider: "dapr"
   config: |
     {
-      "component": "pubsub"
+      "component": "pubsub",
+      "topic": "audit-channel"
     }