feat: make job deployment asynchronous

Makefile

@@ -5,7 +5,7 @@ NAME = "github.com/odpf/optimus"
 LAST_COMMIT := $(shell git rev-parse --short HEAD)
 LAST_TAG := "$(shell git rev-list --tags --max-count=1)"
 OPMS_VERSION := "$(shell git describe --tags ${LAST_TAG})-next"
-PROTON_COMMIT := "da1f6413e587ac1b13ad3429a133526514e2f5fe"
+PROTON_COMMIT := "4cd69522575a01a496bc5babb95d42f38cc8c2cd"
 
 .PHONY: build test test-ci generate-proto unit-test-ci smoke-test integration-test vet coverage clean install lint
 

api/handler/v1beta1/job_spec.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/odpf/salt/log"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -298,14 +299,51 @@ func (sv *JobSpecServiceServer) RefreshJobs(req *pb.RefreshJobsRequest, respStre
 		mu:     new(sync.Mutex),
 	})
 
-	if err := sv.jobSvc.Refresh(respStream.Context(), req.ProjectName, req.NamespaceNames, req.JobNames, observers); err != nil {
+	err := sv.jobSvc.Refresh(respStream.Context(), req.ProjectName, req.NamespaceNames, req.JobNames, observers)
+	if err != nil {
 		return status.Errorf(codes.Internal, "failed to refresh jobs: \n%s", err.Error())
 	}
 
 	sv.l.Info("finished job refresh", "time", time.Since(startTime))
 	return nil
 }
 
+func (sv *JobSpecServiceServer) GetDeployJobsStatus(ctx context.Context, req *pb.GetDeployJobsStatusRequest) (*pb.GetDeployJobsStatusResponse, error) {
+	deployID, err := uuid.Parse(req.DeployId)
+	if err != nil {
+		return nil, err
+	}
+
+	jobDeployment, err := sv.jobSvc.GetDeployment(ctx, models.DeploymentID(deployID))
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to get job deployment: \n%s", err.Error())
+	}
+
+	switch jobDeployment.Status {
+	case models.JobDeploymentStatusSucceed:
+		return &pb.GetDeployJobsStatusResponse{
+			Status:       jobDeployment.Status.String(),
+			SuccessCount: int32(jobDeployment.Details.SuccessCount),
+		}, nil
+	case models.JobDeploymentStatusFailed:
+		var deployJobFailures []*pb.DeployJobFailure
+		for _, failure := range jobDeployment.Details.Failures {
+			deployJobFailures = append(deployJobFailures, &pb.DeployJobFailure{JobName: failure.JobName, Message: failure.Message})
+		}
+
+		return &pb.GetDeployJobsStatusResponse{
+			Status:       jobDeployment.Status.String(),
+			SuccessCount: int32(jobDeployment.Details.SuccessCount),
+			FailureCount: int32(jobDeployment.Details.FailureCount),
+			Failures:     deployJobFailures,
+		}, nil
+	default:
+		return &pb.GetDeployJobsStatusResponse{
+			Status: jobDeployment.Status.String(),
+		}, nil
+	}
+}
+
 func NewJobSpecServiceServer(l log.Logger, jobService models.JobService, adapter ProtoAdapter,
 	projectService service.ProjectService, namespaceService service.NamespaceService, progressObserver progress.Observer) *JobSpecServiceServer {
 	return &JobSpecServiceServer{

api/handler/v1beta1/job_spec_test.go

@@ -603,4 +603,149 @@ func TestJobSpecificationOnServer(t *testing.T) {
 			assert.Contains(t, err.Error(), errorMsg)
 		})
 	})
