feat: stop time in dagger

goto · Oct 7, 2024 · 4301fdc · 4301fdc
1 parent e346119
commit 4301fdc
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 4 deletions.
diff --git a/modules/dagger/config.go b/modules/dagger/config.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/goto/entropy/core/module"
 	"github.com/goto/entropy/modules"
@@ -85,6 +86,10 @@ var (
 	validateConfig = validator.FromJSONSchema(configSchemaRaw)
 )
 
+type StartParams struct {
+	StopTime *time.Time `json:"stop_time"`
+}
+
 type UsageSpec struct {
 	CPU    string `json:"cpu,omitempty" validate:"required"`
 	Memory string `json:"memory,omitempty" validate:"required"`
@@ -114,6 +119,7 @@ type Config struct {
 	State         string            `json:"state"`
 	JobState      string            `json:"job_state"`
 	ResetOffset   string            `json:"reset_offset"`
+	StopTime      *time.Time        `json:"stop_time,omitempty"`
 }
 
 type ChartValues struct {

diff --git a/modules/dagger/driver_plan.go b/modules/dagger/driver_plan.go
@@ -14,10 +14,11 @@ import (
 
 const SourceKafkaConsumerAutoOffsetReset = "SOURCE_KAFKA_CONSUMER_CONFIG_AUTO_OFFSET_RESET"
 const (
-	JobStateRunning   = "running"
-	JobStateSuspended = "suspended"
-	StateDeployed     = "DEPLOYED"
-	StateUserStopped  = "USER_STOPPED"
+	JobStateRunning    = "running"
+	JobStateSuspended  = "suspended"
+	StateDeployed      = "DEPLOYED"
+	StateUserStopped   = "USER_STOPPED"
+	StateSystemStopped = "SYSTEM_STOPPED"
 )
 
 func (dd *daggerDriver) Plan(_ context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) {
@@ -113,7 +114,16 @@ func (dd *daggerDriver) planChange(exr module.ExpandedResource, act module.Actio
 		curConf.State = StateDeployed
 		curConf.JobState = JobStateRunning
 
+		var startParams StartParams
+		if err := json.Unmarshal(act.Params, &startParams); err != nil {
+			return nil, errors.ErrInvalid.WithMsgf("invalid params for start action").WithCausef(err.Error())
+		}
+		if startParams.StopTime != nil {
+			curConf.StopTime = startParams.StopTime
+		}
+
 	}
+
 	immediately := dd.timeNow()
 
 	exr.Resource.Spec.Configs = modules.MustJSON(curConf)

diff --git a/modules/dagger/driver_sync.go b/modules/dagger/driver_sync.go
@@ -61,6 +61,16 @@ func (dd *daggerDriver) Sync(ctx context.Context, exr module.ExpandedResource) (
 		return &finalState, nil
 	}
 
+	finalState.NextSyncAt = conf.StopTime
+	if conf.StopTime != nil && conf.StopTime.Before(dd.timeNow()) {
+		conf.JobState = JobStateSuspended
+		conf.State = StateSystemStopped
+		if err := dd.releaseSync(ctx, exr.Resource, false, *conf, flinkOut.KubeCluster); err != nil {
+			return nil, err
+		}
+		finalState.NextSyncAt = nil
+	}
+
 	finalOut, err := dd.refreshOutput(ctx, exr.Resource, *conf, *out, flinkOut.KubeCluster)
 	if err != nil {
 		return nil, err