feat: Introduce new overwrite flag to jobs (dapr#8701)

* feat: Introcude new overwrite flag to jobs

The new flags can be use to specify if the job can overwrite an existing one.
The default value is `false`

Signed-off-by: Javier Aliaga <[email protected]>

* fix: Lint errors

Signed-off-by: Javier Aliaga <[email protected]>

* fix: Actors reminder overrides existing jobs when using scheduler as storage

Signed-off-by: Javier Aliaga <[email protected]>

* fix: Reviewers comments

Move overwrite tests to its own files

Signed-off-by: Javier Aliaga <[email protected]>

* chore: Use grpc error code if job already exist

Signed-off-by: Javier Aliaga <[email protected]>

* chore: Add Copyright headers

Signed-off-by: Javier Aliaga <[email protected]>

* chore: Schedule job return correct http code

Convert grc error code to http error code on http requests for schedule jobs

Signed-off-by: Javier Aliaga <[email protected]>

* fix: linting

Signed-off-by: Javier Aliaga <[email protected]>

* Update pkg/scheduler/server/api.go

Co-authored-by: Josh van Leeuwen <[email protected]>
Signed-off-by: Javier Aliaga <[email protected]>

* Update pkg/scheduler/server/api.go

Co-authored-by: Josh van Leeuwen <[email protected]>
Signed-off-by: Javier Aliaga <[email protected]>

* fix: Review comments

Better assertions
Import ordering

Signed-off-by: Javier Aliaga <[email protected]>

* chore: Add test and fix lint

Signed-off-by: Javier Aliaga <[email protected]>

---------

Signed-off-by: Javier Aliaga <[email protected]>
Co-authored-by: Josh van Leeuwen <[email protected]>
Co-authored-by: Dapr Bot <[email protected]>
Co-authored-by: Cassie Coyle <[email protected]>

Author: 4 people

Makefile

@@ -404,6 +404,9 @@ test-integration-parallel: test-deps
 lint: check-linter
 	$(GOLANGCI_LINT) run --build-tags=$(GOLANGCI_LINT_TAGS) --timeout=20m --max-same-issues 0 --max-issues-per-linter 0
 
+.PHONY: lint-fix
+lint-fix: check-linter
+	$(GOLANGCI_LINT) run --build-tags=$(GOLANGCI_LINT_TAGS) --timeout=20m --max-same-issues 0 --max-issues-per-linter 0 --fix
 
 ################################################################################
 # Target: check-linter                                                         #

dapr/proto/internals/v1/jobs.proto

@@ -30,4 +30,5 @@ message JobHTTPRequest {
   optional string due_time = 4 [json_name = "dueTime"];
   optional string ttl = 5 [json_name = "ttl"];
   google.protobuf.Value data = 6 [json_name = "data"];
+  optional bool overwrite = 7 [json_name = "overwrite"];
 }

dapr/proto/runtime/v1/dapr.proto

@@ -1265,6 +1265,9 @@ message Job {
   // payload is the serialized job payload that will be sent to the recipient
   // when the job is triggered.
   google.protobuf.Any data = 6 [json_name = "data"];
+
+  // If true, allows this job to overwrite an existing job with the same name.
+  bool overwrite = 7  [json_name = "overwrite"];
 }
 
 // ScheduleJobRequest is the message to create/schedule the job.

dapr/proto/scheduler/v1/scheduler.proto

@@ -46,6 +46,9 @@ message Job {
   // By default, the failure policy is FailurePolicyConstant with a 1s interval
   // and 3 maximum retries.
   optional FailurePolicy failure_policy = 6;
+
+  // If true, allows this job to overwrite an existing job with the same name.
+  bool overwrite = 7;
 }
 
 // TargetJob is the message used by the daprd sidecar to schedule a job

pkg/actors/internal/reminders/storage/scheduler/scheduler.go

@@ -120,6 +120,7 @@ func (s *scheduler) Create(ctx context.Context, reminder *api.CreateReminderRequ
 			Ttl:           ttl,
 			Data:          reminder.Data,
 			FailurePolicy: failurePolicy,
+			Overwrite:     true,
 		},
 		Metadata: &schedulerv1pb.JobMetadata{
 			AppId:     s.appID,

pkg/api/errors/scheduler.go

@@ -17,9 +17,11 @@ import (
 	"net/http"
 
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/dapr/dapr/pkg/messages/errorcodes"
 	kiterrors "github.com/dapr/kit/errors"
+	"github.com/dapr/kit/grpccodes"
 )
 
 func SchedulerURLName(metadata map[string]string) error {
@@ -36,9 +38,16 @@ func SchedulerURLName(metadata map[string]string) error {
 }
 
 func SchedulerScheduleJob(metadata map[string]string, err error) error {
+	code := status.Code(err)
+	if code == codes.Unknown {
+		code = codes.Internal
+	}
+
+	httpCode := grpccodes.HTTPStatusFromCode(code)
+
 	return kiterrors.NewBuilder(
-		codes.Internal,
-		http.StatusInternalServerError,
+		code,
+		httpCode,
 		"failed to schedule job due to: "+err.Error(),
 		"",
 		string(errorcodes.SchedulerScheduleJob.Category),

pkg/api/universal/jobs.go

@@ -45,12 +45,13 @@ func (a *Universal) ScheduleJobAlpha1HTTP(ctx context.Context, job *internalsv1p
 
 	//nolint:protogetter
 	return a.scheduleJob(ctx, &runtimev1pb.Job{
-		Name:     job.GetName(),
-		Schedule: job.Schedule,
-		Repeats:  job.Repeats,
-		DueTime:  job.DueTime,
-		Ttl:      job.Ttl,
-		Data:     data,
+		Name:      job.GetName(),
+		Schedule:  job.Schedule,
+		Repeats:   job.Repeats,
+		DueTime:   job.DueTime,
+		Ttl:       job.Ttl,
+		Data:      data,
+		Overwrite: job.GetOverwrite(),
 	})
 }
 
@@ -85,11 +86,12 @@ func (a *Universal) scheduleJob(ctx context.Context, job *runtimev1pb.Job) (*run
 		},
 		//nolint:protogetter
 		Job: &schedulerv1pb.Job{
-			Schedule: job.Schedule,
-			Data:     job.GetData(),
-			Repeats:  job.Repeats,
-			DueTime:  job.DueTime,
-			Ttl:      job.Ttl,
+			Schedule:  job.Schedule,
+			Data:      job.GetData(),
+			Repeats:   job.Repeats,
+			DueTime:   job.DueTime,
+			Ttl:       job.Ttl,
+			Overwrite: job.GetOverwrite(),
 		},
 	}