Merge branch 'main' into warden-api

cli/server/configs.go

@@ -23,6 +23,7 @@ type serverConfig struct {
 	Compass     compassConfig    `mapstructure:"compass"`
 	Optimus     optimusConfig    `mapstructure:"optimus"`
 	StencilAddr string           `mapstructure:"stencil_addr"`
+	Dlq         dlqConfig        `mapstructure:"dlq"`
 	Warden      wardenConfig     `mapstructure:"warden"`
 }
 
@@ -54,6 +55,11 @@ type optimusConfig struct {
 	Addr string `mapstructure:"addr"`
 }
 
+type dlqConfig struct {
+	DlqJobImage    string `mapstructure:"dlq_job_image"`
+	PrometheusHost string `mapstructure:"prometheus_host"`
+}
+
 type serveConfig struct {
 	Host           string `mapstructure:"host" default:""`
 	Port           int    `mapstructure:"port" default:"8080"`

cli/server/server.go

@@ -5,7 +5,6 @@ import (
 
 	"buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/compass/v1beta1/compassv1beta1grpc"
 	entropyv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/entropy/v1beta1/entropyv1beta1grpc"
-	optimusv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/optimus/core/v1beta1/corev1beta1grpc"
 	shieldv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/shield/v1beta1/shieldv1beta1grpc"
 	sirenv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/siren/v1beta1/sirenv1beta1grpc"
 	"github.com/MakeNowJust/heredoc"
@@ -17,6 +16,8 @@ import (
 
 	"github.com/goto/dex/internal/server"
 	"github.com/goto/dex/internal/server/gcs"
+	"github.com/goto/dex/internal/server/v1/dlq"
+	"github.com/goto/dex/internal/server/v1/optimus"
 	"github.com/goto/dex/pkg/logger"
 	"github.com/goto/dex/pkg/telemetry"
 	"github.com/goto/dex/warden"
@@ -97,26 +98,28 @@ func runServer(baseCtx context.Context, nrApp *newrelic.Application, zapLog *zap
 		return err
 	}
 
-	optimusConn, err := grpc.Dial(cfg.Optimus.Addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		return err
-	}
 	gcsClient, err := gcs.NewClient(cfg.Service.GCSKeyFilePath)
 	if err != nil {
 		return err
 	}
 
 	wardenClient := warden.NewClient(cfg.Warden.Addr)
+	dlqConfig := &dlq.DlqJobConfig{
+		// TODO: map cfg.Dlq\
+		DlqJobImage:    cfg.Dlq.DlqJobImage,
+		PrometheusHost: cfg.Dlq.PrometheusHost,
+	}
 
 	return server.Serve(ctx, cfg.Service.Addr(), nrApp, zapLog,
 		shieldv1beta1.NewShieldServiceClient(shieldConn),
+		&optimus.ClientBuilder{},
 		entropyv1beta1.NewResourceServiceClient(entropyConn),
 		sirenv1beta1.NewSirenServiceClient(sirenConn),
 		compassv1beta1grpc.NewCompassServiceClient(compassConn),
-		optimusv1beta1.NewJobSpecificationServiceClient(optimusConn),
 		&gcs.Client{StorageClient: gcsClient},
 		cfg.Odin.Addr,
 		cfg.StencilAddr,
 		wardenClient,
+		dlqConfig,
 	)
 }

entropy/common.go

@@ -0,0 +1,6 @@
+package entropy
+
+type UsageSpec struct {
+	CPU    string `json:"cpu,omitempty" validate:"required"`
+	Memory string `json:"memory,omitempty" validate:"required"`
+}

entropy/entropy.go → entropy/firehose.go

@@ -2,7 +2,7 @@ package entropy
 
 import "time"
 
-type Config struct {
+type FirehoseConfig struct {
 	// Stopped flag when set forces the firehose to be stopped on next sync.
 	Stopped bool `json:"stopped"`
 
@@ -27,19 +27,14 @@ type Config struct {
 	// ResetOffset represents the value to which kafka consumer offset was set to
 	ResetOffset string `json:"reset_offset,omitempty"`
 
-	Limits        UsageSpec     `json:"limits,omitempty"`
-	Requests      UsageSpec     `json:"requests,omitempty"`
-	Telegraf      *Telegraf     `json:"telegraf,omitempty"`
-	ChartValues   *ChartValues  `json:"chart_values,omitempty"`
-	InitContainer InitContainer `json:"init_container,omitempty"`
+	Limits        UsageSpec             `json:"limits,omitempty"`
+	Requests      UsageSpec             `json:"requests,omitempty"`
+	Telegraf      *FirehoseTelegraf     `json:"telegraf,omitempty"`
+	ChartValues   *FirehoseChartValues  `json:"chart_values,omitempty"`
+	InitContainer FirehoseInitContainer `json:"init_container,omitempty"`
 }
 
-type UsageSpec struct {
-	CPU    string `json:"cpu,omitempty" validate:"required"`
-	Memory string `json:"memory,omitempty" validate:"required"`
-}
-
-type InitContainer struct {
+type FirehoseInitContainer struct {
 	Enabled bool `json:"enabled"`
 
 	Args    []string `json:"args"`
@@ -50,31 +45,31 @@ type InitContainer struct {
 	PullPolicy string `json:"pull_policy"`
 }
 
-type Telegraf struct {
-	Enabled bool           `json:"enabled,omitempty"`
-	Image   map[string]any `json:"image,omitempty"`
-	Config  TelegrafConf   `json:"config,omitempty"`
+type FirehoseTelegraf struct {
+	Enabled bool                 `json:"enabled,omitempty"`
+	Image   map[string]any       `json:"image,omitempty"`
+	Config  FirehoseTelegrafConf `json:"config,omitempty"`
 }
 
-type TelegrafConf struct {
+type FirehoseTelegrafConf struct {
 	Output               map[string]any    `json:"output"`
 	AdditionalGlobalTags map[string]string `json:"additional_global_tags"`
 }
 
-type ChartValues struct {
+type FirehoseChartValues struct {
 	ImageTag        string `json:"image_tag" validate:"required"`
 	ChartVersion    string `json:"chart_version" validate:"required"`
 	ImagePullPolicy string `json:"image_pull_policy" validate:"required"`
 }
 
-type ScaleParams struct {
+type FirehoseScaleParams struct {
 	Replicas int `json:"replicas"`
 }
 
-type StartParams struct {
+type FirehoseStartParams struct {
 	StopTime *time.Time `json:"stop_time"`
 }
 
-type ResetParams struct {
+type FirehoseResetParams struct {
 	To string `json:"to"`
 }

entropy/job.go

@@ -0,0 +1,42 @@
+package entropy
+
+type JobConfig struct {
+	Replicas   int32             `json:"replicas"`
+	Namespace  string            `json:"namespace"`
+	Name       string            `json:"name,omitempty"`
+	Containers []JobContainer    `json:"containers,omitempty"`
+	JobLabels  map[string]string `json:"job_labels,omitempty"`
+	Volumes    []JobVolume       `json:"volumes,omitempty"`
+	TTLSeconds *int32            `json:"ttl_seconds,omitempty"`
+}
+
+type JobVolume struct {
+	Name string
+	Kind string // secret or config-map. secret is for gcs/bq credential
+}
+
+type JobContainer struct {
+	Name              string            `json:"name"`
+	Image             string            `json:"image"`
+	ImagePullPolicy   string            `json:"image_pull_policy,omitempty"`
+	Command           []string          `json:"command,omitempty"`
+	Args              []string          `json:"args,omitempty"`
+	SecretsVolumes    []JobSecret       `json:"secrets_volumes,omitempty"`
+	ConfigMapsVolumes []JobConfigMap    `json:"config_maps_volumes,omitempty"`
+	Limits            UsageSpec         `json:"limits,omitempty"`
+	Requests          UsageSpec         `json:"requests,omitempty"`
+	EnvConfigMaps     []string          `json:"env_config_maps,omitempty"`
+	EnvVariables      map[string]string `json:"env_variables,omitempty"`
+	PreStopCmd        []string          `json:"pre_stop_cmd,omitempty"`
+	PostStartCmd      []string          `json:"post_start_cmd,omitempty"`
+}
+
+type JobSecret struct {
+	Name  string `json:"name"`
+	Mount string `json:"mount"`
+}
+
+type JobConfigMap struct {
+	Name  string `json:"name"`
+	Mount string `json:"mount"`
+}

entropy/kind.go

@@ -0,0 +1,6 @@
+package entropy
+
+const (
+	ResourceKindFirehose = "firehose"
+	ResourceKindJob      = "job"
+)

internal/server/server.go

@@ -7,7 +7,6 @@ import (
 
 	"buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/compass/v1beta1/compassv1beta1grpc"
 	entropyv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/entropy/v1beta1/entropyv1beta1grpc"
-	optimusv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/optimus/core/v1beta1/corev1beta1grpc"
 	shieldv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/shield/v1beta1/shieldv1beta1grpc"
 	sirenv1beta1 "buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/siren/v1beta1/sirenv1beta1grpc"
 	"github.com/go-chi/chi/v5"
@@ -33,14 +32,15 @@ import (
 func Serve(ctx context.Context, addr string,
 	nrApp *newrelic.Application, logger *zap.Logger,
 	shieldClient shieldv1beta1.ShieldServiceClient,
+	optimusClient optimusv1.OptimusClientBuilder,
 	entropyClient entropyv1beta1.ResourceServiceClient,
 	sirenClient sirenv1beta1.SirenServiceClient,
 	compassClient compassv1beta1grpc.CompassServiceClient,
-	optimusClient optimusv1beta1.JobSpecificationServiceClient,
 	gcsClient gcs.BlobStorageClient,
 	odinAddr string,
 	stencilAddr string,
 	wardenClient *warden.Client,
+	dlqConfig *dlqv1.DlqJobConfig,
 ) error {
 	alertSvc := alertsv1.NewService(sirenClient)
 
@@ -64,9 +64,9 @@ func Serve(ctx context.Context, addr string,
 		r.Get("/alertTemplates", alertSvc.HandleListTemplates())
 		r.Route("/subscriptions", alertsv1.SubscriptionRoutes(sirenClient, shieldClient))
 		r.Route("/alerts", alertsv1.AlertRoutes(sirenClient, shieldClient))
-		r.Route("/optimus", optimusv1.Routes(optimusClient))
+		r.Route("/optimus", optimusv1.Routes(shieldClient, optimusClient))
 		r.Route("/projects", projectsv1.Routes(shieldClient))
-		r.Route("/dlq", dlqv1.Routes(entropyClient, gcsClient))
+		r.Route("/dlq", dlqv1.Routes(entropyClient, gcsClient, dlqConfig))
 		r.Route("/firehoses", firehosev1.Routes(entropyClient, shieldClient, alertSvc, compassClient, odinAddr, stencilAddr))
 		r.Route("/kubernetes", kubernetesv1.Routes(entropyClient))
 		r.Route("/iam", iamv1.Routes(shieldClient, wardenClient))

internal/server/v1/dlq/errors.go

@@ -0,0 +1,8 @@
+package dlq
+
+import "errors"
+
+var (
+	ErrFirehoseNamespaceNotFound = errors.New("could not find firehose namespace from resource output")
+	ErrFirehoseNamespaceInvalid  = errors.New("invalid firehose namespace from resource output")
+)

internal/server/v1/dlq/handler.go

@@ -29,7 +29,7 @@ func (h *Handler) ListFirehoseDLQ(w http.ResponseWriter, r *http.Request) {
 		log.Println(err)
 		return
 	}
-	conf := &entropy.Config{}
+	conf := &entropy.FirehoseConfig{}
 	err = utils.ProtoStructToGoVal(resp.GetResource().GetSpec().GetConfigs(), conf)
 	if err != nil {
 		utils.WriteErr(w, err)
@@ -62,6 +62,9 @@ func (*Handler) listDlqJobs(w http.ResponseWriter, _ *http.Request) {
 }
 
 func (*Handler) createDlqJob(w http.ResponseWriter, _ *http.Request) {
+	// transform request body into DlqJob (validation?)
+	// call service.CreateDLQJob
+
 	utils.WriteJSON(w, http.StatusOK, map[string]interface{}{
 		"dlq_job": nil,
 	})