fix: give env vars in migrator:

Arta Asadi · Arta Asadi · commit 86bcd9fe122b · 2025-04-16T13:03:40.000+02:00
diff --git a/jobs/post-install-job/job/migrations/tasks/migrator.go b/jobs/post-install-job/job/migrations/tasks/migrator.go
@@ -11,6 +11,7 @@ import (
 	"github.com/opengovern/opensecurity/jobs/post-install-job/db"
 	"github.com/opengovern/opensecurity/services/tasks/db/models"
 	"github.com/opengovern/opensecurity/services/tasks/worker"
+	"github.com/opengovern/opensecurity/services/tasks/worker/consts"
 	"github.com/xhit/go-str2duration/v2"
 	"gopkg.in/yaml.v3"
 	"gorm.io/gorm/clause"
@@ -27,6 +28,16 @@ import (
 	"go.uber.org/zap"
 )
 
+var (
+	ESAddress  = os.Getenv("ELASTICSEARCH_ADDRESS")
+	ESUsername = os.Getenv("ELASTICSEARCH_USERNAME")
+	ESPassword = os.Getenv("ELASTICSEARCH_PASSWORD")
+	ESIsOnAks  = os.Getenv("ELASTICSEARCH_ISONAKS")
+
+	InventoryBaseURL = os.Getenv("CORE_BASEURL")
+	NatsURL          = os.Getenv("NATS_URL")
+)
+
 type Migration struct {
 }
 
@@ -108,6 +119,18 @@ func (m Migration) Run(ctx context.Context, conf config.MigratorConfig, logger *
 			return err
 		}
 
+		defaultEnvVars := defaultEnvs(&task)
+		envVarsJsonData, err := json.Marshal(defaultEnvVars)
+		if err != nil {
+			return err
+		}
+
+		var envVarsJsonb pgtype.JSONB
+		err = envVarsJsonb.Set(envVarsJsonData)
+		if err != nil {
+			return err
+		}
+
 		timeoutFloat, err := parseToTotalSeconds(task.Timeout)
 		if err != nil {
 			return err
@@ -128,6 +151,7 @@ func (m Migration) Run(ctx context.Context, conf config.MigratorConfig, logger *
 			Timeout:             timeoutFloat,
 			NatsConfig:          natsJsonb,
 			ScaleConfig:         scaleJsonb,
+			EnvVars:             envVarsJsonb,
 		}).Error; err != nil {
 			return err
 		}
@@ -327,3 +351,21 @@ func loadCloudqlBinary(itDbm db.Database, logger *zap.Logger, task worker.Task)
 
 	return nil
 }
+
+func defaultEnvs(taskConfig *worker.Task) map[string]string {
+	return map[string]string{
+		consts.NatsURLEnv:                    NatsURL,
+		consts.NatsConsumerEnv:               taskConfig.NatsConfig.Consumer,
+		consts.NatsStreamNameEnv:             taskConfig.NatsConfig.Stream,
+		consts.NatsTopicNameEnv:              taskConfig.NatsConfig.Topic,
+		consts.NatsResultTopicNameEnv:        taskConfig.NatsConfig.ResultTopic,
+		consts.ElasticSearchAddressEnv:       ESAddress,
+		consts.ElasticSearchUsernameEnv:      ESUsername,
+		consts.ElasticSearchPasswordEnv:      ESPassword,
+		consts.ElasticSearchIsOnAksNameEnv:   ESIsOnAks,
+		consts.ElasticSearchIsOpenSearch:     "false",
+		consts.ElasticSearchAwsRegionEnv:     "",
+		consts.ElasticSearchAssumeRoleArnEnv: "",
+		consts.InventoryBaseURL:              InventoryBaseURL,
+	}
+}
diff --git a/services/tasks/api/tasks.go b/services/tasks/api/tasks.go
@@ -1,16 +1,48 @@
 package api
 
