Push error messages to error topic when NATS mqtrigger fail (#724)

fission/main.go

@@ -185,11 +185,13 @@ func main() {
 	mqtMQTypeFlag := cli.StringFlag{Name: "mqtype", Value: "nats-streaming", Usage: "Message queue type, e.g. nats-streaming, azure-storage-queue (optional)"}
 	mqtTopicFlag := cli.StringFlag{Name: "topic", Usage: "Message queue Topic the trigger listens on"}
 	mqtRespTopicFlag := cli.StringFlag{Name: "resptopic", Usage: "Topic that the function response is sent on (optional; response discarded if unspecified)"}
+	mqtErrorTopicFlag := cli.StringFlag{Name: "errortopic", Usage: "Topic that the function error messages are sent to (optional; errors discarded if unspecified"}
+	mqtMaxRetries := cli.IntFlag{Name: "maxretries", Value: 0, Usage: "Maximum number of times the function will be retried upon failure (optional; default is 0)"}
 	mqtMsgContentType := cli.StringFlag{Name: "contenttype, c", Value: "application/json", Usage: "Content type of messages that publish to the topic (optional)"}
 	mqtSubcommands := []cli.Command{
-		{Name: "create", Aliases: []string{"add"}, Usage: "Create Message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtFnNameFlag, fnNamespaceFlag, mqtMQTypeFlag, mqtTopicFlag, mqtRespTopicFlag, mqtMsgContentType, specSaveFlag}, Action: mqtCreate},
+		{Name: "create", Aliases: []string{"add"}, Usage: "Create Message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtFnNameFlag, fnNamespaceFlag, mqtMQTypeFlag, mqtTopicFlag, mqtRespTopicFlag, mqtErrorTopicFlag, mqtMaxRetries, mqtMsgContentType, specSaveFlag}, Action: mqtCreate},
 		{Name: "get", Usage: "Get message queue trigger", Flags: []cli.Flag{triggerNamespaceFlag}, Action: mqtGet},
-		{Name: "update", Usage: "Update message queue trigger", Flags: []cli.Flag{mqtNameFlag, triggerNamespaceFlag, mqtTopicFlag, mqtRespTopicFlag, mqtFnNameFlag, mqtMsgContentType}, Action: mqtUpdate},
+		{Name: "update", Usage: "Update message queue trigger", Flags: []cli.Flag{mqtNameFlag, triggerNamespaceFlag, mqtTopicFlag, mqtRespTopicFlag, mqtErrorTopicFlag, mqtMaxRetries, mqtFnNameFlag, mqtMsgContentType}, Action: mqtUpdate},
 		{Name: "delete", Usage: "Delete message queue trigger", Flags: []cli.Flag{mqtNameFlag, triggerNamespaceFlag}, Action: mqtDelete},
 		{Name: "list", Usage: "List message queue triggers", Flags: []cli.Flag{mqtMQTypeFlag, triggerNamespaceFlag}, Action: mqtList},
 	}

fission/mqtrigger.go

@@ -69,6 +69,14 @@ func mqtCreate(c *cli.Context) error {
 		log.Fatal("Listen topic should not equal to response topic")
 	}
 
+	errorTopic := c.String("errortopic")
+
+	maxRetries := c.Int("maxretries")
+
+	if maxRetries < 0 {
+		log.Fatal("Maximum number of retries must be a natural number, default is 0")
+	}
+
 	contentType := c.String("contenttype")
 	if len(contentType) == 0 {
 		contentType = "application/json"
@@ -89,6 +97,8 @@ func mqtCreate(c *cli.Context) error {
 			MessageQueueType: mqType,
 			Topic:            topic,
 			ResponseTopic:    respTopic,
+			ErrorTopic:       errorTopic,
+			MaxRetries:       maxRetries,
 			ContentType:      contentType,
 		},
 	}
@@ -122,6 +132,8 @@ func mqtUpdate(c *cli.Context) error {
 
 	topic := c.String("topic")
 	respTopic := c.String("resptopic")
+	errorTopic := c.String("errortopic")
+	maxRetries := c.Int("maxretries")
 	fnName := c.String("function")
 	contentType := c.String("contenttype")
 
@@ -144,6 +156,14 @@ func mqtUpdate(c *cli.Context) error {
 		mqt.Spec.ResponseTopic = respTopic
 		updated = true
 	}
+	if len(errorTopic) > 0 {
+		mqt.Spec.ErrorTopic = errorTopic
+		updated = true
+	}
+	if maxRetries > -1 {
+		mqt.Spec.MaxRetries = maxRetries
+		updated = true
+	}
 	if len(fnName) > 0 {
 		mqt.Spec.FunctionReference.Name = fnName
 		updated = true
@@ -154,7 +174,7 @@ func mqtUpdate(c *cli.Context) error {
 	}
 
 	if !updated {
-		log.Fatal("Nothing to update. Use --topic, --resptopic, or --function.")
+		log.Fatal("Nothing to update. Use --topic, --resptopic, --errortopic, --maxretries or --function.")
 	}
 
 	_, err = client.MessageQueueTriggerUpdate(mqt)
@@ -191,11 +211,11 @@ func mqtList(c *cli.Context) error {
 
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
 
-	fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
+	fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n",
-		"NAME", "FUNCTION_NAME", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC", "PUB_MSG_CONTENT_TYPE")
+		"NAME", "FUNCTION_NAME", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC", "ERROR_TOPIC", "MAX_RETRIES", "PUB_MSG_CONTENT_TYPE")
 	for _, mqt := range mqts {
-		fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
+		fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n",
-			mqt.Metadata.Name, mqt.Spec.FunctionReference.Name, mqt.Spec.MessageQueueType, mqt.Spec.Topic, mqt.Spec.ResponseTopic, mqt.Spec.ContentType)
+			mqt.Metadata.Name, mqt.Spec.FunctionReference.Name, mqt.Spec.MessageQueueType, mqt.Spec.Topic, mqt.Spec.ResponseTopic, mqt.Spec.ErrorTopic, mqt.Spec.MaxRetries, mqt.Spec.ContentType)
 	}
 	w.Flush()
 

mqtrigger/messageQueue/nats.go

@@ -107,35 +107,67 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
 		log.Printf("Making HTTP request to %v", url)
 
 		headers := map[string]string{
-			"X-Fission-MQTrigger-Topic":     trigger.Spec.Topic,
+			"X-Fission-MQTrigger-Topic":      trigger.Spec.Topic,
-			"X-Fission-MQTrigger-RespTopic": trigger.Spec.ResponseTopic,
+			"X-Fission-MQTrigger-RespTopic":  trigger.Spec.ResponseTopic,
-			"Content-Type":                  trigger.Spec.ContentType,
+			"X-Fission-MQTrigger-ErrorTopic": trigger.Spec.ErrorTopic,
+			"Content-Type":                   trigger.Spec.ContentType,
 		}
 
 		// Create request
 		req, err := http.NewRequest("POST", url, bytes.NewReader(msg.Data))
+
+		if err != nil {
+			log.Errorf("Could not issue POST request with message to url %v", url)
+			return
+		}
+
 		for k, v := range headers {
 			req.Header.Add(k, v)
 		}
 
-		// Make the request
+		var resp *http.Response
-		resp, err := http.DefaultClient.Do(req)
+		for attempt := 0; attempt <= trigger.Spec.MaxRetries; attempt++ {
-		if err != nil {
+			// Make the request
-			log.Warningf("Request failed: %v", url)
+			resp, err = http.DefaultClient.Do(req)
+			if err != nil {
+				log.Error("Error invoking function for trigger %v: %v", trigger.Metadata.Name, err)
+				continue
+			}
+			if resp == nil {
+				continue
+			}
+			if err == nil && resp.StatusCode == http.StatusOK {
+				// Success, quit retrying
+				break
+			}
+		}
+
+		if resp == nil {
+			log.Warning("Every retry failed; final retry gave empty response.")
 			return
 		}
+
 		defer resp.Body.Close()
 
-		body, err := ioutil.ReadAll(resp.Body)
+		body, bodyErr := ioutil.ReadAll(resp.Body)
-		if err != nil {
+		if bodyErr != nil {
-			log.Warningf("Request body error: %v", string(body))
+			log.Warningf("Response body error: %v", bodyErr)
 			return
 		}
-		if resp.StatusCode != 200 {
+
-			log.Printf("Request returned failure: %v", resp.StatusCode)
+		// Only the latest error response will be published to error topic
-			return
+		if err != nil || resp.StatusCode != 200 {
+			if len(trigger.Spec.ErrorTopic) > 0 && len(body) > 0 {
+				publishErr := nats.nsConn.Publish(trigger.Spec.ErrorTopic, body)
+				if publishErr != nil {
+					log.Error("Failed to publish error to error topic: %v", publishErr)
+					// TODO: We will ack this message after max retries to prevent re-processing but
+					// this may cause message loss
+				}
+			}
 		}
-		// trigger acks message only if a request done successfully
+
+		// Trigger acks message only if a request was processed successfully
 		err = msg.Ack()
 		if err != nil {
 			log.Warningf("Failed to ack message: %v", err)
@@ -148,4 +180,5 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
 			}
 		}
 	}
+
 }

pkg/apis/fission.io/v1/typefields.go

@@ -306,6 +306,8 @@ type (
 		MessageQueueType  MessageQueueType  `json:"messageQueueType"`
 		Topic             string            `json:"topic"`
 		ResponseTopic     string            `json:"respTopic,omitempty"`
+		ErrorTopic        string            `json:"errorTopic"`
+		MaxRetries        int               `json:"maxRetries"`
 		ContentType       string            `json:"contentType"`
 	}
 

test/tests/mqtrigger/main_error.js

@@ -0,0 +1,6 @@
+module.exports = async function(context) {
+    return {
+        status: 400,
+        body: "Hello, World!"
+    };
+}

test/tests/mqtrigger/test_mqtrigger_error.sh

@@ -0,0 +1,71 @@
+#!/bin/bash
+
+#
+# Create a function and trigger it using NATS
+# To run this on Minikube, uncomment line 18
+
+set -euo pipefail
+set +x
+
+ROOT=$(dirname $0)/../..
+DIR=$(dirname $0)
+
+clusterID="fissionMQTrigger"
+topic="foo.bar"
+resptopic="foo.foo"
+errortopic="foo.error"
+maxretries=1
+# FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(minikube ip):4222"
+expectedRespOutput="[foo.error]: 'Hello, World!'"
+
+log "Pre-test cleanup"
+fission env delete --name nodejs || true
+
+log "Creating nodejs env"
+fission env create --name nodejs --image fission/node-env
+#trap "fission env delete --name nodejs" EXIT
+
+log "Creating function"
+fn=hello-$(date +%s)
+fission fn create --name $fn --env nodejs --code $DIR/main_error.js --method GET
+##trap "fission fn delete --name $fn" EXIT
+
+log "Creating message queue trigger"
+mqt=mqt-$(date +%s)
+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
+#trap "fission mqtrigger delete --name $mqt" EXIT
+
+# wait until nats trigger is created
+sleep 5
+
+#
+# Send a message
+#
+log "Sending message"
+go run $DIR/stan-pub.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id clientPub $topic ""
+
+#
+# Wait for message on error topic
+#
+log "Waiting for response"
+TIMEOUT=timeout
+if [ $(uname -s) == 'Darwin' ]
+then
+    # If this fails on mac os, do "brew install coreutils".
+    TIMEOUT=gtimeout
+fi
+response=$(go run $DIR/stan-sub.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id clientSub $errortopic 2>&1)
+
+log "Subscriber received response: $response"
+
+fission mqtrigger delete --name $mqt
+# kubectl delete functions --all
+
+if [[ "$response" != "$expectedRespOutput" ]]; then
+    log "$response is not equal to $expectedRespOutput"
+    exit 1
+else
+    log "Responses match."
+fi