feat: list the topic and dates for a dlq-bucket

.golangci.yml

@@ -140,4 +140,4 @@ issues:
         - forbidigo
         - contextcheck
 severity:
-  default-severity: error
+  default-severity: error

Makefile

@@ -41,6 +41,10 @@ generate-mocks:
 	@mockery --srcpkg=buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/siren/v1beta1/sirenv1beta1grpc --name=SirenServiceClient
 	@mockery --srcpkg=buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/shield/v1beta1/shieldv1beta1grpc --name=ShieldServiceClient
 	@mockery --srcpkg=buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/optimus/core/v1beta1/corev1beta1grpc --name=JobSpecificationServiceClient
+	@mockery --srcpkg=buf.build/gen/go/gotocompany/proton/grpc/go/gotocompany/entropy/v1beta1/entropyv1beta1grpc --name=ResourceServiceClient
+	@mockery --srcpkg=./internal/server/gcs --name=BlobStorageClient
+	@mockery --srcpkg=./internal/server/gcs --name=BlobObjectClient
+	@mockery --srcpkg=./internal/server/gcs --name=ObjectIterator
 
 clean: tidy
 	@echo "Cleaning up build directories..."
@@ -64,3 +68,4 @@ download:
 setup:
 	@go install github.com/vektra/mockery/[email protected]
 	@go install mvdan.cc/[email protected]
+	@go install github.com/daixiang0/[email protected]

cli/server/configs.go

@@ -50,8 +50,9 @@ type optimusConfig struct {
 }
 
 type serveConfig struct {
-	Host string `mapstructure:"host" default:""`
-	Port int    `mapstructure:"port" default:"8080"`
+	Host           string `mapstructure:"host" default:""`
+	Port           int    `mapstructure:"port" default:"8080"`
+	GCSKeyFilePath string `mapstructure:"gcs_key_file_path"`
 }
 
 func (serveCfg serveConfig) Addr() string {

cli/server/server.go

@@ -16,6 +16,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/goto/dex/internal/server"
+	"github.com/goto/dex/internal/server/gcs"
 	"github.com/goto/dex/pkg/logger"
 	"github.com/goto/dex/pkg/telemetry"
 )
@@ -99,12 +100,17 @@ func runServer(baseCtx context.Context, nrApp *newrelic.Application, zapLog *zap
 	if err != nil {
 		return err
 	}
+	gcsClient, err := gcs.NewClient(cfg.Service.GCSKeyFilePath)
+	if err != nil {
+		return err
+	}
 	return server.Serve(ctx, cfg.Service.Addr(), nrApp, zapLog,
 		shieldv1beta1.NewShieldServiceClient(shieldConn),
 		entropyv1beta1.NewResourceServiceClient(entropyConn),
 		sirenv1beta1.NewSirenServiceClient(sirenConn),
 		compassv1beta1grpc.NewCompassServiceClient(compassConn),
 		optimusv1beta1.NewJobSpecificationServiceClient(optimusConn),
+		&gcs.Client{StorageClient: gcsClient},
 		cfg.Odin.Addr,
 		cfg.StencilAddr,
 	)

dex_server.yml

@@ -53,7 +53,7 @@ shield:
 
 # [Entropy](https://github.com/goto/entropy) client related configurations
 entropy:
-  addr: localhost:8010
+  addr: http://g-godata-systems-dex.golabs.io
   firehose_version: v0.5.0
 
 # [Siren](https://github.com/goto/siren) client related configurations
@@ -62,3 +62,9 @@ siren:
 
 odin:
   addr: localhost:8030
+
+compass:
+  addr: compass-grpc.integration.golabs.io:80
+
+optimus:
+  addr: optimus.staging.golabs.io:80

entropy/entropy.go

@@ -0,0 +1,81 @@
+package entropy
+
+import "time"
+
+type Config struct {
+	// Stopped flag when set forces the firehose to be stopped on next sync.
+	Stopped bool `json:"stopped"`
+
+	// StopTime can be set to schedule the firehose to be stopped at given time.
+	StopTime *time.Time `json:"stop_time,omitempty"`
+
+	// Replicas is the number of firehose instances to run.
+	Replicas int `json:"replicas"`
+
+	// Namespace is the target namespace where firehose should be deployed.
+	// Inherits from driver config.
+	Namespace string `json:"namespace,omitempty"`
+
+	// DeploymentID will be used as the release-name for the deployment.
+	// Must be shorter than 53 chars if set. If not set, one will be generated
+	// automatically.
+	DeploymentID string `json:"deployment_id,omitempty"`
+
+	// EnvVariables contains all the firehose environment config values.
+	EnvVariables map[string]string `json:"env_variables,omitempty"`
+
+	// 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"`
+}
+
+type UsageSpec struct {
+	CPU    string `json:"cpu,omitempty" validate:"required"`
+	Memory string `json:"memory,omitempty" validate:"required"`
+}
+
+type InitContainer struct {
+	Enabled bool `json:"enabled"`
+
+	Args    []string `json:"args"`
+	Command []string `json:"command"`
+
+	Repository string `json:"repository"`
+	ImageTag   string `json:"image_tag"`
+	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 TelegrafConf struct {
+	Output               map[string]any    `json:"output"`
+	AdditionalGlobalTags map[string]string `json:"additional_global_tags"`
+}
+
+type ChartValues 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 {
+	Replicas int `json:"replicas"`
+}
+
+type StartParams struct {
+	StopTime *time.Time `json:"stop_time"`
+}
+
+type ResetParams struct {
+	To       string `json:"to"`
+	Datetime string `json:"datetime"`
+}