Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

use zap for logging (#1112)

Browse Source 
Use zap for logging
This commit is contained in:
Jon Carl
2019-03-14 21:11:15 +05:30
committed by Vishal
parent c717f182b6
commit 0fc864f230
98 changed files with 2031 additions and 1223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mqtrigger/messageQueue/asq.go
+44 -27
View File
@@ -19,7 +19,6 @@ package messageQueue
import (
"bytes"
"encoding/base64"
"errors"
"fmt"
"io/ioutil"
"net/http"
@@ -29,12 +28,12 @@ import (
"sync"
"time"
"github.com/Azure/azure-sdk-for-go/storage"
"github.com/pkg/errors"
"go.uber.org/zap"
"github.com/fission/fission"
"github.com/fission/fission/crd"
log "github.com/sirupsen/logrus"
"github.com/Azure/azure-sdk-for-go/storage"
)
// TODO: some of these constants should probably be environment variables
@@ -55,6 +54,7 @@ const (
// AzureStorageConnection represents an Azure storage connection.
type AzureStorageConnection struct {
logger *zap.Logger
routerURL string
service AzureQueueService
httpClient AzureHTTPClient
@@ -173,7 +173,7 @@ func newAzureQueueService(client storage.Client) AzureQueueService {
}
}
func newAzureStorageConnection(routerURL string, config MessageQueueConfig) (MessageQueue, error) {
func newAzureStorageConnection(logger *zap.Logger, routerURL string, config MessageQueueConfig) (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")
@@ -184,13 +184,14 @@ func newAzureStorageConnection(routerURL string, config MessageQueueConfig) (Mes
return nil, errors.New("Required environment variable 'AZURE_STORAGE_ACCOUNT_KEY' is not set")
}
log.Infof("Creating Azure storage connection to storage account '%s'.", account)
logger.Info("creating Azure storage connection to storage account", zap.String("account", account))
client, err := storage.NewBasicClient(account, key)
if err != nil {
return nil, fmt.Errorf("Failed to Azure create storage client: %v", err)
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{
@@ -200,10 +201,10 @@ func newAzureStorageConnection(routerURL string, config MessageQueueConfig) (Mes
}
func (asc AzureStorageConnection) subscribe(trigger *crd.MessageQueueTrigger) (messageQueueSubscription, error) {
log.Infof("Subscribing to Azure storage queue '%s'.", trigger.Spec.Topic)
asc.logger.Info("subscribing to Azure storage queue", zap.String("queue", trigger.Spec.Topic))
if trigger.Spec.FunctionReference.Type != fission.FunctionReferenceTypeFunctionName {
return nil, fmt.Errorf("Unsupported function reference type (%v) for trigger %v", trigger.Spec.FunctionReference.Type, trigger.Metadata.Name)
return nil, fmt.Errorf("unsupported function reference type (%v) for trigger %q", trigger.Spec.FunctionReference.Type, trigger.Metadata.Name)
}
subscription := &AzureQueueSubscription{
@@ -226,7 +227,7 @@ func (asc AzureStorageConnection) subscribe(trigger *crd.MessageQueueTrigger) (m
func (asc AzureStorageConnection) unsubscribe(subscription messageQueueSubscription) error {
sub := subscription.(*AzureQueueSubscription)
log.Infof("Unsubscribing from Azure storage queue '%s'.", sub.queueName)
asc.logger.Info("unsubscribing from Azure storage queue", zap.String("queue", sub.queueName))
// Let the worker know we've unsubscribed
sub.unsubscribe <- true
@@ -245,7 +246,7 @@ func runAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubsc
timer := time.NewTimer(AzureQueuePollingInterval)
for {
log.Infof("Waiting for %v before polling Azure storage queue '%s'.", AzureQueuePollingInterval, sub.queueName)
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()
@@ -261,18 +262,18 @@ func runAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubsc
}
func pollAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubscription, wg *sync.WaitGroup) {
log.Infof("Polling messages for Azure storage queue '%s'.", sub.queueName)
conn.logger.Info("polling for messages from Azure storage queue", zap.String("queue", sub.queueName))
err := sub.queue.Create(nil)
if err != nil {
log.Errorf("Failed to create message queue '%s': %v", sub.queueName, err)
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 {
log.Errorf("Failed to create message queue '%s': %v", sub.queueName, err)
conn.logger.Error("failed to create message queue", zap.Error(err), zap.String("queue", sub.queueName))
return
}
@@ -281,7 +282,7 @@ func pollAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubs
VisibilityTimeout: int(AzureMessageVisibilityTimeout / time.Second),
})
if err != nil {
log.Errorf("Failed to retrieve messages from Azure storage queue '%s': %v", sub.queueName, err)
conn.logger.Error("failed to retrieve messages from Azure storage queue", zap.Error(err), zap.String("queue", sub.queueName))
break
}
if len(messages) == 0 {
@@ -301,15 +302,15 @@ func pollAzureQueueSubscription(conn AzureStorageConnection, sub *AzureQueueSubs
func invokeTriggeredFunction(conn AzureStorageConnection, sub *AzureQueueSubscription, message AzureMessage) {
defer message.Delete(nil)
log.Printf("Making HTTP request to %s.", sub.functionURL)
conn.logger.Info("making HTTP request to invoke function", zap.String("function_url", sub.functionURL))
for i := 0; i <= AzureQueueRetryLimit; i++ {
if i > 0 {
log.Infof("Retry #%d for request to %s.", i, sub.functionURL)
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 {
log.Errorf("Failed to create HTTP request to %s: %v", sub.functionURL, err)
conn.logger.Error("failed to create HTTP request to invoke function", zap.Error(err), zap.String("function_url", sub.functionURL))
continue
}
@@ -324,19 +325,22 @@ func invokeTriggeredFunction(conn AzureStorageConnection, sub *AzureQueueSubscri
response, err := conn.httpClient.Do(request)
if err != nil {
log.Errorf("Request to %s failed: %v", sub.functionURL, err)
conn.logger.Error("sending function invocation request failed", zap.Error(err), zap.String("function_url", sub.functionURL))
continue
}
defer response.Body.Close()
body, err := ioutil.ReadAll(response.Body)
if err != nil {
log.Errorf("Failed to read response body from %s: %v.", sub.functionURL, err)
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 {
log.Printf("Request to %s returned failure: %s (%d).", sub.functionURL, string(body), response.StatusCode)
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
}
@@ -344,14 +348,19 @@ func invokeTriggeredFunction(conn AzureStorageConnection, sub *AzureQueueSubscri
outputQueue := conn.service.GetQueue(sub.outputQueueName)
err = outputQueue.Create(nil)
if err != nil {
log.Errorf("Failed to create output queue '%s': %v.", sub.outputQueueName, err)
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 {
log.Errorf("Failed to post response body from %s to output queue '%s': %v.", sub.functionURL, sub.outputQueueName, err)
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
}
}
@@ -360,20 +369,28 @@ func invokeTriggeredFunction(conn AzureStorageConnection, sub *AzureQueueSubscri
return
}
log.Errorf("Request to %s failed after %d retries; moving message to poison queue.", sub.functionURL, AzureQueueRetryLimit)
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 {
log.Errorf("Failed to create poison queue '%s': %v", poisonQueueName, err)
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 {
log.Errorf("Failed to post response body from %s to output queue '%s': %v", sub.functionURL, poisonQueueName, err)
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
}
}
mqtrigger/messageQueue/asq_test.go
+32 -8
View File
@@ -18,26 +18,33 @@ package messageQueue
import (
"fmt"
"io/ioutil"
"log"
"net/http"
"os"
"strings"
"testing"
"time"
"github.com/fission/fission"
"github.com/fission/fission/crd"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/Azure/azure-sdk-for-go/storage"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission"
"github.com/fission/fission/crd"
)
const (
DummyRouterURL = "http://localhost"
)
func panicIf(err error) {
if err != nil {
log.Panicf("Error: %v", err)
}
}
type azureQueueServiceMock struct {
mock.Mock
}
@@ -103,7 +110,10 @@ func (m *azureHTTPClientMock) Do(req *http.Request) (*http.Response, error) {
}
func TestNewStorageConnectionMissingAccountName(t *testing.T) {
connection, err := newAzureStorageConnection(DummyRouterURL, MessageQueueConfig{
logger, err := zap.NewDevelopment()
panicIf(err)
connection, err := newAzureStorageConnection(logger, DummyRouterURL, MessageQueueConfig{
MQType: fission.MessageQueueTypeASQ,
Url: "",
})
@@ -112,8 +122,11 @@ func TestNewStorageConnectionMissingAccountName(t *testing.T) {
}
func TestNewStorageConnectionMissingAccessKey(t *testing.T) {
logger, err := zap.NewDevelopment()
panicIf(err)
_ = os.Setenv("AZURE_STORAGE_ACCOUNT_NAME", "accountname")
connection, err := newAzureStorageConnection(DummyRouterURL, MessageQueueConfig{
connection, err := newAzureStorageConnection(logger, DummyRouterURL, MessageQueueConfig{
MQType: fission.MessageQueueTypeASQ,
Url: "",
})
@@ -123,9 +136,12 @@ func TestNewStorageConnectionMissingAccessKey(t *testing.T) {
}
func TestNewStorageConnection(t *testing.T) {
logger, err := zap.NewDevelopment()
panicIf(err)
_ = os.Setenv("AZURE_STORAGE_ACCOUNT_NAME", "accountname")
_ = os.Setenv("AZURE_STORAGE_ACCOUNT_KEY", "bm90IGEga2V5")
connection, err := newAzureStorageConnection(DummyRouterURL, MessageQueueConfig{
connection, err := newAzureStorageConnection(logger, DummyRouterURL, MessageQueueConfig{
MQType: "azure-storage-queue",
Url: "",
})
@@ -276,8 +292,12 @@ func TestAzureStorageQueuePoisonMessage(t *testing.T) {
service.On("GetQueue", QueueName).Return(queue).Once()
service.On("GetQueue", QueueName+AzurePoisonQueueSuffix).Return(poisonQueue).Once()
logger, err := zap.NewDevelopment()
panicIf(err)
// Create the storage connection and subscribe to the trigger
connection := AzureStorageConnection{
logger: logger,
routerURL: DummyRouterURL,
service: service,
httpClient: httpClient,
@@ -420,8 +440,12 @@ func runAzureStorageQueueTest(t *testing.T, count int, output bool) {
service.On("GetQueue", OutputQueueName).Return(outputQueue).Times(count)
}
logger, err := zap.NewDevelopment()
panicIf(err)
// Create the storage connection and subscribe to the trigger
connection := AzureStorageConnection{
logger: logger,
routerURL: DummyRouterURL,
service: service,
httpClient: httpClient,
mqtrigger/messageQueue/kafka.go
+51 -27
View File
@@ -26,7 +26,7 @@ import (
sarama "github.com/Shopify/sarama"
cluster "github.com/bsm/sarama-cluster"
log "github.com/sirupsen/logrus"
"go.uber.org/zap"
"github.com/fission/fission"
"github.com/fission/fission/crd"
@@ -34,30 +34,35 @@ import (
type (
Kafka struct {
logger *zap.Logger
routerUrl string
brokers []string
version sarama.KafkaVersion
}
)
func makeKafkaMessageQueue(routerUrl string, mqCfg MessageQueueConfig) (MessageQueue, error) {
func makeKafkaMessageQueue(logger *zap.Logger, routerUrl string, mqCfg MessageQueueConfig) (MessageQueue, error) {
if len(routerUrl) == 0 || len(mqCfg.Url) == 0 {
return nil, errors.New("The router URL or MQ URL is empty")
return nil, errors.New("the router URL or MQ URL is empty")
}
mqKafkaVersion := os.Getenv("MESSAGE_QUEUE_KAFKA_VERSION")
// Parse version string
kafkaVersion, err := sarama.ParseKafkaVersion(mqKafkaVersion)
if err != nil {
log.Warningf("Error parsing version string %q: %v. Falling back to %q", mqKafkaVersion, err, kafkaVersion)
logger.Warn("error parsing kafka version string - falling back to default",
zap.Error(err),
zap.String("failed_version", mqKafkaVersion),
zap.Any("default_version", kafkaVersion))
}
kafka := Kafka{
logger: logger.Named("kafka"),
routerUrl: routerUrl,
brokers: strings.Split(mqCfg.Url, ","),
version: kafkaVersion,
}
log.Infof("Created Queue %v", kafka)
logger.Info("created kafka queue", zap.Any("kafka", kafka))
return kafka, nil
}
@@ -66,8 +71,8 @@ func isTopicValidForKafka(topic string) bool {
}
func (kafka Kafka) subscribe(trigger *crd.MessageQueueTrigger) (messageQueueSubscription, error) {
log.Infof("Inside kakfa subscribe %q", trigger)
log.Infof("brokers set to %q", kafka.brokers)
kafka.logger.Info("inside kakfa subscribe", zap.Any("trigger", trigger))
kafka.logger.Info("brokers set", zap.Strings("brokers", kafka.brokers))
// Create new consumer
consumerConfig := cluster.NewConfig()
@@ -75,7 +80,7 @@ func (kafka Kafka) subscribe(trigger *crd.MessageQueueTrigger) (messageQueueSubs
consumerConfig.Group.Return.Notifications = true
consumerConfig.Config.Version = kafka.version
consumer, err := cluster.NewConsumer(kafka.brokers, string(trigger.Metadata.UID), []string{trigger.Spec.Topic}, consumerConfig)
log.Infof("Created a new consumer: %#v", consumer)
kafka.logger.Info("created a new consumer", zap.Any("consumer", consumer))
if err != nil {
panic(err)
}
@@ -87,7 +92,7 @@ func (kafka Kafka) subscribe(trigger *crd.MessageQueueTrigger) (messageQueueSubs
producerConfig.Producer.Return.Successes = true
producerConfig.Version = kafka.version
producer, err := sarama.NewSyncProducer(kafka.brokers, producerConfig)
log.Infof("Created a new producer %q", producer)
kafka.logger.Info("created a new producer", zap.Any("consumer", producer))
if err != nil {
panic(err)
}
@@ -95,21 +100,21 @@ func (kafka Kafka) subscribe(trigger *crd.MessageQueueTrigger) (messageQueueSubs
// consume errors
go func() {
for err := range consumer.Errors() {
log.Printf("Error: %s\n", err.Error())
kafka.logger.Error("consumer error", zap.Error(err))
}
}()
// consume notifications
go func() {
for ntf := range consumer.Notifications() {
log.Printf("Rebalanced: %+v\n", ntf)
kafka.logger.Info("consumer notification", zap.Any("notification", ntf))
}
}()
// consume messages
go func() {
for msg := range consumer.Messages() {
log.Infof("Calling message handler with value " + string(msg.Value[:]))
kafka.logger.Info("calling message handler", zap.String("message", string(msg.Value[:])))
if kafkaMsgHandler(&kafka, producer, trigger, msg) {
consumer.MarkOffset(msg, "") // mark message as processed
}
@@ -127,12 +132,13 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
var value string = string(msg.Value[:])
// Support other function ref types
if trigger.Spec.FunctionReference.Type != fission.FunctionReferenceTypeFunctionName {
log.Fatalf("Unsupported function reference type (%v) for trigger %v",
trigger.Spec.FunctionReference.Type, trigger.Metadata.Name)
kafka.logger.Fatal("unsupported function reference type for trigger",
zap.Any("function_reference_type", trigger.Spec.FunctionReference.Type),
zap.String("trigger", trigger.Metadata.Name))
}
url := kafka.routerUrl + "/" + strings.TrimPrefix(fission.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.Metadata.Namespace), "/")
log.Printf("Making HTTP request to %v", url)
kafka.logger.Info("making HTTP request", zap.String("url", url))
// Generate the Headers
fissionHeaders := map[string]string{
@@ -145,7 +151,9 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
// Create request
req, err := http.NewRequest("POST", url, strings.NewReader(value))
if err != nil {
log.Warningf("Request creation failed: %v", url)
kafka.logger.Error("failed to create HTTP request to invoke function",
zap.Error(err),
zap.String("function_url", url))
return false
}
@@ -156,7 +164,8 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
req.Header.Add(string(h.Key), string(h.Value))
}
} else {
log.Warningf("Headers are not supported by Kafka version %q, needs v0.11+: no record headers to add in HTTP request", kafka.version)
kafka.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
zap.Any("current_version", kafka.version))
}
for k, v := range fissionHeaders {
@@ -169,7 +178,10 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
// Make the request
resp, err = http.DefaultClient.Do(req)
if err != nil {
log.Errorf("Error invoking function for trigger %v: %v", trigger.Metadata.Name, err)
kafka.logger.Error("sending function invocation request failed",
zap.Error(err),
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name))
continue
}
if resp == nil {
@@ -182,18 +194,23 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
}
if resp == nil {
log.Warning("Every retry failed; final retry gave empty response.")
kafka.logger.Warn("every function invocation retry failed; final retry gave empty response",
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name))
return false
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
log.Infof("Got response " + string(body))
kafka.logger.Info("got response from function invocation",
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name),
zap.String("body", string(body)))
if err != nil {
errorHandler(trigger, producer, fmt.Sprintf("Request body error: %v", string(body)))
errorHandler(kafka.logger, trigger, producer, fmt.Sprintf("request body error: %v", string(body)))
return false
}
if resp.StatusCode != 200 {
errorHandler(trigger, producer, fmt.Sprintf("Request returned failure: %v", resp.StatusCode))
errorHandler(kafka.logger, trigger, producer, fmt.Sprintf("request returned failure: %v", resp.StatusCode))
return false
}
if len(trigger.Spec.ResponseTopic) > 0 {
@@ -207,7 +224,8 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
}
}
} else {
log.Warningf("Headers are not supported by Kafka version %q, needs v0.11+: dropping the headers", kafka.version)
kafka.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
zap.Any("current_version", kafka.version))
}
_, _, err := producer.SendMessage(&sarama.ProducerMessage{
@@ -216,24 +234,30 @@ func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *crd.Me
Headers: kafkaRecordHeaders,
})
if err != nil {
log.Warningf("Failed to publish message to topic %s: %v", trigger.Spec.ResponseTopic, err)
kafka.logger.Warn("failed to publish response body from function invocation to topic",
zap.Error(err),
zap.String("topic", trigger.Spec.Topic),
zap.String("function_url", url))
return false
}
}
return true
}
func errorHandler(trigger *crd.MessageQueueTrigger, producer sarama.SyncProducer, body string) {
func errorHandler(logger *zap.Logger, trigger *crd.MessageQueueTrigger, producer sarama.SyncProducer, body string) {
if len(trigger.Spec.ErrorTopic) > 0 {
_, _, err := producer.SendMessage(&sarama.ProducerMessage{
Topic: trigger.Spec.ErrorTopic,
Value: sarama.StringEncoder(body),
})
if err != nil {
log.Warningf("Failed to publish message to error topic %s: %v", trigger.Spec.ErrorTopic, err)
logger.Warn("failed to publish message to error topic",
zap.Error(err),
zap.String("topic", trigger.Spec.Topic))
return
}
} else {
log.Printf(body)
logger.Error("message received to publish to error topic, but no error topic was set",
zap.String("message", body))
}
}
mqtrigger/messageQueue/messageQueue.go
+18 -15
View File
@@ -18,9 +18,10 @@ package messageQueue
import (
"errors"
"fmt"
"time"
log "github.com/sirupsen/logrus"
"go.uber.org/zap"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission"
@@ -50,6 +51,7 @@ type (
}
MessageQueueTriggerManager struct {
logger *zap.Logger
reqChan chan request
mqCfg MessageQueueConfig
triggers map[string]*triggerSubscription
@@ -73,27 +75,28 @@ type (
}
)
func MakeMessageQueueTriggerManager(fissionClient *crd.FissionClient, routerUrl string, mqConfig MessageQueueConfig) *MessageQueueTriggerManager {
func MakeMessageQueueTriggerManager(logger *zap.Logger, fissionClient *crd.FissionClient, routerUrl string, mqConfig MessageQueueConfig) *MessageQueueTriggerManager {
var messageQueue MessageQueue
var err error
mqTriggerMgr := MessageQueueTriggerManager{
logger: logger.Named("message_queue_trigger_manager"),
reqChan: make(chan request),
triggers: make(map[string]*triggerSubscription),
fissionClient: fissionClient,
}
switch mqConfig.MQType {
case fission.MessageQueueTypeNats:
messageQueue, err = makeNatsMessageQueue(routerUrl, mqConfig)
messageQueue, err = makeNatsMessageQueue(logger, routerUrl, mqConfig)
case fission.MessageQueueTypeASQ:
messageQueue, err = newAzureStorageConnection(routerUrl, mqConfig)
messageQueue, err = newAzureStorageConnection(logger, routerUrl, mqConfig)
case fission.MessageQueueTypeKafka:
messageQueue, err = makeKafkaMessageQueue(routerUrl, mqConfig)
messageQueue, err = makeKafkaMessageQueue(logger, routerUrl, mqConfig)
default:
err = errors.New("No matched message queue type found")
err = fmt.Errorf("no supported message queue type found for %q", mqConfig.MQType)
}
if err != nil {
log.Fatalf("Failed to connect to remote message queue server: %v", err)
logger.Fatal("failed to connect to remote message queue server", zap.Error(err))
}
mqTriggerMgr.messageQueue = messageQueue
go mqTriggerMgr.service()
@@ -109,7 +112,7 @@ func (mqt *MessageQueueTriggerManager) service() {
var err error
k := crd.CacheKey(&req.triggerSub.trigger.Metadata)
if _, ok := mqt.triggers[k]; ok {
err = errors.New("Trigger already exists")
err = errors.New("trigger already exists")
} else {
mqt.triggers[k] = req.triggerSub
}
@@ -164,11 +167,11 @@ func (mqt *MessageQueueTriggerManager) syncTriggers() {
newTriggers, err := mqt.fissionClient.MessageQueueTriggers(metav1.NamespaceAll).List(metav1.ListOptions{})
if err != nil {
if fission.IsNetworkError(err) {
log.Printf("Encounter network error, retry again: %v", err)
mqt.logger.Info("encountered network error, will retry", zap.Error(err))
time.Sleep(5 * time.Second)
continue
}
log.Fatalf("Failed to read message queue trigger list: %v", err)
mqt.logger.Fatal("failed to read message queue trigger list", zap.Error(err))
}
newTriggerMap := make(map[string]*crd.MessageQueueTrigger)
for index := range newTriggers.Items {
@@ -188,7 +191,7 @@ func (mqt *MessageQueueTriggerManager) syncTriggers() {
// actually subscribe using the message queue client impl
sub, err := mqt.messageQueue.subscribe(trigger)
if err != nil {
log.Warnf("Failed to subscribe to message queue trigger %s: %v", trigger.Metadata.Name, err)
mqt.logger.Warn("failed to subscribe to message queue trigger", zap.Error(err), zap.String("trigger_name", trigger.Metadata.Name))
continue
}
@@ -200,10 +203,10 @@ func (mqt *MessageQueueTriggerManager) syncTriggers() {
// add to our list
err = mqt.addTrigger(&triggerSub)
if err != nil {
log.Fatalf("Message queue trigger %s addition failed: %v", trigger.Metadata.Name, err)
mqt.logger.Fatal("adding message queue trigger failed", zap.Error(err), zap.String("trigger_name", trigger.Metadata.Name))
}
log.Infof("Message queue trigger %s created", trigger.Metadata.Name)
mqt.logger.Info("message queue trigger created", zap.String("trigger_name", trigger.Metadata.Name))
}
// remove old triggers
@@ -213,11 +216,11 @@ func (mqt *MessageQueueTriggerManager) syncTriggers() {
}
err := mqt.messageQueue.unsubscribe(triggerSub.subscription)
if err != nil {
log.Warnf("Failed to unsubscribe to trigger %s: %v", triggerSub.trigger.Metadata.Name, err)
mqt.logger.Warn("failed to unsubscribe from message queue trigger", zap.Error(err), zap.String("trigger_name", triggerSub.trigger.Metadata.Name))
continue
}
mqt.delTrigger(&triggerSub.trigger.Metadata)
log.Infof("Message queue trigger %s deleted", triggerSub.trigger.Metadata.Name)
mqt.logger.Info("message queue trigger deleted", zap.String("trigger_name", triggerSub.trigger.Metadata.Name))
}
// TODO replace with a watch
mqtrigger/messageQueue/nats.go
+36 -14
View File
@@ -18,7 +18,6 @@ package messageQueue
import (
"bytes"
"errors"
"fmt"
"io/ioutil"
"net/http"
@@ -26,7 +25,7 @@ import (
ns "github.com/nats-io/go-nats-streaming"
nsUtil "github.com/nats-io/nats-streaming-server/util"
log "github.com/sirupsen/logrus"
"go.uber.org/zap"
"github.com/fission/fission"
"github.com/fission/fission/crd"
@@ -41,17 +40,19 @@ const (
type (
Nats struct {
logger *zap.Logger
nsConn ns.Conn
routerUrl string
}
)
func makeNatsMessageQueue(routerUrl string, mqCfg MessageQueueConfig) (MessageQueue, error) {
func makeNatsMessageQueue(logger *zap.Logger, routerUrl string, mqCfg MessageQueueConfig) (MessageQueue, error) {
conn, err := ns.Connect(natsClusterID, natsClientID, ns.NatsURL(mqCfg.Url))
if err != nil {
return nil, err
}
nats := Nats{
logger: logger.Named("nats"),
nsConn: conn,
routerUrl: routerUrl,
}
@@ -62,7 +63,7 @@ func (nats Nats) subscribe(trigger *crd.MessageQueueTrigger) (messageQueueSubscr
subj := trigger.Spec.Topic
if !isTopicValidForNats(subj) {
return nil, errors.New(fmt.Sprintf("Not a valid topic: %s", trigger.Spec.Topic))
return nil, fmt.Errorf("not a valid topic: %q", trigger.Spec.Topic)
}
opts := []ns.SubscriptionOption{
@@ -96,15 +97,16 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
// Support other function ref types
if trigger.Spec.FunctionReference.Type != fission.FunctionReferenceTypeFunctionName {
log.Fatalf("Unsupported function reference type (%v) for trigger %v",
trigger.Spec.FunctionReference.Type, trigger.Metadata.Name)
nats.logger.Fatal("unsupported function reference type for trigger",
zap.Any("function_reference_type", trigger.Spec.FunctionReference.Type),
zap.String("trigger", trigger.Metadata.Name))
}
// 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.
url := nats.routerUrl + "/" + strings.TrimPrefix(fission.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.Metadata.Namespace), "/")
log.Printf("Making HTTP request to %v", url)
nats.logger.Info("making HTTP request", zap.String("url", url))
headers := map[string]string{
"X-Fission-MQTrigger-Topic": trigger.Spec.Topic,
@@ -117,7 +119,9 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
req, err := http.NewRequest("POST", url, bytes.NewReader(msg.Data))
if err != nil {
log.Errorf("Could not issue POST request with message to url %v", url)
nats.logger.Error("failed to create HTTP request to invoke function",
zap.Error(err),
zap.String("function_url", url))
return
}
@@ -130,7 +134,10 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
// Make the request
resp, err = http.DefaultClient.Do(req)
if err != nil {
log.Errorf("Error invoking function for trigger %v: %v", trigger.Metadata.Name, err)
nats.logger.Error("sending function invocation request failed",
zap.Error(err),
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name))
continue
}
if resp == nil {
@@ -143,7 +150,9 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
}
if resp == nil {
log.Warning("Every retry failed; final retry gave empty response.")
nats.logger.Warn("every function invocation retry failed; final retry gave empty response",
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name))
return
}
@@ -151,7 +160,10 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
body, bodyErr := ioutil.ReadAll(resp.Body)
if bodyErr != nil {
log.Warningf("Response body error: %v", bodyErr)
nats.logger.Error("error reading function invocation response",
zap.Error(err),
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name))
return
}
@@ -160,7 +172,11 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
if len(trigger.Spec.ErrorTopic) > 0 && len(body) > 0 {
publishErr := nats.nsConn.Publish(trigger.Spec.ErrorTopic, body)
if publishErr != nil {
log.Errorf("Failed to publish error to error topic: %v", publishErr)
nats.logger.Error("failed to publish function invocation error to error topic",
zap.Error(publishErr),
zap.String("topic", trigger.Spec.ErrorTopic),
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name))
// TODO: We will ack this message after max retries to prevent re-processing but
// this may cause message loss
}
@@ -171,13 +187,19 @@ func msgHandler(nats *Nats, trigger *crd.MessageQueueTrigger) func(*ns.Msg) {
// Trigger acks message only if a request was processed successfully
err = msg.Ack()
if err != nil {
log.Warningf("Failed to ack message: %v", err)
nats.logger.Error("failed to ack message after successful function invocation from trigger",
zap.Error(err),
zap.String("function_url", url),
zap.String("trigger", trigger.Metadata.Name))
}
if len(trigger.Spec.ResponseTopic) > 0 {
err = nats.nsConn.Publish(trigger.Spec.ResponseTopic, body)
if err != nil {
log.Warningf("Failed to publish message to topic %s: %v", trigger.Spec.ResponseTopic, err)
nats.logger.Error("failed to publish message with function invocation response to topic",
zap.Error(err),
zap.String("topic", trigger.Spec.ResponseTopic),
zap.String("trigger", trigger.Metadata.Name))
}
}
}