Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(maestro): fixes sink idle event #2672

Merged
merged 14 commits into from
Sep 29, 2023
Merged

fix(maestro): fixes sink idle event #2672

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
9696b66
feat(maestro): fix update event and add redundancy to update when the…
Sep 28, 2023
d4b5959
feat(maestro): add unit test for delete
Sep 28, 2023
a36a4eb
feat(maestro): add unit test for delete
Sep 28, 2023
56dbc5c
feat(maestro): fix delete flow.
Sep 28, 2023
d447525
feat(maestro): fixes sink activity.
Sep 28, 2023
abba61b
feat(maestro): fixes not encoding
Sep 28, 2023
26b3ae7
feat(maestro): fixes create flow.
Sep 28, 2023
62ce7f1
Merge branch 'develop' into fix-eng-1118-update-event
Sep 29, 2023
9b6a1de
feat(maestro): fix on handle activity.
Sep 29, 2023
57e96fa
feat(maestro): fix on handle idle.
Sep 29, 2023
88969d3
feat(maestro): fix on handle idle.
Sep 29, 2023
f707c99
Merge branch 'develop' into fix-eng-1118-update-event
Sep 29, 2023
8e9de7b
feat(maestro): unique sink ids for not racing.
Sep 29, 2023
c69d7c9
feat(maestro): fix failing test.
Sep 29, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 5 additions & 6 deletions maestro/service/deploy_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,13 +146,12 @@ func (d *eventService) HandleSinkIdle(ctx context.Context, event maestroredis.Si
_, err = d.deploymentService.NotifyCollector(ctx, event.OwnerID, event.SinkID, "deploy", "", "")
if err != nil {
d.logger.Error("error trying to notify collector", zap.Error(err))
err2 := d.deploymentService.UpdateStatus(ctx, event.OwnerID, event.SinkID, "provisioning_error", err.Error())
if err2 != nil {
d.logger.Warn("error during notifying provisioning error, customer will not be notified of error")
}
return err
}
err2 := d.deploymentService.UpdateStatus(ctx, event.OwnerID, event.SinkID, "provisioning_error", err.Error())
if err2 != nil {
d.logger.Warn("error during notifying provisioning error, customer will not be notified of error")
d.logger.Error("error during update status", zap.Error(err))
return err
}

return nil
}
11 changes: 6 additions & 5 deletions maestro/service/handle_sinker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ func TestEventService_HandleSinkActivity(t *testing.T) {
args: args{
event: redis.SinkerUpdateEvent{
OwnerID: "owner2",
SinkID: "sink2",
SinkID: "sink22",
State: "active",
Size: "22",
Timestamp: time.Now(),
Expand All @@ -51,7 +51,7 @@ func TestEventService_HandleSinkActivity(t *testing.T) {
"MY_SECRET", NewTestProducer(logger), NewTestKubeCtr(logger))
d := NewEventService(logger, deploymentService, nil)
err := d.HandleSinkCreate(context.Background(), redis.SinksUpdateEvent{
SinkID: "sink2",
SinkID: "sink22",
Owner: "owner2",
Backend: "prometheus",
Config: types.Metadata{
Expand Down Expand Up @@ -103,7 +103,7 @@ func TestEventService_HandleSinkIdle(t *testing.T) {
args: args{
event: redis.SinkerUpdateEvent{
OwnerID: "owner2",
SinkID: "sink2",
SinkID: "sink222",
State: "idle",
Size: "22",
Timestamp: time.Now(),
Expand All @@ -112,10 +112,11 @@ func TestEventService_HandleSinkIdle(t *testing.T) {
},
}
logger := zap.NewNop()
deploymentService := deployment.NewDeploymentService(logger, NewFakeRepository(logger), "kafka:9092", "MY_SECRET", NewTestProducer(logger), nil)
deploymentService := deployment.NewDeploymentService(logger, NewFakeRepository(logger), "kafka:9092", "MY_SECRET", NewTestProducer(logger),
NewTestKubeCtr(logger))
d := NewEventService(logger, deploymentService, NewTestKubeCtr(logger))
err := d.HandleSinkCreate(context.Background(), redis.SinksUpdateEvent{
SinkID: "sink2",
SinkID: "sink222",
Owner: "owner2",
Backend: "prometheus",
Config: types.Metadata{
Expand Down
17 changes: 9 additions & 8 deletions maestro/service/handle_sinks_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ func Test_eventService_HandleSinkCreate(t *testing.T) {
name: "create event",
args: args{
event: redis.SinksUpdateEvent{
SinkID: "sink1",
SinkID: "crt-sink1",
Owner: "owner1",
Config: types.Metadata{
"exporter": types.Metadata{
Expand All @@ -46,7 +46,7 @@ func Test_eventService_HandleSinkCreate(t *testing.T) {
name: "create event without config",
args: args{
event: redis.SinksUpdateEvent{
SinkID: "sink1",
SinkID: "crt-sink1",
Owner: "owner1",
Config: nil,
Backend: "prometheus",
Expand Down Expand Up @@ -82,7 +82,7 @@ func TestEventService_HandleSinkUpdate(t *testing.T) {
name: "update event when there is none in db",
args: args{
event: redis.SinksUpdateEvent{
SinkID: "sink1",
SinkID: "upd-sink1",
Owner: "owner1",
Config: types.Metadata{
"exporter": types.Metadata{
Expand All @@ -104,7 +104,7 @@ func TestEventService_HandleSinkUpdate(t *testing.T) {
name: "update event success",
args: args{
event: redis.SinksUpdateEvent{
SinkID: "sink1",
SinkID: "upd-sink1",
Owner: "owner1",
Backend: "prometheus",
Config: types.Metadata{
Expand All @@ -124,8 +124,9 @@ func TestEventService_HandleSinkUpdate(t *testing.T) {
},
}
logger := zap.NewNop()
deploymentService := deployment.NewDeploymentService(logger, NewFakeRepository(logger), "kafka:9092", "MY_SECRET", NewTestProducer(logger), nil)
d := NewEventService(logger, deploymentService, nil)
deploymentService := deployment.NewDeploymentService(logger, NewFakeRepository(logger), "kafka:9092", "MY_SECRET", NewTestProducer(logger),
NewTestKubeCtr(logger))
d := NewEventService(logger, deploymentService, NewTestKubeCtr(logger))
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ctx := context.WithValue(context.Background(), "test", tt.name)
Expand Down Expand Up @@ -170,7 +171,7 @@ func TestEventService_HandleSinkDelete(t *testing.T) {
name: "delete event success",
args: args{
event: redis.SinksUpdateEvent{
SinkID: "sink2",
SinkID: "sink2-1",
Owner: "owner2",
Backend: "prometheus",
Config: types.Metadata{
Expand All @@ -192,7 +193,7 @@ func TestEventService_HandleSinkDelete(t *testing.T) {
deploymentService := deployment.NewDeploymentService(logger, NewFakeRepository(logger), "kafka:9092", "MY_SECRET", NewTestProducer(logger), nil)
d := NewEventService(logger, deploymentService, nil)
err := d.HandleSinkCreate(context.Background(), redis.SinksUpdateEvent{
SinkID: "sink2",
SinkID: "sink2-1",
Owner: "owner2",
Backend: "prometheus",
Config: types.Metadata{
Expand Down
Loading