-
Notifications
You must be signed in to change notification settings - Fork 14
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Stream restart bug fix by introducing manager (#412)
- Loading branch information
1 parent
5523f71
commit 408b950
Showing
5 changed files
with
241 additions
and
90 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| package stream | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/razorpay/metro/internal/subscriber" | ||
| "github.com/razorpay/metro/internal/subscription" | ||
| "github.com/razorpay/metro/pkg/httpclient" | ||
| "github.com/razorpay/metro/pkg/logger" | ||
| ) | ||
|
|
||
| // PushStreamManager manages push stream | ||
| type PushStreamManager struct { | ||
| ctx context.Context | ||
| cancelFunc func() | ||
| doneCh chan struct{} | ||
| config *httpclient.Config | ||
| ps *PushStream | ||
| } | ||
|
|
||
| // NewPushStreamManager return a push stream manager obj which is used to manage push stream | ||
| func NewPushStreamManager(ctx context.Context, nodeID string, subName string, subscriptionCore subscription.ICore, subscriberCore subscriber.ICore, config *httpclient.Config) (*PushStreamManager, error) { | ||
| ps, err := newPushStream(ctx, nodeID, subName, subscriptionCore, subscriberCore, config) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| ctx, cancelFunc := context.WithCancel(ctx) | ||
| return &PushStreamManager{ | ||
| ctx: ctx, | ||
| cancelFunc: cancelFunc, | ||
| ps: ps, | ||
| doneCh: make(chan struct{}), | ||
| config: config, | ||
| }, nil | ||
| } | ||
|
|
||
| // Run starts the push stream manager that is used to manage underlying stream | ||
| func (psm *PushStreamManager) Run() { | ||
| defer close(psm.doneCh) | ||
|
|
||
| logger.Ctx(psm.ctx).Infow("push stream manager: started running stream manager", "subscription", psm.ps.subscription.Name) | ||
| psm.startPushStream() | ||
|
|
||
| go func() { | ||
| for { | ||
| select { | ||
| case <-psm.ctx.Done(): | ||
| if err := psm.ps.Stop(); err != nil { | ||
| logger.Ctx(psm.ctx).Infow("push stream manager: error stopping stream", "subscription", psm.ps.subscription.Name, "error", err.Error()) | ||
| } | ||
| return | ||
| case err := <-psm.ps.GetErrorChannel(): | ||
| logger.Ctx(psm.ctx).Infow("push stream manager: restarting stream handler", "subscription", psm.ps.subscription.Name, "error", err.Error()) | ||
| psm.restartPushStream() | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| // Stop stops the stream manager along with the underlying stream | ||
| func (psm *PushStreamManager) Stop() { | ||
| logger.Ctx(psm.ctx).Infow("push stream manager: stop invoked", "subscription", psm.ps.subscription.Name) | ||
| psm.cancelFunc() | ||
| <-psm.doneCh | ||
| } | ||
|
|
||
| func newPushStream(ctx context.Context, nodeID string, subName string, subscriptionCore subscription.ICore, subscriberCore subscriber.ICore, config *httpclient.Config) (*PushStream, error) { | ||
| ps, err := NewPushStream(ctx, nodeID, subName, subscriptionCore, subscriberCore, config) | ||
| if err != nil { | ||
| logger.Ctx(ctx).Errorw("push stream manager: Failed to setup push stream for subscription", "logFields", map[string]interface{}{ | ||
| "subscription": subName, | ||
| "nodeID": nodeID, | ||
| }) | ||
| return nil, err | ||
| } | ||
| return ps, nil | ||
| } | ||
|
|
||
| func (psm *PushStreamManager) startPushStream() { | ||
| // run the stream in a separate go routine, this go routine is not part of the worker error group | ||
| // as the worker should continue to run if a single subscription stream exists with error | ||
| go func(ctx context.Context) { | ||
| err := psm.ps.Start() | ||
| if err != nil { | ||
| logger.Ctx(ctx).Errorw( | ||
| "push stream manager: stream exited", | ||
| "subscription", psm.ps.subscription.Name, | ||
| "error", err.Error(), | ||
| ) | ||
| } | ||
| }(psm.ctx) | ||
| } | ||
|
|
||
| func (psm *PushStreamManager) restartPushStream() { | ||
| if err := psm.ps.Stop(); err != nil { | ||
| logger.Ctx(psm.ctx).Errorw("push stream manager: stream stop error", "subscription", psm.ps.subscription.Name, "error", err.Error()) | ||
| return | ||
| } | ||
| psm.ps, _ = newPushStream(psm.ctx, psm.ps.nodeID, psm.ps.subscription.Name, psm.ps.subscriptionCore, psm.ps.subscriberCore, psm.config) | ||
| psm.startPushStream() | ||
| workerComponentRestartCount.WithLabelValues(env, "stream", psm.ps.subscription.Topic, psm.ps.subscription.Name).Inc() | ||
| } |
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,119 @@ | ||
| package stream | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/golang/mock/gomock" | ||
| "github.com/google/uuid" | ||
| "github.com/razorpay/metro/internal/subscriber" | ||
| mocks2 "github.com/razorpay/metro/internal/subscriber/mocks" | ||
| mocks1 "github.com/razorpay/metro/internal/subscription/mocks/core" | ||
| "github.com/razorpay/metro/pkg/httpclient" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestNewPushStreamManager(t *testing.T) { | ||
| ctrl := gomock.NewController(t) | ||
| ctx := context.Background() | ||
|
|
||
| tests := []struct { | ||
| wantErr bool | ||
| }{ | ||
| { | ||
| wantErr: false, | ||
| }, | ||
| { | ||
| wantErr: true, | ||
| }, | ||
| } | ||
|
|
||
| for _, test := range tests { | ||
| got, err := NewPushStreamManager( | ||
| ctx, | ||
| uuid.New().String(), | ||
| subName, | ||
| getSubscriptionCoreMock(ctrl, test.wantErr), | ||
| getSubscriberCoreMock(ctx, ctrl), | ||
| &httpclient.Config{}, | ||
| ) | ||
| assert.Equal(t, test.wantErr, err != nil) | ||
| assert.Equal(t, got == nil, test.wantErr) | ||
| } | ||
| } | ||
|
|
||
| func TestPushStreamManager_Run(t *testing.T) { | ||
| ctrl := gomock.NewController(t) | ||
| ctx := context.Background() | ||
|
|
||
| psm, err := NewPushStreamManager( | ||
| ctx, | ||
| uuid.New().String(), | ||
| subName, | ||
| getSubscriptionCoreMock(ctrl, false), | ||
| getSubscriberCoreMock(ctx, ctrl), | ||
| &httpclient.Config{}, | ||
| ) | ||
| assert.NoError(t, err) | ||
| assert.NotNil(t, psm) | ||
|
|
||
| psm.Run() | ||
| <-time.NewTicker(1 * time.Second).C | ||
|
|
||
| streamObj := psm.ps | ||
| psm.ps.GetErrorChannel() <- fmt.Errorf("Something went wrong") | ||
| <-time.NewTicker(1 * time.Second).C | ||
| assert.NotNil(t, psm.ps) | ||
| assert.NotEqual(t, psm.ps, streamObj) | ||
| } | ||
|
|
||
| func TestPushStreamManager_Stop(t *testing.T) { | ||
| ctrl := gomock.NewController(t) | ||
| ctx := context.Background() | ||
|
|
||
| psm, err := NewPushStreamManager( | ||
| ctx, | ||
| uuid.New().String(), | ||
| subName, | ||
| getSubscriptionCoreMock(ctrl, false), | ||
| getSubscriberCoreMock(ctx, ctrl), | ||
| &httpclient.Config{}, | ||
| ) | ||
| assert.NoError(t, err) | ||
| assert.NotNil(t, psm) | ||
| psm.Run() | ||
|
|
||
| // Stop the stream manager and it should be stopped without any error | ||
| psm.Stop() | ||
| assert.NotNil(t, psm.ctx.Err()) | ||
| assert.Equal(t, psm.ctx.Err(), context.Canceled) | ||
| } | ||
|
|
||
| func getSubscriberCoreMock(ctx context.Context, ctrl *gomock.Controller) *mocks2.MockICore { | ||
| subscriberCoreMock := mocks2.NewMockICore(ctrl) | ||
| subModel := getMockSubModel("") | ||
|
|
||
| subscriberCoreMock.EXPECT().NewSubscriber( | ||
| ctx, | ||
| gomock.Any(), | ||
| subModel, | ||
| defaultTimeoutMs, | ||
| defaultMaxOutstandingMsgs, | ||
| defaultMaxOuttandingBytes, | ||
| gomock.AssignableToTypeOf(make(chan *subscriber.PullRequest)), | ||
| gomock.AssignableToTypeOf(make(chan *subscriber.AckMessage)), | ||
| gomock.AssignableToTypeOf(make(chan *subscriber.ModAckMessage))).AnyTimes().Return(getMockSubscriber(ctx, ctrl), nil) | ||
| return subscriberCoreMock | ||
| } | ||
|
|
||
| func getSubscriptionCoreMock(ctrl *gomock.Controller, wantErr bool) *mocks1.MockICore { | ||
| subscriptionCoreMock := mocks1.NewMockICore(ctrl) | ||
| if wantErr { | ||
| subscriptionCoreMock.EXPECT().Get(gomock.Any(), subName).AnyTimes().Return(nil, fmt.Errorf("Something went wrong")) | ||
| } else { | ||
| subscriptionCoreMock.EXPECT().Get(gomock.Any(), subName).AnyTimes().Return(getMockSubModel(""), nil) | ||
| } | ||
| return subscriptionCoreMock | ||
| } |
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
Oops, something went wrong.