External nats streaming (#1576)

Support for external NATS streaming instead of the one deployed with Fission

Co-authored-by: Shaun Cutts <shaunc@factfiber.com>

charts/README.md

@@ -89,8 +89,12 @@ Parameter | Description | Default
 `logger.fluentdImage` | Logger fluentbit image | `fluent/fluent-bit`
 `logger.fluentdImageTag` | Logger fluentbit image tag | `1.0.4`
 `nats.enabled` | Nats streaming enabled | `true`
+`nats.external` | Use external Nats installation | `false`
+`nats.hostaddress` | Address of NATS cluster | `nats-streaming:4222`
 `nats.authToken` | Nats streaming auth token | `defaultFissionAuthToken`
 `nats.clusterID` | Nats streaming clusterID | `fissionMQTrigger`
+`nats.clientID`  | Client name registered with nats streaming | `fission`
+`nats.queueGroup` | Queue group registered with nats streaming | `fission-messageQueueNatsTrigger`
 `natsStreamingPort` | Nats streaming service port | `31316`
 `azureStorageQueue.enabled` | Azure storage account name | `false`
 `azureStorageQueue.key` | Azure storage account name | `""`

charts/fission-all/templates/deployment.yaml

@@ -507,6 +507,7 @@ spec:
 #      serviceAccountName: fission-svc
 
 {{- if .Values.nats.enabled }}
+{{- if not .Values.nats.external }}
 ---
 apiVersion: apps/v1
 kind: Deployment
@@ -555,6 +556,7 @@ spec:
 {{- if .Values.extraCoreComponentPodConfig }}
 {{ toYaml .Values.extraCoreComponentPodConfig | indent 6 -}}
 {{- end }}
+{{- end }}
 ---
 apiVersion: apps/v1
 kind: Deployment
@@ -585,8 +587,18 @@ spec:
         env:
         - name: MESSAGE_QUEUE_TYPE
           value: nats-streaming
+        - name: MESSAGE_QUEUE_CLUSTER_ID
+          value: {{ .Values.nats.clusterID }}
+        - name: MESSAGE_QUEUE_QUEUE_GROUP
+          value: {{ .Values.nats.queueGroup }}
+        - name: MESSAGE_QUEUE_CLIENT_ID
+          value: {{ .Values.nats.clientID }}
         - name: MESSAGE_QUEUE_URL
-          value: nats://{{ .Values.nats.authToken }}@nats-streaming:4222
+        {{- if .Values.nats.authToken }}
+          value: nats://{{ .Values.nats.authToken }}@{{ .Values.nats.hostaddress }}
+        {{- else }}
+          value: nats://{{ .Values.nats.hostaddress }}
+        {{- end }}
         - name: TRACE_JAEGER_COLLECTOR_ENDPOINT
           value: "{{ .Values.traceCollectorEndpoint }}"
         - name: TRACING_SAMPLING_RATE

charts/fission-all/templates/svc.yaml

@@ -41,7 +41,7 @@ spec:
   selector:
     svc: controller
 
-{{- if .Values.nats.enabled }}
+{{- if and .Values.nats.enabled (not .Values.nats.external) }}
 ---
 apiVersion: v1
 kind: Service

charts/fission-all/values.yaml

@@ -29,6 +29,7 @@ controllerPort: 31313
 routerPort: 31314
 
 ## Port at which NATS streaming service should be exposed
+## (only if nats enabled and not external)
 natsStreamingPort: 31316
 
 ## Set to false if you create the namespaces manually
@@ -142,10 +143,28 @@ router:
 ## Message queue trigger config
 ### NATS Streaming, enabled by default
 nats:
+  # whether or not to use NATS
   enabled: true
+
+  # if true, don't install NATS, but
+  external: false
+
+  # Address of NATS server (domain:port)
+  # change from default for external NATS
+  hostaddress: 'nats-streaming:4222'
+
+  # Authorization token to use with NATS
   authToken: "defaultFissionAuthToken"
+
+  # NATS streaming clusterID
   clusterID: "fissionMQTrigger"
 
+  # Client name registered with NATS streaming 
+  clientID: "fission"
+
+  # Queue group registered with NATS streaming 
+  queueGroup: "fission-messageQueueNatsTrigger"
+
 ## Azure-storage-queue: enable and configure the details
 azureStorageQueue:
   enabled: false

pkg/mqtrigger/messageQueue/nats/nats.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
+	"os"
 	"strings"
 
 	nsUtil "github.com/nats-io/nats-streaming-server/util"
@@ -34,16 +35,32 @@ import (
 	"github.com/fission/fission/pkg/utils"
 )
 
+var natsClusterID string
+var natsQueueGroup string
+var natsClientID string
+
 func init() {
+	natsClusterID = os.Getenv("MESSAGE_QUEUE_CLUSTER_ID")
+	if natsClusterID == "" {
+		natsClusterID = defaultNatsClusterID
+	}
+	natsClientID = os.Getenv("MESSAGE_QUEUE_CLIENT_ID")
+	if natsClientID == "" {
+		natsClientID = defaultNatsClientID
+	}
+	natsQueueGroup = os.Getenv("MESSAGE_QUEUE_QUEUE_GROUP")
+	if natsQueueGroup == "" {
+		natsQueueGroup = defaultNatsQueueGroup
+	}
 	factory.Register(fv1.MessageQueueTypeNats, &Factory{})
 	validator.Register(fv1.MessageQueueTypeNats, IsTopicValid)
 }
 
 const (
-	natsClusterID  = "fissionMQTrigger"
+	natsProtocol          = "nats://"
-	natsProtocol   = "nats://"
+	defaultNatsClusterID  = "fissionMQTrigger"
-	natsClientID   = "fission"
+	defaultNatsClientID   = "fission"
-	natsQueueGroup = "fission-messageQueueNatsTrigger"
+	defaultNatsQueueGroup = "fission-messageQueueNatsTrigger"
 )
 
 type (