Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove deprecated Fission Nats connector (#2403)

Browse Source 
We are removing Fission deprecated Nats connector and
planning to adopt Keda going forward to have better
delegated functionality and more rich support.

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2022-04-06 13:41:38 +05:30
committed by GitHub
parent 8b3a3f29a8
commit 4e91579ef2
23 changed files with 9 additions and 1028 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/kind_CI.sh
+1 -3
View File
@@ -81,8 +81,6 @@ main() {
$ROOT/test/tests/test_specs/test_spec_archive/test_spec_archive.sh \
$ROOT/test/tests/test_environments/test_tensorflow_serving_env.sh \
$ROOT/test/tests/test_environments/test_go_env.sh \
$ROOT/test/tests/mqtrigger/nats/test_mqtrigger.sh \
$ROOT/test/tests/mqtrigger/nats/test_mqtrigger_error.sh \
$ROOT/test/tests/test_huge_response/test_huge_response.sh \
$ROOT/test/tests/test_kubectl/test_kubectl.sh \
$ROOT/test/tests/websocket/test_ws.sh
@@ -126,4 +124,4 @@ main
echo "Total Failures" $FAILURES
if [[ $FAILURES != '0' ]]; then
exit 1
fi
fi
test/test_utils.sh
-4
View File
@@ -203,7 +203,6 @@ set_environment() {
# fission env
export FISSION_URL=http://$(kubectl -n $ns get svc controller -o jsonpath='{...ip}')
export FISSION_ROUTER=$(kubectl -n $ns get svc router -o jsonpath='{...ip}')
export FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(kubectl -n $ns get svc nats-streaming -o jsonpath='{...ip}:{.spec.ports[0].port}')"
# ingress controller env
export INGRESS_CONTROLLER=$(kubectl -n ingress-nginx get svc ingress-nginx -o jsonpath='{...ip}')
@@ -480,7 +479,6 @@ dump_logs() {
dump_fission_logs $ns $fns executor
dump_fission_logs $ns $fns storagesvc
dump_fission_logs $ns $fns mqtrigger
dump_fission_logs $ns $fns mqtrigger-nats-streaming
dump_function_pod_logs $ns $fns
dump_builder_pod_logs $bns
dump_fission_crds
@@ -536,8 +534,6 @@ run_all_tests() {
$ROOT/test/tests/test_specs/test_spec_archive/test_spec_archive.sh \
$ROOT/test/tests/test_environments/test_tensorflow_serving_env.sh \
$ROOT/test/tests/test_environments/test_go_env.sh \
$ROOT/test/tests/mqtrigger/nats/test_mqtrigger.sh \
$ROOT/test/tests/mqtrigger/nats/test_mqtrigger_error.sh \
$ROOT/test/tests/test_huge_response/test_huge_response.sh \
$ROOT/test/tests/test_kubectl/test_kubectl.sh
$ROOT/test/tests/websocket/test_ws.sh
test/tests/mqtrigger/nats/main.js
-6
View File
@@ -1,6 +0,0 @@
module.exports = async function(context) {
return {
status: 200,
body: "Hello, World!"
};
}
test/tests/mqtrigger/nats/main_error.js
-6
View File
@@ -1,6 +0,0 @@
module.exports = async function(context) {
return {
status: 400,
body: "Hello, World!"
};
}
test/tests/mqtrigger/nats/stan-pub/main.go
-143
View File
@@ -1,143 +0,0 @@
// This file originally came from official Nats.io GitHub repository.
// You can reach the original file with the following link:
// https://github.com/nats-io/stan.go/blob/master/examples/stan-pub/main.go
// Copyright 2016-2019 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package main
import (
"flag"
"fmt"
"log"
"os"
"sync"
"time"
nats "github.com/nats-io/nats.go"
"github.com/nats-io/stan.go"
)
var usageStr = `
Usage: stan-pub [options] <subject> <message>
Options:
-s, --server <url> NATS Streaming server URL(s)
-c, --cluster <cluster name> NATS Streaming cluster name
-id, --clientid <client ID> NATS Streaming client ID
-a, --async Asynchronous publish mode
-cr, --creds <credentials> NATS 2.0 Credentials
`
// NOTE: Use tls scheme for TLS, e.g. stan-pub -s tls://demo.nats.io:4443 foo hello
func usage() {
fmt.Printf("%s\n", usageStr)
os.Exit(0)
}
func main() {
var (
clusterID string
clientID string
URL string
async bool
userCreds string
)
flag.StringVar(&URL, "s", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&URL, "server", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&clusterID, "c", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clusterID, "cluster", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clientID, "id", "stan-pub", "The NATS Streaming client ID to connect with")
flag.StringVar(&clientID, "clientid", "stan-pub", "The NATS Streaming client ID to connect with")
flag.BoolVar(&async, "a", false, "Publish asynchronously")
flag.BoolVar(&async, "async", false, "Publish asynchronously")
flag.StringVar(&userCreds, "cr", "", "Credentials File")
flag.StringVar(&userCreds, "creds", "", "Credentials File")
log.SetFlags(0)
flag.Usage = usage
flag.Parse()
args := flag.Args()
if len(args) < 1 {
usage()
}
// Connect Options.
opts := []nats.Option{nats.Name("NATS Streaming Example Publisher")}
// Use UserCredentials
if userCreds != "" {
opts = append(opts, nats.UserCredentials(userCreds))
}
// Connect to NATS
nc, err := nats.Connect(URL, opts...)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
sc, err := stan.Connect(clusterID, clientID, stan.NatsConn(nc))
if err != nil {
log.Fatalf("Can't connect: %v.\nMake sure a NATS Streaming Server is running at: %s", err, URL)
}
defer sc.Close()
subj, msg := args[0], []byte(args[1])
ch := make(chan bool)
var glock sync.Mutex
var guid string
acb := func(lguid string, err error) {
glock.Lock()
log.Printf("Received ACK for guid %s\n", lguid)
defer glock.Unlock()
if err != nil {
log.Fatalf("Error in server ack for guid %s: %v\n", lguid, err)
}
if lguid != guid {
log.Fatalf("Expected a matching guid in ack callback, got %s vs %s\n", lguid, guid)
}
ch <- true
}
if !async {
err = sc.Publish(subj, msg)
if err != nil {
log.Fatalf("Error during publish: %v\n", err)
}
log.Printf("Published [%s] : '%s'\n", subj, msg)
} else {
glock.Lock()
guid, err = sc.PublishAsync(subj, msg, acb)
if err != nil {
log.Fatalf("Error during async publish: %v\n", err)
}
glock.Unlock()
if guid == "" {
log.Fatal("Expected non-empty guid to be returned.")
}
log.Printf("Published [%s] : '%s' [guid: %s]\n", subj, msg, guid)
select {
case <-ch:
break
case <-time.After(5 * time.Second):
log.Fatal("timeout")
}
}
}
test/tests/mqtrigger/nats/stan-sub/main.go
-188
View File
@@ -1,188 +0,0 @@
// This file originally came from official Nats.io GitHub repository.
// You can reach the original file with the following link:
// https://github.com/nats-io/stan.go/blob/master/examples/stan-sub/main.go
// Copyright 2016-2019 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package main
import (
"flag"
"fmt"
"log"
"os"
"os/signal"
"syscall"
"time"
nats "github.com/nats-io/nats.go"
"github.com/nats-io/stan.go"
"github.com/nats-io/stan.go/pb"
)
var usageStr = `
Usage: stan-sub [options] <subject>
Options:
-s, --server <url> NATS Streaming server URL(s)
-c, --cluster <cluster name> NATS Streaming cluster name
-id, --clientid <client ID> NATS Streaming client ID
-cr, --creds <credentials> NATS 2.0 Credentials
Subscription Options:
--qgroup <name> Queue group
--all Deliver all available messages
--last Deliver starting with last published message
--since <time_ago> Deliver messages in last interval (e.g. 1s, 1hr)
--seq <seqno> Start at seqno
--new_only Only deliver new messages
--durable <name> Durable subscriber name
--unsub Unsubscribe the durable on exit
`
// NOTE: Use tls scheme for TLS, e.g. stan-sub -s tls://demo.nats.io:4443 foo
func usage() {
log.Fatalf(usageStr)
}
func printMsg(m *stan.Msg, i int) {
log.Printf("[#%d] Received: %s\n", i, m)
}
func main() {
var (
clusterID, clientID string
URL string
userCreds string
showTime bool
qgroup string
unsubscribe bool
startSeq uint64
startDelta string
deliverAll bool
newOnly bool
deliverLast bool
durable string
)
flag.StringVar(&URL, "s", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&URL, "server", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&clusterID, "c", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clusterID, "cluster", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clientID, "id", "stan-sub", "The NATS Streaming client ID to connect with")
flag.StringVar(&clientID, "clientid", "stan-sub", "The NATS Streaming client ID to connect with")
flag.BoolVar(&showTime, "t", false, "Display timestamps")
// Subscription options
flag.Uint64Var(&startSeq, "seq", 0, "Start at sequence no.")
flag.BoolVar(&deliverAll, "all", true, "Deliver all")
flag.BoolVar(&newOnly, "new_only", false, "Only new messages")
flag.BoolVar(&deliverLast, "last", false, "Start with last value")
flag.StringVar(&startDelta, "since", "", "Deliver messages since specified time offset")
flag.StringVar(&durable, "durable", "", "Durable subscriber name")
flag.StringVar(&qgroup, "qgroup", "", "Queue group name")
flag.BoolVar(&unsubscribe, "unsub", false, "Unsubscribe the durable on exit")
flag.BoolVar(&unsubscribe, "unsubscribe", false, "Unsubscribe the durable on exit")
flag.StringVar(&userCreds, "cr", "", "Credentials File")
flag.StringVar(&userCreds, "creds", "", "Credentials File")
log.SetFlags(0)
flag.Usage = usage
flag.Parse()
args := flag.Args()
if len(args) < 1 {
log.Printf("Error: A subject must be specified.")
usage()
}
// Connect Options.
opts := []nats.Option{nats.Name("NATS Streaming Example Subscriber")}
// Use UserCredentials
if userCreds != "" {
opts = append(opts, nats.UserCredentials(userCreds))
}
// Connect to NATS
nc, err := nats.Connect(URL, opts...)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
sc, err := stan.Connect(clusterID, clientID, stan.NatsConn(nc),
stan.SetConnectionLostHandler(func(_ stan.Conn, reason error) {
log.Fatalf("Connection lost, reason: %v", reason)
}))
if err != nil {
log.Fatalf("Can't connect: %v.\nMake sure a NATS Streaming Server is running at: %s", err, URL)
}
log.Printf("Connected to %s clusterID: [%s] clientID: [%s]\n", URL, clusterID, clientID)
// Process Subscriber Options.
startOpt := stan.StartAt(pb.StartPosition_NewOnly)
if startSeq != 0 {
startOpt = stan.StartAtSequence(startSeq)
} else if deliverLast {
startOpt = stan.StartWithLastReceived()
} else if deliverAll && !newOnly {
startOpt = stan.DeliverAllAvailable()
} else if startDelta != "" {
ago, err := time.ParseDuration(startDelta)
if err != nil {
sc.Close()
log.Fatal(err)
}
startOpt = stan.StartAtTimeDelta(ago)
}
subj, i := args[0], 0
mcb := func(msg *stan.Msg) {
i++
printMsg(msg, i)
}
sub, err := sc.QueueSubscribe(subj, qgroup, mcb, startOpt, stan.DurableName(durable))
if err != nil {
sc.Close()
log.Fatal(err)
}
log.Printf("Listening on [%s], clientID=[%s], qgroup=[%s] durable=[%s]\n", subj, clientID, qgroup, durable)
if showTime {
log.SetFlags(log.LstdFlags)
}
// Wait for a SIGINT (perhaps triggered by user with CTRL-C)
// Run cleanup when signal is received
signalChan := make(chan os.Signal, 1)
cleanupDone := make(chan bool)
signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
go func() {
for sig := range signalChan {
fmt.Printf("\nReceived signal %s, unsubscribing and closing connection...\n\n", sig.String())
// Do not unsubscribe a durable on exit, except if asked to.
if durable == "" || unsubscribe {
err := sub.Unsubscribe()
if err != nil {
log.Fatal(err)
}
}
sc.Close()
cleanupDone <- true
}
}()
<-cleanupDone
}
test/tests/mqtrigger/nats/test_mqtrigger.sh
-72
View File
@@ -1,72 +0,0 @@
#!/bin/bash
#test:disabled
#
# Create a function and trigger it using NATS
#
set -euo pipefail
source $(dirname $0)/../../../utils.sh
set +x
TEST_ID=$(generate_test_id)
echo "TEST_ID = $TEST_ID"
ROOT=$(dirname $0)/../../..
DIR=$(dirname $0)
clusterID="fissionMQTrigger"
pubClientID="clientPub-$TEST_ID"
subClientID="clientSub-$TEST_ID"
topic="foo.bar$TEST_ID"
resptopic="foo.foo$TEST_ID"
#FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(minikube ip):4222"
expectedRespOutput="subject:\"$resptopic\" data:\"Hello, World!\""
env=nodejs-$TEST_ID
fn=hello-$TEST_ID
mqt=mqt-$TEST_ID
cleanup() {
log "Cleaning up..."
clean_resource_by_id $TEST_ID
}
if [ -z "${TEST_NOCLEANUP:-}" ]; then
trap cleanup EXIT
else
log "TEST_NOCLEANUP is set; not cleaning up test artifacts afterwards."
fi
log "Creating nodejs env"
fission env create --name $env --image $NODE_RUNTIME_IMAGE
log "Creating function"
fission fn create --name $fn --env $env --code $DIR/main.js --method GET
log "Creating message queue trigger"
fission mqtrigger create --name $mqt --function $fn --mqtype "nats-streaming" --topic $topic --resptopic $resptopic
# wait until nats trigger is created
sleep 5
#
# Send a message
#
log "Sending message"
go run $DIR/stan-pub/main.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $pubClientID $topic ""
#
# Wait for message on response topic
#
log "Waiting for response"
response=$(timeout 10s go run $DIR/stan-sub/main.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $subClientID $resptopic 2>&1 || true)
log "Output from subscriber"
echo "$response"
echo "$response" | grep "$expectedRespOutput"
log "Deleting message queue trigger"
fission mqtrigger delete --name $mqt
log "Subscriber received expected response: $response"
log "Test PASSED"
test/tests/mqtrigger/nats/test_mqtrigger_error.sh
-75
View File
@@ -1,75 +0,0 @@
#!/bin/bash
#test:disabled
#
# Create a function and trigger it using NATS
# To run this on Minikube, uncomment line 24
set -euo pipefail
set +x
source $(dirname $0)/../../../utils.sh
TEST_ID=$(generate_test_id)
echo "TEST_ID = $TEST_ID"
ROOT=$(dirname $0)/../../..
DIR=$(dirname $0)
clusterID="fissionMQTrigger"
pubClientID="clientPub-$TEST_ID"
subClientID="clientSub-$TEST_ID"
topic="foo.bar$TEST_ID"
resptopic="foo.foo$TEST_ID"
errortopic="foo.error$TEST_ID"
maxretries=1
#FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(minikube ip):4222"
expectedRespOutput="subject:\"$errortopic\" data:\"Hello, World!\""
env=nodejs-$TEST_ID
fn=hello-$TEST_ID
mqt=mqt-$TEST_ID
cleanup() {
log "Cleaning up..."
clean_resource_by_id $TEST_ID
}
if [ -z "${TEST_NOCLEANUP:-}" ]; then
trap cleanup EXIT
else
log "TEST_NOCLEANUP is set; not cleaning up test artifacts afterwards."
fi
log "Creating nodejs env"
fission env create --name $env --image $NODE_RUNTIME_IMAGE
log "Creating function"
fission fn create --name $fn --env $env --code $DIR/main_error.js --method GET
log "Creating message queue trigger"
fission mqtrigger create --name $mqt --function $fn --mqtype "nats-streaming" --topic $topic --resptopic $resptopic --errortopic $errortopic --maxretries $maxretries
log "Updated mqtrigger list"
fission mqtrigger list
# wait until nats trigger is created
sleep 5
#
# Send a message
#
log "Sending message"
go run $DIR/stan-pub/main.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $pubClientID $topic ""
#
# Wait for message on error topic
#
log "Waiting for response"
response=$(timeout 10s go run $DIR/stan-sub/main.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $subClientID $errortopic 2>&1 || true)
log "Output from subscriber"
echo "$response"
echo "$response" | grep "$expectedRespOutput"
log "Deleting message queue trigger"
fission mqtrigger delete --name $mqt
log "Test PASSED"