forked from flyteorg/flyte
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement GCP notifications processor (flyteorg#259)
- Loading branch information
1 parent
6891d9f
commit bca6228
Showing
8 changed files
with
287 additions
and
62 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| package implementations | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/NYTimes/gizmo/pubsub" | ||
| "github.com/flyteorg/flyteadmin/pkg/async" | ||
| "github.com/flyteorg/flyteadmin/pkg/async/notifications/interfaces" | ||
| "github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin" | ||
| "github.com/flyteorg/flytestdlib/logger" | ||
| "github.com/flyteorg/flytestdlib/promutils" | ||
| "github.com/golang/protobuf/proto" | ||
| ) | ||
|
|
||
| // TODO: Add a counter that encompasses the publisher stats grouped by project and domain. | ||
| type GcpProcessor struct { | ||
| sub pubsub.Subscriber | ||
| email interfaces.Emailer | ||
| systemMetrics processorSystemMetrics | ||
| } | ||
|
|
||
| func NewGcpProcessor(sub pubsub.Subscriber, emailer interfaces.Emailer, scope promutils.Scope) interfaces.Processor { | ||
| return &GcpProcessor{ | ||
| sub: sub, | ||
| email: emailer, | ||
| systemMetrics: newProcessorSystemMetrics(scope.NewSubScope("gcp_processor")), | ||
| } | ||
| } | ||
|
|
||
| func (p *GcpProcessor) StartProcessing() { | ||
| for { | ||
| logger.Warningf(context.Background(), "Starting GCP notifications processor") | ||
| err := p.run() | ||
| logger.Errorf(context.Background(), "error with running GCP processor err: [%v] ", err) | ||
| time.Sleep(async.RetryDelay) | ||
| } | ||
| } | ||
|
|
||
| func (p *GcpProcessor) run() error { | ||
| var emailMessage admin.EmailMessage | ||
|
|
||
| for msg := range p.sub.Start() { | ||
| p.systemMetrics.MessageTotal.Inc() | ||
|
|
||
| if err := proto.Unmarshal(msg.Message(), &emailMessage); err != nil { | ||
| logger.Debugf(context.Background(), "failed to unmarshal to notification object message [%s] with err: %v", string(msg.Message()), err) | ||
| p.systemMetrics.MessageDecodingError.Inc() | ||
| p.markMessageDone(msg) | ||
| continue | ||
| } | ||
|
|
||
| if err := p.email.SendEmail(context.Background(), emailMessage); err != nil { | ||
| p.systemMetrics.MessageProcessorError.Inc() | ||
| logger.Errorf(context.Background(), "Error sending an email message for message [%s] with emailM with err: %v", emailMessage.String(), err) | ||
| } else { | ||
| p.systemMetrics.MessageSuccess.Inc() | ||
| } | ||
|
|
||
| p.markMessageDone(msg) | ||
| } | ||
|
|
||
| // According to https://github.com/NYTimes/gizmo/blob/f2b3deec03175b11cdfb6642245a49722751357f/pubsub/pubsub.go#L36-L39, | ||
| // the channel backing the subscriber will just close if there is an error. The call to Err() is needed to identify | ||
| // there was an error in the channel or there are no more messages left (resulting in no errors when calling Err()). | ||
| if err := p.sub.Err(); err != nil { | ||
| p.systemMetrics.ChannelClosedError.Inc() | ||
| logger.Warningf(context.Background(), "The stream for the subscriber channel closed with err: %v", err) | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (p *GcpProcessor) markMessageDone(message pubsub.SubscriberMessage) { | ||
| if err := message.Done(); err != nil { | ||
| p.systemMetrics.MessageDoneError.Inc() | ||
| logger.Errorf(context.Background(), "failed to mark message as Done() in processor with err: %v", err) | ||
| } | ||
| } | ||
|
|
||
| func (p *GcpProcessor) StopProcessing() error { | ||
| // Note: If the underlying channel is already closed, then Stop() will return an error. | ||
| if err := p.sub.Stop(); err != nil { | ||
| p.systemMetrics.StopError.Inc() | ||
| logger.Errorf(context.Background(), "Failed to stop the subscriber channel gracefully with err: %v", err) | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } |
113 changes: 113 additions & 0 deletions
113
pkg/async/notifications/implementations/gcp_processor_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| package implementations | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/NYTimes/gizmo/pubsub/pubsubtest" | ||
| "github.com/flyteorg/flyteadmin/pkg/async/notifications/mocks" | ||
| "github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin" | ||
| "github.com/flyteorg/flytestdlib/promutils" | ||
| "github.com/pkg/errors" | ||
| dto "github.com/prometheus/client_model/go" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| var ( | ||
| testGcpSubscriber pubsubtest.TestSubscriber | ||
| mockGcpEmailer mocks.MockEmailer | ||
| ) | ||
|
|
||
| // This method should be invoked before every test to Subscriber. | ||
| func initializeGcpSubscriber() { | ||
| testGcpSubscriber.GivenStopError = nil | ||
| testGcpSubscriber.GivenErrError = nil | ||
| testGcpSubscriber.FoundError = nil | ||
| testGcpSubscriber.ProtoMessages = nil | ||
| testGcpSubscriber.JSONMessages = nil | ||
| } | ||
|
|
||
| func TestGcpProcessor_StartProcessing(t *testing.T) { | ||
| initializeGcpSubscriber() | ||
| testGcpSubscriber.ProtoMessages = append(testGcpSubscriber.ProtoMessages, testSubscriberProtoMessages...) | ||
|
|
||
| testGcpProcessor := NewGcpProcessor(&testGcpSubscriber, &mockGcpEmailer, promutils.NewTestScope()) | ||
|
|
||
| sendEmailValidationFunc := func(ctx context.Context, email admin.EmailMessage) error { | ||
| assert.Equal(t, email.Body, testEmail.Body) | ||
| assert.Equal(t, email.RecipientsEmail, testEmail.RecipientsEmail) | ||
| assert.Equal(t, email.SubjectLine, testEmail.SubjectLine) | ||
| assert.Equal(t, email.SenderEmail, testEmail.SenderEmail) | ||
| return nil | ||
| } | ||
| mockGcpEmailer.SetSendEmailFunc(sendEmailValidationFunc) | ||
| assert.Nil(t, testGcpProcessor.(*GcpProcessor).run()) | ||
|
|
||
| // Check fornumber of messages processed. | ||
| m := &dto.Metric{} | ||
| err := testGcpProcessor.(*GcpProcessor).systemMetrics.MessageSuccess.Write(m) | ||
| assert.Nil(t, err) | ||
| assert.Equal(t, "counter:<value:1 > ", m.String()) | ||
| } | ||
|
|
||
| func TestGcpProcessor_StartProcessingNoMessages(t *testing.T) { | ||
| initializeGcpSubscriber() | ||
|
|
||
| testGcpProcessor := NewGcpProcessor(&testGcpSubscriber, &mockGcpEmailer, promutils.NewTestScope()) | ||
|
|
||
| // Expect no errors are returned. | ||
| assert.Nil(t, testGcpProcessor.(*GcpProcessor).run()) | ||
|
|
||
| // Check fornumber of messages processed. | ||
| m := &dto.Metric{} | ||
| err := testGcpProcessor.(*GcpProcessor).systemMetrics.MessageSuccess.Write(m) | ||
| assert.Nil(t, err) | ||
| assert.Equal(t, "counter:<value:0 > ", m.String()) | ||
| } | ||
|
|
||
| func TestGcpProcessor_StartProcessingError(t *testing.T) { | ||
| initializeGcpSubscriber() | ||
|
|
||
| ret := errors.New("err() returned an error") | ||
| // The error set by GivenErrError is returned by Err(). | ||
| // Err() is checked before Run() returning. | ||
| testGcpSubscriber.GivenErrError = ret | ||
|
|
||
| testGcpProcessor := NewGcpProcessor(&testGcpSubscriber, &mockGcpEmailer, promutils.NewTestScope()) | ||
| assert.Equal(t, ret, testGcpProcessor.(*GcpProcessor).run()) | ||
| } | ||
|
|
||
| func TestGcpProcessor_StartProcessingEmailError(t *testing.T) { | ||
| initializeGcpSubscriber() | ||
| emailError := errors.New("error sending email") | ||
| sendEmailErrorFunc := func(ctx context.Context, email admin.EmailMessage) error { | ||
| return emailError | ||
| } | ||
| mockGcpEmailer.SetSendEmailFunc(sendEmailErrorFunc) | ||
| testGcpSubscriber.ProtoMessages = append(testGcpSubscriber.ProtoMessages, testSubscriberProtoMessages...) | ||
|
|
||
| testGcpProcessor := NewGcpProcessor(&testGcpSubscriber, &mockGcpEmailer, promutils.NewTestScope()) | ||
|
|
||
| // Even if there is an error in sending an email StartProcessing will return no errors. | ||
| assert.Nil(t, testGcpProcessor.(*GcpProcessor).run()) | ||
|
|
||
| // Check for an email error stat. | ||
| m := &dto.Metric{} | ||
| err := testGcpProcessor.(*GcpProcessor).systemMetrics.MessageProcessorError.Write(m) | ||
| assert.Nil(t, err) | ||
| assert.Equal(t, "counter:<value:1 > ", m.String()) | ||
| } | ||
|
|
||
| func TestGcpProcessor_StopProcessing(t *testing.T) { | ||
| initializeGcpSubscriber() | ||
| testGcpProcessor := NewGcpProcessor(&testGcpSubscriber, &mockGcpEmailer, promutils.NewTestScope()) | ||
| assert.Nil(t, testGcpProcessor.StopProcessing()) | ||
| } | ||
|
|
||
| func TestGcpProcessor_StopProcessingError(t *testing.T) { | ||
| initializeGcpSubscriber() | ||
| stopError := errors.New("stop() returns an error") | ||
| testGcpSubscriber.GivenStopError = stopError | ||
| testGcpProcessor := NewGcpProcessor(&testGcpSubscriber, &mockGcpEmailer, promutils.NewTestScope()) | ||
| assert.Equal(t, stopError, testGcpProcessor.StopProcessing()) | ||
| } |
32 changes: 32 additions & 0 deletions
32
pkg/async/notifications/implementations/processor_metrics.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| package implementations | ||
|
|
||
| import ( | ||
| "github.com/flyteorg/flytestdlib/promutils" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| ) | ||
|
|
||
| type processorSystemMetrics struct { | ||
| Scope promutils.Scope | ||
| MessageTotal prometheus.Counter | ||
| MessageDoneError prometheus.Counter | ||
| MessageDecodingError prometheus.Counter | ||
| MessageDataError prometheus.Counter | ||
| MessageProcessorError prometheus.Counter | ||
| MessageSuccess prometheus.Counter | ||
| ChannelClosedError prometheus.Counter | ||
| StopError prometheus.Counter | ||
| } | ||
|
|
||
| func newProcessorSystemMetrics(scope promutils.Scope) processorSystemMetrics { | ||
| return processorSystemMetrics{ | ||
| Scope: scope, | ||
| MessageTotal: scope.MustNewCounter("message_total", "overall count of messages processed"), | ||
| MessageDecodingError: scope.MustNewCounter("message_decoding_error", "count of messages with decoding errors"), | ||
| MessageDataError: scope.MustNewCounter("message_data_error", "count of message data processing errors experience when preparing the message to be notified."), | ||
| MessageDoneError: scope.MustNewCounter("message_done_error", "count of message errors when marking it as done with underlying processor"), | ||
| MessageProcessorError: scope.MustNewCounter("message_processing_error", "count of errors when interacting with notification processor"), | ||
| MessageSuccess: scope.MustNewCounter("message_ok", "count of messages successfully processed by underlying notification mechanism"), | ||
| ChannelClosedError: scope.MustNewCounter("channel_closed_error", "count of channel closing errors"), | ||
| StopError: scope.MustNewCounter("stop_error", "count of errors in Stop() method"), | ||
| } | ||
| } |
Oops, something went wrong.