Feature/shared ingress uses config nats broker (#777)

* share ingress uses config map now

* added unit test for ingress

* refactored tests

Author: astelmashenko

cmd/ingress/main.go

@@ -19,18 +19,21 @@ package main
 import (
 	"os"
 
+	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/signals"
 
 	"knative.dev/eventing-natss/pkg/broker/ingress"
+	"knative.dev/eventing-natss/pkg/common/configloader/fsloader"
 )
 
 func main() {
 	component := "natsjs-broker-ingress"
 
 	ctx := signals.NewContext()
 	ctx = sharedmain.WithHealthProbesDisabled(ctx)
+	ctx = fsloader.WithLoader(ctx, configmap.Load)
 
 	ns := os.Getenv("NAMESPACE")
 	if ns != "" {

config/broker/500-shared-ingress.yaml

@@ -42,10 +42,12 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.name
-            - name: NATS_URL
-              value: "nats://nats.nats-io.svc.cluster.local:4222"
             - name: PORT
               value: "8080"
+          volumeMounts:
+            - name: config-nats-broker
+              mountPath: /etc/config-nats-broker
+              readOnly: true
           ports:
             - containerPort: 8080
               name: http
@@ -81,6 +83,10 @@ spec:
                 - ALL
             seccompProfile:
               type: RuntimeDefault
+      volumes:
+        - name: config-nats-broker
+          configMap:
+            name: config-nats-broker
 ---
 apiVersion: v1
 kind: Service

pkg/broker/ingress/controller.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	nats "github.com/nats-io/nats.go"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
@@ -36,7 +37,9 @@ import (
 
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
 
+	"knative.dev/eventing-natss/pkg/broker/constants"
 	"knative.dev/eventing-natss/pkg/broker/contract"
+	"knative.dev/eventing-natss/pkg/common/configloader/fsloader"
 	commonnats "knative.dev/eventing-natss/pkg/common/nats"
 )
 
@@ -49,29 +52,23 @@ const (
 )
 
 type envConfig struct {
-	NatsURL string `envconfig:"NATS_URL" required:"true"`
-	Port    int    `envconfig:"PORT" default:"8080"`
+	Port int `envconfig:"PORT" default:"8080"`
 }
 
 // NewController creates a new shared ingress controller
 func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	logger := logging.FromContext(ctx)
 
-	// Load environment configuration
-	env := &envConfig{}
-	if err := envconfig.Process("", env); err != nil {
+	env, err := loadEnvConfig()
+	if err != nil {
 		logger.Fatalw("Failed to process environment variables", zap.Error(err))
 	}
 
-	logger.Infow("Starting shared broker ingress",
-		zap.String("nats_url", env.NatsURL),
-		zap.Int("port", env.Port),
-	)
+	logger.Infow("Starting shared broker ingress", zap.Int("port", env.Port))
 
-	// Create NATS connection using URL from environment variable
-	natsConn, err := commonnats.NewNatsConnFromURL(ctx, env.NatsURL)
+	natsConn, err := buildNatsConn(ctx)
 	if err != nil {
-		logger.Fatalw("Failed to create NATS connection", zap.Error(err))
+		logger.Fatalw("Failed to initialize NATS connection", zap.Error(err))
 	}
 
 	// Create JetStream context
@@ -154,6 +151,29 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	return impl
 }
 
+// loadEnvConfig reads environment variables into envConfig.
+func loadEnvConfig() (*envConfig, error) {
+	env := &envConfig{}
+	return env, envconfig.Process("", env)
+}
+
+// buildNatsConn loads NATS configuration from the mounted ConfigMap and creates a connection.
+func buildNatsConn(ctx context.Context) (*nats.Conn, error) {
+	fsLoader, err := fsloader.Get(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get ConfigMap loader from context: %w", err)
+	}
+	cmData, err := fsLoader(constants.SettingsConfigMapMountPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load NATS ConfigMap: %w", err)
+	}
+	natsConfig, err := commonnats.LoadEventingNatsConfig(cmData)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse NATS configuration: %w", err)
+	}
+	return commonnats.NewNatsConn(ctx, natsConfig)
+}
+
 // filterContractConfigMap returns true if the object is the contract ConfigMap
 func filterContractConfigMap(obj interface{}) bool {
 	if obj == nil {