feat(firehose): add field stop_by_time to stop firehose by give datetime (#85)

* feat(firehose): add field stop_by_time to stop firehose by give datetime

* fix(firehose): tests with updated firehose spec config schema

* feat(firehose): created scheduled sync job when stop time is set for firehose

* fix(firehose): schedule sync job and change stop_by_time to stop_time field

* refactor: refactor Plan struct to use Resource by value

* feat: use run-at timestamp value in sync jobid

Co-authored-by: Shivaprasad <[email protected]>

Author: rohilsurana

.gitignore

@@ -25,3 +25,5 @@ tmp/
 expt/
 
 entropy.dev.yaml
+
+requests.http

cli/serve.go

@@ -90,6 +90,10 @@ func runServer(baseCtx context.Context, nrApp *newrelic.Application, zapLog *zap
 		return err
 	}
 
+	if err := asyncWorker.Register(core.JobKindScheduledSyncResource, service.HandleSyncJob); err != nil {
+		return err
+	}
+
 	return entropyserver.Serve(ctx, cfg.Service.addr(), nrApp, zapLog, service)
 }
 

core/mocks/loggable_module.go

@@ -5,6 +5,7 @@ package module
 import (
 	"context"
 	"encoding/json"
+	"time"
 
 	"github.com/odpf/entropy/core/resource"
 	"github.com/odpf/entropy/pkg/errors"
@@ -16,7 +17,7 @@ type Module interface {
 	// Plan SHOULD validate the action on the current version of the resource,
 	// return the resource with config/status/state changes (if any) applied.
 	// Plan SHOULD NOT have side effects on anything other than the resource.
-	Plan(ctx context.Context, spec Spec, act ActionRequest) (*resource.Resource, error)
+	Plan(ctx context.Context, spec Spec, act ActionRequest) (*Plan, error)
 
 	// Sync is called repeatedly by Entropy core until the returned state is
 	// a terminal status. Module implementation is free to execute an action
@@ -27,6 +28,12 @@ type Module interface {
 	Sync(ctx context.Context, spec Spec) (*resource.State, error)
 }
 
+// Plan represents the changes to be staged and later synced by module.
+type Plan struct {
+	Resource      resource.Resource
+	ScheduleRunAt time.Time
+}
+
 // Spec represents the context for Plan() or Sync() invocations.
 type Spec struct {
 	Resource     resource.Resource             `json:"resource"`

core/mocks/module.go

@@ -59,7 +59,7 @@ func (mr *Registry) get(kind string) (Descriptor, bool) {
 	return desc, found
 }
 
-func (mr *Registry) Plan(ctx context.Context, spec Spec, act ActionRequest) (*resource.Resource, error) {
+func (mr *Registry) Plan(ctx context.Context, spec Spec, act ActionRequest) (*Plan, error) {
 	kind := spec.Resource.Kind
 
 	desc, found := mr.get(kind)

core/module/module.go

@@ -6,19 +6,23 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/odpf/entropy/core/module"
 	"github.com/odpf/entropy/core/resource"
 	"github.com/odpf/entropy/pkg/errors"
 	"github.com/odpf/entropy/pkg/worker"
 )
 
-const JobKindSyncResource = "sync_resource"
+const (
+	JobKindSyncResource          = "sync_resource"
+	JobKindScheduledSyncResource = "sched_sync_resource"
+)
 
 type syncJobPayload struct {
 	ResourceURN string    `json:"resource_urn"`
 	UpdatedAt   time.Time `json:"updated_at"`
 }
 
-func (s *Service) enqueueSyncJob(ctx context.Context, res resource.Resource) error {
+func (s *Service) enqueueSyncJob(ctx context.Context, res resource.Resource, runAt time.Time, jobType string) error {
 	data := syncJobPayload{
 		ResourceURN: res.URN,
 		UpdatedAt:   res.UpdatedAt,
@@ -29,12 +33,17 @@ func (s *Service) enqueueSyncJob(ctx context.Context, res resource.Resource) err
 		return err
 	}
 
-	return s.worker.Enqueue(ctx, worker.Job{
-		ID:      fmt.Sprintf("sync-%s-%d", res.URN, res.UpdatedAt.Unix()),
-		Kind:    JobKindSyncResource,
-		RunAt:   time.Now(),
+	job := worker.Job{
+		ID:      fmt.Sprintf(jobType+"-%s-%d", res.URN, runAt.Unix()),
+		Kind:    jobType,
+		RunAt:   runAt,
 		Payload: payload,
-	})
+	}
+
+	if err := s.worker.Enqueue(ctx, job); err != nil && !errors.Is(err, worker.ErrJobExists) {
+		return err
+	}
+	return nil
 }
 
 // HandleSyncJob is meant to be invoked by asyncWorker when an enqueued job is
@@ -94,7 +103,7 @@ func (s *Service) syncChange(ctx context.Context, urn string) (*resource.Resourc
 			return nil, err
 		}
 	} else {
-		if err := s.upsert(ctx, *res, false); err != nil {
+		if err := s.upsert(ctx, module.Plan{Resource: *res}, false); err != nil {
 			return nil, err
 		}
 	}

core/module/registry.go

@@ -55,70 +55,77 @@ func (s *Service) ApplyAction(ctx context.Context, urn string, act module.Action
 }
 
 func (s *Service) execAction(ctx context.Context, res resource.Resource, act module.ActionRequest) (*resource.Resource, error) {
-	plannedRes, err := s.planChange(ctx, res, act)
+	planned, err := s.planChange(ctx, res, act)
 	if err != nil {
 		return nil, err
 	}
 
 	if isCreate(act.Name) {
-		plannedRes.CreatedAt = s.clock()
-		plannedRes.UpdatedAt = plannedRes.CreatedAt
+		planned.Resource.CreatedAt = s.clock()
+		planned.Resource.UpdatedAt = planned.Resource.CreatedAt
 	} else {
-		plannedRes.CreatedAt = res.CreatedAt
-		plannedRes.UpdatedAt = s.clock()
+		planned.Resource.CreatedAt = res.CreatedAt
+		planned.Resource.UpdatedAt = s.clock()
 	}
 
-	if err := s.upsert(ctx, *plannedRes, isCreate(act.Name)); err != nil {
+	if err := s.upsert(ctx, *planned, isCreate(act.Name)); err != nil {
 		return nil, err
 	}
-	return plannedRes, nil
+	return &planned.Resource, nil
 }
 
 func isCreate(actionName string) bool {
 	return actionName == module.CreateAction
 }
 
-func (s *Service) planChange(ctx context.Context, res resource.Resource, act module.ActionRequest) (*resource.Resource, error) {
+func (s *Service) planChange(ctx context.Context, res resource.Resource, act module.ActionRequest) (*module.Plan, error) {
 	modSpec, err := s.generateModuleSpec(ctx, res)
 	if err != nil {
 		return nil, err
 	}
 
-	plannedRes, err := s.rootModule.Plan(ctx, *modSpec, act)
+	planned, err := s.rootModule.Plan(ctx, *modSpec, act)
 	if err != nil {
 		if errors.Is(err, errors.ErrInvalid) {
 			return nil, err
 		}
 		return nil, errors.ErrInternal.WithMsgf("plan() failed").WithCausef(err.Error())
-	} else if err := plannedRes.Validate(isCreate(act.Name)); err != nil {
+	} else if err := planned.Resource.Validate(isCreate(act.Name)); err != nil {
 		return nil, err
 	}
 
-	return plannedRes, nil
+	return planned, nil
 }
 
-func (s *Service) upsert(ctx context.Context, res resource.Resource, isCreate bool) error {
-	hook := func(ctx context.Context) error {
-		if res.State.IsTerminal() {
+func (s *Service) upsert(ctx context.Context, plan module.Plan, isCreate bool) error {
+	var hooks []resource.MutationHook
+	hooks = append(hooks, func(ctx context.Context) error {
+		if plan.Resource.State.IsTerminal() {
 			// no need to enqueue if resource has reached terminal state.
 			return nil
 		}
 
-		return s.enqueueSyncJob(ctx, res)
+		return s.enqueueSyncJob(ctx, plan.Resource, s.clock(), JobKindSyncResource)
+	})
+
+	if !plan.ScheduleRunAt.IsZero() {
+		hooks = append(hooks, func(ctx context.Context) error {
+			return s.enqueueSyncJob(ctx, plan.Resource, plan.ScheduleRunAt, JobKindScheduledSyncResource)
+		})
 	}
 
 	var err error
 	if isCreate {
-		err = s.store.Create(ctx, res, hook)
+		err = s.store.Create(ctx, plan.Resource, hooks...)
 	} else {
-		err = s.store.Update(ctx, res, hook)
+		err = s.store.Update(ctx, plan.Resource, hooks...)
 	}
 
 	if err != nil {
 		if isCreate && errors.Is(err, errors.ErrConflict) {
-			return errors.ErrConflict.WithMsgf("resource with urn '%s' already exists", res.URN)
+			return errors.ErrConflict.WithMsgf("resource with urn '%s' already exists", plan.Resource.URN)
 		} else if !isCreate && errors.Is(err, errors.ErrNotFound) {
-			return errors.ErrNotFound.WithMsgf("resource with urn '%s' does not exist", res.URN)
+			return errors.ErrNotFound.WithMsgf("resource with urn '%s' does not exist", plan.Resource.URN)
 		}
 		return errors.ErrInternal.WithCausef(err.Error())
 	}