fix(policy): run migrations on db only once for all policy services (#1040)

Closes #742 

1. Runs migrations just once and IFF at least one service requires a DB
connection
2. Resolves pass by value error where struct pointer `*db.Client`
outside for loop was not reassigned within `startService` function due
to copy of value. Returns a `*db.Client` instead of resolving with
`**db.Client` parameter for better readability.

Service starting logs after these changes:
```shell
time=2024-06-26T16:54:22.597-07:00 level=INFO msg="starting services"
time=2024-06-26T16:54:22.597-07:00 level=INFO msg="skipping migrations" namespace=wellknown service=wellknownconfiguration.WellKnownService runMigrations=true reason="service does not require a database"
time=2024-06-26T16:54:22.597-07:00 level=INFO msg="started service" namespace=wellknown service=wellknownconfiguration.WellKnownService
time=2024-06-26T16:54:22.597-07:00 level=INFO msg="skipping migrations" namespace=entityresolution service=entityresolution.EntityResolutionService runMigrations=true reason="service does not require a database"
time=2024-06-26T16:54:22.597-07:00 level=INFO msg="started service" namespace=entityresolution service=entityresolution.EntityResolutionService
time=2024-06-26T16:54:22.597-07:00 level=INFO msg="creating database client" namespace=policy
time=2024-06-26T16:54:22.611-07:00 level=INFO msg="running database migrations"
time=2024-06-26T16:54:22.612-07:00 level=INFO msg="running migration up" schema=opentdf_policy database=opentdf
time=2024-06-26T16:54:22.617-07:00 level=INFO msg="migration db info " "current version"=20240618000000
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="migration up complete" "post-op version"=20240618000000
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="database migrations complete" applied=0
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=policy service=policy.attributes.AttributesService
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="skipping migrations" namespace=policy service=policy.namespaces.NamespaceService runMigrations=true reason="required migrations already ran"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=policy service=policy.namespaces.NamespaceService
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="skipping migrations" namespace=policy service=policy.resourcemapping.ResourceMappingService runMigrations=true reason="required migrations already ran"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=policy service=policy.resourcemapping.ResourceMappingService
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="skipping migrations" namespace=policy service=policy.subjectmapping.SubjectMappingService runMigrations=true reason="required migrations already ran"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=policy service=policy.subjectmapping.SubjectMappingService
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="skipping migrations" namespace=policy service=policy.kasregistry.KeyAccessServerRegistryService runMigrations=true reason="required migrations already ran"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=policy service=policy.kasregistry.KeyAccessServerRegistryService
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="skipping migrations" namespace=health service=grpc.health.v1.Health runMigrations=true reason="service does not require a database"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=health service=grpc.health.v1.Health
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="skipping migrations" namespace=authorization service=authorization.AuthorizationService runMigrations=true reason="service does not require a database"
time=2024-06-26T16:54:22.619-07:00 level=DEBUG msg="authorization service client config" config="{ClientID:tdf-authorization-svc ClientSecret:secret TokenURL:http://localhost:8888/auth/realms/opentdf/protocol/openid-connect/token Scopes:[] EndpointParams:map[] AuthStyle:0 authStyleCache:{v:{v:<nil>}}}"
time=2024-06-26T16:54:22.619-07:00 level=DEBUG msg="authorization service token source created" token_source="&{new:0x14000412450 mu:{state:0 sema:0} t:<nil> expiryDelta:100}"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=authorization service=authorization.AuthorizationService
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="skipping migrations" namespace=kas service=kas.AccessService runMigrations=true reason="service does not require a database"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="started service" namespace=kas service=kas.AccessService
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="starting opentdf"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="services running"
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="service running" namespace=wellknown service=wellknownconfiguration.WellKnownService database=false
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="service running" namespace=entityresolution service=entityresolution.EntityResolutionService database=false
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="service running" namespace=policy service=policy.attributes.AttributesService database=true
time=2024-06-26T16:54:22.619-07:00 level=INFO msg="service running" namespace=policy service=policy.namespaces.NamespaceService database=true
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="service running" namespace=policy service=policy.resourcemapping.ResourceMappingService database=true
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="service running" namespace=policy service=policy.subjectmapping.SubjectMappingService database=true
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="service running" namespace=policy service=policy.kasregistry.KeyAccessServerRegistryService database=true
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="service running" namespace=health service=grpc.health.v1.Health database=false
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="service running" namespace=authorization service=authorization.AuthorizationService database=false
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="service running" namespace=kas service=kas.AccessService database=false
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="starting http server" address=:8080
time=2024-06-26T16:54:22.620-07:00 level=INFO msg="starting in process grpc server"
```

---------

Co-authored-by: Ryan Schumacher <[email protected]>

Author: jakedoublev

service/pkg/db/db_migration.go

@@ -55,6 +55,9 @@ func migrationInit(ctx context.Context, c *Client, migrations *embed.FS) (*goose
 // RunMigrations runs the migrations for the schema
 // Schema will be created if it doesn't exist
 func (c *Client) RunMigrations(ctx context.Context, migrations *embed.FS) (int, error) {
+	if migrations == nil {
+		return 0, fmt.Errorf("migrations FS is required to run migrations")
+	}
 	slog.Info("running migration up", slog.String("schema", c.config.Schema), slog.String("database", c.config.Database))
 
 	// Create schema if it doesn't exist

service/pkg/server/services.go

@@ -67,61 +67,82 @@ func startServices(ctx context.Context, cfg config.Config, otdf *server.OpenTDFS
 			continue
 		}
 
-		// Use a single database client per namespace
+		// Use a single database client per namespace and run migrations once per namespace
 		var d *db.Client
+		runMigrations := cfg.DB.RunMigrations
 
 		for _, r := range registers {
-			s, err := startService(ctx, cfg, r, otdf, eng, client, d, logger)
+			s, db, err := startService(ctx, &cfg, r, otdf, eng, client, d, &runMigrations, logger)
 			if err != nil {
 				return closeServices, services, err
 			}
+			d = db
 			services = append(services, s)
 		}
 	}
 
 	return closeServices, services, nil
 }
 
-func startService(ctx context.Context, cfg config.Config, s serviceregistry.Service, otdf *server.OpenTDFServer, eng *opa.Engine, client *sdk.SDK, d *db.Client, logger *logger.Logger) (serviceregistry.Service, error) {
-	// Create the database client if required
+func startService(
+	ctx context.Context,
+	cfg *config.Config,
+	s serviceregistry.Service,
+	otdf *server.OpenTDFServer,
+	eng *opa.Engine,
+	client *sdk.SDK,
+	d *db.Client,
+	runMigrations *bool,
+	logger *logger.Logger,
+) (serviceregistry.Service, *db.Client, error) {
+	// Create the database client only if required
 	if s.DB.Required && d == nil {
 		var err error
 
-		// Conditionally set the db client if the service requires it
-		// Currently, we are dynamically registering namespaces and don't offer the ability to apply
-		// config at the NS layer. This poses a problem where services under a NS want to share a
-		// database connection.
-		// TODO: this should be reassessed with how we handle registering a single namespace
 		logger.Info("creating database client", slog.String("namespace", s.Namespace))
-		// Make sure we only create a single db client per namespace
 		d, err = db.New(ctx, cfg.DB,
 			db.WithService(s.Namespace),
 			db.WithMigrations(s.DB.Migrations),
 		)
 		if err != nil {
-			return s, fmt.Errorf("issue creating database client for %s: %w", s.Namespace, err)
+			return s, d, fmt.Errorf("issue creating database client for %s: %w", s.Namespace, err)
 		}
 	}
 
-	// Run migrations if required
-	if cfg.DB.RunMigrations && d != nil {
-		if s.DB.Migrations == nil {
-			return s, fmt.Errorf("migrations FS is required when runMigrations is enabled")
-		}
-
+	// Run migrations IFF a service requires it and they're configured to run but haven't run yet
+	shouldRun := s.DB.Required && *runMigrations
+	if shouldRun {
 		logger.Info("running database migrations")
 		appliedMigrations, err := d.RunMigrations(ctx, s.DB.Migrations)
 		if err != nil {
-			return s, fmt.Errorf("issue running database migrations: %w", err)
+			return s, d, fmt.Errorf("issue running database migrations: %w", err)
 		}
 		logger.Info("database migrations complete",
 			slog.Int("applied", appliedMigrations),
 		)
-	} else {
+		// Only run migrations once
+		*runMigrations = false
+	}
+
+	if !shouldRun {
+		requiredAlreadyRan := s.DB.Required && cfg.DB.RunMigrations && !*runMigrations
+		noDBRequired := !s.DB.Required
+		migrationsDisabled := !cfg.DB.RunMigrations
+
+		reason := "undetermined"
+		if requiredAlreadyRan { //nolint:gocritic // This is more readable than a switch
+			reason = "required migrations already ran"
+		} else if noDBRequired {
+			reason = "service does not require a database"
+		} else if migrationsDisabled {
+			reason = "migrations are disabled"
+		}
+
 		logger.Info("skipping migrations",
 			slog.String("namespace", s.Namespace),
-			slog.String("reason", "runMigrations is false"),
-			slog.Bool("runMigrations", false),
+			slog.String("service", s.ServiceDesc.ServiceName),
+			slog.Bool("configured runMigrations", cfg.DB.RunMigrations),
+			slog.String("reason", reason),
 		)
 	}
 
@@ -146,7 +167,7 @@ func startService(ctx context.Context, cfg config.Config, s serviceregistry.Serv
 	// Register the service with the gRPC gateway
 	if err := handler(ctx, otdf.Mux, impl); err != nil {
 		logger.Error("failed to start service", slog.String("namespace", s.Namespace), slog.String("error", err.Error()))
-		return s, err
+		return s, d, err
 	}
 
 	logger.Info("started service", slog.String("namespace", s.Namespace), slog.String("service", s.ServiceDesc.ServiceName))
@@ -158,5 +179,5 @@ func startService(ctx context.Context, cfg config.Config, s serviceregistry.Serv
 			d.Close()
 		}
 	}
-	return s, nil
+	return s, d, nil
 }