feat: add stop flag support

modules/firehose/config.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/goto/entropy/core/resource"
 	"github.com/goto/entropy/pkg/errors"
@@ -27,6 +28,8 @@ var (
 )
 
 type Config struct {
+	Stopped      bool              `json:"stopped"`
+	StopTime     *time.Time        `json:"stop_time,omitempty"`
 	Telegraf     map[string]any    `json:"telegraf,omitempty"`
 	Replicas     int               `json:"replicas"`
 	Namespace    string            `json:"namespace,omitempty"`

modules/firehose/driver.go

@@ -3,6 +3,7 @@ package firehose
 import (
 	"context"
 	"encoding/json"
+	"strings"
 
 	"github.com/goto/entropy/core/module"
 	"github.com/goto/entropy/modules/kubernetes"
@@ -18,6 +19,12 @@ const (
 	stepKafkaReset    = "consumer_reset"
 )
 
+const (
+	chartRepo = "https://goto.github.io/charts/"
+	chartName = "firehose"
+	imageRepo = "gotocompany/firehose"
+)
+
 var defaultDriverConf = driverConf{
 	Namespace: "firehose",
 	Telegraf: map[string]any{
@@ -67,12 +74,6 @@ type transientData struct {
 }
 
 func (*firehoseDriver) getHelmRelease(conf Config) *helm.ReleaseConfig {
-	const (
-		chartRepo = "https://odpf.github.io/charts/"
-		chartName = "firehose"
-		imageRepo = "odpf/firehose"
-	)
-
 	rc := helm.DefaultReleaseConfig()
 	rc.Name = conf.DeploymentID
 	rc.Repository = chartRepo
@@ -95,6 +96,29 @@ func (*firehoseDriver) getHelmRelease(conf Config) *helm.ReleaseConfig {
 	return rc
 }
 
+func mergeChartValues(cur, newVal *chartValues) (*chartValues, error) {
+	if newVal == nil {
+		return cur, nil
+	}
+
+	merged := chartValues{
+		ImageTag:        cur.ImageTag,
+		ChartVersion:    cur.ChartVersion,
+		ImagePullPolicy: cur.ImagePullPolicy,
+	}
+
+	newTag := strings.TrimSpace(newVal.ImageTag)
+	if newTag != "" {
+		if strings.Contains(newTag, ":") && !strings.HasPrefix(newTag, imageRepo) {
+			return nil, errors.ErrInvalid.
+				WithMsgf("unknown image repo: '%s', must start with '%s'", newTag, imageRepo)
+		}
+		merged.ImageTag = strings.TrimPrefix(newTag, imageRepo)
+	}
+
+	return &merged, nil
+}
+
 func readOutputData(exr module.ExpandedResource) (*Output, error) {
 	var curOut Output
 	if err := json.Unmarshal(exr.Resource.State.Output, &curOut); err != nil {

modules/firehose/driver_plan.go

@@ -38,9 +38,14 @@ func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.Act
 			return nil, err
 		}
 
+		chartVals, err := mergeChartValues(curConf.ChartValues, newConf.ChartValues)
+		if err != nil {
+			return nil, err
+		}
+
 		// restore configs that are not user-controlled.
 		newConf.DeploymentID = curConf.DeploymentID
-		newConf.ChartValues = curConf.ChartValues
+		newConf.ChartValues = chartVals
 		newConf.Namespace = curConf.Namespace
 		newConf.Telegraf = curConf.Telegraf
 
@@ -67,6 +72,7 @@ func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.Act
 
 	case UpgradeAction:
 		// upgrade the chart values to the latest project-level config.
+		// Note: upgrade/downgrade will happen based on module-level configs.
 		curConf.ChartValues = &fd.conf.ChartValues
 	}
 
@@ -91,10 +97,15 @@ func (fd *firehoseDriver) planCreate(exr module.ExpandedResource, act module.Act
 		return nil, err
 	}
 
+	chartVals, err := mergeChartValues(&fd.conf.ChartValues, conf.ChartValues)
+	if err != nil {
+		return nil, err
+	}
+
 	// set project defaults.
 	conf.Telegraf = fd.conf.Telegraf
 	conf.Namespace = fd.conf.Namespace
-	conf.ChartValues = &fd.conf.ChartValues
+	conf.ChartValues = chartVals
 
 	exr.Resource.Spec.Configs = mustJSON(conf)
 	exr.Resource.State = resource.State{

modules/firehose/driver_plan_test.go

@@ -71,6 +71,7 @@ func TestFirehoseDriver_Plan(t *testing.T) {
 					Project: "ABCDEFGHIJKLMNOPQRSTUVWXYZ",
 					Spec: resource.Spec{
 						Configs: mustJSON(map[string]any{
+							"stopped":       false,
 							"replicas":      1,
 							"namespace":     "firehose",
 							"deployment_id": "firehose-ABCDEFGHIJKLMNOPQRSTUVWXYZ-abcdefghij-9bf099",
@@ -137,6 +138,7 @@ func TestFirehoseDriver_Plan(t *testing.T) {
 					Project: "foo",
 					Spec: resource.Spec{
 						Configs: mustJSON(map[string]any{
+							"stopped":       false,
 							"replicas":      1,
 							"namespace":     "firehose",
 							"deployment_id": "firehose-foo-fh1",
@@ -224,6 +226,7 @@ func TestFirehoseDriver_Plan(t *testing.T) {
 					Project: "foo",
 					Spec: resource.Spec{
 						Configs: mustJSON(map[string]any{
+							"stopped":       false,
 							"replicas":      10,
 							"deployment_id": "firehose-deployment-x",
 							"env_variables": map[string]string{
@@ -376,6 +379,7 @@ func TestFirehoseDriver_Plan(t *testing.T) {
 					Project: "foo",
 					Spec: resource.Spec{
 						Configs: mustJSON(map[string]any{
+							"stopped":       false,
 							"replicas":      1,
 							"deployment_id": "firehose-deployment-x",
 							"chart_values": map[string]string{
@@ -412,6 +416,7 @@ func TestFirehoseDriver_Plan(t *testing.T) {
 					Project: "foo",
 					Spec: resource.Spec{
 						Configs: mustJSON(map[string]any{
+							"stopped":       false,
 							"replicas":      1,
 							"deployment_id": "firehose-deployment-x",
 							"chart_values": map[string]string{

modules/firehose/driver_sync.go

@@ -41,7 +41,7 @@ func (fd *firehoseDriver) Sync(ctx context.Context, exr module.ExpandedResource)
 			// we want to stop the current deployment. we do this by setting
 			// replicas to 0. But this value will not be persisted to DB since
 			// config changes during Sync() are not saved.
-			if pendingStep == stepReleaseStop {
+			if pendingStep == stepReleaseStop || conf.Stopped {
 				conf.Replicas = 0
 			}