feat: support slack based alerts on failure and sla miss

Author: kushsharma

Makefile

@@ -27,7 +27,7 @@ pack-files:
 generate-proto: ## regenerate protos
 	@echo " > cloning protobuf from odpf/proton"
 	@rm -rf proton/
-	@git -c advice.detachedHead=false clone https://github.com/odpf/proton --depth 1 --quiet --branch main
+	@git -c advice.detachedHead=false clone https://github.com/odpf/proton --depth 1 --quiet --branch optimus-job-event
 	@echo " > generating protobuf"
 	@echo " > info: make sure correct version of dependencies are installed using 'install'"
 	@buf generate

README.md

@@ -68,6 +68,12 @@ brew install odpf/taps/optimus
 optimus version
 ```
 
+### Stability and Compatibility
+Optimus is currently undergoing heavy development with frequent, breaking API changes.
+
+ > ⭐ Current major version is zero (v0.x.x) to accommodate rapid development and fast iteration while getting early feedback from users (feedback on APIs are appreciated). 
+ > The public API could change without a major version update before v1.0.0 release.
+
 ## Credits
 
 This project exists thanks to all the [contributors](https://github.com/odpf/optimus/graphs/contributors).

api/handler/v1/adapter.go

@@ -73,11 +73,22 @@ func (adapt *Adapter) FromJobProto(spec *pb.JobSpecification) (models.JobSpec, e
 	retryDelay := time.Duration(0)
 	retryCount := 0
 	retryExponentialBackoff := false
-	if spec.Behavior != nil && spec.Behavior.Retry != nil {
-		retryCount = int(spec.Behavior.Retry.Count)
-		retryExponentialBackoff = spec.Behavior.Retry.ExponentialBackoff
-		if spec.Behavior.Retry.Delay != nil && spec.Behavior.Retry.Delay.IsValid() {
-			retryDelay = spec.Behavior.Retry.Delay.AsDuration()
+	var notifiers []models.JobSpecNotifier
+	if spec.Behavior != nil {
+		if spec.Behavior.Retry != nil {
+			retryCount = int(spec.Behavior.Retry.Count)
+			retryExponentialBackoff = spec.Behavior.Retry.ExponentialBackoff
+			if spec.Behavior.Retry.Delay != nil && spec.Behavior.Retry.Delay.IsValid() {
+				retryDelay = spec.Behavior.Retry.Delay.AsDuration()
+			}
+		}
+
+		for _, notify := range spec.Behavior.Notify {
+			notifiers = append(notifiers, models.JobSpecNotifier{
+				On:       models.JobEventType(strings.ToLower(notify.On.String())),
+				Config:   notify.Config,
+				Channels: notify.Channels,
+			})
 		}
 	}
 	return models.JobSpec{
@@ -100,6 +111,7 @@ func (adapt *Adapter) FromJobProto(spec *pb.JobSpecification) (models.JobSpec, e
 				Delay:              retryDelay,
 				ExponentialBackoff: retryExponentialBackoff,
 			},
+			Notify: notifiers,
 		},
 		Task: models.JobSpecTask{
 			Unit:   execUnit,
@@ -150,6 +162,15 @@ func (adapt *Adapter) ToJobProto(spec models.JobSpec) (*pb.JobSpecification, err
 		return nil, err
 	}
 
+	var notifyProto []*pb.JobSpecification_Behavior_Notifiers
+	for _, notify := range spec.Behavior.Notify {
+		notifyProto = append(notifyProto, &pb.JobSpecification_Behavior_Notifiers{
+			On:       pb.JobEvent_Type(pb.JobEvent_Type_value[strings.ToUpper(string(notify.On))]),
+			Channels: notify.Channels,
+			Config:   notify.Config,
+		})
+	}
+
 	conf := &pb.JobSpecification{
 		Version:          int32(spec.Version),
 		Name:             spec.Name,
@@ -173,6 +194,7 @@ func (adapt *Adapter) ToJobProto(spec models.JobSpec) (*pb.JobSpecification, err
 				Delay:              ptypes.DurationProto(spec.Behavior.Retry.Delay),
 				ExponentialBackoff: spec.Behavior.Retry.ExponentialBackoff,
 			},
+			Notify: notifyProto,
 		},
 	}
 	if spec.Schedule.EndDate != nil {
@@ -185,7 +207,7 @@ func (adapt *Adapter) ToJobProto(spec models.JobSpec) (*pb.JobSpecification, err
 		})
 	}
 
-	taskConfigs := []*pb.JobConfigItem{}
+	var taskConfigs []*pb.JobConfigItem
 	for _, c := range spec.Task.Config {
 		taskConfigs = append(taskConfigs, &pb.JobConfigItem{
 			Name:  strings.ToUpper(c.Name),

api/handler/v1/adapter_test.go

@@ -41,6 +41,16 @@ func TestAdapter(t *testing.T) {
 		allTasksRepo.On("GetByName", "sample-task").Return(execUnit1, nil)
 		defer allTasksRepo.AssertExpectations(t)
 
+		hookUnit1 := new(mock.HookPlugin)
+		hookUnit1.On("GetHookSchema", context.Background(), models.GetHookSchemaRequest{}).Return(models.GetHookSchemaResponse{
+			Name: "sample-hook",
+		}, nil)
+		defer hookUnit1.AssertExpectations(t)
+
+		allHookRepo := new(mock.SupportedHookRepo)
+		allHookRepo.On("GetByName", "sample-hook").Return(hookUnit1, nil)
+		defer allHookRepo.AssertExpectations(t)
+
 		jobSpec := models.JobSpec{
 			Name: "test-job",
 			Schedule: models.JobSpecSchedule{
@@ -55,6 +65,15 @@ func TestAdapter(t *testing.T) {
 					Delay:              0,
 					ExponentialBackoff: true,
 				},
+				Notify: []models.JobSpecNotifier{
+					{
+						On: models.JobEventTypeFailure,
+						Config: map[string]string{
+							"key": "val",
+						},
+						Channels: []string{"slack://@devs"},
+					},
+				},
 			},
 			Task: models.JobSpecTask{
 				Unit: execUnit1,
@@ -79,9 +98,20 @@ func TestAdapter(t *testing.T) {
 				},
 			),
 			Dependencies: map[string]models.JobSpecDependency{},
+			Hooks: []models.JobSpecHook{
+				{
+					Config: models.JobSpecConfigs{
+						{
+							Name:  "PROJECT",
+							Value: "this",
+						},
+					},
+					Unit: hookUnit1,
+				},
+			},
 		}
 
-		adapter := v1.NewAdapter(allTasksRepo, nil, nil)
+		adapter := v1.NewAdapter(allTasksRepo, allHookRepo, nil)
 		inProto, err := adapter.ToJobProto(jobSpec)
 		assert.Nil(t, err)
 		original, err := adapter.FromJobProto(inProto)