add kinsumer autoscale module

justtrackio · Jan 13, 2025 · 239e1b7 · 239e1b7
1 parent a6a2b1d
commit 239e1b7
Show file tree

Hide file tree

Showing 10 changed files with 910 additions and 2 deletions.
diff --git a/pkg/application/options.go b/pkg/application/options.go
@@ -190,6 +190,14 @@ func WithKernelExitHandler(handler kernelPkg.ExitHandler) Option {
 	}
 }
 
+func WithKinsumerAutoscaleModule(kinsumerInputName string) Option {
+	return func(app *App) {
+		app.addKernelOption(func(config cfg.GosoConf) kernelPkg.Option {
+			return kernelPkg.WithModuleMultiFactory(stream.KinsumerAutoscaleModuleFactory(kinsumerInputName))
+		})
+	}
+}
+
 func WithLoggerGroupTag(app *App) {
 	app.addLoggerOption(func(config cfg.GosoConf, logger log.GosoLogger) error {
 		if !config.IsSet("app_group") {

diff --git a/pkg/cloud/aws/ecs/client.go b/pkg/cloud/aws/ecs/client.go
@@ -18,6 +18,8 @@ import (
 type Client interface {
 	DescribeContainerInstances(ctx context.Context, params *ecs.DescribeContainerInstancesInput, optFns ...func(*ecs.Options)) (*ecs.DescribeContainerInstancesOutput, error)
 	DescribeTasks(ctx context.Context, params *ecs.DescribeTasksInput, optFns ...func(*ecs.Options)) (*ecs.DescribeTasksOutput, error)
+	DescribeServices(ctx context.Context, params *ecs.DescribeServicesInput, optFns ...func(*ecs.Options)) (*ecs.DescribeServicesOutput, error)
+	UpdateService(ctx context.Context, params *ecs.UpdateServiceInput, optFns ...func(*ecs.Options)) (*ecs.UpdateServiceOutput, error)
 }
 
 type ClientSettings struct {

diff --git a/pkg/cloud/aws/ecs/mocks/Client.go b/pkg/cloud/aws/ecs/mocks/Client.go
diff --git a/pkg/cloud/aws/kinesis/record_writer.go b/pkg/cloud/aws/kinesis/record_writer.go
@@ -5,8 +5,6 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/justtrackio/gosoline/pkg/funk"
-
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/kinesis"
 	"github.com/aws/aws-sdk-go-v2/service/kinesis/types"
@@ -15,6 +13,7 @@ import (
 	"github.com/justtrackio/gosoline/pkg/cfg"
 	"github.com/justtrackio/gosoline/pkg/clock"
 	"github.com/justtrackio/gosoline/pkg/exec"
+	"github.com/justtrackio/gosoline/pkg/funk"
 	"github.com/justtrackio/gosoline/pkg/log"
 	"github.com/justtrackio/gosoline/pkg/metric"
 	"github.com/justtrackio/gosoline/pkg/uuid"
@@ -178,6 +177,7 @@ func (w *recordWriter) putRecordsBatch(ctx context.Context, batch []*Record) err
 
 		if len(failedRecords) == 0 && attempt > 1 {
 			logger.Warn("PutRecords successful after %d attempts in %s", attempt, took)
+
 			break
 		}