From 03dc8824d2074d0852fe7123e41ac5baef5709a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bruno=20Miguel=20Cust=C3=B3dio?= Date: Tue, 10 Dec 2019 11:06:47 +0000 Subject: [PATCH] Support customising the NATS Streaming channel. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Bruno Miguel Custódio --- gateway/Gopkg.lock | 6 +++--- gateway/Gopkg.toml | 2 +- gateway/README.md | 1 + gateway/server.go | 2 +- gateway/types/readconfig.go | 11 +++++++++++ gateway/types/readconfig_test.go | 15 +++++++++++++++ .../openfaas/nats-queue-worker/handler/handler.go | 9 +++++++-- 7 files changed, 39 insertions(+), 7 deletions(-) diff --git a/gateway/Gopkg.lock b/gateway/Gopkg.lock index 4f72a18e..c7fa1b21 100644 --- a/gateway/Gopkg.lock +++ b/gateway/Gopkg.lock @@ -112,15 +112,15 @@ version = "0.12.0" [[projects]] - digest = "1:28692259e850b777eb6db1ee8937ecd14d4d1172204d583f8b71ac11455a737e" + digest = "1:b7c06bdc590871ac5c88f8a85acc9f892c994d6327de311ec5871aaecb6f9489" name = "github.com/openfaas/nats-queue-worker" packages = [ "handler", "nats", ] pruneopts = "UT" - revision = "8dff9cb169398f4f131565702c7f695e76247ed6" - version = "0.8.2" + revision = "dea1c90b8cc66dc73597b7531a4fd29a32b5f88c" + version = "0.9.0" [[projects]] digest = "1:eb04f69c8991e52eff33c428bd729e04208bf03235be88e4df0d88497c6861b9" diff --git a/gateway/Gopkg.toml b/gateway/Gopkg.toml index 69c390a9..297182c7 100644 --- a/gateway/Gopkg.toml +++ b/gateway/Gopkg.toml @@ -12,7 +12,7 @@ [[constraint]] name = "github.com/openfaas/nats-queue-worker" - version = "0.8.2" + version = "0.9.0" [[constraint]] name = "github.com/prometheus/client_golang" diff --git a/gateway/README.md b/gateway/README.md index 25352d5c..eda360a0 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -73,6 +73,7 @@ The gateway can be configured through the following environment variables: | `faas_nats_address` | The host at which NATS Streaming can be reached. Required for asynchronous mode | | `faas_nats_port` | The port at which NATS Streaming can be reached. Required for asynchronous mode | | `faas_nats_cluster_name` | The name of the target NATS Streaming cluster. Defaults to `faas-cluster` for backwards-compatibility | +| `faas_nats_channel` | The name of the NATS Streaming channel to use. Defaults to `faas-request` for backwards-compatibility | | `faas_prometheus_host` | Host to connect to Prometheus. Default: `"prometheus"` | | `faas_promethus_port` | Port to connect to Prometheus. Default: `9090` | | `direct_functions` | `true` or `false` - functions are invoked directly over overlay network by DNS name without passing through the provider | diff --git a/gateway/server.go b/gateway/server.go index 8c50cd49..30ad21cc 100644 --- a/gateway/server.go +++ b/gateway/server.go @@ -135,7 +135,7 @@ func main() { defaultNATSConfig := natsHandler.NewDefaultNATSConfig(maxReconnect, interval) - natsQueue, queueErr := natsHandler.CreateNATSQueue(*config.NATSAddress, *config.NATSPort, *config.NATSClusterName, defaultNATSConfig) + natsQueue, queueErr := natsHandler.CreateNATSQueue(*config.NATSAddress, *config.NATSPort, *config.NATSClusterName, *config.NATSChannel, defaultNATSConfig) if queueErr != nil { log.Fatalln(queueErr) } diff --git a/gateway/types/readconfig.go b/gateway/types/readconfig.go index 12ccf382..2a2fdb4a 100644 --- a/gateway/types/readconfig.go +++ b/gateway/types/readconfig.go @@ -106,6 +106,14 @@ func (ReadConfig) Read(hasEnv HasEnv) (*GatewayConfig, error) { cfg.NATSClusterName = &v } + faasNATSChannel := hasEnv.Getenv("faas_nats_channel") + if len(faasNATSChannel) > 0 { + cfg.NATSChannel = &faasNATSChannel + } else { + v := "faas-request" + cfg.NATSChannel = &v + } + prometheusPort := hasEnv.Getenv("faas_prometheus_port") if len(prometheusPort) > 0 { prometheusPortVal, err := strconv.Atoi(prometheusPort) @@ -197,6 +205,9 @@ type GatewayConfig struct { // The name of the NATS Streaming cluster. Required for async mode. NATSClusterName *string + // NATSChannel is the name of the NATS Streaming channel used for asynchronous function invocations. + NATSChannel *string + // Host to connect to Prometheus. PrometheusHost string diff --git a/gateway/types/readconfig_test.go b/gateway/types/readconfig_test.go index c307aa09..e3364a59 100644 --- a/gateway/types/readconfig_test.go +++ b/gateway/types/readconfig_test.go @@ -255,6 +255,12 @@ func TestRead_UseNATS(t *testing.T) { t.Fail() } + wantNATSChannel := "faas-request" + if *config.NATSChannel != wantNATSChannel { + t.Logf("faas_nats_channel: want %s, got %s", wantNATSChannel, *config.NATSChannel) + t.Fail() + } + defaults.Setenv("faas_nats_cluster_name", "example-nats-cluster") config, _ = readConfig.Read(defaults) @@ -263,6 +269,15 @@ func TestRead_UseNATS(t *testing.T) { t.Logf("faas_nats_cluster_name: want %s, got %s", wantNATSClusterName, *config.NATSClusterName) t.Fail() } + + defaults.Setenv("faas_nats_channel", "foo") + config, _ = readConfig.Read(defaults) + + wantNATSChannel = "foo" + if *config.NATSChannel != wantNATSChannel { + t.Logf("faas_nats_channel: want %s, got %s", wantNATSChannel, *config.NATSChannel) + t.Fail() + } } func TestRead_UseNATSBadPort(t *testing.T) { diff --git a/gateway/vendor/github.com/openfaas/nats-queue-worker/handler/handler.go b/gateway/vendor/github.com/openfaas/nats-queue-worker/handler/handler.go index 59b83ca3..66291c3a 100644 --- a/gateway/vendor/github.com/openfaas/nats-queue-worker/handler/handler.go +++ b/gateway/vendor/github.com/openfaas/nats-queue-worker/handler/handler.go @@ -7,18 +7,23 @@ import ( ) // CreateNATSQueue ready for asynchronous processing -func CreateNATSQueue(address string, port int, clusterName string, clientConfig NATSConfig) (*NATSQueue, error) { +func CreateNATSQueue(address string, port int, clusterName, channel string, clientConfig NATSConfig) (*NATSQueue, error) { var err error natsURL := fmt.Sprintf("nats://%s:%d", address, port) log.Printf("Opening connection to %s\n", natsURL) clientID := clientConfig.GetClientID() + // If 'channel' is empty, use the previous default. + if channel == "" { + channel = "faas-request" + } + queue1 := NATSQueue{ ClientID: clientID, ClusterID: clusterName, NATSURL: natsURL, - Topic: "faas-request", + Topic: channel, maxReconnect: clientConfig.GetMaxReconnect(), reconnectDelay: clientConfig.GetReconnectDelay(), ncMutex: &sync.RWMutex{},