StartProjections (#822)

Author: FoseFx

dev/docker-compose.yaml

@@ -1,5 +1,3 @@
-version: '3'
-
 services:
 
   postgres:

libs/common/grpc.go

@@ -101,7 +101,7 @@ func StartNewGRPCServer(ctx context.Context, addr string, registerServerHook fun
 	}
 
 	// Shut down Setup()'s resources
-	Shutdown()
+	Shutdown(nil)
 }
 
 // DefaultInterceptors returns the slice of default Interceptors for the GRPC service

libs/common/setup.go

@@ -9,15 +9,18 @@ import (
 	"github.com/rs/zerolog/log"
 	"hwutil"
 	"net/http"
+	"os"
 	"strings"
 	"telemetry"
+	"time"
 )
 
 var (
 	Mode                    string // Mode is set in Setup()
 	InsecureFakeTokenEnable = false
 	InstanceOrganizationID  *uuid.UUID
 	shutdownOpenTelemetryFn func() // cleanup function
+	contextCancel           func() // Setup() yields the "root" context, which can be canceled using this function
 )
 
 const DevelopmentMode = "development"
@@ -57,8 +60,9 @@ var skipAuthForMethods []string
 
 // Setup loads the .env file and sets up logging,
 // also sets up tokens when the service requires auth.
