feat: multi namespace deployment support

.optimus.sample.yaml

@@ -27,13 +27,13 @@ log:
 # for configuring optimus namespace
 #namespace:
 #  name: sample_namespace
-#  jobs:
-#    # folder where job specifications are stored
+#  job:
+#    # relative path pointing to folder where job specifications are stored
 #    path: "job"
 #  datastore:
 #    # optimus is capable of supporting multiple datastores
 #    type: bigquery
-#    # path where resource spec for BQ are stored
+#    # relative path where resource spec for BQ are stored
 #    path: "bq"
 #  # namespace variables usable in specifications
 #  config: {}

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 := "2a30976f7b40884ddd90e1792576c0941426e8bc"
+PROTON_COMMIT := "a8eb38041e37c152daba7fbb8eb9721f56e902e6"
 
 .PHONY: build test test-ci generate-proto unit-test-ci smoke-test integration-test vet coverage clean install lint
 

api/handler/v1beta1/job.go

@@ -2,7 +2,9 @@ package v1beta1
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"io"
 	"sync"
 	"time"
 
@@ -13,48 +15,114 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-func (sv *RuntimeServiceServer) DeployJobSpecification(req *pb.DeployJobSpecificationRequest, respStream pb.RuntimeService_DeployJobSpecificationServer) error {
+func (sv *RuntimeServiceServer) DeployJobSpecification(stream pb.RuntimeService_DeployJobSpecificationServer) error {
 	startTime := time.Now()
+	errNamespaces := []string{}
 
-	namespaceSpec, err := sv.namespaceService.Get(respStream.Context(), req.GetProjectName(), req.GetNamespaceName())
-	if err != nil {
-		return mapToGRPCErr(sv.l, err, "unable to get namespace")
+	for {
+		req, err := stream.Recv()
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			stream.Send(&pb.DeployJobSpecificationResponse{
+				Success: false,
+				Ack:     true,
+				Message: err.Error(),
+			})
+			return err // immediate error returned (grpc error level)
+		}
+		namespaceSpec, err := sv.namespaceService.Get(stream.Context(), req.GetProjectName(), req.GetNamespaceName())
+		if err != nil {
+			stream.Send(&pb.DeployJobSpecificationResponse{
+				Success: false,
+				Ack:     true,
+				Message: err.Error(),
+			})
+			errNamespaces = append(errNamespaces, req.NamespaceName)
+			continue
+		}
+
+		jobsToKeep, err := sv.getJobsToKeep(stream.Context(), namespaceSpec, req)
+		if err != nil {
+			stream.Send(&pb.DeployJobSpecificationResponse{
+				Success: false,
+				Ack:     true,
+				Message: err.Error(),
+			})
+			errNamespaces = append(errNamespaces, req.NamespaceName)
+			continue
+		}
+
+		observers := new(progress.ObserverChain)
+		observers.Join(sv.progressObserver)
+		observers.Join(&jobSyncObserver{
+			stream: stream,
+			log:    sv.l,
+			mu:     new(sync.Mutex),
+		})
+
+		// delete specs not sent for deployment from internal repository
+		if err := sv.jobSvc.KeepOnly(stream.Context(), namespaceSpec, jobsToKeep, observers); err != nil {
+			stream.Send(&pb.DeployJobSpecificationResponse{
+				Success: false,
+				Ack:     true,
+				Message: fmt.Sprintf("failed to delete jobs: \n%s", err.Error()),
+			})
+			errNamespaces = append(errNamespaces, req.NamespaceName)
+			continue
+		}
+		if err := sv.jobSvc.Sync(stream.Context(), namespaceSpec, observers); err != nil {
+			stream.Send(&pb.DeployJobSpecificationResponse{
+				Success: false,
+				Ack:     true,
+				Message: fmt.Sprintf("failed to sync jobs: \n%s", err.Error()),
+			})
+			errNamespaces = append(errNamespaces, req.NamespaceName)
+			continue
+		}
+		runtimeDeployJobSpecificationCounter.Add(float64(len(req.Jobs)))
+		stream.Send(&pb.DeployJobSpecificationResponse{
+			Success: true,
+			Ack:     true,
+			Message: "success",
+		})
+	}
+	sv.l.Info("finished job deployment", "time", time.Since(startTime))
+	if len(errNamespaces) > 0 {
+		sv.l.Warn("there's error while deploying namespaces: %v", errNamespaces)
+		return fmt.Errorf("error when deploying: %v", errNamespaces)
+	}
+	return nil
+}
+
+func (sv *RuntimeServiceServer) getJobsToKeep(ctx context.Context, namespaceSpec models.NamespaceSpec, req *pb.DeployJobSpecificationRequest) ([]models.JobSpec, error) {
+	jobs := req.GetJobs()
+	if len(jobs) == 0 {
+		return []models.JobSpec{}, nil
 	}
 
 	var jobsToKeep []models.JobSpec
-	for _, reqJob := range req.GetJobs() {
+	for _, reqJob := range jobs {
 		adaptJob, err := sv.adapter.FromJobProto(reqJob)
 		if err != nil {
-			return status.Errorf(codes.Internal, "%s: cannot adapt job %s", err.Error(), reqJob.GetName())
+			sv.l.Error(fmt.Sprintf("%s: cannot adapt job %s", err.Error(), reqJob.GetName()))
+			continue
 		}
 
-		err = sv.jobSvc.Create(respStream.Context(), namespaceSpec, adaptJob)
+		err = sv.jobSvc.Create(ctx, namespaceSpec, adaptJob)
 		if err != nil {
-			return status.Errorf(codes.Internal, "%s: failed to save %s", err.Error(), adaptJob.Name)
+			sv.l.Error(fmt.Sprintf("%s: failed to save %s", err.Error(), adaptJob.Name))
+			continue
 		}
 		jobsToKeep = append(jobsToKeep, adaptJob)
 	}
 
-	observers := new(progress.ObserverChain)
-	observers.Join(sv.progressObserver)
-	observers.Join(&jobSyncObserver{
-		stream: respStream,
-		log:    sv.l,
-		mu:     new(sync.Mutex),
-	})
-
-	// delete specs not sent for deployment from internal repository
-	if err := sv.jobSvc.KeepOnly(respStream.Context(), namespaceSpec, jobsToKeep, observers); err != nil {
-		return status.Errorf(codes.Internal, "failed to delete jobs: \n%s", err.Error())
+	if jobsToKeep == nil {
+		return nil, errors.New("job spec creation is failed")
 	}
 
-	if err := sv.jobSvc.Sync(respStream.Context(), namespaceSpec, observers); err != nil {
-		return status.Errorf(codes.Internal, "failed to sync jobs: \n%s", err.Error())
-	}
-
-	runtimeDeployJobSpecificationCounter.Add(float64(len(req.Jobs)))
-	sv.l.Info("finished job deployment", "time", time.Since(startTime))
-	return nil
+	return jobsToKeep, nil
 }
 
 func (sv *RuntimeServiceServer) ListJobSpecification(ctx context.Context, req *pb.ListJobSpecificationRequest) (*pb.ListJobSpecificationResponse, error) {