+import (
+	"time"
+)
+
 type TaskListResponse struct {
 	Items      []TaskResponse `json:"items"`
 	TotalCount int            `json:"total_count"`
 }
 
 type TaskResponse struct {
-	ID          string  `json:"id"`
-	Name        string  `json:"name"`
-	Description string  `json:"description"`
-	ImageUrl    string  `json:"image_url"`
-	Timeout     float64 `json:"timeout"`
+	ID              string `json:"id"`
+	Name            string `json:"name"`
+	Description     string `json:"description"`
+	ImageUrl        string `json:"image_url"`
+	SchedulesNumber int    `json:"schedules_number"`
+}
+
+type TaskDetailsResponse struct {
+	ID           string              `json:"id"`
+	Name         string              `json:"name"`
+	Description  string              `json:"description"`
+	ImageUrl     string              `json:"image_url"`
+	RunSchedules []RunScheduleObject `json:"run_schedules"`
+	Credentials  []string            `json:"credentials"`
+	EnvVars      []string            `json:"env_vars"`
+	ScaleConfig  ScaleConfig         `json:"scale_config"`
+}
+
+type ScaleConfig struct {
+	Stream       string `json:"stream" yaml:"stream"`
+	Consumer     string `json:"consumer" yaml:"consumer"`
+	LagThreshold string `json:"lag_threshold" yaml:"lag_threshold"`
+	MinReplica   int32  `json:"min_replica" yaml:"min_replica"`
+	MaxReplica   int32  `json:"max_replica" yaml:"max_replica"`
+
+	PollingInterval int32 `json:"polling_interval" yaml:"polling_interval"`
+	CooldownPeriod  int32 `json:"cooldown_period" yaml:"cooldown_period"`
+}
+
+type RunScheduleObject struct {
+	LastRun   *time.Time     `json:"last_run"`
+	Params    map[string]any `json:"params"`
+	Frequency float64        `json:"frequency"`
 }
 
 type RunTaskRequest struct {
diff --git a/services/tasks/db/models/task.go b/services/tasks/db/models/task.go
@@ -27,6 +27,7 @@ type Task struct {
 	Timeout             float64
 	NatsConfig          pgtype.JSONB
 	ScaleConfig         pgtype.JSONB
+	EnvVars             pgtype.JSONB
 }
 
 type TaskBinary struct {
diff --git a/services/tasks/http.go b/services/tasks/http.go
@@ -3,6 +3,8 @@ package tasks
 import (
 	"crypto/rsa"
 	"encoding/json"
+	"fmt"
+	"github.com/jackc/pgtype"
 	api2 "github.com/opengovern/og-util/pkg/api"
 	"github.com/opengovern/og-util/pkg/httpserver"
 	"github.com/opengovern/og-util/pkg/vault"
@@ -98,12 +100,17 @@ func (r *httpRoutes) ListTasks(ctx echo.Context) error {
 	}
 	var taskResponses []api.TaskResponse
 	for _, task := range items {
+		runSchedules, err := r.db.GetTaskRunSchedules(task.ID)
+		if err != nil {
+			r.logger.Error("failed to get task run schedules", zap.Error(err))
+			return ctx.JSON(http.StatusInternalServerError, "failed to get task run schedules")
+		}
 		taskResponses = append(taskResponses, api.TaskResponse{
-			ID:          task.ID,
-			Name:        task.Name,
-			Description: task.Description,
-			ImageUrl:    task.ImageUrl,
-			Timeout:     task.Timeout,
+			ID:              task.ID,
+			Name:            task.Name,
+			Description:     task.Description,
+			ImageUrl:        task.ImageUrl,
+			SchedulesNumber: len(runSchedules),
 		})
 	}
 
@@ -129,18 +136,68 @@ func (r *httpRoutes) GetTask(ctx echo.Context) error {
 		r.logger.Error("failed to get task results", zap.Error(err))
 		return ctx.JSON(http.StatusInternalServerError, "failed to get task results")
 	}
-	var taskResponse api.TaskResponse
-	taskResponse = api.TaskResponse{
-		ID:          task.ID,
-		Name:        task.Name,
-		Description: task.Description,
-		ImageUrl:    task.ImageUrl,
-		Timeout:     task.Timeout,
+	runSchedules, err := r.db.GetTaskRunSchedules(task.ID)
+	if err != nil {
+		r.logger.Error("failed to get task run schedules", zap.Error(err))
+		return ctx.JSON(http.StatusInternalServerError, "failed to get task run schedules")
+	}
+
+	var runSchedulesObjects []api.RunScheduleObject
+	for _, runSchedule := range runSchedules {
+		params, err := JSONBToMap(runSchedule.Params)
+		if err != nil {
+			r.logger.Error("failed to get task run params", zap.Error(err))
+			return ctx.JSON(http.StatusInternalServerError, "failed to get task run params")
+		}
+		runSchedulesObjects = append(runSchedulesObjects, api.RunScheduleObject{
+			LastRun:   runSchedule.LastRun,
+			Params:    params,
+			Frequency: runSchedule.Frequency,
+		})
+	}
+
+	var credentials []string
+	configSecrets, err := r.db.GetTaskConfigSecret(task.ID)
+	if err != nil {
+		r.logger.Error("failed to get task config secret", zap.Error(err))
+		return ctx.JSON(http.StatusInternalServerError, "failed to get task config secret")
+	}
+	if configSecrets != nil {
+		mapData, err := r.vault.Decrypt(ctx.Request().Context(), configSecrets.Secret)
+		if err != nil {
+			r.logger.Error("failed to decrypt secret", zap.Error(err))
+			return echo.NewHTTPError(http.StatusInternalServerError, "failed to decrypt config")
+		}
+		for k := range mapData {
+			credentials = append(credentials, k)
+		}
+	}
+
+	taskResponse := api.TaskDetailsResponse{
+		ID:           task.ID,
+		Name:         task.Name,
+		Description:  task.Description,
+		ImageUrl:     task.ImageUrl,
+		RunSchedules: runSchedulesObjects,
+		Credentials:  credentials,
 	}
 
 	return ctx.JSON(http.StatusOK, taskResponse)
 }
 
+func JSONBToMap(jsonb pgtype.JSONB) (map[string]any, error) {
+	if jsonb.Status != pgtype.Present {
+		return nil, fmt.Errorf("JSONB data is not present")
+	}
+
+	var result map[string]any
+	if err := json.Unmarshal(jsonb.Bytes, &result); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal JSONB: %w", err)
+	}
+
+	return result, nil
+}
+
 // RunTask godoc
 //
 //	@Summary	Run a new task
diff --git a/services/tasks/scheduler/service.go b/services/tasks/scheduler/service.go
@@ -63,7 +63,7 @@ func (s *MainScheduler) Start(ctx context.Context) error {
 		if !ok {
 			return fmt.Errorf("current namespace lookup failed")
 		}
-		err = worker.CreateWorker(ctx, s.cfg, s.kubeClient, &task, currentNamespace)
+		err = worker.CreateWorker(ctx, s.kubeClient, &task, currentNamespace)
 		if err != nil {
 			return err
 		}
diff --git a/services/tasks/worker/worker.go b/services/tasks/worker/worker.go
@@ -5,37 +5,35 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/jackc/pgtype"
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
-	"github.com/opengovern/opensecurity/services/tasks/config"
 	"github.com/opengovern/opensecurity/services/tasks/db/models"
-	"github.com/opengovern/opensecurity/services/tasks/worker/consts"
 	"golang.org/x/net/context"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"os"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"strconv"
 )
 
-var (
-	ESAddress  = os.Getenv("ELASTICSEARCH_ADDRESS")
-	ESUsername = os.Getenv("ELASTICSEARCH_USERNAME")
-	ESPassword = os.Getenv("ELASTICSEARCH_PASSWORD")
-	ESIsOnAks  = os.Getenv("ELASTICSEARCH_ISONAKS")
-
-	InventoryBaseURL = os.Getenv("CORE_BASEURL")
-)
-
-func CreateWorker(ctx context.Context, cfg config.Config, kubeClient client.Client, taskConfig *models.Task, namespace string) error {
+func CreateWorker(ctx context.Context, kubeClient client.Client, taskConfig *models.Task, namespace string) error {
 	soNatsUrl, _ := os.LookupEnv("SCALED_OBJECT_NATS_URL")
 
-	env, err := defaultEnvs(cfg, taskConfig)
-	if err != nil {
-		return err
+	var envVars map[string]string
+	if taskConfig.EnvVars.Status == pgtype.Present {
+		if err := json.Unmarshal(taskConfig.EnvVars.Bytes, &envVars); err != nil {
+			return err
+		}
+	}
+
+	var env []corev1.EnvVar
+	for k, v := range envVars {
+		env = append(env, corev1.EnvVar{
+			Name:  k,
+			Value: v,
+		})
 	}
 
 	var deployment appsv1.Deployment
-	err = kubeClient.Get(ctx, client.ObjectKey{
+	err := kubeClient.Get(ctx, client.ObjectKey{
 		Namespace: namespace,
 		Name:      taskConfig.ID,
 	}, &deployment)
@@ -77,15 +75,15 @@ func CreateWorker(ctx context.Context, cfg config.Config, kubeClient client.Clie
 				},
 			},
 		}
-		err := kubeClient.Create(ctx, &deployment)
+		err = kubeClient.Create(ctx, &deployment)
 		if err != nil {
 			return err
 		}
 	}
 
 	var scaleConfig ScaleConfig
 	if taskConfig.ScaleConfig.Status == pgtype.Present {
-		if err := json.Unmarshal(taskConfig.ScaleConfig.Bytes, &scaleConfig); err != nil {
+		if err = json.Unmarshal(taskConfig.ScaleConfig.Bytes, &scaleConfig); err != nil {
 			return err
 		}
 	}
@@ -139,67 +137,3 @@ func CreateWorker(ctx context.Context, cfg config.Config, kubeClient client.Clie
 
 	return nil
 }
-
-func defaultEnvs(cfg config.Config, taskConfig *models.Task) ([]corev1.EnvVar, error) {
-	var natsConfig NatsConfig
-	if taskConfig.NatsConfig.Status == pgtype.Present {
-		if err := json.Unmarshal(taskConfig.NatsConfig.Bytes, &natsConfig); err != nil {
-			return nil, err
-		}
-	}
-
-	return []corev1.EnvVar{
-		{
-			Name:  consts.NatsURLEnv,
-			Value: cfg.NATS.URL,
-		},
-		{
-			Name:  consts.NatsConsumerEnv,
-			Value: natsConfig.Consumer,
-		},
-		{
-			Name:  consts.NatsStreamNameEnv,
-			Value: natsConfig.Stream,
-		},
-		{
-			Name:  consts.NatsTopicNameEnv,
-			Value: natsConfig.Topic,
-		},
-		{
-			Name:  consts.NatsResultTopicNameEnv,
-			Value: natsConfig.ResultTopic,
-		},
-		{
-			Name:  consts.ElasticSearchAddressEnv,
-			Value: ESAddress,
-		},
-		{
-			Name:  consts.ElasticSearchUsernameEnv,
-			Value: ESUsername,
-		},
-		{
-			Name:  consts.ElasticSearchPasswordEnv,
-			Value: ESPassword,
-		},
-		{
-			Name:  consts.ElasticSearchIsOnAksNameEnv,
-			Value: ESIsOnAks,
-		},
-		{
-			Name:  consts.ElasticSearchIsOpenSearch,
-			Value: strconv.FormatBool(cfg.ElasticSearch.IsOpenSearch),
-		},
-		{
-			Name:  consts.ElasticSearchAwsRegionEnv,
-			Value: "",
-		},
-		{
-			Name:  consts.ElasticSearchAssumeRoleArnEnv,
-			Value: "",
-		},
-		{
-			Name:  consts.InventoryBaseURL,
-			Value: InventoryBaseURL,
-		},
-	}, nil
-}

Original file line number	Diff line number	Diff line change
`@@ -27,6 +27,7 @@ type Task struct {`
`27`	`27`	`Timeout float64`
`28`	`28`	`NatsConfig pgtype.JSONB`
`29`	`29`	`ScaleConfig pgtype.JSONB`
	`30`	`+ EnvVars pgtype.JSONB`
`30`	`31`	`}`
`31`	`32`
`32`	`33`	`type TaskBinary struct {`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ func (s *MainScheduler) Start(ctx context.Context) error {`
`63`	`63`	`if !ok {`
`64`	`64`	`return fmt.Errorf("current namespace lookup failed")`
`65`	`65`	`}`
`66`		`- err = worker.CreateWorker(ctx, s.cfg, s.kubeClient, &task, currentNamespace)`
	`66`	`+ err = worker.CreateWorker(ctx, s.kubeClient, &task, currentNamespace)`
`67`	`67`	`if err != nil {`
`68`	`68`	`return err`
`69`	`69`	`}`