Remove deprecated Fission Azure Storage Queue connector (#2404)
We are removing Fission deprecated Azure Storage Queue connector and planning to adopt Keda going forward to have better delegated functionality and more rich support. Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
@@ -76,7 +76,6 @@ const (
|
||||
)
|
||||
|
||||
const (
|
||||
MessageQueueTypeASQ = "azure-storage-queue"
|
||||
MessageQueueTypeKafka = "kafka"
|
||||
)
|
||||
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
Reference in New Issue
Block a user