refactor: split config

.optimus.sample.yaml

@@ -1,46 +1,19 @@
 version: 1
 
+########################################
+# COMMON CONFIG (APPLIED FOR SERVER & CLIENT)
+########################################
 # logging configuration
 log:
   # debug, info, warning, error, fatal - default 'info'
   level: info
 
-#
-# cli configurations
-#
-
-# used to connect optimus service
-#host: localhost:9100
 
-# for configuring optimus project
-#project:
-#  name: sample_project
-#  # project variables usable in specifications
-#  config:
-#    environment: integration
-#    scheduler_host: http://example.io/
-#    # storage_path is used for storing compiled job specifications that can be
-#    # consumed by schedulers like Airflow
-#    # it supports multiple schemes like: file://, gcs://
-#    storage_path: file://absolute_path_to_a_directory
 
-# for configuring optimus namespace
-#namespace:
-#  name: sample_namespace
-#  jobs:
-#    # 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
-#    path: "bq"
-#  # namespace variables usable in specifications
-#  config: {}
 
-#
-# server configurations
-#
+########################################
+# SERVER CONFIG
+########################################
 
 # for configuring optimus service
 #serve:
@@ -83,4 +56,41 @@ log:
 #  profile_addr: ":9110"
 #
 #  # jaeger collector address to send application traces
-#  jaeger_addr: "http://localhost:14268/api/traces"
+#  jaeger_addr: "http://localhost:14268/api/traces"
+
+
+
+
+########################################
+# PROJECT CONFIG
+########################################
+
+# used to connect optimus service
+#host: localhost:9100
+
+# for configuring optimus project
+#project:
+#  name: sample_project
+#  # project variables usable in specifications
+#  config:
+#    environment: integration
+#    scheduler_host: http://example.io/
+#    # storage_path is used for storing compiled job specifications that can be
+#    # consumed by schedulers like Airflow
+#    # it supports multiple schemes like: file://, gcs://
+#    storage_path: file://absolute_path_to_a_directory
+
+# for configuring optimus namespace
+#namespace:
+#  name: sample_namespace
+#  job:
+#    # relative path pointing to folder where job specifications are stored
+#    path: "job"
+#  datastore:
+#    # optimus is capable of supporting multiple datastores
+#    type: bigquery
+#    # 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) {