From 0274ceb312ae742424bdaf1d48873b09fccd7665 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Fri, 20 Dec 2019 02:19:11 +0800 Subject: [PATCH] Add message queue nats-streaming example (#1472) --- .../message-queue/nats-streaming/README.md | 42 +++++++++++++++ .../message-queue/nats-streaming/hello.go | 13 +++++ .../nats-streaming/mqtrigger/go.mod | 12 +++++ .../nats-streaming/mqtrigger/go.sum | 22 ++++++++ .../nats-streaming/mqtrigger/main.go | 38 +++++++++++++ .../message-queue/nats-streaming/specs/README | 42 +++++++++++++++ .../nats-streaming/specs/env-go.yaml | 17 ++++++ .../specs/fission-deployment-config.yaml | 7 +++ .../nats-streaming/specs/function-hello.yaml | 54 +++++++++++++++++++ .../specs/function-publisher.yaml | 27 ++++++++++ .../specs/package-publisher.yaml | 25 +++++++++ 11 files changed, 299 insertions(+) create mode 100644 examples/message-queue/nats-streaming/README.md create mode 100644 examples/message-queue/nats-streaming/hello.go create mode 100644 examples/message-queue/nats-streaming/mqtrigger/go.mod create mode 100644 examples/message-queue/nats-streaming/mqtrigger/go.sum create mode 100644 examples/message-queue/nats-streaming/mqtrigger/main.go create mode 100644 examples/message-queue/nats-streaming/specs/README create mode 100644 examples/message-queue/nats-streaming/specs/env-go.yaml create mode 100644 examples/message-queue/nats-streaming/specs/fission-deployment-config.yaml create mode 100644 examples/message-queue/nats-streaming/specs/function-hello.yaml create mode 100644 examples/message-queue/nats-streaming/specs/function-publisher.yaml create mode 100644 examples/message-queue/nats-streaming/specs/package-publisher.yaml diff --git a/examples/message-queue/nats-streaming/README.md b/examples/message-queue/nats-streaming/README.md new file mode 100644 index 00000000..53a5a7b8 --- /dev/null +++ b/examples/message-queue/nats-streaming/README.md @@ -0,0 +1,42 @@ +# Message Queue Trigger Demonstration - NATS Streaming + +## Create spec + +```bash +$ fission spec init +$ fission env create --name go --image fission/go-env-1.12:1.7.1 --builder fission/go-builder-1.12:1.7.1 --period 5 --spec +$ fission pkg create --name publisher --src mqtrigger/* --spec +$ fission fn create --name publisher --env go --pkg publisher --entrypoint "Handler" --spec +$ fission fn create --name hello --env go --src https://raw.githubusercontent.com/fission/fission/master/examples/go/hello.go --entrypoint "Handler" --spec +$ fission mqt create --name mqtrigger --function hello --mqtype nats-streaming --topic foobar --spec +``` + +## Apply CRDs + +```bash +$ fission spec apply + +# wait for package build status become succeeded +$ fission pkg list +NAME BUILD_STATUS ENV LASTUPDATEDAT +hello-98476132-84ff-4e74-8b0f-2d1005871d1c succeeded go 19 Dec 19 17:31 UTC +publisher succeeded go 19 Dec 19 17:19 UTC + +# you can rebuild the package if it shows failed +$ fission pkg rebuild --name + +$ fission fn test --name publisher +Publish Success + +$ kubectl -n fission-function get pod -l functionName=hello +NAME READY STATUS RESTARTS AGE +poolmgr-go-default-610954-55664ccc68-b258c 2/2 Running 0 18m + +# you should be able to see the function prints message +$ kubectl -n fission-function logs -f -c go poolmgr-go-default-610954-55664ccc68-b258c +{"level":"info","ts":1576775701.7085218,"caller":"go/server.go:209","msg":"listening on 8888 ..."} +{"level":"info","ts":1576776720.3545933,"logger":"specialize_v2_handler","caller":"go/server.go:171","msg":"specializing ..."} +{"level":"info","ts":1576776720.3546736,"logger":"specialize_v2_handler","caller":"go/server.go:62","msg":"loading plugin","location":"/userfunc/15382797-f381-48af-9189-561f45f9285c/hello-98476132-84ff-4e74-8b0f-2d1005871d1c-7693uh-pwsz5u"} +{"level":"info","ts":1576776720.3640525,"logger":"specialize_v2_handler","caller":"go/server.go:180","msg":"done"} +2019/12/19 17:32:00 Hello, world! +``` diff --git a/examples/message-queue/nats-streaming/hello.go b/examples/message-queue/nats-streaming/hello.go new file mode 100644 index 00000000..6fd7af5c --- /dev/null +++ b/examples/message-queue/nats-streaming/hello.go @@ -0,0 +1,13 @@ +package main + +import ( + "log" + "net/http" +) + +// Handler is the entry point for this fission function +func Handler(w http.ResponseWriter, r *http.Request) { + log.Print("Hello, world!") + msg := "Hello, world!\n" + w.Write([]byte(msg)) +} diff --git a/examples/message-queue/nats-streaming/mqtrigger/go.mod b/examples/message-queue/nats-streaming/mqtrigger/go.mod new file mode 100644 index 00000000..eb7b65a7 --- /dev/null +++ b/examples/message-queue/nats-streaming/mqtrigger/go.mod @@ -0,0 +1,12 @@ +module github.com/fission/mqtrigger + +go 1.12 + +require ( + github.com/gogo/protobuf v1.3.1 // indirect + github.com/nats-io/go-nats v1.7.2 // indirect + github.com/nats-io/go-nats-streaming v0.4.4 + github.com/nats-io/nkeys v0.1.3 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + github.com/satori/go.uuid v1.2.0 // indirect +) diff --git a/examples/message-queue/nats-streaming/mqtrigger/go.sum b/examples/message-queue/nats-streaming/mqtrigger/go.sum new file mode 100644 index 00000000..d0a8fa7e --- /dev/null +++ b/examples/message-queue/nats-streaming/mqtrigger/go.sum @@ -0,0 +1,22 @@ +github.com/gogo/protobuf v1.3.1 h1:DqDEcV5aeaTmdFBePNpYsp3FlcVH/2ISVVM9Qf8PSls= +github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o= +github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/nats-io/go-nats v1.7.2 h1:cJujlwCYR8iMz5ofZSD/p2WLW8FabhkQ2lIEVbSvNSA= +github.com/nats-io/go-nats v1.7.2/go.mod h1:+t7RHT5ApZebkrQdnn6AhQJmhJJiKAvJUio1PiiCtj0= +github.com/nats-io/go-nats-streaming v0.4.4 h1:1I3lkZDRdQYXb+holjdqZ2J6xyekrD06o9Fd8rWlgP4= +github.com/nats-io/go-nats-streaming v0.4.4/go.mod h1:gfq4R3c9sKAINOpelo0gn/b9QDMBZnmrttcsNF+lqyo= +github.com/nats-io/nkeys v0.1.3 h1:6JrEfig+HzTH85yxzhSVbjHRJv9cn0p6n3IngIcM5/k= +github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= +github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc= +golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/examples/message-queue/nats-streaming/mqtrigger/main.go b/examples/message-queue/nats-streaming/mqtrigger/main.go new file mode 100644 index 00000000..52e7f2ac --- /dev/null +++ b/examples/message-queue/nats-streaming/mqtrigger/main.go @@ -0,0 +1,38 @@ +package main + +import ( + "fmt" + nats "github.com/nats-io/go-nats-streaming" + uuid "github.com/satori/go.uuid" + "log" + "net/http" +) + +const ( + authToken = "defaultFissionAuthToken" + host = "nats-streaming.fission" + clusterID = "fissionMQTrigger" + topic = "foobar" +) + +func Handler(w http.ResponseWriter, r *http.Request) { + addr := fmt.Sprintf("nats://%v@%v:4222", authToken, host) + nc, err := nats.Connect(clusterID, uuid.NewV4().String(), nats.NatsURL(addr)) + if err != nil { + log.Fatal(err) + } + + log.Printf("Publishing message to topic '%v'\n", topic) + + err = nc.Publish(topic, []byte("dummy")) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(err.Error())) + log.Printf("error sending message to topic: %v", err.Error()) + return + } + nc.Close() + + w.WriteHeader(http.StatusOK) + w.Write([]byte("Publish Success")) +} diff --git a/examples/message-queue/nats-streaming/specs/README b/examples/message-queue/nats-streaming/specs/README new file mode 100644 index 00000000..1db3f9a5 --- /dev/null +++ b/examples/message-queue/nats-streaming/specs/README @@ -0,0 +1,42 @@ + +Fission Specs +============= + +This is a set of specifications for a Fission app. This includes functions, +environments, and triggers; we collectively call these things "resources". + +How to use these specs +---------------------- + +These specs are handled with the 'fission spec' command. See 'fission spec --help'. + +'fission spec apply' will "apply" all resources specified in this directory to your +cluster. That means it checks what resources exist on your cluster, what resources are +specified in the specs directory, and reconciles the difference by creating, updating or +deleting resources on the cluster. + +'fission spec apply' will also package up your source code (or compiled binaries) and +upload the archives to the cluster if needed. It uses 'ArchiveUploadSpec' resources in +this directory to figure out which files to archive. + +You can use 'fission spec apply --watch' to watch for file changes and continuously keep +the cluster updated. + +You can add YAMLs to this directory by writing them manually, but it's easier to generate +them. Use 'fission function create --spec' to generate a function spec, +'fission environment create --spec' to generate an environment spec, and so on. + +You can edit any of the files in this directory, except 'fission-deployment-config.yaml', +which contains a UID that you should never change. To apply your changes simply use +'fission spec apply'. + +fission-deployment-config.yaml +------------------------------ + +fission-deployment-config.yaml contains a UID. This UID is what fission uses to correlate +resources on the cluster to resources in this directory. + +All resources created by 'fission spec apply' are annotated with this UID. Resources on +the cluster that are _not_ annotated with this UID are never modified or deleted by +fission. + diff --git a/examples/message-queue/nats-streaming/specs/env-go.yaml b/examples/message-queue/nats-streaming/specs/env-go.yaml new file mode 100644 index 00000000..c17d8d00 --- /dev/null +++ b/examples/message-queue/nats-streaming/specs/env-go.yaml @@ -0,0 +1,17 @@ +apiVersion: fission.io/v1 +kind: Environment +metadata: + creationTimestamp: null + name: go + namespace: default +spec: + builder: + command: build + image: fission/go-builder-1.12:1.7.1 + imagepullsecret: "" + keeparchive: false + poolsize: 3 + resources: {} + runtime: + image: fission/go-env-1.12:1.7.1 + version: 2 diff --git a/examples/message-queue/nats-streaming/specs/fission-deployment-config.yaml b/examples/message-queue/nats-streaming/specs/fission-deployment-config.yaml new file mode 100644 index 00000000..6c962999 --- /dev/null +++ b/examples/message-queue/nats-streaming/specs/fission-deployment-config.yaml @@ -0,0 +1,7 @@ +# This file is generated by the 'fission spec init' command. +# See the README in this directory for background and usage information. +# Do not edit the UID below: that will break 'fission spec apply' +apiVersion: fission.io/v1 +kind: DeploymentConfig +name: nats-streaming +uid: 9d6b82e4-3d73-49de-9007-1e47fbc377de diff --git a/examples/message-queue/nats-streaming/specs/function-hello.yaml b/examples/message-queue/nats-streaming/specs/function-hello.yaml new file mode 100644 index 00000000..85954c0e --- /dev/null +++ b/examples/message-queue/nats-streaming/specs/function-hello.yaml @@ -0,0 +1,54 @@ +include: +- hello.go +kind: ArchiveUploadSpec +name: hello-go-zqZW + +--- +apiVersion: fission.io/v1 +kind: Package +metadata: + creationTimestamp: null + name: hello-98476132-84ff-4e74-8b0f-2d1005871d1c + namespace: default +spec: + deployment: + checksum: {} + environment: + name: go + namespace: default + source: + checksum: {} + type: url + url: archive://hello-go-zqZW +status: + buildstatus: pending + lastUpdateTimestamp: "2019-12-19T17:31:23.43918Z" + +--- +apiVersion: fission.io/v1 +kind: Function +metadata: + creationTimestamp: null + name: hello + namespace: default +spec: + InvokeStrategy: + ExecutionStrategy: + ExecutorType: poolmgr + MaxScale: 0 + MinScale: 0 + SpecializationTimeout: 120 + TargetCPUPercent: 0 + StrategyType: execution + configmaps: null + environment: + name: go + namespace: default + functionTimeout: 60 + package: + functionName: Handler + packageref: + name: hello-98476132-84ff-4e74-8b0f-2d1005871d1c + namespace: default + resources: {} + secrets: null diff --git a/examples/message-queue/nats-streaming/specs/function-publisher.yaml b/examples/message-queue/nats-streaming/specs/function-publisher.yaml new file mode 100644 index 00000000..a3161f83 --- /dev/null +++ b/examples/message-queue/nats-streaming/specs/function-publisher.yaml @@ -0,0 +1,27 @@ +apiVersion: fission.io/v1 +kind: Function +metadata: + creationTimestamp: null + name: publisher + namespace: default +spec: + InvokeStrategy: + ExecutionStrategy: + ExecutorType: poolmgr + MaxScale: 0 + MinScale: 0 + SpecializationTimeout: 120 + TargetCPUPercent: 0 + StrategyType: execution + configmaps: null + environment: + name: go + namespace: default + functionTimeout: 60 + package: + functionName: Handler + packageref: + name: publisher + namespace: default + resources: {} + secrets: null diff --git a/examples/message-queue/nats-streaming/specs/package-publisher.yaml b/examples/message-queue/nats-streaming/specs/package-publisher.yaml new file mode 100644 index 00000000..8d47d3d3 --- /dev/null +++ b/examples/message-queue/nats-streaming/specs/package-publisher.yaml @@ -0,0 +1,25 @@ +include: +- mqtrigger/* +kind: ArchiveUploadSpec +name: mqtrigger-Z3l6 + +--- +apiVersion: fission.io/v1 +kind: Package +metadata: + creationTimestamp: null + name: publisher + namespace: default +spec: + deployment: + checksum: {} + environment: + name: go + namespace: default + source: + checksum: {} + type: url + url: archive://mqtrigger-Z3l6 +status: + buildstatus: pending + lastUpdateTimestamp: "2019-12-19T17:18:45.81397Z"