fix(eventbus): set nats routes with pod DNS names. Fixes #1026 (#1033)

Signed-off-by: Derek Wang <whynowy@gmail.com>
argoproj · Jan 23, 2021 · 4c93e07 · 4c93e07
1 parent 74b9fc2
commit 4c93e07
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 4 deletions.
diff --git a/controllers/eventbus/installer/nats.go b/controllers/eventbus/installer/nats.go
@@ -509,16 +509,16 @@ func (i *natsInstaller) buildConfigMap() (*corev1.ConfigMap, error) {
 		return nil, err
 	}
 	peers := []string{}
+	routes := []string{}
 	for j := 0; j < replicas; j++ {
 		peers = append(peers, fmt.Sprintf("\"%s-%s\"", ssName, strconv.Itoa(j)))
+		routes = append(routes, fmt.Sprintf("nats://%s-%s.%s.%s.svc:%s", ssName, strconv.Itoa(j), svcName, i.eventBus.Namespace, strconv.Itoa(int(clusterPort))))
 	}
 	conf := fmt.Sprintf(`http: %s
 include ./auth.conf
 cluster {
   port: %s
-  routes [
-   nats://%s:%s
-  ]
+  routes: [%s]
   cluster_advertise: $CLUSTER_ADVERTISE
   connect_retries: 10
 }
@@ -534,7 +534,7 @@ streaming {
   store_limits {
     max_age: %s
   }
-}`, strconv.Itoa(int(monitorPort)), strconv.Itoa(int(clusterPort)), svcName, strconv.Itoa(int(clusterPort)), clusterID, strings.Join(peers, ","), maxAge)
+}`, strconv.Itoa(int(monitorPort)), strconv.Itoa(int(clusterPort)), strings.Join(routes, ","), clusterID, strings.Join(peers, ","), maxAge)
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: i.eventBus.Namespace,

diff --git a/controllers/eventbus/installer/nats_test.go b/controllers/eventbus/installer/nats_test.go
@@ -3,6 +3,7 @@ package installer
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -255,3 +256,33 @@ func TestBuildServiceAccountStatefulSetSpec(t *testing.T) {
 		assert.True(t, len(ss.Spec.Template.Spec.ServiceAccountName) > 0)
 	})
 }
+
+func TestBuildConfigMap(t *testing.T) {
+	t.Run("test build config map", func(t *testing.T) {
+		cl := fake.NewClientBuilder().Build()
+		installer := &natsInstaller{
+			client:         cl,
+			eventBus:       testEventBus,
+			streamingImage: testStreamingImage,
+			metricsImage:   testMetricsImage,
+			labels:         testLabels,
+			logger:         logging.NewArgoEventsLogger(),
+		}
+		cm, err := installer.buildConfigMap()
+		assert.NoError(t, err)
+		assert.NotNil(t, cm)
+		conf, ok := cm.Data[configMapKey]
+		assert.True(t, ok)
+		assert.True(t, strings.Contains(conf, "routes:"))
+		svcName := generateServiceName(testEventBus)
+		ssName := generateStatefulSetName(testEventBus)
+		r := fmt.Sprintf("nats://%s-%s.%s.%s.svc:%s", ssName, "0", svcName, testNamespace, strconv.Itoa(int(clusterPort)))
+		lines := strings.Split(conf, `\n`)
+		for _, l := range lines {
+			l = strings.Trim(l, " ")
+			if strings.HasPrefix(l, "routes:") {
+				assert.True(t, strings.Contains(l, r))
+			}
+		}
+	})
+}