From 1ed59f0789783cb3f9e9826a25c221c45a9a786d Mon Sep 17 00:00:00 2001 From: Nafisa Shazia Date: Tue, 26 Jun 2018 09:27:57 -0700 Subject: [PATCH] Push error messages to error topic when NATS mqtrigger fail (#724) --- fission/main.go | 6 +- fission/mqtrigger.go | 30 +++++++-- mqtrigger/messageQueue/nats.go | 61 +++++++++++++---- pkg/apis/fission.io/v1/typefields.go | 2 + test/tests/mqtrigger/main_error.js | 6 ++ test/tests/mqtrigger/test_mqtrigger_error.sh | 71 ++++++++++++++++++++ 6 files changed, 155 insertions(+), 21 deletions(-) create mode 100644 test/tests/mqtrigger/main_error.js create mode 100755 test/tests/mqtrigger/test_mqtrigger_error.sh diff --git a/fission/main.go b/fission/main.go index af1573a0..3d1e11b2 100644 --- a/fission/main.go +++ b/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}, } diff --git a/fission/mqtrigger.go b/fission/mqtrigger.go index b09ad18b..780f9ca5 100644 --- a/fission/mqtrigger.go +++ b/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", - "NAME", "FUNCTION_NAME", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC", "PUB_MSG_CONTENT_TYPE") + 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", "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", - mqt.Metadata.Name, mqt.Spec.FunctionReference.Name, mqt.Spec.MessageQueueType, mqt.Spec.Topic, mqt.Spec.ResponseTopic, mqt.Spec.ContentType) + 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.ErrorTopic, mqt.Spec.MaxRetries, mqt.Spec.ContentType) } w.Flush() diff --git a/mqtrigger/messageQueue/nats.go b/mqtrigger/messageQueue/nats.go index 4a05bbc5..5d455aab 100644 --- a/mqtrigger/messageQueue/nats.go +++ b/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-RespTopic": trigger.Spec.ResponseTopic, - "Content-Type": trigger.Spec.ContentType, + "X-Fission-MQTrigger-Topic": trigger.Spec.Topic, + "X-Fission-MQTrigger-RespTopic": trigger.Spec.ResponseTopic, + "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 - resp, err := http.DefaultClient.Do(req) - if err != nil { - log.Warningf("Request failed: %v", url) + var resp *http.Response + for attempt := 0; attempt <= trigger.Spec.MaxRetries; attempt++ { + // Make the request + 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) - if err != nil { - log.Warningf("Request body error: %v", string(body)) + body, bodyErr := ioutil.ReadAll(resp.Body) + if bodyErr != nil { + log.Warningf("Response body error: %v", bodyErr) return } - if resp.StatusCode != 200 { - log.Printf("Request returned failure: %v", resp.StatusCode) - return + + // Only the latest error response will be published to error topic + 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) { } } } + } diff --git a/pkg/apis/fission.io/v1/typefields.go b/pkg/apis/fission.io/v1/typefields.go index 44087ac0..81a87172 100644 --- a/pkg/apis/fission.io/v1/typefields.go +++ b/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"` } diff --git a/test/tests/mqtrigger/main_error.js b/test/tests/mqtrigger/main_error.js new file mode 100644 index 00000000..37466bc9 --- /dev/null +++ b/test/tests/mqtrigger/main_error.js @@ -0,0 +1,6 @@ +module.exports = async function(context) { + return { + status: 400, + body: "Hello, World!" + }; +} \ No newline at end of file diff --git a/test/tests/mqtrigger/test_mqtrigger_error.sh b/test/tests/mqtrigger/test_mqtrigger_error.sh new file mode 100755 index 00000000..e3fbca80 --- /dev/null +++ b/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 \ No newline at end of file