+
+	t.Run("GetDeployJobsStatus", func(t *testing.T) {
+		projectName := "a-data-project"
+		projectSpec := models.ProjectSpec{
+			ID:   models.ProjectID(uuid.New()),
+			Name: projectName,
+			Config: map[string]string{
+				"bucket": "gs://some_folder",
+			},
+		}
+		t.Run("should get on progress job deployment successfully", func(t *testing.T) {
+			jobService := new(mock.JobService)
+			defer jobService.AssertExpectations(t)
+
+			jobSpecServiceServer := v1.NewJobSpecServiceServer(
+				log,
+				jobService,
+				nil,
+				nil,
+				nil,
+				nil,
+			)
+			deployID := uuid.New()
+			jobDeployment := models.JobDeployment{
+				ID:      models.DeploymentID(deployID),
+				Project: projectSpec,
+				Status:  models.JobDeploymentStatusInProgress,
+			}
+			getDeployJobsStatusResponse := &pb.GetDeployJobsStatusResponse{
+				Status:       jobDeployment.Status.String(),
+				SuccessCount: 0,
+			}
+
+			jobService.On("GetDeployment", ctx, models.DeploymentID(deployID)).Return(jobDeployment, nil)
+
+			getDeployJobsStatusRequest := &pb.GetDeployJobsStatusRequest{DeployId: deployID.String()}
+			actual, err := jobSpecServiceServer.GetDeployJobsStatus(ctx, getDeployJobsStatusRequest)
+
+			assert.Nil(t, err)
+			assert.Equal(t, getDeployJobsStatusResponse, actual)
+		})
+		t.Run("should get succeeded job deployment successfully", func(t *testing.T) {
+			jobService := new(mock.JobService)
+			defer jobService.AssertExpectations(t)
+
+			jobSpecServiceServer := v1.NewJobSpecServiceServer(
+				log,
+				jobService,
+				nil,
+				nil,
+				nil,
+				nil,
+			)
+			deployID := uuid.New()
+			jobDeployment := models.JobDeployment{
+				ID:      models.DeploymentID(deployID),
+				Project: projectSpec,
+				Status:  models.JobDeploymentStatusSucceed,
+				Details: models.JobDeploymentDetail{
+					SuccessCount: 5,
+				},
+			}
+			getDeployJobsStatusResponse := &pb.GetDeployJobsStatusResponse{
+				Status:       jobDeployment.Status.String(),
+				SuccessCount: int32(jobDeployment.Details.SuccessCount),
+			}
+
+			jobService.On("GetDeployment", ctx, models.DeploymentID(deployID)).Return(jobDeployment, nil)
+
+			getDeployJobsStatusRequest := &pb.GetDeployJobsStatusRequest{DeployId: deployID.String()}
+			actual, err := jobSpecServiceServer.GetDeployJobsStatus(ctx, getDeployJobsStatusRequest)
+
+			assert.Nil(t, err)
+			assert.Equal(t, getDeployJobsStatusResponse, actual)
+		})
+		t.Run("should get failed job deployment successfully", func(t *testing.T) {
+			jobService := new(mock.JobService)
+			defer jobService.AssertExpectations(t)
+
+			jobSpecServiceServer := v1.NewJobSpecServiceServer(
+				log,
+				jobService,
+				nil,
+				nil,
+				nil,
+				nil,
+			)
+			deployID := uuid.New()
+			jobDeployment := models.JobDeployment{
+				ID:      models.DeploymentID(deployID),
+				Project: projectSpec,
+				Status:  models.JobDeploymentStatusFailed,
+				Details: models.JobDeploymentDetail{
+					SuccessCount: 4,
+					Failures: []models.JobDeploymentFailure{
+						{
+							JobName: "job-a",
+							Message: "internal error",
+						},
+					},
+				},
+			}
+			getDeployJobsStatusResponse := &pb.GetDeployJobsStatusResponse{
+				Status:       jobDeployment.Status.String(),
+				SuccessCount: int32(jobDeployment.Details.SuccessCount),
+				Failures: []*pb.DeployJobFailure{
+					{
+						JobName: jobDeployment.Details.Failures[0].JobName,
+						Message: jobDeployment.Details.Failures[0].Message,
+					},
+				},
+			}
+
+			jobService.On("GetDeployment", ctx, models.DeploymentID(deployID)).Return(jobDeployment, nil)
+
+			getDeployJobsStatusRequest := &pb.GetDeployJobsStatusRequest{DeployId: deployID.String()}
+			actual, err := jobSpecServiceServer.GetDeployJobsStatus(ctx, getDeployJobsStatusRequest)
+
+			assert.Nil(t, err)
+			assert.Equal(t, getDeployJobsStatusResponse, actual)
+		})
+		t.Run("should failed when unable to get job deployment", func(t *testing.T) {
+			jobService := new(mock.JobService)
+			defer jobService.AssertExpectations(t)
+
+			jobSpecServiceServer := v1.NewJobSpecServiceServer(
+				log,
+				jobService,
+				nil,
+				nil,
+				nil,
+				nil,
+			)
+			deployID := uuid.New()
+
+			errorMsg := "internal error"
+			jobService.On("GetDeployment", ctx, models.DeploymentID(deployID)).Return(models.JobDeployment{}, errors.New(errorMsg))
+
+			getDeployJobsStatusRequest := &pb.GetDeployJobsStatusRequest{DeployId: deployID.String()}
+			actual, err := jobSpecServiceServer.GetDeployJobsStatus(ctx, getDeployJobsStatusRequest)
+
+			assert.Nil(t, actual)
+			assert.Contains(t, err.Error(), errorMsg)
+		})
+	})
 }

api/handler/v1beta1/observer.go

@@ -135,7 +135,7 @@ func (obs *jobRefreshObserver) Notify(e progress.Event) {
 		}
 		if evt.Err != nil {
 			resp.Success = false
-			resp.Message = evt.Err.Error()
+			resp.Value = evt.Err.Error()
 		}
 
 		if err := obs.stream.Send(resp); err != nil {
@@ -144,7 +144,7 @@ func (obs *jobRefreshObserver) Notify(e progress.Event) {
 	case *models.ProgressJobSpecUnknownDependencyUsed:
 		resp := &pb.RefreshJobsResponse{
 			JobName: evt.Job,
-			Message: evt.String(),
+			Value:   evt.String(),
 			Success: false,
 			Type:    evt.Type(),
 		}
@@ -154,16 +154,25 @@ func (obs *jobRefreshObserver) Notify(e progress.Event) {
 	case *models.ProgressJobDependencyResolution:
 		resp := &pb.RefreshJobsResponse{
 			JobName: evt.Job,
-			Message: evt.String(),
+			Value:   evt.String(),
 			Success: true,
 			Type:    evt.Type(),
 		}
 		if evt.Err != nil {
 			resp.Success = false
-			resp.Message = evt.Err.Error()
+			resp.Value = evt.Err.Error()
 		}
 		if err := obs.stream.Send(resp); err != nil {
 			obs.log.Error("failed to send failed dependency resolution notification", "evt", evt.String(), "error", err)
 		}
+	case *models.ProgressJobDeploymentRequestCreated:
+		resp := &pb.RefreshJobsResponse{
+			Value:   evt.ID().UUID().String(),
+			Success: true,
+			Type:    evt.Type(),
+		}
+		if err := obs.stream.Send(resp); err != nil {
+			obs.log.Error("failed to send job deployment request created", "evt", evt.String(), "error", err)
+		}
 	}
 }