diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index cc05952e..4c77eb38 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -839,3 +839,46 @@ spec: {{- if .Values.extraCoreComponentPodConfig }} {{ toYaml .Values.extraCoreComponentPodConfig | indent 6 -}} {{- end }} + +{{- if .Values.mqt_keda.enabled }} +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: mqtrigger-keda + labels: + chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" + svc: mqtrigger-keda + messagequeue: keda +spec: + replicas: 1 + selector: + matchLabels: + svc: mqtrigger-keda + messagequeue: keda + template: + metadata: + labels: + svc: mqtrigger-keda + messagequeue: keda + spec: + containers: + - name: mqtrigger-keda + image: {{ include "fission-bundleImage" . | quote }} + imagePullPolicy: {{ .Values.pullPolicy }} + command: ["/fission-bundle"] + args: ["--mqt_keda", "--routerUrl", "http://router.{{ .Release.Namespace }}"] + env: + - name: TRACE_JAEGER_COLLECTOR_ENDPOINT + value: "{{ .Values.traceCollectorEndpoint }}" + - name: TRACING_SAMPLING_RATE + value: {{ .Values.traceSamplingRate | default "0.5" | quote }} + - name: DEBUG_ENV + value: {{ .Values.debugEnv | quote }} + - name: KAFKA_IMAGE + value: {{ .Values.mqt_keda.connector_images.kafka }} + serviceAccountName: fission-svc +{{- if .Values.extraCoreComponentPodConfig }} +{{ toYaml .Values.extraCoreComponentPodConfig | indent 6 -}} +{{- end }} +{{- end }} diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index c75bb720..08f2e8d4 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -300,3 +300,9 @@ canaryDeployment: # Use these flags to enable opentracing, the variable is endpoint of Jaeger collector in the format shown below #traceCollectorEndpoint: "http://jaeger-collector.jaeger.svc:14268/api/traces?format=jaeger.thrift" #traceSamplingRate: 0.75 + +## Message Queue Trigger Kind, KEDA: enable and configuration +mqt_keda: + enabled: true + connector_images: + kafka: fission/keda-kafka diff --git a/cmd/fission-bundle/main.go b/cmd/fission-bundle/main.go index e1949734..3c7c902f 100644 --- a/cmd/fission-bundle/main.go +++ b/cmd/fission-bundle/main.go @@ -35,6 +35,7 @@ import ( "github.com/fission/fission/pkg/info" "github.com/fission/fission/pkg/kubewatcher" functionLogger "github.com/fission/fission/pkg/logger" + mqt "github.com/fission/fission/pkg/mqtrigger" "github.com/fission/fission/pkg/router" "github.com/fission/fission/pkg/storagesvc" "github.com/fission/fission/pkg/timer" @@ -78,6 +79,14 @@ func runMessageQueueMgr(logger *zap.Logger, routerUrl string) { } } +// KEDA based MessageQueue Trigger Manager +func runMQManager(logger *zap.Logger, routerURL string) { + err := mqt.StartScalerManager(logger, routerURL) + if err != nil { + logger.Fatal("error starting mqt scaler manager", zap.Error(err)) + } +} + func runStorageSvc(logger *zap.Logger, port int, storage storagesvc.Storage) { err := storagesvc.Start(logger, storage, port) if err != nil { @@ -139,6 +148,8 @@ func registerTraceExporter(logger *zap.Logger, arguments map[string]interface{}) serviceName = "Fission-BuilderMgr" } else if arguments["--storageServicePort"] != nil { serviceName = "Fission-StorageSvc" + } else if arguments["--mqt_keda"] == true { + serviceName = "Fission-Keda-MQTrigger" } exporter, err := jaeger.NewExporter(jaeger.Options{ @@ -201,6 +212,7 @@ Usage: fission-bundle --builderMgr [--storageSvcUrl=] [--envbuilder-namespace=] fission-bundle --timer [--routerUrl=] fission-bundle --mqt [--routerUrl=] + fission-bundle --mqt_keda [--routerUrl=] fission-bundle --logger fission-bundle --version Options: @@ -217,6 +229,7 @@ Options: --kubewatcher Start Kubernetes events watcher. --timer Start Timer. --mqt Start message queue trigger. + --mqt_keda Start message queue trigger of kind KEDA --builderMgr Start builder manager. --version Print version information ` @@ -282,6 +295,10 @@ Options: runMessageQueueMgr(logger, routerUrl) } + if arguments["--mqt_keda"] == true { + runMQManager(logger, routerUrl) + } + if arguments["--builderMgr"] == true { runBuilderMgr(logger, storageSvcUrl, envBuilderNs) } diff --git a/go.mod b/go.mod index 04959081..ad4050dd 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/Azure/azure-sdk-for-go v12.4.0-beta+incompatible github.com/Microsoft/go-winio v0.4.14 // indirect github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 // indirect - github.com/Shopify/sarama v1.21.0 + github.com/Shopify/sarama v1.23.1 github.com/aws/aws-sdk-go v1.32.7 // indirect github.com/blend/go-sdk v1.1.1 // indirect github.com/bsm/sarama-cluster v2.1.15+incompatible @@ -29,7 +29,6 @@ require ( github.com/go-openapi/spec v0.17.2 github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 // indirect github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef // indirect - github.com/google/gofuzz v1.0.0 // indirect github.com/googleapis/gnostic v0.3.0 // indirect github.com/gophercloud/gophercloud v0.2.0 // indirect github.com/gorilla/mux v1.7.0 @@ -39,6 +38,7 @@ require ( github.com/hashicorp/go-multierror v0.0.0-20180717150148-3d5d8f294aa0 github.com/imdario/mergo v0.3.5 github.com/influxdata/influxdb v1.2.0 + github.com/json-iterator/go v1.1.9 // indirect github.com/kr/pretty v0.1.0 // indirect github.com/life1347/color v1.7.0 github.com/marstr/guid v0.0.0-20170427235115-8bdf7d1a087c // indirect @@ -53,6 +53,7 @@ require ( github.com/opencontainers/image-spec v1.0.1 // indirect github.com/opencontainers/runc v0.1.1 // indirect github.com/ory/dockertest v3.3.5+incompatible + github.com/pierrec/lz4 v2.0.5+incompatible // indirect github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.0.0 github.com/prometheus/common v0.4.1 @@ -64,6 +65,7 @@ require ( github.com/stretchr/testify v1.5.1 github.com/ulikunitz/xz v0.0.0-20180703112113-636d36a76670 // indirect github.com/wcharczuk/go-chart v2.0.1+incompatible + github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c go.opencensus.io v0.22.0 go.uber.org/atomic v1.3.2 // indirect go.uber.org/multierr v1.1.0 // indirect @@ -75,6 +77,7 @@ require ( gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/ini.v1 v1.57.0 // indirect + gopkg.in/jcmturner/goidentity.v3 v3.0.0 // indirect k8s.io/api v0.0.0-20190620084959-7cf5895f2711 k8s.io/apiextensions-apiserver v0.0.0-20190620085554-14e95df34f1f k8s.io/apimachinery v0.0.0-20190612205821-1799e75a0719 diff --git a/go.sum b/go.sum index 2db47568..173221f0 100644 --- a/go.sum +++ b/go.sum @@ -17,8 +17,8 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= github.com/DataDog/datadog-go v0.0.0-20180822151419-281ae9f2d895/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ= -github.com/DataDog/zstd v1.3.5 h1:DtpNbljikUepEPD16hD4LvIcmhnhdLTiW/5pHgbmp14= -github.com/DataDog/zstd v1.3.5/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo= +github.com/DataDog/zstd v1.3.6-0.20190409195224-796139022798 h1:2T/jmrHeTezcCM58lvEQXs0UpQJCo5SoGAcg+mbSTIg= +github.com/DataDog/zstd v1.3.6-0.20190409195224-796139022798/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo= github.com/Microsoft/go-winio v0.4.14 h1:+hMXMk01us9KgxGb7ftKQt2Xpf5hH/yky+TDA+qxleU= github.com/Microsoft/go-winio v0.4.14/go.mod h1:qXqCSQ3Xa7+6tgxaGTIe4Kpcdsi+P8jBhyzoq1bpyYA= github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb06e3pkSAbeQ52E9H9iFoQsEEwGN64994WTCIhntQ= @@ -29,8 +29,8 @@ github.com/PuerkitoBio/purell v1.1.0/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbt github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 h1:d+Bc7a5rLufV/sSk/8dngufqelfh6jnri85riMAaF/M= github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE= github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo= -github.com/Shopify/sarama v1.21.0 h1:0GKs+e8mn1RRUzfg9oUXv3v7ZieQLmOZF/bfnmmGhM8= -github.com/Shopify/sarama v1.21.0/go.mod h1:yuqtN/pe8cXRWG5zPaO7hCfNJp5MwmkoJEoLjkm5tCQ= +github.com/Shopify/sarama v1.23.1 h1:XxJBCZEoWJtoWjf/xRbmGUpAmTZGnuuF0ON0EvxxBrs= +github.com/Shopify/sarama v1.23.1/go.mod h1:XLH1GYJnLVE0XCr6KdJGVJRTwY30moWNJ4sERjXX6fs= github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc= github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= github.com/airbrake/gobrake v3.6.1+incompatible/go.mod h1:wM4gu3Cn0W0K7GUuVWnlXZU11AGBXMILnrdOU8Kn00o= @@ -176,6 +176,8 @@ github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db h1:woRePGFeVFfLKN/pOkfl+p/TAqKOfFu+7KPlMVpok/w= github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/btree v0.0.0-20160524151835-7d79101e329e/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= @@ -225,6 +227,8 @@ github.com/hashicorp/go-multierror v0.0.0-20180717150148-3d5d8f294aa0/go.mod h1: github.com/hashicorp/go-retryablehttp v0.5.3/go.mod h1:9B5zBasrRhHXnJnui7y6sL7es7NDiJgTc6Er0maI1Xs= github.com/hashicorp/go-uuid v1.0.0 h1:RS8zrF7PhGwyNPOtxSClXXj9HA8feRnJzgnI1RJCSnM= github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/hashicorp/go-uuid v1.0.1 h1:fv1ep09latC32wFoVwnqcnKJGnMSdBanPczbHAYm1BE= +github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+dAcgU= github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= @@ -240,12 +244,16 @@ github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NH github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= github.com/influxdata/influxdb v1.2.0 h1:ZSB1cdZP9/8yyFzZhyaHimPL55Qo2kRDv2VhgnCePJ4= github.com/influxdata/influxdb v1.2.0/go.mod h1:qZna6X/4elxqT3yI9iZYdZrWWdeFOOprn86kgg4+IzY= +github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 h1:FUwcHNlEqkqLjLBdCp5PRlCFijNjvcYANOZXzCfXwCM= +github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o= github.com/jmespath/go-jmespath v0.3.0 h1:OS12ieG61fsCg5+qLJ+SsW9NicxNkg3b25OyT2yCeUc= github.com/jmespath/go-jmespath v0.3.0/go.mod h1:9QtRXoHjLGCJ5IBSaohpXITPlowMeeYCZ7fLUTSywik= github.com/jonboulle/clockwork v0.0.0-20141017032234-72f9bd7c4e0c/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo= github.com/json-iterator/go v0.0.0-20180701071628-ab8a2e0c74be/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= github.com/json-iterator/go v1.1.6 h1:MrUvLMLTMxbqFJ9kzlvat/rYZqZnW3u4wkLzWTaFwKs= github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= +github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns= +github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU= github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= @@ -287,8 +295,10 @@ github.com/minio/minio-go v6.0.14+incompatible/go.mod h1:7guKYtitv8dktvNUGrhzmNl github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y= github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI= github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= @@ -338,6 +348,7 @@ github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144T github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU= +github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc= github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I= github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY= github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= @@ -347,6 +358,7 @@ github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pquerna/cachecontrol v0.0.0-20171018203845-0dec1b30a021/go.mod h1:prYjPmNq4d1NPVmpShWobRqXY3q7Vp+80DqgxxUrUIA= @@ -419,6 +431,10 @@ github.com/ulikunitz/xz v0.0.0-20180703112113-636d36a76670 h1:HQWT4ta3wW5GZ790Ga github.com/ulikunitz/xz v0.0.0-20180703112113-636d36a76670/go.mod h1:2bypXElzHzzJZwzH67Y6wb67pO62Rzfn7BSiF4ABRW8= github.com/wcharczuk/go-chart v2.0.1+incompatible h1:0pz39ZAycJFF7ju/1mepnk26RLVLBCWz1STcD3doU0A= github.com/wcharczuk/go-chart v2.0.1+incompatible/go.mod h1:PF5tmL4EIx/7Wf+hEkpCqYi5He4u90sw+0+6FhrryuE= +github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk= +github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I= +github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0= +github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y= github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU= github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q= go.etcd.io/bbolt v1.3.3 h1:MUGmc65QhB3pIlaQ5bB4LwqSj6GIonVJXpZiaKNyaKk= @@ -442,6 +458,7 @@ golang.org/x/crypto v0.0.0-20181025213731-e84da0312774/go.mod h1:6SG95UA2DQfeDnf golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190211182817-74369b46fc67/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE= golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc= golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= @@ -501,6 +518,7 @@ golang.org/x/sys v0.0.0-20190209173611-3b5209105503/go.mod h1:STP8DvDyc/dI5b8T5h golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -571,6 +589,16 @@ gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= gopkg.in/ini.v1 v1.57.0 h1:9unxIsFcTt4I55uWluz+UmL95q4kdJ0buvQ1ZIqVQww= gopkg.in/ini.v1 v1.57.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= +gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw= +gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN5hSM= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q= +gopkg.in/jcmturner/goidentity.v3 v3.0.0 h1:1duIyWiTaYvVx3YX2CYtpJbUFd7/UuPYCfgXtQ3VTbI= +gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4= +gopkg.in/jcmturner/gokrb5.v7 v7.2.3 h1:hHMV/yKPwMnJhPuPx7pH2Uw/3Qyf+thJYlisUc44010= +gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM= +gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU= +gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8= gopkg.in/natefinch/lumberjack.v2 v2.0.0-20150622162204-20b71e5b60d7/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k= gopkg.in/square/go-jose.v2 v2.0.0-20180411045311-89060dee6a84/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= diff --git a/pkg/apis/core/v1/types.go b/pkg/apis/core/v1/types.go index 4b0f160d..6a71c400 100644 --- a/pkg/apis/core/v1/types.go +++ b/pkg/apis/core/v1/types.go @@ -645,6 +645,34 @@ type ( // Content type of payload ContentType string `json:"contentType"` + + // The period to check each trigger source on every ScaledObject, and scale the deployment up or down accordingly + // +optional + PollingInterval *int32 `json:"pollingInterval,omitempty"` + + // The period to wait after the last trigger reported active before scaling the deployment back to 0 + // +optional + CooldownPeriod *int32 `json:"cooldownPeriod,omitempty"` + + // Minimum number of replicas KEDA will scale the deployment down to + // +optional + MinReplicaCount *int32 `json:"minReplicaCount,omitempty"` + + // Maximum number of replicas KEDA will scale the deployment up to + // +optional + MaxReplicaCount *int32 `json:"maxReplicaCount,omitempty"` + + // ScalerTrigger fields + // +optional + Metadata map[string]string `json:"metadata"` + + // Secret name + // +optional + Secret string `json:"secret,omitempty"` + + // Kind of Message Queue Trigger to be created, by default its fission + // +optional + MqtKind string `json:"mqtkind,omitempty"` } // TimeTrigger invokes the specific function at a time or diff --git a/pkg/apis/core/v1/zz_generated.deepcopy.go b/pkg/apis/core/v1/zz_generated.deepcopy.go index d71d38e3..00d58eae 100644 --- a/pkg/apis/core/v1/zz_generated.deepcopy.go +++ b/pkg/apis/core/v1/zz_generated.deepcopy.go @@ -710,6 +710,33 @@ func (in *MessageQueueTriggerList) DeepCopyObject() runtime.Object { func (in *MessageQueueTriggerSpec) DeepCopyInto(out *MessageQueueTriggerSpec) { *out = *in in.FunctionReference.DeepCopyInto(&out.FunctionReference) + if in.PollingInterval != nil { + in, out := &in.PollingInterval, &out.PollingInterval + *out = new(int32) + **out = **in + } + if in.CooldownPeriod != nil { + in, out := &in.CooldownPeriod, &out.CooldownPeriod + *out = new(int32) + **out = **in + } + if in.MinReplicaCount != nil { + in, out := &in.MinReplicaCount, &out.MinReplicaCount + *out = new(int32) + **out = **in + } + if in.MaxReplicaCount != nil { + in, out := &in.MaxReplicaCount, &out.MaxReplicaCount + *out = new(int32) + **out = **in + } + if in.Metadata != nil { + in, out := &in.Metadata, &out.Metadata + *out = make(map[string]string, len(*in)) + for key, val := range *in { + (*out)[key] = val + } + } return } diff --git a/pkg/crd/client.go b/pkg/crd/client.go index 7fe7d7ed..31fb1bf9 100644 --- a/pkg/crd/client.go +++ b/pkg/crd/client.go @@ -23,6 +23,7 @@ import ( apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" _ "k8s.io/client-go/plugin/pkg/client/auth" "k8s.io/client-go/rest" @@ -107,3 +108,29 @@ func (fc *FissionClient) WaitForCRDs() error { } } } + +// GetDynamicClient creates and returns new dynamic client or returns an error +func GetDynamicClient() (dynamic.Interface, error) { + var config *rest.Config + var err error + + // get the config, either from kubeconfig or using our + // in-cluster service account + kubeConfig := os.Getenv("KUBECONFIG") + if len(kubeConfig) != 0 { + config, err = clientcmd.BuildConfigFromFlags("", kubeConfig) + if err != nil { + return nil, err + } + } else { + config, err = rest.InClusterConfig() + if err != nil { + return nil, err + } + } + dynamicClient, err := dynamic.NewForConfig(config) + if err != nil { + return nil, err + } + return dynamicClient, nil +} diff --git a/pkg/fission-cli/cmd/mqtrigger/command.go b/pkg/fission-cli/cmd/mqtrigger/command.go index ef5c3616..ad15e91d 100644 --- a/pkg/fission-cli/cmd/mqtrigger/command.go +++ b/pkg/fission-cli/cmd/mqtrigger/command.go @@ -33,7 +33,9 @@ func Commands() *cobra.Command { Required: []flag.Flag{flag.MqtFnName, flag.MqtTopic}, Optional: []flag.Flag{flag.MqtName, flag.MqtMQType, flag.MqtRespTopic, flag.MqtErrorTopic, flag.MqtMaxRetries, flag.MqtMsgContentType, - flag.NamespaceFunction, flag.SpecSave, flag.SpecDry}, + flag.NamespaceFunction, flag.SpecSave, flag.SpecDry, flag.MqtPollingInterval, + flag.MqtCooldownPeriod, flag.MqtMinReplicaCount, flag.MqtMaxReplicaCount, flag.MqtSecret, + flag.MqtMetadata, flag.MqtKind}, }) updateCmd := &cobra.Command{ @@ -45,7 +47,9 @@ func Commands() *cobra.Command { wrapper.SetFlags(updateCmd, flag.FlagSet{ Required: []flag.Flag{flag.MqtName}, Optional: []flag.Flag{flag.MqtFnName, flag.MqtTopic, flag.MqtRespTopic, flag.MqtErrorTopic, - flag.MqtMaxRetries, flag.MqtMsgContentType, flag.NamespaceTrigger}, + flag.MqtMaxRetries, flag.MqtMsgContentType, flag.NamespaceTrigger, flag.MqtPollingInterval, + flag.MqtCooldownPeriod, flag.MqtMinReplicaCount, flag.MqtMaxReplicaCount, flag.MqtMetadata, + flag.MqtSecret, flag.MqtKind}, }) deleteCmd := &cobra.Command{ diff --git a/pkg/fission-cli/cmd/mqtrigger/create.go b/pkg/fission-cli/cmd/mqtrigger/create.go index 71e90ad3..1c651e99 100644 --- a/pkg/fission-cli/cmd/mqtrigger/create.go +++ b/pkg/fission-cli/cmd/mqtrigger/create.go @@ -93,6 +93,34 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { return err } + pollingInterval := int32(input.Int(flagkey.MqtPollingInterval)) + if pollingInterval < 0 { + return errors.New("Polling interval must be greater than or equal to 0") + } + + cooldownPeriod := int32(input.Int(flagkey.MqtCooldownPeriod)) + if cooldownPeriod < 0 { + return errors.New("CooldownPeriod interval is the period to wait after the last trigger reported active before scaling the deployment back to 0, it must be greater than or equal to 0") + } + + minReplicaCount := int32(input.Int(flagkey.MqtMinReplicaCount)) + if minReplicaCount < 0 { + return errors.New("MinReplicaCount must be greater than or equal to 0") + } + + maxReplicaCount := int32(input.Int(flagkey.MqtMaxReplicaCount)) + if maxReplicaCount < 0 { + return errors.New("MaxReplicaCount must be greater than or equal to 0") + } + + metadata := make(map[string]string) + metadataParams := input.StringSlice(flagkey.MqtMetadata) + _ = util.UpdateMapFromStringSlice(&metadata, metadataParams) + + secret := input.String(flagkey.MqtSecret) + + mqtKind := input.String(flagkey.MqtKind) + if input.Bool(flagkey.SpecSave) { specDir := util.GetSpecDir(input) fr, err := spec.ReadSpecs(specDir) @@ -131,6 +159,13 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { ErrorTopic: errorTopic, MaxRetries: maxRetries, ContentType: contentType, + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: metadata, + Secret: secret, + MqtKind: mqtKind, }, } diff --git a/pkg/fission-cli/cmd/mqtrigger/update.go b/pkg/fission-cli/cmd/mqtrigger/update.go index d81f0f03..d0454e26 100644 --- a/pkg/fission-cli/cmd/mqtrigger/update.go +++ b/pkg/fission-cli/cmd/mqtrigger/update.go @@ -26,6 +26,7 @@ import ( "github.com/fission/fission/pkg/fission-cli/cliwrapper/cli" "github.com/fission/fission/pkg/fission-cli/cmd" flagkey "github.com/fission/fission/pkg/fission-cli/flag/key" + "github.com/fission/fission/pkg/fission-cli/util" ) type UpdateSubCommand struct { @@ -60,7 +61,13 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { maxRetries := input.Int(flagkey.MqtMaxRetries) fnName := input.String(flagkey.MqtFnName) contentType := input.String(flagkey.MqtMsgContentType) - + pollingInterval := int32(input.Int(flagkey.MqtPollingInterval)) + cooldownPeriod := int32(input.Int(flagkey.MqtCooldownPeriod)) + minReplicaCount := int32(input.Int(flagkey.MqtMinReplicaCount)) + maxReplicaCount := int32(input.Int(flagkey.MqtMaxReplicaCount)) + metadataParams := input.StringSlice(flagkey.MqtMetadata) + secret := input.String(flagkey.MqtSecret) + mqtKind := input.String(flagkey.MqtKind) // TODO : Find out if we can make a call to checkIfFunctionExists, in the same ns more importantly. err = checkMQTopicAvailability(mqt.Spec.MessageQueueType, topic, respTopic) @@ -81,7 +88,7 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { mqt.Spec.ErrorTopic = errorTopic updated = true } - if maxRetries > -1 { + if input.IsSet(flagkey.MqtMaxRetries) { mqt.Spec.MaxRetries = maxRetries updated = true } @@ -89,10 +96,39 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { mqt.Spec.FunctionReference.Name = fnName updated = true } - if len(contentType) > 0 { + if input.IsSet(flagkey.MqtMsgContentType) { mqt.Spec.ContentType = contentType updated = true } + if input.IsSet(flagkey.MqtPollingInterval) { + mqt.Spec.PollingInterval = &pollingInterval + updated = true + } + if input.IsSet(flagkey.MqtCooldownPeriod) { + mqt.Spec.CooldownPeriod = &cooldownPeriod + updated = true + } + if input.IsSet(flagkey.MqtMinReplicaCount) { + mqt.Spec.MinReplicaCount = &minReplicaCount + updated = true + } + if input.IsSet(flagkey.MqtMaxReplicaCount) { + mqt.Spec.MaxReplicaCount = &maxReplicaCount + updated = true + } + + if input.IsSet(flagkey.MqtMetadata) { + updated = updated || util.UpdateMapFromStringSlice(&mqt.Spec.Metadata, metadataParams) + } + if input.IsSet(flagkey.MqtSecret) { + mqt.Spec.Secret = secret + updated = true + } + + if input.IsSet(flagkey.MqtKind) { + mqt.Spec.MqtKind = mqtKind + updated = true + } if !updated { return errors.New("Nothing changed, see 'help' for more details") diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go index 65e192e4..db8a23cc 100644 --- a/pkg/fission-cli/flag/flag.go +++ b/pkg/fission-cli/flag/flag.go @@ -128,14 +128,21 @@ var ( TtFnName = Flag{Type: String, Name: flagkey.TtFnName, Usage: "Function name"} TtRound = Flag{Type: Int, Name: flagkey.TtRound, Usage: "Get next N rounds of invocation time", DefaultValue: 1} - MqtName = Flag{Type: String, Name: flagkey.MqtName, Usage: "Message queue trigger name"} - MqtFnName = Flag{Type: String, Name: flagkey.MqtFnName, Usage: "Function name"} - MqtMQType = Flag{Type: String, Name: flagkey.MqtMQType, Usage: "Message queue type, e.g. nats-streaming, azure-storage-queue, kafka", DefaultValue: "nats-streaming"} - MqtTopic = Flag{Type: String, Name: flagkey.MqtTopic, Usage: "Message queue Topic the trigger listens on"} - MqtRespTopic = Flag{Type: String, Name: flagkey.MqtRespTopic, Usage: "Topic that the function response is sent on (response discarded if unspecified)"} - MqtErrorTopic = Flag{Type: String, Name: flagkey.MqtErrorTopic, Usage: "Topic that the function error messages are sent to (errors discarded if unspecified"} - MqtMaxRetries = Flag{Type: Int, Name: flagkey.MqtMaxRetries, Usage: "Maximum number of times the function will be retried upon failure", DefaultValue: 0} - MqtMsgContentType = Flag{Type: String, Name: flagkey.MqtMsgContentType, Short: "c", Usage: "Content type of messages that publish to the topic", DefaultValue: "application/json"} + MqtName = Flag{Type: String, Name: flagkey.MqtName, Usage: "Message queue trigger name"} + MqtFnName = Flag{Type: String, Name: flagkey.MqtFnName, Usage: "Function name"} + MqtMQType = Flag{Type: String, Name: flagkey.MqtMQType, Usage: "Message queue type, e.g. nats-streaming, azure-storage-queue, kafka", DefaultValue: "nats-streaming"} + MqtTopic = Flag{Type: String, Name: flagkey.MqtTopic, Usage: "Message queue Topic the trigger listens on"} + MqtRespTopic = Flag{Type: String, Name: flagkey.MqtRespTopic, Usage: "Topic that the function response is sent on (response discarded if unspecified)"} + MqtErrorTopic = Flag{Type: String, Name: flagkey.MqtErrorTopic, Usage: "Topic that the function error messages are sent to (errors discarded if unspecified"} + MqtMaxRetries = Flag{Type: Int, Name: flagkey.MqtMaxRetries, Usage: "Maximum number of times the function will be retried upon failure", DefaultValue: 0} + MqtMsgContentType = Flag{Type: String, Name: flagkey.MqtMsgContentType, Short: "c", Usage: "Content type of messages that publish to the topic", DefaultValue: "application/json"} + MqtPollingInterval = Flag{Type: Int, Name: flagkey.MqtPollingInterval, Usage: "Interval to check the message source for up/down scaling operation of consumers", DefaultValue: 30} + MqtCooldownPeriod = Flag{Type: Int, Name: flagkey.MqtCooldownPeriod, Usage: "The period to wait after the last trigger reported active before scaling the consumer back to 0", DefaultValue: 300} + MqtMinReplicaCount = Flag{Type: Int, Name: flagkey.MqtMinReplicaCount, Usage: "Minimum number of replicas of consumers to scale down to", DefaultValue: 0} + MqtMaxReplicaCount = Flag{Type: Int, Name: flagkey.MqtMaxReplicaCount, Usage: "Maximum number of replicas of consumers to scale up to", DefaultValue: 100} + MqtMetadata = Flag{Type: StringSlice, Name: flagkey.MqtMetadata, Usage: "Metadata needed for connecting to source system in format: --metadata key1=value1 --metadata key2=value2"} + MqtSecret = Flag{Type: String, Name: flagkey.MqtSecret, Usage: "Name of secret object", DefaultValue: ""} + MqtKind = Flag{Type: String, Name: flagkey.MqtKind, Usage: "Kind of Message Queue Trigger, e.g. fission, keda", DefaultValue: "fission"} EnvName = Flag{Type: String, Name: flagkey.EnvName, Usage: "Environment name"} EnvPoolsize = Flag{Type: Int, Name: flagkey.EnvPoolsize, Usage: "Size of the pool", DefaultValue: 3} diff --git a/pkg/fission-cli/flag/key/key.go b/pkg/fission-cli/flag/key/key.go index ec3d8d07..0f9d82cc 100644 --- a/pkg/fission-cli/flag/key/key.go +++ b/pkg/fission-cli/flag/key/key.go @@ -81,14 +81,21 @@ const ( TtFnName = "function" TtRound = "round" - MqtName = resourceName - MqtFnName = "function" - MqtMQType = "mqtype" - MqtTopic = "topic" - MqtRespTopic = "resptopic" - MqtErrorTopic = "errortopic" - MqtMaxRetries = "maxretries" - MqtMsgContentType = "contenttype" + MqtName = resourceName + MqtFnName = "function" + MqtMQType = "mqtype" + MqtTopic = "topic" + MqtRespTopic = "resptopic" + MqtErrorTopic = "errortopic" + MqtMaxRetries = "maxretries" + MqtMsgContentType = "contenttype" + MqtPollingInterval = "pollinginterval" + MqtCooldownPeriod = "cooldownperiod" + MqtMinReplicaCount = "minreplicacount" + MqtMaxReplicaCount = "maxreplicacount" + MqtMetadata = "metadata" + MqtSecret = "secret" + MqtKind = "mqtkind" EnvName = resourceName EnvPoolsize = "poolsize" diff --git a/pkg/fission-cli/util/util.go b/pkg/fission-cli/util/util.go index 3c329eee..5bd5080a 100644 --- a/pkg/fission-cli/util/util.go +++ b/pkg/fission-cli/util/util.go @@ -315,3 +315,18 @@ func GetSpecDir(input cli.Input) string { } return specDir } + +// UpdateMapFromStringSlice parses key, val from "key=val" string array and updates passed map +func UpdateMapFromStringSlice(dataMap *map[string]string, params []string) bool { + updated := false + for _, m := range params { + keyValue := strings.SplitN(m, "=", 2) + if len(keyValue) == 2 { + key := keyValue[0] + value := keyValue[1] + (*dataMap)[key] = value + updated = true + } + } + return updated +} diff --git a/pkg/mqtrigger/kafka/Dockerfile b/pkg/mqtrigger/kafka/Dockerfile new file mode 100644 index 00000000..209f4f54 --- /dev/null +++ b/pkg/mqtrigger/kafka/Dockerfile @@ -0,0 +1,19 @@ +FROM golang:1.12-alpine as builder + +RUN apk add bash ca-certificates git gcc g++ libc-dev + +ARG GOPKG=github.com/fission/fission + +ENV GO111MODULE=on + +WORKDIR /go/src/${GOPKG} +COPY ./ ./ + +WORKDIR /go/src/${GOPKG}/pkg/mqtrigger/kafka +RUN go build -a -o /go/bin/main + +FROM alpine:3.12 as base +RUN apk add --update ca-certificates +COPY --from=builder /go/bin/main / + +ENTRYPOINT ["/main"] \ No newline at end of file diff --git a/pkg/mqtrigger/kafka/main.go b/pkg/mqtrigger/kafka/main.go new file mode 100644 index 00000000..fe1b5dcd --- /dev/null +++ b/pkg/mqtrigger/kafka/main.go @@ -0,0 +1,453 @@ +package main + +import ( + "context" + "crypto/sha256" + "crypto/sha512" + "crypto/tls" + "crypto/x509" + "fmt" + "hash" + "io/ioutil" + "log" + "net/http" + "os" + "os/signal" + "strings" + "sync" + "syscall" + "time" + + "github.com/Shopify/sarama" + "github.com/pkg/errors" + "github.com/xdg/scram" + "go.uber.org/zap" + + "github.com/fission/fission/pkg/mqtrigger/util" +) + +type kafkaMetadata struct { + bootstrapServers []string + consumerGroup string + + // auth + authMode kafkaAuthMode + username string + password string + + // ssl + cert string + key string + ca string +} + +type kafkaAuthMode string + +const ( + kafkaAuthModeNone kafkaAuthMode = "none" + kafkaAuthModeSaslPlaintext kafkaAuthMode = "sasl_plaintext" + kafkaAuthModeSaslScramSha256 kafkaAuthMode = "sasl_scram_sha256" + kafkaAuthModeSaslScramSha512 kafkaAuthMode = "sasl_scram_sha512" + kafkaAuthModeSaslSSL kafkaAuthMode = "sasl_ssl" + kafkaAuthModeSaslSSLPlain kafkaAuthMode = "sasl_ssl_plain" +) + +var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() } +var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() } + +type XDGSCRAMClient struct { + *scram.Client + *scram.ClientConversation + scram.HashGeneratorFcn +} + +func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) { + x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID) + if err != nil { + return err + } + x.ClientConversation = x.Client.NewConversation() + return nil +} + +func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) { + response, err = x.ClientConversation.Step(challenge) + return +} + +func (x *XDGSCRAMClient) Done() bool { + return x.ClientConversation.Done() +} + +func parseKafkaMetadata(logger *zap.Logger) (kafkaMetadata, error) { + meta := kafkaMetadata{} + + // brokerList marked as deprecated, bootstrapServers is the new one to use + if os.Getenv("BROKER_LIST") != "" && os.Getenv("BOOTSTRAP_SERVERS") != "" { + return meta, errors.New("cannot specify both bootstrapServers and brokerList (deprecated)") + } + if os.Getenv("BROKER_LIST") == "" && os.Getenv("BOOTSTRAP_SERVERS") == "" { + return meta, errors.New("no bootstrapServers or brokerList (deprecated) given") + } + if os.Getenv("BOOTSTRAP_SERVERS") != "" { + meta.bootstrapServers = strings.Split(os.Getenv("BOOTSTRAP_SERVERS"), ",") + } + if os.Getenv("BROKER_LIST") != "" { + logger.Info("WARNING: usage of brokerList is deprecated. use bootstrapServers instead.") + meta.bootstrapServers = strings.Split(os.Getenv("BROKER_LIST"), ",") + } + if os.Getenv("CONSUMER_GROUP") == "" { + return meta, errors.New("No consumerGroup given") + } + meta.consumerGroup = os.Getenv("CONSUMER_GROUP") + + meta.authMode = kafkaAuthModeNone + mode := kafkaAuthMode(strings.TrimSpace((os.Getenv("AUTH_MODE")))) + if mode == "" { + mode = kafkaAuthModeNone + } + + if mode != kafkaAuthModeNone && mode != kafkaAuthModeSaslPlaintext && mode != kafkaAuthModeSaslSSL && mode != kafkaAuthModeSaslSSLPlain && mode != kafkaAuthModeSaslScramSha256 && mode != kafkaAuthModeSaslScramSha512 { + return meta, fmt.Errorf("err auth mode %s given", mode) + } + + meta.authMode = mode + + if meta.authMode != kafkaAuthModeNone && meta.authMode != kafkaAuthModeSaslSSL { + if os.Getenv("USERNAME") == "" { + return meta, errors.New("no username given") + } + meta.username = strings.TrimSpace(os.Getenv("USERNAME")) + + if os.Getenv("PASSWORD") == "" { + return meta, errors.New("no password given") + } + meta.password = strings.TrimSpace(os.Getenv("PASSWORD")) + } + + if meta.authMode == kafkaAuthModeSaslSSL { + if os.Getenv("CA") == "" { + return meta, errors.New("no ca given") + } + meta.ca = os.Getenv("CA") + + if os.Getenv("CERT") == "" { + return meta, errors.New("no cert given") + } + meta.cert = os.Getenv("CERT") + + if os.Getenv("KEY") == "" { + return meta, errors.New("no key given") + } + meta.key = os.Getenv("KEY") + } + + return meta, nil +} + +func getConfig(metadata kafkaMetadata) (*sarama.Config, error) { + config := sarama.NewConfig() + config.Version = sarama.V1_0_0_0 + + if ok := metadata.authMode == kafkaAuthModeSaslPlaintext || metadata.authMode == kafkaAuthModeSaslSSLPlain || metadata.authMode == kafkaAuthModeSaslScramSha256 || metadata.authMode == kafkaAuthModeSaslScramSha512; ok { + config.Net.SASL.Enable = true + config.Net.SASL.User = metadata.username + config.Net.SASL.Password = metadata.password + } + + if metadata.authMode == kafkaAuthModeSaslSSLPlain { + config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypePlaintext) + + tlsConfig := &tls.Config{ + InsecureSkipVerify: true, + ClientAuth: 0, + } + + config.Net.TLS.Enable = true + config.Net.TLS.Config = tlsConfig + config.Net.DialTimeout = 10 * time.Second + } + + if metadata.authMode == kafkaAuthModeSaslSSL { + cert, err := tls.X509KeyPair([]byte(metadata.cert), []byte(metadata.key)) + if err != nil { + return nil, fmt.Errorf("error parse X509KeyPair: %s", err) + } + + caCertPool := x509.NewCertPool() + caCertPool.AppendCertsFromPEM([]byte(metadata.ca)) + + tlsConfig := &tls.Config{ + Certificates: []tls.Certificate{cert}, + RootCAs: caCertPool, + } + + config.Net.TLS.Enable = true + config.Net.TLS.Config = tlsConfig + } + + if metadata.authMode == kafkaAuthModeSaslScramSha256 { + config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA256} } + config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA256) + } + + if metadata.authMode == kafkaAuthModeSaslScramSha512 { + config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} } + config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512) + } + + if metadata.authMode == kafkaAuthModeSaslPlaintext { + config.Net.SASL.Mechanism = sarama.SASLTypePlaintext + config.Net.TLS.Enable = true + } + return config, nil +} + +// Connector represents a Sarama consumer group consumer +type Connector struct { + ready chan bool + logger *zap.Logger + producer sarama.SyncProducer + fissionTriggerFields util.FissionMetadata +} + +// Setup is run at the beginning of a new session, before ConsumeClaim +func (connector *Connector) Setup(sarama.ConsumerGroupSession) error { + close(connector.ready) + return nil +} + +// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited +func (connector *Connector) Cleanup(sarama.ConsumerGroupSession) error { + return nil +} + +// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages() +func (connector *Connector) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { + for message := range claim.Messages() { + connector.logger.Info(fmt.Sprintf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)) + success := handleFissionFunction(message, connector.fissionTriggerFields, connector.producer, connector.logger) + if success { + session.MarkMessage(message, "") + } + } + return nil +} + +func getProducer(metadata kafkaMetadata) (sarama.SyncProducer, error) { + config, err := getConfig(metadata) + if err != nil { + return nil, err + } + + config.Producer.RequiredAcks = sarama.WaitForAll + config.Producer.Retry.Max = 10 + config.Producer.Return.Successes = true + producer, err := sarama.NewSyncProducer(metadata.bootstrapServers, config) + if err != nil { + return nil, err + } + return producer, nil +} + +func handleFissionFunction(msg *sarama.ConsumerMessage, triggerFields util.FissionMetadata, producer sarama.SyncProducer, logger *zap.Logger) bool { + var value string = string(msg.Value[:]) + // Generate the Headers + fissionHeaders := map[string]string{ + "X-Fission-MQTrigger-Topic": triggerFields.Topic, + "X-Fission-MQTrigger-RespTopic": triggerFields.ResponseTopic, + "X-Fission-MQTrigger-ErrorTopic": triggerFields.ErrorTopic, + "Content-Type": triggerFields.ContentType, + } + + // Create request + req, err := http.NewRequest("POST", triggerFields.FunctionURL, strings.NewReader(value)) + if err != nil { + logger.Error("failed to create HTTP request to invoke function", + zap.Error(err), + zap.String("function_url", triggerFields.FunctionURL)) + return false + } + + // Set the headers came from Kafka record + // Using Header.Add() as msg.Headers may have keys with more than one value + for _, h := range msg.Headers { + req.Header.Add(string(h.Key), string(h.Value)) + } + + for k, v := range fissionHeaders { + req.Header.Set(k, v) + } + + // Make the request + var resp *http.Response + for attempt := 0; attempt <= triggerFields.MaxRetries; attempt++ { + // Make the request + resp, err = http.DefaultClient.Do(req) + if err != nil { + logger.Error("sending function invocation request failed", + zap.Error(err), + zap.String("function_url", triggerFields.FunctionURL), + zap.String("trigger", triggerFields.TriggerName)) + continue + } + if resp == nil { + continue + } + if err == nil && resp.StatusCode == http.StatusOK { + // Success, quit retrying + break + } + } + + if resp == nil { + logger.Warn("every function invocation retry failed; final retry gave empty response", + zap.String("function_url", triggerFields.FunctionURL), + zap.String("trigger", triggerFields.TriggerName)) + return false + } + defer resp.Body.Close() + body, err := ioutil.ReadAll(resp.Body) + + logger.Debug("got response from function invocation", + zap.String("function_url", triggerFields.FunctionURL), + zap.String("trigger", triggerFields.TriggerName), + zap.String("body", string(body))) + + if err != nil { + errorHandler(logger, triggerFields, producer, + errors.Wrapf(err, "request body error: %v", string(body))) + return false + } + if resp.StatusCode != 200 { + errorHandler(logger, triggerFields, producer, + fmt.Errorf("request returned failure: %v", resp.StatusCode)) + return false + } + + if len(triggerFields.ResponseTopic) > 0 { + // Generate Kafka record headers + var kafkaRecordHeaders []sarama.RecordHeader + + for k, v := range resp.Header { + // One key may have multiple values + for _, v := range v { + kafkaRecordHeaders = append(kafkaRecordHeaders, sarama.RecordHeader{Key: []byte(k), Value: []byte(v)}) + } + } + + _, _, err = producer.SendMessage(&sarama.ProducerMessage{ + Topic: triggerFields.ResponseTopic, + Value: sarama.StringEncoder(body), + Headers: kafkaRecordHeaders, + }) + if err != nil { + logger.Warn("failed to publish response body from function invocation to topic", + zap.Error(err), + zap.String("topic", triggerFields.Topic), + zap.String("function_url", triggerFields.FunctionURL)) + return false + } + } + + return true +} + +func errorHandler(logger *zap.Logger, triggerFields util.FissionMetadata, producer sarama.SyncProducer, err error) { + if len(triggerFields.ErrorTopic) > 0 { + _, _, e := producer.SendMessage(&sarama.ProducerMessage{ + Topic: triggerFields.ErrorTopic, + Value: sarama.StringEncoder(err.Error()), + }) + if e != nil { + logger.Error("failed to publish message to error topic", + zap.Error(e), + zap.String("trigger", triggerFields.TriggerName), + zap.String("message", err.Error()), + zap.String("topic", triggerFields.Topic)) + } + } else { + logger.Error("message received to publish to error topic, but no error topic was set", + zap.String("message", err.Error()), zap.String("trigger", triggerFields.TriggerName), zap.String("function_url", triggerFields.FunctionURL)) + } +} + +func main() { + logger, err := zap.NewProduction() + if err != nil { + log.Fatalf("can't initialize zap logger: %v", err) + } + defer logger.Sync() + + metadata, err := parseKafkaMetadata(logger) + if err != nil { + logger.Error("Failed to fetch kafka metadata", zap.Error(err)) + return + } + + triggerFields, err := util.ParseFissionMetadata() + if err != nil { + logger.Error("Failed to parse fission trigger fields", zap.Error(err)) + return + } + + config, err := getConfig(metadata) + if err != nil { + logger.Error("Failed to create kafka config", zap.Error(err)) + return + } + + producer, err := getProducer(metadata) + if err != nil { + logger.Error("Failed to create kafka producer", zap.Error(err)) + return + } + defer producer.Close() + + connector := Connector{ + ready: make(chan bool), + logger: logger, + producer: producer, + fissionTriggerFields: triggerFields, + } + + ctx, cancel := context.WithCancel(context.Background()) + client, err := sarama.NewConsumerGroup(metadata.bootstrapServers, metadata.consumerGroup, config) + if err != nil { + logger.Error("Error creating consumer group client", zap.Error(err)) + return + } + + wg := &sync.WaitGroup{} + wg.Add(1) + + go func() { + defer wg.Done() + for { + if err := client.Consume(ctx, []string{triggerFields.Topic}, &connector); err != nil { + logger.Error("Error from consumer", zap.Error(err)) + } + // check if context was cancelled, signaling that the consumer should stop + if ctx.Err() != nil { + return + } + connector.ready = make(chan bool) + } + }() + + <-connector.ready // Await till the consumer has been set up + logger.Info("Sarama consumer up and running!...") + sigterm := make(chan os.Signal, 1) + signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM) + select { + case <-ctx.Done(): + logger.Info("terminating: context cancelled") + case <-sigterm: + logger.Info("terminating: via signal") + } + cancel() + wg.Wait() + if err = client.Close(); err != nil { + logger.Error("Error closing client", zap.Error(err)) + } +} diff --git a/pkg/mqtrigger/scalermanager.go b/pkg/mqtrigger/scalermanager.go new file mode 100644 index 00000000..3c104c35 --- /dev/null +++ b/pkg/mqtrigger/scalermanager.go @@ -0,0 +1,540 @@ +package mqtrigger + +import ( + "context" + "fmt" + "os" + "regexp" + "strconv" + "strings" + "time" + + "github.com/pkg/errors" + "go.uber.org/zap" + appsv1 "k8s.io/api/apps/v1" + apiv1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/kubernetes" + k8sCache "k8s.io/client-go/tools/cache" + + fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/utils" +) + +var ( + scaledObjectGVR = schema.GroupVersionResource{ + Group: "keda.k8s.io", + Version: "v1alpha1", + Resource: "scaledobjects", + } + authTriggerGVR = schema.GroupVersionResource{ + Group: "keda.k8s.io", + Version: "v1alpha1", + Resource: "triggerauthentications", + } + matchFirstCap = regexp.MustCompile("(.)([A-Z][a-z]+)") + matchAllCap = regexp.MustCompile("([a-z0-9])([A-Z])") +) + +func getScaledObjectClient(namespace string) (dynamic.ResourceInterface, error) { + dynamicClient, err := crd.GetDynamicClient() + if err != nil { + return nil, err + } + return dynamicClient.Resource(scaledObjectGVR).Namespace(namespace), nil +} + +func getAuthTriggerClient(namespace string) (dynamic.ResourceInterface, error) { + dynamicClient, err := crd.GetDynamicClient() + if err != nil { + return nil, err + } + return dynamicClient.Resource(scaledObjectGVR).Namespace(namespace), nil +} + +// StartScalerManager watches for changes in MessageQueueTrigger and, +// Based on changes, it Creates, Updates and Deletes Objects of Kind ScaledObjects, AuthenticationTriggers and Deployments +func StartScalerManager(logger *zap.Logger, routerURL string) error { + fissionClient, kubeClient, _, err := crd.MakeFissionClient() + if err != nil { + return err + } + err = fissionClient.WaitForCRDs() + if err != nil { + return errors.Wrap(err, "error waiting for CRDs") + } + crdClient := fissionClient.CoreV1().RESTClient() + resyncPeriod := 30 * time.Second + listWatch := k8sCache.NewListWatchFromClient(crdClient, "messagequeuetriggers", metav1.NamespaceAll, fields.Everything()) + _, controller := k8sCache.NewInformer(listWatch, &fv1.MessageQueueTrigger{}, resyncPeriod, k8sCache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + go func() { + mqt := obj.(*fv1.MessageQueueTrigger) + if mqt.Spec.MqtKind == "fission" { + return + } + logger.Debug("Create deployment for Scaler Object", zap.Any("mqt", mqt.ObjectMeta), zap.Any("mqt.Spec", mqt.Spec)) + + authenticationRef := "" + if len(mqt.Spec.Secret) > 0 { + authenticationRef = fmt.Sprintf("%s-auth-trigger", mqt.ObjectMeta.Name) + err = createAuthTrigger(mqt, authenticationRef, kubeClient) + if err != nil { + logger.Error("Failed to create Authentication Trigger", zap.Error(err)) + return + } + } + + if err = createDeployment(mqt, routerURL, kubeClient); err != nil { + logger.Error("Failed to create Deployment", zap.Error(err)) + if len(authenticationRef) > 0 { + err = deleteAuthTrigger(authenticationRef, mqt.ObjectMeta.Namespace) + if err != nil { + logger.Error("Failed to delete Authentication Trigger", zap.Error(err)) + } + } + return + } + + if err = createScaledObject(mqt, authenticationRef); err != nil { + logger.Error("Failed to create ScaledObject", zap.Error(err)) + if len(authenticationRef) > 0 { + if err = deleteAuthTrigger(authenticationRef, mqt.ObjectMeta.Namespace); err != nil { + logger.Error("Failed to delete Authentication Trigger", zap.Error(err)) + } + } + if err = deleteDeployment(mqt.ObjectMeta.Name, kubeClient); err != nil { + logger.Error("Failed to delete Deployment", zap.Error(err)) + } + } + }() + }, + UpdateFunc: func(obj interface{}, newObj interface{}) { + go func() { + mqt := obj.(*fv1.MessageQueueTrigger) + newMqt := newObj.(*fv1.MessageQueueTrigger) + updated := checkAndUpdateTriggerFields(mqt, newMqt) + if mqt.Spec.MqtKind == "fission" { + return + } + if !updated { + logger.Warn(fmt.Sprintf("%s remains unchanged. No changes found in trigger fields", mqt.ObjectMeta.Name)) + return + } + + authenticationRef := "" + if len(newMqt.Spec.Secret) > 0 && newMqt.Spec.Secret != mqt.Spec.Secret { + authenticationRef = fmt.Sprintf("%s-auth-trigger", mqt.ObjectMeta.Name) + if err = updateAuthTrigger(mqt, authenticationRef, kubeClient); err != nil { + logger.Error("Failed to update Authentication Trigger", zap.Error(err)) + return + } + } + + if err = updateDeployment(mqt, routerURL, kubeClient); err != nil { + logger.Error("Failed to Update Deployment", zap.Error(err)) + return + } + + if err = updateScaledObject(mqt, authenticationRef); err != nil { + logger.Error("Failed to Update ScaledObject", zap.Error(err)) + return + } + }() + }, + }) + controller.Run(context.Background().Done()) + return nil +} + +func toEnvVar(str string) string { + envVar := matchFirstCap.ReplaceAllString(str, "${1}_${2}") + envVar = matchAllCap.ReplaceAllString(envVar, "${1}_${2}") + return strings.ToUpper(envVar) +} + +func getEnvVarlist(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient kubernetes.Interface) ([]apiv1.EnvVar, error) { + url := routerURL + "/" + strings.TrimPrefix(utils.UrlForFunction(mqt.Spec.FunctionReference.Name, mqt.ObjectMeta.Namespace), "/") + envVars := []apiv1.EnvVar{ + { + Name: "TOPIC", + Value: mqt.Spec.Topic, + }, + { + Name: "FUNCTION_URL", + Value: url, + }, + { + Name: "ERROR_TOPIC", + Value: mqt.Spec.ErrorTopic, + }, + { + Name: "RESPONSE_TOPIC", + Value: mqt.Spec.ResponseTopic, + }, + { + Name: "TRIGGER_NAME", + Value: mqt.ObjectMeta.Name, + }, + { + Name: "MAX_RETRIES", + Value: strconv.Itoa(mqt.Spec.MaxRetries), + }, + { + Name: "CONTENT_TYPE", + Value: mqt.Spec.ContentType, + }, + } + // Metadata Fields + for key, value := range mqt.Spec.Metadata { + envVars = append(envVars, apiv1.EnvVar{ + Name: toEnvVar(key), + Value: value, + }) + } + + // Add Auth Fields + secretName := mqt.Spec.Secret + if len(secretName) > 0 { + secret, err := kubeClient.CoreV1().Secrets(apiv1.NamespaceDefault).Get(secretName, metav1.GetOptions{}) + if err != nil { + return nil, err + } + for key, value := range secret.Data { + envVars = append(envVars, apiv1.EnvVar{ + Name: toEnvVar(key), + Value: string(value), + }) + } + } + return envVars, nil +} + +func checkAndUpdateTriggerFields(mqt, newMqt *fv1.MessageQueueTrigger) bool { + updated := false + if len(newMqt.Spec.Topic) > 0 && newMqt.Spec.Topic != mqt.Spec.Topic { + mqt.Spec.Topic = newMqt.Spec.Topic + updated = true + } + if len(newMqt.Spec.ResponseTopic) > 0 && newMqt.Spec.ResponseTopic != mqt.Spec.ResponseTopic { + mqt.Spec.ResponseTopic = newMqt.Spec.ResponseTopic + updated = true + } + if len(newMqt.Spec.ErrorTopic) > 0 && newMqt.Spec.ErrorTopic != mqt.Spec.ErrorTopic { + mqt.Spec.ErrorTopic = newMqt.Spec.ErrorTopic + updated = true + } + if newMqt.Spec.MaxRetries >= 0 && newMqt.Spec.MaxRetries != mqt.Spec.MaxRetries { + mqt.Spec.MaxRetries = newMqt.Spec.MaxRetries + updated = true + } + if len(newMqt.Spec.FunctionReference.Name) > 0 && newMqt.Spec.FunctionReference.Name != mqt.Spec.FunctionReference.Name { + mqt.Spec.FunctionReference.Name = newMqt.Spec.FunctionReference.Name + updated = true + } + if len(newMqt.Spec.ContentType) > 0 && newMqt.Spec.ContentType != mqt.Spec.ContentType { + mqt.Spec.ContentType = newMqt.Spec.ContentType + updated = true + } + if *newMqt.Spec.PollingInterval >= 0 && *newMqt.Spec.PollingInterval != *mqt.Spec.PollingInterval { + mqt.Spec.PollingInterval = newMqt.Spec.PollingInterval + updated = true + } + if *newMqt.Spec.CooldownPeriod >= 0 && *newMqt.Spec.CooldownPeriod != *mqt.Spec.CooldownPeriod { + mqt.Spec.CooldownPeriod = newMqt.Spec.CooldownPeriod + updated = true + } + if *newMqt.Spec.MinReplicaCount >= 0 && *newMqt.Spec.MinReplicaCount != *mqt.Spec.MinReplicaCount { + mqt.Spec.MinReplicaCount = newMqt.Spec.MinReplicaCount + updated = true + } + if *newMqt.Spec.MaxReplicaCount >= 0 && *newMqt.Spec.MaxReplicaCount != *mqt.Spec.MaxReplicaCount { + mqt.Spec.MaxReplicaCount = newMqt.Spec.MaxReplicaCount + updated = true + } + if len(newMqt.Spec.FunctionReference.Name) > 0 && newMqt.Spec.FunctionReference.Name != mqt.Spec.FunctionReference.Name { + newMqt.Spec.FunctionReference.Name = mqt.Spec.FunctionReference.Name + updated = true + } + + for key, value := range newMqt.Spec.Metadata { + if val, ok := mqt.Spec.Metadata[key]; ok && val != value { + mqt.Spec.Metadata[key] = value + updated = true + } + } + + if len(newMqt.Spec.Secret) > 0 && newMqt.Spec.Secret != mqt.Spec.Secret { + mqt.Spec.Secret = newMqt.Spec.Secret + updated = true + } + + if newMqt.Spec.MqtKind != mqt.Spec.MqtKind { + mqt.Spec.MqtKind = newMqt.Spec.MqtKind + updated = true + } + + return updated +} + +func getResourceVersion(scaledObjectName string, kedaClient dynamic.ResourceInterface) (version string, err error) { + scaledObject, err := kedaClient.Get(scaledObjectName, metav1.GetOptions{}) + if err != nil { + return "", err + } + return scaledObject.GetResourceVersion(), nil +} + +func getAuthTriggerSpec(mqt *fv1.MessageQueueTrigger, authenticationRef string, kubeClient kubernetes.Interface) (*unstructured.Unstructured, error) { + secret, err := kubeClient.CoreV1().Secrets(apiv1.NamespaceDefault).Get(mqt.Spec.Secret, metav1.GetOptions{}) + if err != nil { + return nil, err + } + var secretTargetRefFields []interface{} + for secretField := range secret.Data { + secretTargetRefFields = append(secretTargetRefFields, map[string]interface{}{ + "name": mqt.Spec.Secret, + "parameter": secretField, + "key": secretField, + }) + } + authTriggerObj := &unstructured.Unstructured{ + Object: map[string]interface{}{ + "kind": "TriggerAuthentication", + "apiVersion": "keda.k8s.io/v1alpha1", + "metadata": map[string]interface{}{ + "name": authenticationRef, + "namespace": mqt.ObjectMeta.Namespace, + "ownerReferences": []interface{}{ + map[string]interface{}{ + "kind": "MessageQueueTrigger", + "apiVersion": "fission.io/v1", + "name": mqt.ObjectMeta.Name, + "uid": mqt.ObjectMeta.UID, + "blockOwnerDeletion": true, + }, + }, + }, + "spec": map[string]interface{}{ + "secretTargetRef": secretTargetRefFields, + }, + }, + } + return authTriggerObj, nil +} + +func createAuthTrigger(mqt *fv1.MessageQueueTrigger, authenticationRef string, kubeClient *kubernetes.Clientset) error { + authTriggerObj, err := getAuthTriggerSpec(mqt, authenticationRef, kubeClient) + if err != nil { + return err + } + authTriggerClient, err := getAuthTriggerClient(mqt.ObjectMeta.Namespace) + if err != nil { + return err + } + _, err = authTriggerClient.Create(authTriggerObj, metav1.CreateOptions{}) + if err != nil { + return err + } + return nil +} + +func updateAuthTrigger(mqt *fv1.MessageQueueTrigger, authenticationRef string, kubeClient *kubernetes.Clientset) error { + authTriggerClient, err := getAuthTriggerClient(mqt.ObjectMeta.Namespace) + if err != nil { + return err + } + oldAuthTriggerObj, err := authTriggerClient.Get(authenticationRef, metav1.GetOptions{}) + if err != nil { + return err + } + resourceVersion := oldAuthTriggerObj.GetResourceVersion() + + authTriggerObj, err := getAuthTriggerSpec(mqt, authenticationRef, kubeClient) + if err != nil { + return err + } + authTriggerObj.SetResourceVersion(resourceVersion) + _, err = authTriggerClient.Update(authTriggerObj, metav1.UpdateOptions{}) + if err != nil { + return err + } + return nil +} + +func deleteAuthTrigger(name, namespace string) error { + authTriggerClient, err := getAuthTriggerClient(namespace) + if err != nil { + return err + } + err = authTriggerClient.Delete(name, &metav1.DeleteOptions{}) + if err != nil { + return err + } + return nil +} + +func getDeploymentSpec(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient *kubernetes.Clientset) (*appsv1.Deployment, error) { + envVars, err := getEnvVarlist(mqt, routerURL, kubeClient) + if err != nil { + return nil, err + } + imageName := fmt.Sprintf("%s_image", string(mqt.Spec.MessageQueueType)) + image := os.Getenv(strings.ToUpper(imageName)) + + blockOwnerDeletion := true + return &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: mqt.ObjectMeta.Name, + Labels: map[string]string{ + "app": mqt.ObjectMeta.Name, + }, + OwnerReferences: []metav1.OwnerReference{ + { + Kind: "MessageQueueTrigger", + APIVersion: "fission.io/v1", + Name: mqt.ObjectMeta.Name, + UID: mqt.ObjectMeta.UID, + BlockOwnerDeletion: &blockOwnerDeletion, + }, + }, + }, + Spec: appsv1.DeploymentSpec{ + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + "app": mqt.ObjectMeta.Name, + }, + }, + Template: apiv1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + "app": mqt.ObjectMeta.Name, + }, + }, + Spec: apiv1.PodSpec{ + Containers: []apiv1.Container{ + { + Name: mqt.ObjectMeta.Name, + Image: image, + ImagePullPolicy: "Always", + Env: envVars, + }, + }, + }, + }, + }, + }, nil +} + +func createDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient *kubernetes.Clientset) error { + deployment, err := getDeploymentSpec(mqt, routerURL, kubeClient) + if err != nil { + return err + } + _, err = kubeClient.AppsV1().Deployments(apiv1.NamespaceDefault).Create(deployment) + if err != nil { + return err + } + return nil +} + +func updateDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient *kubernetes.Clientset) error { + deployment, err := getDeploymentSpec(mqt, routerURL, kubeClient) + if err != nil { + return err + } + _, err = kubeClient.AppsV1().Deployments(apiv1.NamespaceDefault).Update(deployment) + if err != nil { + return err + } + return nil +} + +func deleteDeployment(name string, kubeClient *kubernetes.Clientset) error { + deletePolicy := metav1.DeletePropagationForeground + if err := kubeClient.AppsV1().Deployments(apiv1.NamespaceDefault).Delete(name, &metav1.DeleteOptions{ + PropagationPolicy: &deletePolicy, + }); err != nil { + return err + } + return nil +} + +func getScaledObject(mqt *fv1.MessageQueueTrigger, authenticationRef string) *unstructured.Unstructured { + return &unstructured.Unstructured{ + Object: map[string]interface{}{ + "kind": "ScaledObject", + "apiVersion": "keda.k8s.io/v1alpha1", + "metadata": map[string]interface{}{ + "name": mqt.ObjectMeta.Name, + "namespace": mqt.ObjectMeta.Namespace, + "ownerReferences": []interface{}{ + map[string]interface{}{ + "kind": "MessageQueueTrigger", + "apiVersion": "fission.io/v1", + "name": mqt.ObjectMeta.Name, + "uid": mqt.ObjectMeta.UID, + "blockOwnerDeletion": true, + }, + }, + }, + "spec": map[string]interface{}{ + "cooldownPeriod": &mqt.Spec.CooldownPeriod, + "maxReplicaCount": &mqt.Spec.MaxReplicaCount, + "minReplicaCount": &mqt.Spec.MinReplicaCount, + "pollingInterval": &mqt.Spec.PollingInterval, + "scaleTargetRef": map[string]interface{}{ + "deploymentName": mqt.ObjectMeta.Name, + }, + "triggers": []interface{}{ + map[string]interface{}{ + "type": mqt.Spec.MessageQueueType, + "metadata": mqt.Spec.Metadata, + "authenticationRef": map[string]interface{}{ + "name": authenticationRef, + }, + }, + }, + }, + }, + } +} + +func createScaledObject(mqt *fv1.MessageQueueTrigger, authenticationRef string) error { + scaledObject := getScaledObject(mqt, authenticationRef) + kedaClient, err := getScaledObjectClient(mqt.ObjectMeta.Namespace) + if err != nil { + return err + } + _, err = kedaClient.Create(scaledObject, metav1.CreateOptions{}) + if err != nil { + return err + } + return nil +} + +func updateScaledObject(mqt *fv1.MessageQueueTrigger, authenticationRef string) error { + kedaClient, err := getScaledObjectClient(mqt.ObjectMeta.Namespace) + if err != nil { + return err + } + oldScaledObject, err := kedaClient.Get(mqt.ObjectMeta.Name, metav1.GetOptions{}) + if err != nil { + return err + } + resourceVersion := oldScaledObject.GetResourceVersion() + + scaledObject := getScaledObject(mqt, authenticationRef) + scaledObject.SetResourceVersion(resourceVersion) + + _, err = kedaClient.Update(scaledObject, metav1.UpdateOptions{}) + if err != nil { + return err + } + return nil +} diff --git a/pkg/mqtrigger/scalermanager_test.go b/pkg/mqtrigger/scalermanager_test.go new file mode 100644 index 00000000..592c070d --- /dev/null +++ b/pkg/mqtrigger/scalermanager_test.go @@ -0,0 +1,622 @@ +package mqtrigger + +import ( + "fmt" + "reflect" + "sort" + "testing" + + fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/stretchr/testify/assert" + apiv1 "k8s.io/api/core/v1" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/dynamic" + dynfake "k8s.io/client-go/dynamic/fake" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" +) + +func Test_toEnvVar(t *testing.T) { + type args struct { + str string + } + tests := []struct { + name string + args args + want string + }{ + {"Empty string", args{""}, ""}, + {"Single word", args{"fission"}, "FISSION"}, + {"CamelCase", args{"responseTopic"}, "RESPONSE_TOPIC"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := toEnvVar(tt.args.str); got != tt.want { + t.Errorf("toEnvVar() = %v, want %v", got, tt.want) + } + }) + } +} + +func Test_getEnvVarlist(t *testing.T) { + // Kafka Test with Valid Secret + pollingInterval := int32(30) + cooldownPeriod := int32(300) + minReplicaCount := int32(0) + maxReplicaCount := int32(100) + + mqt := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test", + Namespace: "default", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test", + }, + MessageQueueType: "kafka", + Topic: "topic", + ResponseTopic: "response-topic", + ErrorTopic: "error-topic", + MaxRetries: 4, + ContentType: "application/json", + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: map[string]string{ + "bootstrapServers": "my-cluster-kafka-brokers.my-kafka-project.svc:9092", + "consumerGroup": "my-group", + "topic": "topic", + }, + Secret: "test-kafka-secrets", + MqtKind: "keda", + }, + } + + data := map[string][]byte{ + "authMode": []byte("sasl_plaintext"), + "username": []byte("admin"), + "password": []byte("admin"), + "ca": []byte("test_ca"), + "cert": []byte("test_cert"), + "key": []byte("test_key"), + } + namespace := apiv1.NamespaceDefault + + routerURL := "http://router.fission/fission-function" + + secret := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-kafka-secrets", + Namespace: namespace, + }, + Data: data, + } + + kubeClient := fake.NewSimpleClientset() + _, err := kubeClient.CoreV1().Secrets(namespace).Create(secret) + if err != nil { + assert.Equal(t, nil, err) + } + + expectedEnvVars := []apiv1.EnvVar{ + { + Name: "TOPIC", + Value: mqt.Spec.Topic, + }, + { + Name: "FUNCTION_URL", + Value: "http://router.fission/fission-function/fission-function/test", + }, + { + Name: "ERROR_TOPIC", + Value: "error-topic", + }, + { + Name: "RESPONSE_TOPIC", + Value: "response-topic", + }, + { + Name: "TRIGGER_NAME", + Value: "Test", + }, + { + Name: "MAX_RETRIES", + Value: "4", + }, + { + Name: "CONTENT_TYPE", + Value: "application/json", + }, + { + Name: "BOOTSTRAP_SERVERS", + Value: "my-cluster-kafka-brokers.my-kafka-project.svc:9092", + }, + { + Name: "CONSUMER_GROUP", + Value: "my-group", + }, + { + Name: "TOPIC", + Value: "topic", + }, + { + Name: "KEY", + Value: "test_key", + }, + { + Name: "AUTH_MODE", + Value: "sasl_plaintext", + }, + { + Name: "USERNAME", + Value: "admin", + }, + { + Name: "PASSWORD", + Value: "admin", + }, + { + Name: "CA", + Value: "test_ca", + }, + { + Name: "CERT", + Value: "test_cert", + }, + } + + // Kafka Test with Invalid Secret Name + mqt2 := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test2", + Namespace: "default", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test2", + }, + MessageQueueType: "kafka", + Topic: "topic", + ResponseTopic: "response-topic", + ErrorTopic: "error-topic", + MaxRetries: 4, + ContentType: "application/json", + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: map[string]string{ + "bootstrapServers": "my-cluster-kafka-brokers.my-kafka-project.svc:9092", + "consumerGroup": "my-group", + "topic": "topic", + }, + Secret: "test-kafka-secrets-invalid", + MqtKind: "keda", + }, + } + + // Test Code + type args struct { + mqt *fv1.MessageQueueTrigger + routerURL string + kubeClient kubernetes.Interface + } + tests := []struct { + name string + args args + want []apiv1.EnvVar + wantErr bool + }{ + {"Test kafka example", args{mqt, routerURL, kubeClient}, expectedEnvVars, false}, + {"Test kafka invalid secret", args{mqt2, routerURL, kubeClient}, nil, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := getEnvVarlist(tt.args.mqt, tt.args.routerURL, tt.args.kubeClient) + sort.Slice(got, func(i, j int) bool { + return got[i].Name < got[j].Name + }) + sort.Slice(tt.want, func(i, j int) bool { + return tt.want[i].Name < tt.want[j].Name + }) + if (err != nil) != tt.wantErr { + t.Errorf("getEnvVarlist() error = %v, wantErr %v", err, tt.wantErr) + return + } + if !reflect.DeepEqual(got, tt.want) { + t.Errorf("getEnvVarlist() = %v, want %v", got, tt.want) + } + }) + } +} + +func Test_checkAndUpdateTriggerFields(t *testing.T) { + pollingInterval := int32(30) + cooldownPeriod := int32(300) + minReplicaCount := int32(0) + maxReplicaCount := int32(100) + + // Test 1 with difference + mqt := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test", + Namespace: "default", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test", + }, + MessageQueueType: "kafka", + Topic: "topic", + ResponseTopic: "response-topic", + ErrorTopic: "error-topic", + MaxRetries: 4, + ContentType: "application/json", + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: map[string]string{ + "bootstrapServers": "my-cluster-kafka-brokers.my-kafka-project.svc:9092", + "consumerGroup": "my-group", + "topic": "topic", + }, + Secret: "test-kafka-secrets", + MqtKind: "keda", + }, + } + newMqt1 := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test", + Namespace: "default", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test2", + }, + MessageQueueType: "kafka", + Topic: "my-topic", + ResponseTopic: "response-topic", + ErrorTopic: "error-topic", + MaxRetries: 2, + ContentType: "application/json", + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: map[string]string{ + "bootstrapServers": "my-cluster-kafka-brokers-2.my-kafka-project.svc:9092", + "consumerGroup": "my-group-2", + "topic": "my-topic", + }, + Secret: "new-test-kafka-secrets", + MqtKind: "keda", + }, + } + + // Test 2 with no difference + mqt2 := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test", + Namespace: "default", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test", + }, + MessageQueueType: "kafka", + Topic: "topic", + ResponseTopic: "response-topic", + ErrorTopic: "error-topic", + MaxRetries: 4, + ContentType: "application/json", + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: map[string]string{ + "bootstrapServers": "my-cluster-kafka-brokers.my-kafka-project.svc:9092", + "consumerGroup": "my-group", + "topic": "topic", + }, + Secret: "test-kafka-secrets", + MqtKind: "keda", + }, + } + newMqt2 := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test", + Namespace: "default", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test", + }, + MaxRetries: 4, + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + MqtKind: "keda", + }, + } + type args struct { + mqt *fv1.MessageQueueTrigger + newMqt *fv1.MessageQueueTrigger + } + tests := []struct { + name string + args args + want bool + }{ + {"With diff", args{mqt, newMqt1}, true}, + {"With no diff", args{mqt2, newMqt2}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := checkAndUpdateTriggerFields(tt.args.mqt, tt.args.newMqt); got != tt.want { + t.Errorf("checkAndUpdateTriggerFields() = %v, want %v", got, tt.want) + } + }) + } +} + +func newUnstructured(apiVersion, kind, namespace, name, resourceVersion string) *unstructured.Unstructured { + return &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": apiVersion, + "kind": kind, + "metadata": map[string]interface{}{ + "namespace": namespace, + "name": name, + "resourceVersion": resourceVersion, + }, + }, + } +} + +func Test_getResourceVersion(t *testing.T) { + scheme := runtime.NewScheme() + client := dynfake.NewSimpleDynamicClient(scheme, newUnstructured("keda.k8s.io/v1alpha1", "ScaledObject", "default", "test-1", "12345")) + dynamicResourceClient := client.Resource(schema.GroupVersionResource{ + Group: "keda.k8s.io", + Version: "v1alpha1", + Resource: "scaledobjects", + }) + type args struct { + scaledObjectName string + kedaClient dynamic.ResourceInterface + } + tests := []struct { + name string + args args + wantVersion string + wantErr bool + }{ + {"Valid Resource", args{"test-1", dynamicResourceClient}, "12345", false}, + {"Invalid Resource", args{"test-2", dynamicResourceClient}, "", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotVersion, err := getResourceVersion(tt.args.scaledObjectName, tt.args.kedaClient) + if (err != nil) != tt.wantErr { + t.Errorf("getResourceVersion() error = %v, wantErr %v", err, tt.wantErr) + return + } + if gotVersion != tt.wantVersion { + t.Errorf("getResourceVersion() = %v, want %v", gotVersion, tt.wantVersion) + } + }) + } +} + +func Test_getAuthTriggerSpec(t *testing.T) { + + // Valid - with Secret + pollingInterval := int32(30) + cooldownPeriod := int32(300) + minReplicaCount := int32(0) + maxReplicaCount := int32(200) + + mqt1 := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test", + Namespace: "default", + UID: "test123", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test", + }, + MessageQueueType: "kafka", + Topic: "topic", + ResponseTopic: "response-topic", + ErrorTopic: "error-topic", + MaxRetries: 4, + ContentType: "application/json", + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: map[string]string{ + "bootstrapServers": "my-cluster-kafka-brokers.my-kafka-project.svc:9092", + "consumerGroup": "my-group", + "topic": "topic", + }, + Secret: "test-kafka-secrets", + MqtKind: "keda", + }, + } + + data := map[string][]byte{ + "authMode": []byte("sasl_plaintext"), + "username": []byte("admin"), + "password": []byte("admin"), + "ca": []byte("test_ca"), + "cert": []byte("test_cert"), + "key": []byte("test_key"), + } + + namespace := apiv1.NamespaceDefault + secret := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-kafka-secrets", + Namespace: namespace, + }, + Data: data, + } + + kubeClient := fake.NewSimpleClientset() + _, err := kubeClient.CoreV1().Secrets(namespace).Create(secret) + if err != nil { + assert.Equal(t, nil, err) + } + + authenticationRef := fmt.Sprintf("%s-auth-trigger", mqt1.ObjectMeta.Name) + + expectedAuthTriggerObj := &unstructured.Unstructured{ + Object: map[string]interface{}{ + "kind": "TriggerAuthentication", + "apiVersion": "keda.k8s.io/v1alpha1", + "metadata": map[string]interface{}{ + "name": authenticationRef, + "namespace": mqt1.ObjectMeta.Namespace, + "ownerReferences": []interface{}{ + map[string]interface{}{ + "kind": "MessageQueueTrigger", + "apiVersion": "fission.io/v1", + "name": mqt1.ObjectMeta.Name, + "uid": mqt1.ObjectMeta.UID, + "blockOwnerDeletion": true, + }, + }, + }, + "spec": map[string]interface{}{ + "secretTargetRef": []interface{}{ + map[string]interface{}{ + "name": mqt1.Spec.Secret, + "parameter": "authMode", + "key": "authMode", + }, + map[string]interface{}{ + "name": mqt1.Spec.Secret, + "parameter": "username", + "key": "username", + }, + map[string]interface{}{ + "name": mqt1.Spec.Secret, + "parameter": "password", + "key": "password", + }, + map[string]interface{}{ + "name": mqt1.Spec.Secret, + "parameter": "ca", + "key": "ca", + }, + map[string]interface{}{ + "name": mqt1.Spec.Secret, + "parameter": "cert", + "key": "cert", + }, + map[string]interface{}{ + "name": mqt1.Spec.Secret, + "parameter": "key", + "key": "key", + }, + }, + }, + }, + } + + // Invalid without secret + mqt2 := &fv1.MessageQueueTrigger{ + ObjectMeta: metav1.ObjectMeta{ + Name: "Test", + Namespace: "default", + UID: "test123", + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: fv1.FunctionReferenceTypeFunctionName, + Name: "test", + }, + MessageQueueType: "kafka", + Topic: "topic", + ResponseTopic: "response-topic", + ErrorTopic: "error-topic", + MaxRetries: 4, + ContentType: "application/json", + PollingInterval: &pollingInterval, + CooldownPeriod: &cooldownPeriod, + MinReplicaCount: &minReplicaCount, + MaxReplicaCount: &maxReplicaCount, + Metadata: map[string]string{ + "bootstrapServers": "my-cluster-kafka-brokers.my-kafka-project.svc:9092", + "consumerGroup": "my-group", + "topic": "topic", + }, + Secret: "test-kafka-no-secret", + MqtKind: "keda", + }, + } + + type args struct { + mqt *fv1.MessageQueueTrigger + authenticationRef string + kubeClient kubernetes.Interface + } + tests := []struct { + name string + args args + want *unstructured.Unstructured + wantErr bool + }{ + {"With secret", args{mqt1, authenticationRef, kubeClient}, expectedAuthTriggerObj, false}, + {"With invalid secret", args{mqt2, authenticationRef, kubeClient}, nil, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := getAuthTriggerSpec(tt.args.mqt, tt.args.authenticationRef, tt.args.kubeClient) + if (err != nil) != tt.wantErr { + t.Errorf("getAuthTriggerSpec() error = %v, wantErr %v", err, tt.wantErr) + return + } + if err != nil && tt.wantErr { + return + } + + gotSpec := got.Object["spec"].(map[string]interface{})["secretTargetRef"].([]interface{}) + + sort.Slice(gotSpec, func(i, j int) bool { + return gotSpec[i].(map[string]interface{})["parameter"].(string) < gotSpec[j].(map[string]interface{})["parameter"].(string) + }) + + wantSpec := tt.want.Object["spec"].(map[string]interface{})["secretTargetRef"].([]interface{}) + + sort.Slice(wantSpec, func(i, j int) bool { + return wantSpec[i].(map[string]interface{})["parameter"].(string) < wantSpec[j].(map[string]interface{})["parameter"].(string) + }) + + if !reflect.DeepEqual(got.Object["kind"], tt.want.Object["kind"]) && + !reflect.DeepEqual(got.Object["apiVersion"], tt.want.Object["apiVersion"]) && + !reflect.DeepEqual(got.Object["metadata"], tt.want.Object["metadata"]) && + !reflect.DeepEqual(gotSpec, wantSpec) { + t.Errorf("getAuthTriggerSpec() = %v, want %v", got, tt.want) + } + + }) + } +} diff --git a/pkg/mqtrigger/util/util.go b/pkg/mqtrigger/util/util.go new file mode 100644 index 00000000..e1665f2b --- /dev/null +++ b/pkg/mqtrigger/util/util.go @@ -0,0 +1,43 @@ +package util + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +// FissionMetadata contains common fission side fields +type FissionMetadata struct { + // fission + Topic string + ResponseTopic string + ErrorTopic string + FunctionURL string + MaxRetries int + ContentType string + TriggerName string +} + +// ParseFissionMetadata parses fission side common fields and returns as fissionMetadata or returns error +func ParseFissionMetadata() (FissionMetadata, error) { + for _, envVars := range []string{"TOPIC", "FUNCTION_URL", "MAX_RETRIES", "CONTENT_TYPE", "TRIGGER_NAME"} { + if os.Getenv(envVars) == "" { + return FissionMetadata{}, fmt.Errorf("environment variable not found: %v", envVars) + } + } + meta := FissionMetadata{ + Topic: os.Getenv("TOPIC"), + ResponseTopic: os.Getenv("RESPONSE_TOPIC"), + ErrorTopic: os.Getenv("ERROR_TOPIC"), + FunctionURL: os.Getenv("FUNCTION_URL"), + ContentType: os.Getenv("CONTENT_TYPE"), + TriggerName: os.Getenv("TRIGGER_NAME"), + } + val, err := strconv.ParseInt(strings.TrimSpace(os.Getenv("MAX_RETRIES")), 0, 64) + if err != nil { + return FissionMetadata{}, fmt.Errorf("failed to parse value from MAX_RETRIES environment variable %v", err) + } + meta.MaxRetries = int(val) + return meta, nil +} diff --git a/test/test_utils.sh b/test/test_utils.sh index e8d2628e..d8aed4b8 100755 --- a/test/test_utils.sh +++ b/test/test_utils.sh @@ -404,7 +404,7 @@ dump_fission_logs() { component=$3 echo --- $component logs --- - kubectl -n $ns get pod -o name | grep $component | xargs kubectl -n $ns logs + kubectl -n $ns get pod -o name | grep $component | xargs -n1 kubectl -n $ns logs echo --- end $component logs --- } diff --git a/test/tests/mqtrigger_kind_keda/kafka/hellokafka.js b/test/tests/mqtrigger_kind_keda/kafka/hellokafka.js new file mode 100644 index 00000000..43307255 --- /dev/null +++ b/test/tests/mqtrigger_kind_keda/kafka/hellokafka.js @@ -0,0 +1,15 @@ +module.exports = async function (context) { + console.log(context.request.body); + console.log("z-custom-name: " + context.request.headers["z-custom-name"]); + console.log( + "x-fission-function-name: " + + context.request.headers["x-fission-function-name"] + ); + let obj = context.request.body; + let headers = context.request.headers; + return { + status: 200, + headers: headers, + body: obj, + }; +}; diff --git a/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/go.mod b/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/go.mod new file mode 100644 index 00000000..10960330 --- /dev/null +++ b/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/go.mod @@ -0,0 +1,8 @@ +module kafka_pub + +go 1.12 + +require ( + github.com/Shopify/sarama v1.23.1 + gopkg.in/jcmturner/goidentity.v3 v3.0.0 // indirect +) diff --git a/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/go.sum b/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/go.sum new file mode 100644 index 00000000..dbad49b8 --- /dev/null +++ b/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/go.sum @@ -0,0 +1,51 @@ +github.com/DataDog/zstd v1.3.6-0.20190409195224-796139022798 h1:2T/jmrHeTezcCM58lvEQXs0UpQJCo5SoGAcg+mbSTIg= +github.com/DataDog/zstd v1.3.6-0.20190409195224-796139022798/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo= +github.com/Shopify/sarama v1.23.1 h1:XxJBCZEoWJtoWjf/xRbmGUpAmTZGnuuF0ON0EvxxBrs= +github.com/Shopify/sarama v1.23.1/go.mod h1:XLH1GYJnLVE0XCr6KdJGVJRTwY30moWNJ4sERjXX6fs= +github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc= +github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/eapache/go-resiliency v1.1.0 h1:1NtRmCAqadE2FN4ZcN6g90TP3uk8cg9rn9eNK2197aU= +github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs= +github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw= +github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU= +github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc= +github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I= +github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/hashicorp/go-uuid v1.0.1 h1:fv1ep09latC32wFoVwnqcnKJGnMSdBanPczbHAYm1BE= +github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 h1:FUwcHNlEqkqLjLBdCp5PRlCFijNjvcYANOZXzCfXwCM= +github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o= +github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41 h1:GeinFsrjWz97fAxVUEd748aV0cYL+I6k44gFJTCVvpU= +github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc= +github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I= +github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5 h1:bselrhR0Or1vomJZC8ZIjWtbDmn9OYFLX5Ik9alpJpE= +golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ= +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-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw= +gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN5hSM= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q= +gopkg.in/jcmturner/goidentity.v3 v3.0.0 h1:1duIyWiTaYvVx3YX2CYtpJbUFd7/UuPYCfgXtQ3VTbI= +gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4= +gopkg.in/jcmturner/gokrb5.v7 v7.2.3 h1:hHMV/yKPwMnJhPuPx7pH2Uw/3Qyf+thJYlisUc44010= +gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM= +gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU= +gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8= diff --git a/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/kafka-pub.go b/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/kafka-pub.go new file mode 100644 index 00000000..e53f9562 --- /dev/null +++ b/test/tests/mqtrigger_kind_keda/kafka/kafka_pub/kafka-pub.go @@ -0,0 +1,38 @@ +package main + +import ( + "fmt" + "net/http" + + sarama "github.com/Shopify/sarama" +) + +// Handler posts a message to Kafka Topic +func Handler(w http.ResponseWriter, r *http.Request) { + brokers := []string{"my-cluster-kafka-brokers.my-kafka-project.svc:9092"} + producerConfig := sarama.NewConfig() + producerConfig.Producer.RequiredAcks = sarama.WaitForAll + producerConfig.Producer.Retry.Max = 100 + producerConfig.Producer.Retry.Backoff = 100 + producerConfig.Producer.Return.Successes = true + producerConfig.Version = sarama.V1_0_0_0 + producer, err := sarama.NewSyncProducer(brokers, producerConfig) + fmt.Println("Created a new producer ", producer) + if err != nil { + panic(err) + } + for i := 0; i < 1000; i++ { + headers := []sarama.RecordHeader{{Key: []byte("Z-Custom-Name"), Value: []byte("Kafka-Header-test")}} + _, _, err = producer.SendMessage(&sarama.ProducerMessage{ + Topic: "topic2", + Value: sarama.StringEncoder("{\"name\": \"testvalue\"}"), + Headers: headers, + }) + + if err != nil { + w.Write([]byte(fmt.Sprintf("Failed to publish message to topic %s: %v", "testtopic", err))) + return + } + } + w.Write([]byte("Successfully sent to testtopic")) +} diff --git a/test/tests/mqtrigger_kind_keda/kafka/test_kafka.sh b/test/tests/mqtrigger_kind_keda/kafka/test_kafka.sh new file mode 100755 index 00000000..955a12c3 --- /dev/null +++ b/test/tests/mqtrigger_kind_keda/kafka/test_kafka.sh @@ -0,0 +1,103 @@ +#!/bin/bash +#test:disabled + +# Create a function and trigger it using Kafka +# This test requires KEDA and Kafka in the cluster +# This test assumes that there are three topics are already created with 3 Partitions and 3 Replicas in the cluster + +set -euo pipefail +source $(dirname $0)/../../../utils.sh +set +x + +TEST_ID=$(generate_test_id) +echo "TEST_ID = $TEST_ID" + +tmp_dir="/tmp/test-$TEST_ID" +mkdir -p $tmp_dir + +nodeenv="node-kafka-$TEST_ID" +goenv="go-kafka-$TEST_ID" +producerfunc="producer-func-$TEST_ID" +consumerfunc="consumer-func-$TEST_ID" +mqt="kafkatest-$TEST_ID" +no_of_topic_partition=3 # Change this if number of partitions are different +topic="topic2" +resptopic="response-topic" +errortopic="error-topic" +bootstrap_server="my-cluster-kafka-brokers.my-kafka-project.svc:9092" +consumer_group="my-group" +cooldownperiod=30 +pollinginterval=30 + +export FISSION_ROUTER=127.0.0.1:11009 + +test_totalpods(){ + echo "Checking total number of scaledpods" + + set +e + while true; do + + cnt=$(kubectl get deployment ${mqt} -o jsonpath='{.status.readyReplicas}') + if [$cnt -ge $no_of_topic_partition]; then + break + fi + sleep 1 + done + set -e +} + +export -f test_totalpods + +cleanup() { + log "Cleaning up..." + clean_resource_by_id $TEST_ID + rm -rf $tmp_dir +} + +if [ -z "${TEST_NOCLEANUP:-}" ]; then + trap cleanup EXIT +else + log "TEST_NOCLEANUP is set; not cleaning up test artifacts afterwards." +fi + +DIR=$(dirname $0) + +log "Creating ${nodeenv} environment" +fission env create --name ${nodeenv} --image ${NODE_RUNTIME_IMAGE} + +log "Creating ${goenv} environment" +fission env create --name ${goenv} --image ${GO_RUNTIME_IMAGE} --builder ${GO_BUILDER_IMAGE} + +log "Creating package for Kafka producer" +cp -r $DIR/kafka_pub $tmp_dir/ +pushd $tmp_dir/kafka_pub +go mod vendor +zip -qr kafka.zip * +pkgName=$(fission package create --name kafka-${TEST_ID} --env ${goenv} --src kafka.zip|cut -f2 -d' '| tr -d \') + +log "pkgName=${pkgName}" +popd + +timeout 300s bash -c "waitBuild $pkgName" +log "Package ${pkgName} created" +log "Creating function ${producerfunc}" +fission fn create --name ${producerfunc} --env ${goenv} --pkg ${pkgName} --entrypoint Handler +log "Creating function ${consumerfunc}" +fission fn create --name ${consumerfunc} --env ${nodeenv} --code $DIR/hellokafka.js + +log "Creating trigger $mqt" +fission mqt create --name ${mqt} --function ${consumerfunc} --mqtype kafka --topic $topic --resptopic $resptopic --errortopic $errortopic --version2=true --metadata bootstrapServers=$bootstrap_server --metadata consumerGroup=$consumer_group --metadata topic=$topic --cooldownperiod=$cooldownperiod --pollinginterval=$pollinginterval + +log "Create Route to producer function" +fission route create --function ${producerfunc} --url /${producerfunc} --method GET + +log "Run Producer function" +funcURL=$FISSION_ROUTER/${producerfunc} +success="Successfully sent to testtopic" +echo $funcURL + +timeout 100s bash -c "test_response $funcURL $success" + +timeout 100s bash -c "test_totalpods" + +log "Test PASSED"