NATS Streaming
This tutorial will demonstrate how to use a NATS Streaming trigger to invoke a function. We’ll assume you have Fission and Kubernetes installed. If not, please head over to the install guide.
You will also need NATS Streaming server setup which is reachable from the Fission Kubernetes cluster.
Installation
If you want to setup NATS Streaming server on the Kubernetes cluster, you can use the information here or you can check the documentation for nats streaming docs.
You can also setup NATS streaming server with this yaml file.(Monitoring is already configured)
NATS streaming keda connector uses NATS monitoring to scale the deployment, to enable monitoring in nats we need to pass flags as below, you can get more information here
-m, --http_port PORT HTTP PORT for monitoring -ms,--https_port PORT Use HTTPS PORT for monitoring
$ kubectl apply -f nats-dep.yaml
NAME READY STATUS RESTARTS AGE
nats-streaming-deployment-646768fcfd-qtpmk 1/1 Running 0 8s
You can find the above file here.
Verify if monitoring endpoint is reachable by exec into any container
$ kubectl create deployment test --image=nginx
$ kubectl exec -it test-844b65666c-8kppc /bin/bash
$ curl nats.default.svc.cluster.local:8222
<html lang="en">
<head>
<link rel="shortcut icon" href="http://nats.io/img/favicon.ico">
<style type="text/css">
body { font-family: "Century Gothic", CenturyGothic, AppleGothic, sans-serif; font-size: 22; }
a { margin-left: 32px; }
</style>
</head>
<body>
<img src="http://nats.io/img/logo.png" alt="NATS">
<br/>
<a href=/varz>varz</a><br/>
<a href=/connz>connz</a><br/>
<a href=/routez>routez</a><br/>
<a href=/gatewayz>gatewayz</a><br/>
<a href=/leafz>leafz</a><br/>
<a href=/subsz>subsz</a><br/>
<br/>
<a href=https://docs.nats.io/nats-server/configuration/monitoring.html>help</a>
</body>
</html>
Overview
Before we dive into details, let’s walk through overall flow of event and functions involved.
- A Go producer function (producer) which acts as a producer and drops a message in a NATS queue named
request. - Fission NATS Streaming trigger activates and invokes another function (consumer) with message received from producer.
- The consumer function (consumer) gets body of message and returns a response.
- Fission NATS streaming trigger takes the response of consumer function (consumer) and drops the message in a response queue named
response. If there is an error, the message is dropped in error queue namederror.
Building the app
Producer Function
The producer function is a go program which creates a message and drops into a NATS streaming queue request.
For brevity all values have been hard coded in the code itself.
There are different ways of loading this function into cluster. One of the ways is to create a deployment.
All the files required are present here.
Steps for deploying producer function:
docker build . -t producer
kind load docker-image producer --name kind
kubectl apply -f deployment.yaml //replicas is set to 0 when deployed
package main
import (
"fmt"
"log"
"strconv"
"github.com/nats-io/nats.go"
"github.com/nats-io/stan.go"
)
func main() {
nc, err := nats.Connect("nats://nats:4222")
if err != nil {
log.Fatal(err)
}
sc, err := stan.Connect("test-cluster", "stan-sub", stan.NatsConn(nc))
if err != nil {
log.Fatal(err)
}
for i := 100; i < 500; i++ {
sc.Publish("hello", []byte("Test"+strconv.Itoa(i)))
}
fmt.Println("Published all the messages")
select {}
}
FROM golang:1.15-alpine as builder
RUN apk add bash ca-certificates git gcc g++ libc-dev
RUN mkdir /app
WORKDIR /app
COPY go.mod .
COPY go.sum .
RUN go mod download
# Copy source code to image
COPY . .
RUN go build -a -o /go/bin/main
FROM alpine:3.14 as base
RUN apk add --update ca-certificates
COPY --from=builder /go/bin/main /
ENTRYPOINT ["/main"]
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: nats-pub
name: nats-pub
spec:
replicas: 0
selector:
matchLabels:
app: nats-pub
template:
metadata:
labels:
app: nats-pub
spec:
containers:
- image: producer
imagePullPolicy: Never
name: producer
Verify that the deployment succeeded before proceeding.
$ kubectl get deployment nats-pub
NAME READY UP-TO-DATE AVAILABLE AGE
nats-pub 0/0 0 0
Consumer function
The consumer function is golang function which takes the body of the request, appends a “Hello” and returns the resulting string. The file is present here.
package main
import (
"fmt"
"io/ioutil"
"net/http"
)
// Handler is the entry point for this fission function
func Handler(w http.ResponseWriter, r *http.Request) {
body, err := ioutil.ReadAll(r.Body)
if err != nil {
http.Error(w, "Error reading request body",
http.StatusInternalServerError)
}
results := string(body)
fmt.Println(results)
w.Write([]byte("Hello " + results))
}
Let’s create the environment and function:
fission environment create --name go --image ghcr.io/fission/go-env-1.23 --builder ghcr.io/fission/go-builder-1.23
fission fn create --name helloworld --env go --src hello.go --entrypoint Handler
Connecting via trigger
We have both the functions ready but the connection between them is the missing glue.
Let’s create a message queue trigger which will invoke the consumerfunc every time there is a message in request queue.
The response will be sent to response queue and in case of consumerfunc invocation fails, the error is written to error queue.
fission mqt create --name natstest --function helloworld --mqtype stan --topic hello --resptopic response --mqtkind keda --errortopic error --maxretries 3 --metadata subject=hello --metadata queueGroup=grp1 --metadata durableName=due --metadata natsServerMonitoringEndpoint=nats.default.svc.cluster.local:8222 --metadata clusterId=test-cluster --metadata natsServer=nats://nats:4222
Parameter list:
- natsServerMonitoringEndpoint - Location of the Nats Streaming Monitoring
- queueGroup - Queue group name of the subscribers
- durableName - Must identify the durability name used by the subscribers
- subject - Name of channel
- natsServer - Location of the Nats Streaming
- clusterId - StanClusterID to form a connection to the NATS Streaming subsystem // it will be same as in producer function
Testing it out
Let’s invoke the producer function so that the queue request gets some messages and we can see the consumer function in action.
$ kubectl scale --replicas=1 deployment/nats-pub
deployment.apps/nats-pub scaled
There are a couple of ways you can verify that the consumerfunc is called:
- Check the logs of
natstestpods:
kubectl logs natstest-b4f6c6579-q2bxd -f
{"level":"info","ts":1616588727.6284368,"caller":"app/main.go:50","msg":"Done processing message","message":"Hello, world!\n"}
{"level":"info","ts":1616588727.6291118,"caller":"app/main.go:36","msg":"Test102"}
{"level":"info","ts":1616588732.3232052,"caller":"app/main.go:50","msg":"Done processing message","message":"Hello, world!\n"}
{"level":"info","ts":1616588732.3247235,"caller":"app/main.go:36","msg":"Test105"}
{"level":"info","ts":1616588735.3536534,"caller":"app/main.go:50","msg":"Done processing message","message":"Hello, world!\n"}
{"level":"info","ts":1616588735.35448,"caller":"app/main.go:36","msg":"Test107"}
{"level":"info","ts":1616588737.6849225,"caller":"app/main.go:50","msg":"Done processing message","message":"Hello, world!\n"}
- Go to nats streaming server queue and check if messages are coming in response queue
Introducing an error
Let’s introduce an error scenario - instead of consumer function returning a 200, you can return 400 which will cause an error:
package main
import (
"net/http"
)
// Handler is the entry point for this fission function
func Handler(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Error reading request body",
http.StatusNotFound)
}
Update the function with new code and invoke the producer function:
fission fn update --name consumerfunc --code hello.go
kubectl scale --replicas=0 deployment/nats-pub
kubectl scale --replicas=1 deployment/nats-pub
We can verify the message in error queue as we did earlier:
{"level":"info","ts":1616589794.4876041,"caller":"app/main.go:62","msg":"NATs consumer up and running!..."}
{"level":"info","ts":1616589794.4877403,"caller":"app/main.go:36","msg":"Test138"}
{"level":"info","ts":1616589998.6383834,"caller":"app/main.go:39","msg":"request returned failure: 400. http_endpoint: http://router.fission/fission-function/helloworld, source: natstest"}
{"level":"info","ts":1616589998.6396098,"caller":"app/main.go:36","msg":"Test129"}
- Go to nats streaming server and check if messages are coming in error queue