-func Setup(serviceName, version string, opts ...SetupOption) {
-	ctx := context.Background()
+func Setup(serviceName, version string, opts ...SetupOption) context.Context {
+	ctx, cancel := context.WithCancel(context.Background())
+	contextCancel = cancel
 
 	// Collect options
 	options := SetupOptions{}
@@ -130,6 +134,8 @@ func Setup(serviceName, version string, opts ...SetupOption) {
 
 		setupAuth(ctx, options.fakeAuthOnly)
 	}
+
+	return ctx
 }
 
 // ResolveAddrFromEnv uses the "APP_PORT", "PORT" and "ADDR" env variables to
@@ -150,7 +156,22 @@ func ResolveAddrFromEnv() string {
 // It should only ever be called after Setup() or SetupWithUnauthenticatedMethods()!
 // StartNewGRPCServer() already calls this function, so there is usually not need to call Shutdown() at all!
 // Keep in mind, that this shuts down the otel exporter, new traces won't be processed!
-func Shutdown() {
-	log.Info().Msg("shutting down otel")
-	shutdownOpenTelemetryFn()
+func Shutdown(err error) {
+	if shutdownOpenTelemetryFn != nil {
+		log.Info().Msg("shutting down otel")
+		shutdownOpenTelemetryFn()
+	}
+
+	if contextCancel != nil {
+		log.Info().Msg("canceling main context")
+		contextCancel()
+	}
+
+	time.Sleep(time.Second * 2) // give other resources some time to react to closed context (not sure if needed)
+
+	exitCode := 0
+	if err != nil {
+		exitCode = 1
+	}
+	os.Exit(exitCode)
 }

libs/hwes/eventstoredb/projections/custom/custom.go

@@ -28,6 +28,17 @@ type EventStoreClient interface {
 
 type eventHandler func(ctx context.Context, evt hwes.Event) (error, *esdb.NackAction)
 
+// ICustomProjection is an interface, implemented by CustomProjection,
+// and thus also implemented by all types that struct-embed CustomProjection
+type ICustomProjection interface {
+	// Subscribe creates and subscribes to a persistent subscription in EventStoreDB
+	// A persistent subscription is a type of subscription where the state is saved on the server-side
+	// This function blocks the thread until the passed context gets canceled
+	// https://developers.eventstore.com/server/v23.10/persistent-subscriptions.html
+	Subscribe(ctx context.Context) error
+	GetSubscriptionGroupName() string
+}
+
 // CustomProjection can be used to develop own projections
 // A projection is an event sourcing pattern to build up
 // a read model based on the underlying event data.
@@ -77,6 +88,14 @@ func NewCustomProjection(esdbClient EventStoreClient, subscriptionGroupName stri
 	}
 }
 
+func (p *CustomProjection) GetSubscriptionGroupName() string {
+	return p.subscriptionGroupName
+}
+
+func (p *CustomProjection) CustomProjection() *CustomProjection {
+	return p
+}
+
 func (p *CustomProjection) RegisterEventListener(eventType string, eventHandler eventHandler) *CustomProjection {
 	if _, found := p.eventHandlers[eventType]; found {
 		zlog.Error().
@@ -100,10 +119,6 @@ func (p *CustomProjection) HandleEvent(ctx context.Context, event hwes.Event) (e
 	return eventHandler(ctx, event)
 }
 
-// Subscribe creates and subscribes to a persistent subscription in EventStoreDB
-// A persistent subscription is a type of subscription where the state is saved on the server-side
-// This function blocks the thread until the passed context gets canceled
-// https://developers.eventstore.com/server/v23.10/persistent-subscriptions.html
 func (p *CustomProjection) Subscribe(ctx context.Context) error {
 	log := zlog.Ctx(ctx)
 

libs/hwes/eventstoredb/projections/start.go

@@ -0,0 +1,30 @@
+package projections
+
+import (
+	"context"
+	"github.com/rs/zerolog/log"
+	"hwes/eventstoredb/projections/custom"
+)
+
+// StartProjections starts the passed projections, onErr may be nil and is called on the first subscription error
+func StartProjections(ctx context.Context, onErr func(err error), projections ...custom.ICustomProjection) {
+	errCh := make(chan error)
+
+	for _, projection := range projections {
+		name := projection.GetSubscriptionGroupName()
+
+		// spawn new green thread, as Subscribe is blocking
+		go func() {
+			if err := projection.Subscribe(ctx); err != nil {
+				log.Err(err).Msgf("could not subscribe to %s", name)
+				errCh <- err // send error back to function thread
+			}
+		}()
+	}
+
+	err := <-errCh // wait for any errors
+
+	if onErr != nil {
+		onErr(err)
+	}
+}

services/ory-svc/main.go

@@ -43,9 +43,8 @@ func newErrAndLog(ctx context.Context, msg string) error {
 }
 
 func main() {
-	common.Setup(ServiceName, Version)
+	ctx := common.Setup(ServiceName, Version)
 
-	ctx := context.Background()
 	log := zlog.Ctx(ctx)
 
 	DaprPubsub = hwutil.GetEnvOr("DAPR_PUBSUB", "pubsub")
@@ -83,7 +82,7 @@ func main() {
 
 	// we don't use common.StartNewGRPCServer,
 	// so we have to call Shutdown manually
-	common.Shutdown()
+	common.Shutdown(nil)
 }
 
 func prepCtxForSvcToSvcCall(parentCtx context.Context, targetDaprAppId string) (context.Context, context.CancelFunc, error) {

services/property-svc/cmd/service/main.go

@@ -2,32 +2,32 @@ package service
 
 import (
 	"common"
-	"context"
 	"flag"
+
 	pb "gen/services/property_svc/v1"
+	daprd "github.com/dapr/go-sdk/service/grpc"
 	"github.com/rs/zerolog/log"
 	"hwdb"
 	"hwes/eventstoredb"
+	"hwes/eventstoredb/projections"
+
 	propertySet "property-svc/internal/property-set/api"
+	psh "property-svc/internal/property-set/handlers"
 	propertyValue "property-svc/internal/property-value/api"
+	pvh "property-svc/internal/property-value/handlers"
 	"property-svc/internal/property-value/projections/property_value_postgres_projection"
 	propertyViews "property-svc/internal/property-view/api"
+	pvih "property-svc/internal/property-view/handlers"
 	"property-svc/internal/property-view/projections/property_rules_postgres"
 	property "property-svc/internal/property/api"
-	"property-svc/internal/property/projections/property_postgres_projection"
-
-	daprd "github.com/dapr/go-sdk/service/grpc"
-	psh "property-svc/internal/property-set/handlers"
-	pvh "property-svc/internal/property-value/handlers"
-	pvih "property-svc/internal/property-view/handlers"
 	ph "property-svc/internal/property/handlers"
+	"property-svc/internal/property/projections/property_postgres_projection"
 )
 
 const ServiceName = "property-svc"
 
 func Main(version string, ready func()) {
-	ctx, cancel := context.WithCancel(context.Background())
-	common.Setup(ServiceName, version, common.WithAuth())
+	ctx := common.Setup(ServiceName, version, common.WithAuth())
 
 	replayMode := flag.Bool("replay", false, "")
 	flag.Parse()
@@ -39,44 +39,22 @@ func Main(version string, ready func()) {
 	eventStore := eventstoredb.SetupEventStoreByEnv()
 	aggregateStore := eventstoredb.NewAggregateStore(eventStore)
 
-	propertyPostgresProjection := property_postgres_projection.
-		NewProjection(eventStore, ServiceName, hwdb.GetDB())
-
-	propertyValuePostgresProjection := property_value_postgres_projection.
-		NewProjection(eventStore, ServiceName, hwdb.GetDB())
-
 	if *replayMode {
-		if err := replay(ctx, eventStore); err != nil {
+		err := replay(ctx, eventStore)
+		if err != nil {
 			log.Err(err).Msg("error during replay")
-			cancel()
 		}
-		// TODO: Find a more generic approach to run common.Shutdown()
-		common.Shutdown()
-		cancel()
+		common.Shutdown(err)
 		return
 	}
 
-	go func() {
-		if err := propertyPostgresProjection.Subscribe(ctx); err != nil {
-			log.Err(err).Msg("error during property-postgres projection subscription")
-			cancel()
-		}
-	}()
-
-	go func() {
-		if err := propertyValuePostgresProjection.Subscribe(ctx); err != nil {
-			log.Err(err).Msg("error during propertyValue-postgres projection subscription")
-			cancel()
-		}
-	}()
-
-	go func() {
-		taskViewsPostgresProjection := property_rules_postgres.NewProjection(eventStore, ServiceName)
-		if err := taskViewsPostgresProjection.Subscribe(ctx); err != nil {
-			log.Err(err).Msg("error during taskViewsPostgresProjection subscription")
-			cancel()
-		}
-	}()
+	go projections.StartProjections(
+		ctx,
+		common.Shutdown,
+		property_postgres_projection.NewProjection(eventStore, ServiceName, hwdb.GetDB()),
+		property_value_postgres_projection.NewProjection(eventStore, ServiceName, hwdb.GetDB()),
+		property_rules_postgres.NewProjection(eventStore, ServiceName),
+	)
 
 	propertyHandlers := ph.NewPropertyHandlers(aggregateStore)
 	propertySetHandlers := psh.NewPropertySetHandlers(aggregateStore)
@@ -96,5 +74,5 @@ func Main(version string, ready func()) {
 		}
 	})
 
-	cancel()
+	common.Shutdown(nil)
 }

services/property-svc/internal/property-view/api/grpc_test.go

@@ -25,13 +25,13 @@ import (
 	"google.golang.org/grpc"
 )
 
-func server(ctx context.Context) (pb.PropertyViewsServiceClient, *hwes_test.AggregateStore, func()) {
+func server() (context.Context, pb.PropertyViewsServiceClient, *hwes_test.AggregateStore, func()) {
 	// Build gRPC service
 	aggregateStore := hwes_test.NewAggregateStore()
 	propertyViewHandlers := handlers.NewPropertyViewHandlers(aggregateStore)
 	grpcService := api.NewPropertyViewService(aggregateStore, propertyViewHandlers)
 
-	common.Setup("property-svc", "test", common.WithFakeAuthOnly())
+	ctx := common.Setup("property-svc", "test", common.WithFakeAuthOnly())
 
 	// Start Server
 	grpcServer := grpc.NewServer(common.DefaultInterceptorChain())
@@ -40,12 +40,11 @@ func server(ctx context.Context) (pb.PropertyViewsServiceClient, *hwes_test.Aggr
 
 	client := pb.NewPropertyViewsServiceClient(conn)
 
-	return client, aggregateStore, closer
+	return ctx, client, aggregateStore, closer
 }
 
 func setup(t *testing.T) (ctx context.Context, client pb.PropertyViewsServiceClient, as *hwes_test.AggregateStore, dbMock pgxmock.PgxPoolIface, teardown func()) {
-	ctx = context.Background()
-	client, as, closer := server(ctx)
+	ctx, client, as, closer := server()
 	ctx = common_test.AuthenticatedUserContext(ctx, uuid.NewString())
 
 	dbMock, err := pgxmock.NewPool()

services/property-svc/internal/property/projections/property_postgres_projection/property_postgres_projection.go

@@ -30,7 +30,8 @@ func NewProjection(es *esdb.Client, serviceName string, db hwdb.DBTX) *Projectio
 	p := &Projection{
 		CustomProjection: custom.NewCustomProjection(es, subscriptionGroupName, &[]string{fmt.Sprintf("%s-", aggregate.PropertyAggregateType)}),
 		db:               db,
-		propertyRepo:     property_repo.New(db)}
+		propertyRepo:     property_repo.New(db),
+	}
 	p.initEventListeners()
 	return p
 }

services/task-svc/main.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"common"
-	"context"
 	pb "gen/services/task_svc/v1"
 	daprd "github.com/dapr/go-sdk/service/grpc"
 	"hwdb"
@@ -22,14 +21,14 @@ const ServiceName = "task-svc"
 var Version string
 
 func main() {
-	common.Setup(ServiceName, Version, common.WithAuth())
+	ctx := common.Setup(ServiceName, Version, common.WithAuth())
 
-	closeDBPool := hwdb.SetupDatabaseFromEnv(context.Background())
+	closeDBPool := hwdb.SetupDatabaseFromEnv(ctx)
 	defer closeDBPool()
 
 	tracking.SetupTracking(ServiceName, 10, 24*time.Hour, 20)
 
-	common.StartNewGRPCServer(context.Background(), common.ResolveAddrFromEnv(), func(server *daprd.Server) {
+	common.StartNewGRPCServer(ctx, common.ResolveAddrFromEnv(), func(server *daprd.Server) {
 		grpcServer := server.GrpcServer()
 		pb.RegisterTaskServiceServer(grpcServer, task.NewServiceServer())
 		pb.RegisterPatientServiceServer(grpcServer, patient.NewServiceServer())