Async Integration Tests using Binaries

cmd/api-server/main.go

@@ -7,23 +7,21 @@ import (
 	"log/slog"
 	"net"
 	"os"
-	"syscall"
 	"time"
 
 	"github.com/openkcm/common-sdk/pkg/commoncfg"
 	"github.com/openkcm/common-sdk/pkg/health"
 	"github.com/openkcm/common-sdk/pkg/logger"
 	"github.com/openkcm/common-sdk/pkg/otlp"
-	"github.com/openkcm/common-sdk/pkg/status"
 	"github.com/samber/oops"
 
 	"github.com/openkcm/cmk/internal/config"
 	"github.com/openkcm/cmk/internal/constants"
 	"github.com/openkcm/cmk/internal/daemon"
 	"github.com/openkcm/cmk/internal/db"
-	"github.com/openkcm/cmk/internal/db/dsn"
 	"github.com/openkcm/cmk/internal/log"
 	"github.com/openkcm/cmk/utils/cmd"
+	statusserver "github.com/openkcm/cmk/utils/status_server"
 )
 
 var (
@@ -35,6 +33,8 @@ var (
 
 // - Starts the status server
 // - Starts the CMK API Server
+//
+//nolint:funlen
 func run(ctx context.Context, cfg *config.Config) error {
 	// Update Version
 	err := commoncfg.UpdateConfigVersion(&cfg.BaseConfig, BuildInfo)
@@ -60,7 +60,23 @@ func run(ctx context.Context, cfg *config.Config) error {
 	}
 
 	// Start status server
-	startStatusServer(ctx, cfg)
+	statusserver.StartStatusServer(ctx, cfg, health.WithCheck(health.Check{
+		Name: "HTTP Server",
+		Check: func(ctx context.Context) error {
+			dialer := &net.Dialer{
+				Timeout: time.Second * 1,
+			}
+			conn, err := dialer.DialContext(ctx, "tcp", cfg.HTTP.Address)
+			if err != nil {
+				return fmt.Errorf("health check: cannot connect to %s: %w", cfg.HTTP.Address, err)
+			}
+			defer func() { _ = conn.Close() }()
+
+			return nil
+		},
+	}))
+
+	go daemon.MonitorKeystorePoolSize(ctx, cfg)
 
 	// Database initialisation
 	dbCon, err := db.StartDB(ctx, cfg)
@@ -91,47 +107,6 @@ func run(ctx context.Context, cfg *config.Config) error {
 	return nil
 }
 
-func startStatusServer(ctx context.Context, cfg *config.Config) {
-	dsnFromConfig, err := dsn.FromDBConfig(cfg.Database)
-	if err != nil {
-		log.Error(ctx, "Could not load DSN from database config", err)
-	}
-
-	healthOptions := []health.Option{
-		health.WithDatabaseChecker(
-			constants.DBDriver,
-			dsnFromConfig,
-		),
-		health.WithCheck(health.Check{
-			Name: "HTTP Server",
-			Check: func(ctx context.Context) error {
-				dialer := &net.Dialer{
-					Timeout: time.Second * 1,
-				}
-
-				conn, err := dialer.DialContext(ctx, "tcp", cfg.HTTP.Address)
-				if err != nil {
-					return fmt.Errorf("health check: cannot connect to %s: %w", cfg.HTTP.Address, err)
-				}
-				defer func() { _ = conn.Close() }()
-
-				return nil
-			},
-		}),
-	}
-
-	go daemon.MonitorKeystorePoolSize(ctx, cfg)
-
-	go func() {
-		err := status.Serve(ctx, &cfg.BaseConfig, healthOptions...)
-		if err != nil {
-			log.Error(ctx, "Failure on the status server", err)
-
-			_ = syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
-		}
-	}()
-}
-
 // main is the entry point for the application. It is intentionally kept small
 // because it is hard to test, which would lower test coverage.
 func main() {

cmd/event-reconciler/main.go

@@ -4,25 +4,21 @@ import (
 	"context"
 	"flag"
 	"os"
-	"syscall"
 
 	"github.com/openkcm/common-sdk/pkg/commoncfg"
-	"github.com/openkcm/common-sdk/pkg/health"
 	"github.com/openkcm/common-sdk/pkg/logger"
 	"github.com/openkcm/common-sdk/pkg/otlp"
-	"github.com/openkcm/common-sdk/pkg/status"
 	"github.com/samber/oops"
 
 	"github.com/openkcm/cmk/internal/clients"
 	"github.com/openkcm/cmk/internal/config"
-	"github.com/openkcm/cmk/internal/constants"
 	"github.com/openkcm/cmk/internal/db"
-	"github.com/openkcm/cmk/internal/db/dsn"
 	eventprocessor "github.com/openkcm/cmk/internal/event-processor"
 	"github.com/openkcm/cmk/internal/log"
 	cmkpluginregistry "github.com/openkcm/cmk/internal/pluginregistry"
 	"github.com/openkcm/cmk/internal/repo/sql"
 	"github.com/openkcm/cmk/utils/cmd"
+	statusserver "github.com/openkcm/cmk/utils/status_server"
 )
 
 var (
@@ -57,7 +53,7 @@ func run(ctx context.Context, cfg *config.Config) error {
 	}
 
 	// Start status server
-	startStatusServer(ctx, cfg)
+	statusserver.StartStatusServer(ctx, cfg)
 
 	// Database initialisation
 	dbCon, err := db.StartDB(ctx, cfg)
@@ -96,29 +92,6 @@ func run(ctx context.Context, cfg *config.Config) error {
 	return nil
 }
 
-func startStatusServer(ctx context.Context, cfg *config.Config) {
-	dsnFromConfig, err := dsn.FromDBConfig(cfg.Database)
-	if err != nil {
-		log.Error(ctx, "Could not load DSN from database config", err)
-	}
-
-	healthOptions := []health.Option{
-		health.WithDatabaseChecker(
-			constants.DBDriver,
-			dsnFromConfig,
-		),
-	}
-
-	go func() {
-		err := status.Serve(ctx, &cfg.BaseConfig, healthOptions...)
-		if err != nil {
-			log.Error(ctx, "Failure on the status server", err)
-
-			_ = syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
-		}
-	}()
-}
-
 func main() {
 	flag.Parse()
 

cmd/task-cli/commands/root.go

@@ -5,12 +5,14 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/openkcm/cmk/internal/config"
 	cliUtils "github.com/openkcm/cmk/utils/cli"
 )
 
-func NewRootCmd(ctx context.Context) *cobra.Command {
+func NewRootCmd(ctx context.Context, cfg *config.Config) *cobra.Command {
 	return cliUtils.NewRootCmdWithInfinitySleep(
 		ctx,
+		cfg,
 		"task",
 		"Async Task CLI",
 		"CLI tool to manage and invoke CMK asynchronous tasks.",

cmd/task-cli/commands/root_test.go

@@ -11,18 +11,19 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/openkcm/cmk/cmd/task-cli/commands"
+	"github.com/openkcm/cmk/internal/config"
 )
 
 func TestRootCmdProvidesSleepFlag(t *testing.T) {
 	ctx := context.Background()
-	cmd := commands.NewRootCmd(ctx)
+	cmd := commands.NewRootCmd(ctx, &config.Config{})
 
 	assert.NotNil(t, cmd.PersistentFlags().Lookup("sleep"))
 }
 
 func TestRootCmdSleepModePrintsStartAndShutdown(t *testing.T) {
 	ctx := context.Background()
-	cmd := commands.NewRootCmd(ctx)
+	cmd := commands.NewRootCmd(ctx, &config.Config{})
 
 	var out bytes.Buffer
 	cmd.SetOut(&out)

cmd/task-cli/main.go

@@ -65,7 +65,7 @@ func run(ctx context.Context, cfg *config.Config) error {
 	asyncClient := asyncApp.Client()
 	asyncInspector := asyncApp.Inspector()
 
-	rootCmd := commands.NewRootCmd(ctx)
+	rootCmd := commands.NewRootCmd(ctx, cfg)
 	rootCmd.AddCommand(commands.NewStatsCmd(ctx, asyncInspector))
 	rootCmd.AddCommand(commands.NewQueuesCmd(ctx, asyncInspector))
 	rootCmd.AddCommand(commands.NewInvokeCmd(ctx, asyncClient))

cmd/task-scheduler/main.go

@@ -2,10 +2,14 @@ package main
 
 import (
 	"context"
-	"log"
+	"errors"
+	"flag"
+	"fmt"
+	"log/slog"
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/openkcm/common-sdk/pkg/commoncfg"
 	"github.com/openkcm/common-sdk/pkg/logger"
@@ -14,31 +18,29 @@ import (
 	"github.com/openkcm/cmk/internal/async"
 	"github.com/openkcm/cmk/internal/config"
 	"github.com/openkcm/cmk/internal/constants"
-	cmklog "github.com/openkcm/cmk/internal/log"
+	"github.com/openkcm/cmk/internal/log"
+	statusserver "github.com/openkcm/cmk/utils/status_server"
 )
 
-const AppName = "scheduler"
-
-func start() error {
-	ctx, cancelOnSignal := signal.NotifyContext(
-		context.Background(),
-		os.Interrupt, syscall.SIGTERM,
-	)
-
-	defer cancelOnSignal()
+var (
+	BuildInfo               = "{}"
+	gracefulShutdownSec     = flag.Int64("graceful-shutdown", 1, "graceful shutdown seconds")
+	gracefulShutdownMessage = flag.String("graceful-shutdown-message", "Graceful shutdown in %d seconds",
+		"graceful shutdown message")
+	ErrRegistryEnabled = errors.New("failed to create registry client")
+)
 
-	cfg, err := config.LoadConfig(commoncfg.WithEnvOverride(constants.APIName + "_task_scheduler"))
-	if err != nil {
-		return oops.In("main").Wrapf(err, "failed to load the config")
-	}
+const AppName = "scheduler"
 
-	// LoggerConfig initialisation
-	err = logger.InitAsDefault(cfg.Logger, cfg.Application)
+func run(ctx context.Context, cfg *config.Config) error {
+	err := logger.InitAsDefault(cfg.Logger, cfg.Application)
 	if err != nil {
 		return oops.In("main").
 			Wrapf(err, "Failed to initialise the logger")
 	}
 
+	statusserver.StartStatusServer(ctx, cfg)
+
 	cronJob, err := async.New(cfg)
 	if err != nil {
 		return oops.In("main").Wrapf(err, "failed to create the scheduler")
@@ -56,14 +58,50 @@ func start() error {
 		return oops.In("main").Wrapf(err, "failed to shutdown the scheduler")
 	}
 
-	cmklog.Info(ctx, "shutting down scheduler")
+	log.Info(ctx, "shutting down scheduler")
 
 	return nil
 }
 
-func main() {
-	err := start()
+// runFuncWithSignalHandling runs the given function with signal handling. When
+// a CTRL-C is received, the context will be cancelled on which the function can
+// act upon.
+// It returns the exitCode
+func runFuncWithSignalHandling(f func(context.Context, *config.Config) error) int {
+	ctx, cancelOnSignal := signal.NotifyContext(
+		context.Background(),
+		os.Interrupt, syscall.SIGTERM,
+	)
+	defer cancelOnSignal()
+
+	cfg, err := config.LoadConfig(commoncfg.WithEnvOverride(constants.APIName + "_task_scheduler"))
 	if err != nil {
-		log.Fatal(err)
+		log.Error(ctx, "Failed to load the configuration", err)
+		_, _ = fmt.Fprintln(os.Stderr, err)
+
+		return 1
 	}
+
+	log.Debug(ctx, "Starting the application", slog.Any("config", *cfg))
+
+	err = f(ctx, cfg)
+	if err != nil {
+		log.Error(ctx, "Failed to start the application", err)
+		_, _ = fmt.Fprintln(os.Stderr, err)
+
+		return 1
+	}
+
+	// graceful shutdown so running goroutines may finish
+	_, _ = fmt.Fprintln(os.Stderr, fmt.Sprintf(*gracefulShutdownMessage, *gracefulShutdownSec))
+	time.Sleep(time.Duration(*gracefulShutdownSec) * time.Second)
+
+	return 0
+}
+
+func main() {
+	flag.Parse()
+
+	exitCode := runFuncWithSignalHandling(run)
+	os.Exit(exitCode)
 }