diff --git a/charts/fission-all/templates/mqt-fission-azurestoragequeue/deployment.yaml b/charts/fission-all/templates/mqt-fission-azurestoragequeue/deployment.yaml deleted file mode 100644 index 9de91ebc..00000000 --- a/charts/fission-all/templates/mqt-fission-azurestoragequeue/deployment.yaml +++ /dev/null @@ -1,62 +0,0 @@ -{{- if .Values.azureStorageQueue.enabled }} -apiVersion: apps/v1 -kind: Deployment -metadata: - name: mqtrigger-azure-storage-queue - labels: - chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" - svc: mqtrigger - messagequeue: azure-storage-queue -spec: - replicas: 1 - selector: - matchLabels: - svc: mqtrigger - messagequeue: azure-storage-queue - template: - metadata: - labels: - svc: mqtrigger - messagequeue: azure-storage-queue - spec: - containers: - - name: mqtrigger - {{- if eq .Values.imageTag "" }} - image: "{{ .Values.image }}" - {{- else }} - image: "{{ .Values.image }}:{{ .Values.imageTag }}" - {{- end }} - imagePullPolicy: {{ .Values.pullPolicy }} - command: ["/fission-bundle"] - args: ["--mqt", "--routerUrl", "http://router.{{ .Release.Namespace }}"] - env: - - name: MESSAGE_QUEUE_TYPE - value: azure-storage-queue - - name: AZURE_STORAGE_ACCOUNT_NAME - value: {{ required "An Azure storage account name is required." .Values.azureStorageQueue.accountName }} - - name: AZURE_STORAGE_ACCOUNT_KEY - valueFrom: - secretKeyRef: - name: azure-storage-account-key - key: key - - name: DEBUG_ENV - value: {{ .Values.debugEnv | quote }} - - name: PPROF_ENABLED - value: {{ .Values.pprof.enabled | quote }} - {{- include "opentracing.envs" . | indent 8 }} - {{- include "opentelemtry.envs" . | indent 8 }} - {{- if .Values.terminationMessagePath }} - terminationMessagePath: {{ .Values.terminationMessagePath }} - {{- end }} - {{- if .Values.terminationMessagePolicy }} - terminationMessagePolicy: {{ .Values.terminationMessagePolicy }} - {{- end }} - {{- with .Values.imagePullSecrets }} - imagePullSecrets: - {{- toYaml . | nindent 8 }} - {{- end }} - serviceAccountName: fission-svc -{{- if .Values.extraCoreComponentPodConfig }} -{{ toYaml .Values.extraCoreComponentPodConfig | indent 6 -}} -{{- end }} -{{- end }} \ No newline at end of file diff --git a/charts/fission-all/templates/mqt-fission-azurestoragequeue/secret.yaml b/charts/fission-all/templates/mqt-fission-azurestoragequeue/secret.yaml deleted file mode 100644 index 78adbfb1..00000000 --- a/charts/fission-all/templates/mqt-fission-azurestoragequeue/secret.yaml +++ /dev/null @@ -1,11 +0,0 @@ -{{- if .Values.azureStorageQueue.enabled }} -apiVersion: v1 -kind: Secret -metadata: - name: azure-storage-account-key - labels: - chart: "{{ .Chart.Name }}-{{ .Chart.Version }}" -type: Opaque -data: - key: {{ required "An Azure storage access key is required." .Values.azureStorageQueue.key | b64enc | quote }} -{{- end }} diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index 34407eab..fd7a9344 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -320,13 +320,6 @@ timer: ## resources: {} -## Azure-storage-queue: enable and configure the details -## -azureStorageQueue: - enabled: false - key: "" - accountName: "" - ## Kafka: enable and configure the details ## kafka: diff --git a/cmd/fission-bundle/mqtrigger/mqtrigger.go b/cmd/fission-bundle/mqtrigger/mqtrigger.go index 14cb601d..92c92d15 100644 --- a/cmd/fission-bundle/mqtrigger/mqtrigger.go +++ b/cmd/fission-bundle/mqtrigger/mqtrigger.go @@ -31,7 +31,6 @@ import ( "github.com/fission/fission/pkg/mqtrigger" "github.com/fission/fission/pkg/mqtrigger/factory" "github.com/fission/fission/pkg/mqtrigger/messageQueue" - _ "github.com/fission/fission/pkg/mqtrigger/messageQueue/azurequeuestorage" _ "github.com/fission/fission/pkg/mqtrigger/messageQueue/kafka" ) diff --git a/cmd/fission-cli/app/app.go b/cmd/fission-cli/app/app.go index 76a99d20..8c914a1a 100644 --- a/cmd/fission-cli/app/app.go +++ b/cmd/fission-cli/app/app.go @@ -39,7 +39,6 @@ import ( "github.com/fission/fission/pkg/fission-cli/flag" flagkey "github.com/fission/fission/pkg/fission-cli/flag/key" "github.com/fission/fission/pkg/fission-cli/util" - _ "github.com/fission/fission/pkg/mqtrigger/messageQueue/azurequeuestorage" _ "github.com/fission/fission/pkg/mqtrigger/messageQueue/kafka" ) diff --git a/go.mod b/go.mod index adfc0f35..5270254b 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,6 @@ go 1.18 require ( contrib.go.opencensus.io/exporter/jaeger v0.2.1 - github.com/Azure/azure-sdk-for-go v62.1.0+incompatible github.com/Shopify/sarama v1.32.0 github.com/dchest/uniuri v0.0.0-20200228104902-7aecb25e1fe5 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815 @@ -101,7 +100,6 @@ require ( github.com/go-openapi/jsonpointer v0.19.5 // indirect github.com/go-openapi/jsonreference v0.19.6 // indirect github.com/go-openapi/swag v0.19.15 // indirect - github.com/gofrs/uuid v4.0.0+incompatible // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 // indirect github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect @@ -149,7 +147,6 @@ require ( github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/sergi/go-diff v1.1.0 // indirect github.com/sirupsen/logrus v1.8.1 // indirect - github.com/stretchr/objx v0.2.0 // indirect github.com/uber/jaeger-client-go v2.28.0+incompatible // indirect github.com/ulikunitz/xz v0.5.9 // indirect github.com/xanzy/ssh-agent v0.3.0 // indirect diff --git a/go.sum b/go.sum index b42371a5..2547a49e 100644 --- a/go.sum +++ b/go.sum @@ -51,8 +51,6 @@ contrib.go.opencensus.io/exporter/jaeger v0.2.1 h1:yGBYzYMewVL0yO9qqJv3Z5+IRhPdU contrib.go.opencensus.io/exporter/jaeger v0.2.1/go.mod h1:Y8IsLgdxqh1QxYxPC5IgXVmBaeLUeQFfBeBi9PbeZd0= dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU= github.com/Azure/azure-sdk-for-go v32.5.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc= -github.com/Azure/azure-sdk-for-go v62.1.0+incompatible h1:FKtAafvCGabH4hzycX4uGq/mSTKN+zaeRTUdnJB5Yag= -github.com/Azure/azure-sdk-for-go v62.1.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc= github.com/Azure/go-ansiterm v0.0.0-20210608223527-2377c96fe795/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8= github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8= github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= @@ -71,7 +69,6 @@ github.com/Azure/go-autorest/autorest/mocks v0.1.0/go.mod h1:OTyCOPRA2IgIlWxVYxB github.com/Azure/go-autorest/autorest/mocks v0.2.0/go.mod h1:OTyCOPRA2IgIlWxVYxBee2F5Gr4kF2zd2J5cFRaIDN0= github.com/Azure/go-autorest/autorest/mocks v0.4.1 h1:K0laFcLE6VLTOwNgSxaGbUcLPuGXlNkbVvq4cW4nIHk= github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k= -github.com/Azure/go-autorest/autorest/to v0.4.0 h1:oXVqrxakqqV1UZdSazDOPOLvOIz+XA683u8EctwboHk= github.com/Azure/go-autorest/autorest/to v0.4.0/go.mod h1:fE8iZBn7LQR7zH/9XU2NcPR4o9jEImooCeWJcYV/zLE= github.com/Azure/go-autorest/logger v0.1.0/go.mod h1:oExouG+K6PryycPJfVSxi/koC6LSNgds39diKLz7Vrc= github.com/Azure/go-autorest/logger v0.2.1 h1:IG7i4p/mDa2Ce4TRyAO8IHnVhAVF3RFU+ZtXWSmf4Tg= @@ -193,7 +190,6 @@ github.com/dchest/uniuri v0.0.0-20200228104902-7aecb25e1fe5 h1:RAV05c0xOkJ3dZGS0 github.com/dchest/uniuri v0.0.0-20200228104902-7aecb25e1fe5/go.mod h1:GgB8SF9nRG+GqaDtLcwJZsQFhcogVCJ79j4EdT0c2V4= github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no= -github.com/dnaeon/go-vcr v1.1.0 h1:ReYa/UBrRyQdant9B4fNHGoCNKw6qh6P0fsdGmZpR7c= github.com/dnaeon/go-vcr v1.1.0/go.mod h1:M7tiix8f0r6mKKJ3Yq/kqU1OYf3MnfmBWVbPx/yU9ko= github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ= github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec= @@ -313,8 +309,6 @@ github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/me github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= github.com/godbus/dbus/v5 v5.0.6/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= -github.com/gofrs/uuid v4.0.0+incompatible h1:1SD/1F5pU8p29ybwgQSwpQk+mwdRrXCYuPhW6m+TnJw= -github.com/gofrs/uuid v4.0.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4= github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o= @@ -723,8 +717,6 @@ github.com/spf13/viper v1.8.1/go.mod h1:o0Pch8wJ9BVSWGQMbra6iw0oQ5oktSIBaujf1rJH github.com/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag1KpM8ahLw8= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= -github.com/stretchr/objx v0.2.0 h1:Hbg2NidpLE8veEBkEZTL3CvlkUIVzuU9jDplZO54c48= -github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= diff --git a/pkg/apis/core/v1/const.go b/pkg/apis/core/v1/const.go index 922c164e..886b60c7 100644 --- a/pkg/apis/core/v1/const.go +++ b/pkg/apis/core/v1/const.go @@ -76,7 +76,6 @@ const ( ) const ( - MessageQueueTypeASQ = "azure-storage-queue" MessageQueueTypeKafka = "kafka" ) diff --git a/pkg/fission-cli/cmd/support/resources/crd.go b/pkg/fission-cli/cmd/support/resources/crd.go index f8d874fd..bbf611fc 100644 --- a/pkg/fission-cli/cmd/support/resources/crd.go +++ b/pkg/fission-cli/cmd/support/resources/crd.go @@ -113,7 +113,7 @@ func (res CrdDumper) Dump(dumpDir string) { case CrdMessageQueueTrigger: var triggers []fv1.MessageQueueTrigger - for _, mqType := range []string{fv1.MessageQueueTypeASQ, fv1.MessageQueueTypeKafka} { + for _, mqType := range []string{fv1.MessageQueueTypeKafka} { l, err := res.client.V1().MessageQueueTrigger().List(mqType, metav1.NamespaceAll) if err != nil { console.Warn(fmt.Sprintf("Error getting %v list: %v", res.crdType, err)) diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go index cad3ac46..f17245ce 100644 --- a/pkg/fission-cli/flag/flag.go +++ b/pkg/fission-cli/flag/flag.go @@ -155,7 +155,7 @@ var ( 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: "For mqtype \"fission\" => azure-storage-queue, kafka\n\t\t\t\t\t For mqtype \"keda\" => kafka, aws-sqs-queue, aws-kinesis-stream, gcp-pubsub, stan, rabbitmq, redis", DefaultValue: "kafka"} + MqtMQType = Flag{Type: String, Name: flagkey.MqtMQType, Usage: "For mqtype \"fission\" => kafka\n\t\t\t\t\t For mqtype \"keda\" => kafka, aws-sqs-queue, aws-kinesis-stream, gcp-pubsub, stan, rabbitmq, redis", DefaultValue: "kafka"} 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"} diff --git a/pkg/mqtrigger/messageQueue/azurequeuestorage/asq.go b/pkg/mqtrigger/messageQueue/azurequeuestorage/asq.go deleted file mode 100644 index 5e9c90c5..00000000 --- a/pkg/mqtrigger/messageQueue/azurequeuestorage/asq.go +++ /dev/null @@ -1,424 +0,0 @@ -/* -Copyright 2017 The Fission Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package azurequeuestorage - -import ( - "bytes" - "encoding/base64" - "fmt" - "io" - "net/http" - "os" - "regexp" - "strconv" - "strings" - "sync" - "time" - - "github.com/Azure/azure-sdk-for-go/storage" - "github.com/pkg/errors" - "go.uber.org/zap" - - fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/mqtrigger/factory" - "github.com/fission/fission/pkg/mqtrigger/messageQueue" - "github.com/fission/fission/pkg/mqtrigger/validator" - "github.com/fission/fission/pkg/utils" -) - -func init() { - factory.Register(fv1.MessageQueueTypeASQ, &Factory{}) - validator.Register(fv1.MessageQueueTypeASQ, IsTopicValid) -} - -// TODO: some of these constants should probably be environment variables -const ( - // AzureQueuePollingInterval is the polling interval (default is 1 minute). - AzureQueuePollingInterval = time.Minute - // AzureQueueRetryLimit is the limit for attempts to retry invoking a function. - AzureQueueRetryLimit = 3 - // AzureMessageFetchCount is the number of messages to fetch at a time. - AzureMessageFetchCount = 10 - // AzureMessageVisibilityTimeout is the visibility timeout for dequeued messages. - AzureMessageVisibilityTimeout = time.Minute - // AzurePoisonQueueSuffix is the suffix used for poison queues. - AzurePoisonQueueSuffix = "-poison" - // AzureFunctionInvocationTimeout is the amount of time to wait for a triggered function to execute. - AzureFunctionInvocationTimeout = 10 * time.Minute -) - -var ( - validAzureQueueName = regexp.MustCompile(`^[a-z0-9][a-z0-9\\-]*[a-z0-9]$`) -) - -// AzureStorageConnection represents an Azure storage connection. -type AzureStorageConnection struct { - logger *zap.Logger - routerURL string - service AzureQueueService - httpClient AzureHTTPClient -} - -// AzureQueueSubscription represents an Azure storage message queue subscription. -type AzureQueueSubscription struct { - queue AzureQueue - queueName string - outputQueueName string - functionURL string - contentType string - unsubscribe chan bool - done chan bool -} - -// AzureQueueService is the interface that abstracts the Azure storage service. -// This exists to enable unit testing. -type AzureQueueService interface { - GetQueue(name string) AzureQueue -} - -// AzureQueue is the interface that abstracts Azure storage queues. -// This exists to enable unit testing. -type AzureQueue interface { - Create(options *storage.QueueServiceOptions) error - NewMessage(text string) AzureMessage - GetMessages(options *storage.GetMessagesOptions) ([]AzureMessage, error) -} - -// AzureMessage is the interface that abstracts Azure storage messages. -// This exists to enable unit testing. -type AzureMessage interface { - Bytes() []byte - Put(options *storage.PutMessageOptions) error - Delete(options *storage.QueueServiceOptions) error -} - -// AzureHTTPClient is the interface that abstract HTTP requests made by the trigger. -// This exists to enable unit testing. -type AzureHTTPClient interface { - Do(req *http.Request) (*http.Response, error) -} - -type Factory struct{} - -func (factory *Factory) Create(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messageQueue.MessageQueue, error) { - return New(logger, mqCfg, routerUrl) -} - -type azureQueueService struct { - service storage.QueueServiceClient -} - -func (qs azureQueueService) GetQueue(name string) AzureQueue { - return azureQueue{ - ref: qs.service.GetQueueReference(name), - } -} - -type azureQueue struct { - ref *storage.Queue -} - -func (qr azureQueue) Create(options *storage.QueueServiceOptions) error { - exists, err := qr.ref.Exists() - if err != nil { - return err - } - if exists { - return nil - } - return qr.ref.Create(options) -} - -func (qr azureQueue) NewMessage(text string) AzureMessage { - return azureMessage{ - ref: qr.ref.GetMessageReference(text), - bytes: []byte(text), - } -} - -func (qr azureQueue) GetMessages(options *storage.GetMessagesOptions) ([]AzureMessage, error) { - msgs, err := qr.ref.GetMessages(options) - if err != nil { - return nil, err - } - messages := make([]AzureMessage, len(msgs)) - for i := range msgs { - bytes, err := base64.StdEncoding.DecodeString(msgs[i].Text) - if err != nil { - return nil, err - } - messages[i] = azureMessage{ - ref: &msgs[i], - bytes: bytes, - } - } - return messages, nil -} - -type azureMessage struct { - ref *storage.Message - bytes []byte -} - -func (m azureMessage) Bytes() []byte { - return m.bytes -} - -func (m azureMessage) Put(options *storage.PutMessageOptions) error { - return m.ref.Put(options) -} - -func (m azureMessage) Delete(options *storage.QueueServiceOptions) error { - return m.ref.Delete(options) -} - -func newAzureQueueService(client storage.Client) AzureQueueService { - return azureQueueService{ - service: client.GetQueueService(), - } -} - -func New(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messageQueue.MessageQueue, error) { - account := os.Getenv("AZURE_STORAGE_ACCOUNT_NAME") - if len(account) == 0 { - return nil, errors.New("Required environment variable 'AZURE_STORAGE_ACCOUNT_NAME' is not set") - } - - key := os.Getenv("AZURE_STORAGE_ACCOUNT_KEY") - if len(key) == 0 { - return nil, errors.New("Required environment variable 'AZURE_STORAGE_ACCOUNT_KEY' is not set") - } - - logger.Info("creating Azure storage connection to storage account", zap.String("account", account)) - - client, err := storage.NewBasicClient(account, key) - if err != nil { - return nil, errors.Wrap(err, "failed to create Azure storage client") - } - return &AzureStorageConnection{ - logger: logger.Named("azue_storage"), - routerURL: routerUrl, - service: newAzureQueueService(client), - httpClient: &http.Client{ - Timeout: AzureFunctionInvocationTimeout, - }, - }, nil -} - -func (asc AzureStorageConnection) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Subscription, error) { - asc.logger.Info("subscribing to Azure storage queue", zap.String("queue", trigger.Spec.Topic)) - - if trigger.Spec.FunctionReference.Type != fv1.FunctionReferenceTypeFunctionName { - return nil, fmt.Errorf("unsupported function reference type (%v) for trigger %q", trigger.Spec.FunctionReference.Type, trigger.ObjectMeta.Name) - } - - subscription := &AzureQueueSubscription{ - queue: asc.service.GetQueue(trigger.Spec.Topic), - queueName: trigger.Spec.Topic, - outputQueueName: trigger.Spec.ResponseTopic, - // with the addition of multi-tenancy, the users can create functions in any namespace. however, - // the triggers can only be created in the same namespace as the function. - // so essentially, function namespace = trigger namespace. - functionURL: asc.routerURL + "/" + strings.TrimPrefix(utils.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.ObjectMeta.Namespace), "/"), - contentType: trigger.Spec.ContentType, - unsubscribe: make(chan bool), - done: make(chan bool), - } - - go runAzureQueueSubscription(asc, subscription) - return subscription, nil -} - -func (asc AzureStorageConnection) Unsubscribe(subscription messageQueue.Subscription) error { - sub := subscription.(*AzureQueueSubscription) - - asc.logger.Info("unsubscribing from Azure storage queue", zap.String("queue", sub.queueName)) - - // Let the worker know we've unsubscribed - sub.unsubscribe <- true - - // Wait until the subscription is done - <-sub.done - return nil -} - -func runAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubscription) { - var wg sync.WaitGroup - - // Process the queue before waiting - pollAzureQueueSubscription(conn, sub, &wg) - - timer := time.NewTimer(AzureQueuePollingInterval) - - for { - conn.logger.Info("waiting before polling Azure storage queue", zap.Duration("interval_length", AzureQueuePollingInterval), zap.String("queue", sub.queueName)) - select { - case <-sub.unsubscribe: - timer.Stop() - wg.Wait() - sub.done <- true - return - case <-timer.C: - pollAzureQueueSubscription(conn, sub, &wg) - timer.Reset(AzureQueuePollingInterval) - continue - } - } -} - -func pollAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubscription, wg *sync.WaitGroup) { - conn.logger.Info("polling for messages from Azure storage queue", zap.String("queue", sub.queueName)) - - err := sub.queue.Create(nil) - if err != nil { - conn.logger.Error("failed to create message queue", zap.Error(err), zap.String("queue", sub.queueName)) - return - } - - for { - err := sub.queue.Create(nil) - if err != nil { - conn.logger.Error("failed to create message queue", zap.Error(err), zap.String("queue", sub.queueName)) - return - } - - messages, err := sub.queue.GetMessages(&storage.GetMessagesOptions{ - NumOfMessages: AzureMessageFetchCount, - VisibilityTimeout: int(AzureMessageVisibilityTimeout / time.Second), - }) - if err != nil { - conn.logger.Error("failed to retrieve messages from Azure storage queue", zap.Error(err), zap.String("queue", sub.queueName)) - break - } - if len(messages) == 0 { - break - } - - wg.Add(len(messages)) - for _, msg := range messages { - go func(conn AzureStorageConnection, sub *AzureQueueSubscription, msg AzureMessage) { - defer wg.Done() - invokeTriggeredFunction(conn, sub, msg) - }(conn, sub, msg) - } - } -} - -func invokeTriggeredFunction(conn AzureStorageConnection, sub *AzureQueueSubscription, message AzureMessage) { - defer func() { - err := message.Delete(nil) - if err != nil { - conn.logger.Error(err.Error()) - } - }() - - conn.logger.Info("making HTTP request to invoke function", zap.String("function_url", sub.functionURL)) - - for i := 0; i <= AzureQueueRetryLimit; i++ { - if i > 0 { - conn.logger.Info("retrying function invocation", zap.Int("retry", i), zap.String("function_url", sub.functionURL)) - } - request, err := http.NewRequest("POST", sub.functionURL, bytes.NewReader(message.Bytes())) - if err != nil { - conn.logger.Error("failed to create HTTP request to invoke function", zap.Error(err), zap.String("function_url", sub.functionURL)) - continue - } - - request.Header.Set("X-Fission-MQTrigger-Topic", sub.queueName) - if len(sub.outputQueueName) > 0 { - request.Header.Set("X-Fission-MQTrigger-RespTopic", sub.outputQueueName) - } - if i > 0 { - request.Header.Set("X-Fission-MQTrigger-RetryCount", strconv.Itoa(i)) - } - request.Header.Set("Content-Type", sub.contentType) - - response, err := conn.httpClient.Do(request) - if err != nil { - conn.logger.Error("sending function invocation request failed", zap.Error(err), zap.String("function_url", sub.functionURL)) - continue - } - defer response.Body.Close() - - body, err := io.ReadAll(response.Body) - if err != nil { - conn.logger.Error("failed to read response body from function invocation", zap.Error(err), zap.String("function_url", sub.functionURL)) - continue - } - - if response.StatusCode < 200 || response.StatusCode >= 300 { - conn.logger.Error("function invocation request returned a failure status code", - zap.String("function_url", sub.functionURL), - zap.String("body", string(body)), - zap.Int("status_code", response.StatusCode)) - continue - } - - if len(sub.outputQueueName) > 0 { - outputQueue := conn.service.GetQueue(sub.outputQueueName) - err = outputQueue.Create(nil) - if err != nil { - conn.logger.Error("failed to create output queue", - zap.Error(err), - zap.String("output_queue", sub.outputQueueName), - zap.String("function_url", sub.functionURL)) - return - } - - outputMessage := outputQueue.NewMessage(string(body)) - err = outputMessage.Put(nil) - if err != nil { - conn.logger.Error("failed to post response body from function invocation to output queue", - zap.String("output_queue", sub.outputQueueName), - zap.String("function_url", sub.functionURL)) - return - } - } - - // Function invocation was successful - return - } - - conn.logger.Error("function invocation retired too many times - moving message to poison queue", - zap.Int("retry_limit", AzureQueueRetryLimit), - zap.String("function_url", sub.functionURL)) - - poisonQueueName := sub.queueName + AzurePoisonQueueSuffix - poisonQueue := conn.service.GetQueue(poisonQueueName) - err := poisonQueue.Create(nil) - if err != nil { - conn.logger.Error("failed to create poison queue", - zap.Error(err), - zap.String("poison_queue_name", poisonQueueName), - zap.String("function_url", sub.functionURL)) - return - } - - poisonMessage := poisonQueue.NewMessage(string(message.Bytes())) - err = poisonMessage.Put(nil) - if err != nil { - conn.logger.Error("failed to post response body from function invocation failure poison queue", - zap.Error(err), - zap.String("poison_queue_name", poisonQueueName), - zap.String("function_url", sub.functionURL)) - return - } -} - -func IsTopicValid(topic string) bool { - return len(topic) >= 3 && len(topic) <= 63 && validAzureQueueName.MatchString(topic) -} diff --git a/pkg/mqtrigger/messageQueue/azurequeuestorage/asq_test.go b/pkg/mqtrigger/messageQueue/azurequeuestorage/asq_test.go deleted file mode 100644 index 4ab34c00..00000000 --- a/pkg/mqtrigger/messageQueue/azurequeuestorage/asq_test.go +++ /dev/null @@ -1,488 +0,0 @@ -/* -Copyright 2017 The Fission Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ -package azurequeuestorage - -import ( - "fmt" - "io" - "log" - "net/http" - "os" - "strings" - "testing" - "time" - - "github.com/Azure/azure-sdk-for-go/storage" - "github.com/stretchr/testify/mock" - "github.com/stretchr/testify/require" - "go.uber.org/zap" - "go.uber.org/zap/zapcore" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/mqtrigger/messageQueue" -) - -const ( - DummyRouterURL = "http://localhost" -) - -func panicIf(err error) { - if err != nil { - log.Panicf("Error: %v", err) - } -} - -type azureQueueServiceMock struct { - mock.Mock -} - -func (m *azureQueueServiceMock) GetQueue(name string) AzureQueue { - args := m.Called(name) - return args.Get(0).(AzureQueue) -} - -type azureQueueMock struct { - mock.Mock -} - -func (m *azureQueueMock) Create(options *storage.QueueServiceOptions) error { - args := m.Called(options) - return args.Error(0) -} - -func (m *azureQueueMock) NewMessage(text string) AzureMessage { - args := m.Called(text) - return args.Get(0).(AzureMessage) -} - -func (m *azureQueueMock) GetMessages(options *storage.GetMessagesOptions) ([]AzureMessage, error) { - args := m.Called(options) - return args.Get(0).([]AzureMessage), args.Error(1) -} - -type azureMessageMock struct { - mock.Mock -} - -func (m *azureMessageMock) Bytes() []byte { - args := m.Called() - return args.Get(0).([]byte) -} - -func (m *azureMessageMock) Put(options *storage.PutMessageOptions) error { - args := m.Called(options) - return args.Error(0) -} - -func (m *azureMessageMock) Delete(options *storage.QueueServiceOptions) error { - args := m.Called(options) - return args.Error(0) -} - -type azureHTTPClientMock struct { - mock.Mock - bodyHandler func(res *http.Response) -} - -func (m *azureHTTPClientMock) Do(req *http.Request) (*http.Response, error) { - args := m.Called(req) - - res := args.Get(0).(*http.Response) - err := args.Error(1) - - if res != nil && m.bodyHandler != nil { - m.bodyHandler(res) - } - return res, err -} - -func TestNewStorageConnectionMissingAccountName(t *testing.T) { - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - panicIf(err) - - connection, err := New(logger, messageQueue.Config{ - MQType: fv1.MessageQueueTypeASQ, - Url: "", - }, DummyRouterURL) - require.Nil(t, connection) - require.Error(t, err, "Required environment variable 'AZURE_STORAGE_ACCOUNT_NAME' is not set") -} - -func TestNewStorageConnectionMissingAccessKey(t *testing.T) { - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - panicIf(err) - - _ = os.Setenv("AZURE_STORAGE_ACCOUNT_NAME", "accountname") - connection, err := New(logger, messageQueue.Config{ - MQType: fv1.MessageQueueTypeASQ, - Url: "", - }, DummyRouterURL) - _ = os.Unsetenv("AZURE_STORAGE_ACCOUNT_NAME") - require.Nil(t, connection) - require.Error(t, err, "Required environment variable 'AZURE_STORAGE_ACCOUNT_KEY' is not set") -} - -func TestNewStorageConnection(t *testing.T) { - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - panicIf(err) - - _ = os.Setenv("AZURE_STORAGE_ACCOUNT_NAME", "accountname") - _ = os.Setenv("AZURE_STORAGE_ACCOUNT_KEY", "bm90IGEga2V5") - connection, err := New(logger, messageQueue.Config{ - MQType: "azure-storage-queue", - Url: "", - }, DummyRouterURL) - _ = os.Unsetenv("AZURE_STORAGE_ACCOUNT_NAME") - _ = os.Unsetenv("AZURE_STORAGE_ACCOUNT_KEY") - require.NoError(t, err) - require.IsType(t, &AzureStorageConnection{}, connection) - - p := connection.(*AzureStorageConnection) - require.Equal(t, DummyRouterURL, p.routerURL) - require.NotNil(t, p.service) -} - -func TestAzureStorageQueueSingleMessage(t *testing.T) { - runAzureStorageQueueTest(t, 1, false) -} - -// TODO: Enable after fixing race condition -// func TestAzureStorageQueueMultipleMessages(t *testing.T) { -// runAzureStorageQueueTest(t, 10, false) -// } - -func TestAzureStorageQueueSingleOutputMessage(t *testing.T) { - runAzureStorageQueueTest(t, 1, true) -} - -// TODO: Enable after fixing race condition -// func TestAzureStorageQueueMultipleOutputMessages(t *testing.T) { -// runAzureStorageQueueTest(t, 10, true) -// } - -func TestAzureStorageQueuePoisonMessage(t *testing.T) { - const ( - TriggerName = "queuetrigger" - QueueName = "inputqueue" - MessageBody = "input" - FunctionName = "badfunc" - ContentType = "text/plain" - ) - - // Mock a HTTP client that returns different failures - httpClient := new(azureHTTPClientMock) - httpClient.On( - "Do", - mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "", ContentType, FunctionName, MessageBody)), - ).Return( - &http.Response{ - StatusCode: http.StatusInternalServerError, - Body: io.NopCloser(strings.NewReader("server error")), - }, - nil, - ).Once() - httpClient.On( - "Do", - mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "1", ContentType, FunctionName, MessageBody)), - ).Return( - &http.Response{ - StatusCode: http.StatusNotFound, - Body: io.NopCloser(strings.NewReader("not found")), - }, - nil, - ).Once() - httpClient.On( - "Do", - mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "2", ContentType, FunctionName, MessageBody)), - ).Return( - &http.Response{ - StatusCode: http.StatusBadRequest, - Body: io.NopCloser(strings.NewReader("bad request")), - }, - nil, - ).Once() - httpClient.On( - "Do", - mock.MatchedBy(httpRequestMatcher(t, QueueName, "", "3", ContentType, FunctionName, MessageBody)), - ).Return( - &http.Response{ - StatusCode: http.StatusForbidden, - Body: io.NopCloser(strings.NewReader("not authorized")), - }, - nil, - ).Once() - - // Mock a queue message with "input" as the message body - message := new(azureMessageMock) - message.On("Bytes").Return([]byte(MessageBody)) - message.On( - "Delete", - mock.MatchedBy( - func(options *storage.QueueServiceOptions) bool { - return options == nil - }, - ), - ).Return(nil) - - // Mock a queue that performs a no-op create, returns a "poison" message, and then returns no more messages - queue := new(azureQueueMock) - queue.On( - "Create", - mock.MatchedBy( - func(options *storage.QueueServiceOptions) bool { - return options == nil - }, - ), - ).Return(nil) - queue.On( - "GetMessages", - mock.MatchedBy( - func(options *storage.GetMessagesOptions) bool { - return options.NumOfMessages == AzureMessageFetchCount && - options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second) - }, - ), - ).Return([]AzureMessage{message}, nil).Once() - queue.On( - "GetMessages", - mock.MatchedBy( - func(options *storage.GetMessagesOptions) bool { - return options.NumOfMessages == AzureMessageFetchCount && - options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second) - }, - ), - ).Return([]AzureMessage{}, nil) - - // Mock a poison queue message that performs a no-op Put - poisonMessage := new(azureMessageMock) - poisonMessage.On( - "Put", - mock.MatchedBy( - func(options *storage.PutMessageOptions) bool { - return options == nil - }, - ), - ).Return(nil) - - // Mock a poison queue that performs a no-op create and creates a new message - poisonQueue := new(azureQueueMock) - poisonQueue.On( - "Create", - mock.MatchedBy( - func(options *storage.QueueServiceOptions) bool { - return options == nil - }, - ), - ).Return(nil) - poisonQueue.On("NewMessage", MessageBody).Return(poisonMessage).Once() - - // Mock the queue service to return the input queue - service := new(azureQueueServiceMock) - service.On("GetQueue", QueueName).Return(queue).Once() - service.On("GetQueue", QueueName+AzurePoisonQueueSuffix).Return(poisonQueue).Once() - - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - panicIf(err) - - // Create the storage connection and subscribe to the trigger - connection := AzureStorageConnection{ - logger: logger, - routerURL: DummyRouterURL, - service: service, - httpClient: httpClient, - } - subscription, err := connection.Subscribe(&fv1.MessageQueueTrigger{ - ObjectMeta: metav1.ObjectMeta{ - Name: TriggerName, - Namespace: metav1.NamespaceDefault, - }, - Spec: fv1.MessageQueueTriggerSpec{ - FunctionReference: fv1.FunctionReference{ - Type: fv1.FunctionReferenceTypeFunctionName, - Name: FunctionName, - }, - MessageQueueType: fv1.MessageQueueTypeASQ, - Topic: QueueName, - ContentType: ContentType, - }, - }) - require.NoError(t, err) - require.NotNil(t, subscription) - - panicIf(connection.Unsubscribe(subscription)) - - mock.AssertExpectationsForObjects(t, httpClient, message, poisonMessage, queue, poisonQueue, service) -} - -func httpRequestMatcher(t *testing.T, queue string, responseQueue string, retry string, contentType string, functionName string, body string) func(*http.Request) bool { - expectedURL := fmt.Sprintf("%s/fission-function/%s", DummyRouterURL, functionName) - return func(req *http.Request) bool { - requestBody, err := io.ReadAll(req.Body) - require.NoError(t, err) - - req.Body = io.NopCloser(strings.NewReader(string(requestBody))) - - return queue == req.Header.Get("X-Fission-MQTrigger-Topic") && - responseQueue == req.Header.Get("X-Fission-MQTrigger-RespTopic") && - retry == req.Header.Get("X-Fission-MQTrigger-RetryCount") && - contentType == req.Header.Get("Content-Type") && - req.URL.String() == expectedURL && - string(requestBody) == body - } -} - -func runAzureStorageQueueTest(t *testing.T, count int, output bool) { - const ( - TriggerName = "queuetrigger" - QueueName = "inputqueue" - OutputQueueName = "outputqueue" - MessageBody = "input" - FunctionName = "testfunc" - FunctionResponse = "output" - ContentType = "text/plain" - ) - - responseTopic := "" - if output { - responseTopic = OutputQueueName - } - - // Mock a HTTP client that returns http.StatusOK with "output" for the body - httpClient := new(azureHTTPClientMock) - httpClient.bodyHandler = func(res *http.Response) { - res.Body = io.NopCloser(strings.NewReader(FunctionResponse)) - } - httpClient.On( - "Do", - mock.MatchedBy(httpRequestMatcher(t, QueueName, responseTopic, "", ContentType, FunctionName, MessageBody)), - ).Return(&http.Response{StatusCode: http.StatusOK}, nil).Times(count) - - // Mock a queue message with "input" as the message body - message := new(azureMessageMock) - message.On("Bytes").Return([]byte(MessageBody)).Times(count) - message.On( - "Delete", - mock.MatchedBy( - func(options *storage.QueueServiceOptions) bool { - return options == nil - }, - ), - ).Return(nil).Times(count) - - // Mock a queue that performs a no-op create, returns a message the specified number of times, and then returns no more messages - queue := new(azureQueueMock) - queue.On( - "Create", - mock.MatchedBy( - func(options *storage.QueueServiceOptions) bool { - return options == nil - }, - ), - ).Return(nil) - queue.On( - "GetMessages", - mock.MatchedBy( - func(options *storage.GetMessagesOptions) bool { - return options.NumOfMessages == AzureMessageFetchCount && - options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second) - }, - ), - ).Return([]AzureMessage{message}, nil).Times(count) - queue.On( - "GetMessages", - mock.MatchedBy( - func(options *storage.GetMessagesOptions) bool { - return options.NumOfMessages == AzureMessageFetchCount && - options.VisibilityTimeout == int(AzureMessageVisibilityTimeout/time.Second) - }, - ), - ).Return([]AzureMessage{}, nil) - - // Mock the output queue if needed - outputMessage := new(azureMessageMock) - outputQueue := new(azureQueueMock) - if output { - outputMessage.On( - "Put", - mock.MatchedBy( - func(options *storage.PutMessageOptions) bool { - return options == nil - }, - ), - ).Return(nil).Times(count) - - outputQueue.On( - "Create", - mock.MatchedBy( - func(options *storage.QueueServiceOptions) bool { - return options == nil - }, - ), - ).Return(nil).Times(count) - outputQueue.On("NewMessage", FunctionResponse).Return(outputMessage).Times(count) - } - - // Mock the queue service to return the input queue - service := new(azureQueueServiceMock) - service.On("GetQueue", QueueName).Return(queue).Once() - if output { - service.On("GetQueue", OutputQueueName).Return(outputQueue).Times(count) - } - - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - panicIf(err) - - // Create the storage connection and subscribe to the trigger - connection := AzureStorageConnection{ - logger: logger, - routerURL: DummyRouterURL, - service: service, - httpClient: httpClient, - } - subscription, err := connection.Subscribe(&fv1.MessageQueueTrigger{ - ObjectMeta: metav1.ObjectMeta{ - Name: TriggerName, - Namespace: metav1.NamespaceDefault, - }, - Spec: fv1.MessageQueueTriggerSpec{ - FunctionReference: fv1.FunctionReference{ - Type: fv1.FunctionReferenceTypeFunctionName, - Name: FunctionName, - }, - MessageQueueType: fv1.MessageQueueTypeASQ, - Topic: QueueName, - ResponseTopic: responseTopic, - ContentType: ContentType, - }, - }) - require.NoError(t, err) - require.NotNil(t, subscription) - - panicIf(connection.Unsubscribe(subscription)) - - mock.AssertExpectationsForObjects(t, httpClient, message, outputMessage, queue, outputQueue, service) -}