diff --git a/internal/datastore/crdb/crdb.go b/internal/datastore/crdb/crdb.go index 68eaba536d..9e4dd9c7d5 100644 --- a/internal/datastore/crdb/crdb.go +++ b/internal/datastore/crdb/crdb.go @@ -320,6 +320,10 @@ func (cds *crdbDatastore) MetricsID() (string, error) { return common.MetricsIDFromURL(cds.dburl) } +func (cds *crdbDatastore) EngineName() string { + return Engine +} + func (cds *crdbDatastore) ReadWriteTx( ctx context.Context, f datastore.TxUserFunc, diff --git a/internal/datastore/crdb/crdb_test.go b/internal/datastore/crdb/crdb_test.go index a750a8e644..1db99869e4 100644 --- a/internal/datastore/crdb/crdb_test.go +++ b/internal/datastore/crdb/crdb_test.go @@ -206,7 +206,7 @@ func TestCRDBDatastoreWithIntegrity(t *testing.T) { //nolint:tparallel t.Parallel() b := testdatastore.RunCRDBForTesting(t, crdbTestVersion()) - test.All(t, crdbFactory.NewTester(test.DatastoreTesterFunc(func(_ testing.TB, revisionQuantization, gcInterval, gcWindow time.Duration, watchBufferLength uint16) (datastore.Datastore, error) { + test.AllWithExceptions(t, crdbFactory.NewTester(test.DatastoreTesterFunc(func(_ testing.TB, revisionQuantization, gcInterval, gcWindow time.Duration, watchBufferLength uint16) (datastore.Datastore, error) { ctx := t.Context() ds := b.NewDatastore(t, func(engine, uri string) datastore.Datastore { ds, err := NewCRDBDatastore( @@ -231,7 +231,7 @@ func TestCRDBDatastoreWithIntegrity(t *testing.T) { //nolint:tparallel }) return ds, nil - }))) + })), test.WithCategories(test.MigrationCategory)) unwrappedTester := test.DatastoreTesterFunc(func(_ testing.TB, revisionQuantization, gcInterval, gcWindow time.Duration, watchBufferLength uint16) (datastore.Datastore, error) { ctx := t.Context() diff --git a/internal/datastore/crdb/enginebuilder.go b/internal/datastore/crdb/enginebuilder.go new file mode 100644 index 0000000000..7387753afa --- /dev/null +++ b/internal/datastore/crdb/enginebuilder.go @@ -0,0 +1,79 @@ +package crdb + +import ( + "context" + "errors" + + "github.com/ccoveille/go-safecast/v2" + + "github.com/authzed/spicedb/internal/datastore/common" + "github.com/authzed/spicedb/internal/datastore/crdb/migrations" + datastorecfg "github.com/authzed/spicedb/pkg/cmd/datastore/dsconfig" + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/migration" +) + +func init() { + datastorecfg.RegisterEngine(Engine, newDatastoreFromConfig) + migration.RegisterMigratableEngine(Engine, migrations.CRDBMigrations, newMigrationDriverFromConfig, "add-schema-tables") +} + +func newMigrationDriverFromConfig(_ context.Context, cfg *migration.Config) (*migrations.CRDBDriver, error) { + return migrations.NewCRDBDriver(cfg.DatastoreURI) +} + +func newDatastoreFromConfig(ctx context.Context, opts datastorecfg.Config) (datastore.Datastore, error) { + if len(opts.ReadReplicaURIs) > 0 { + return nil, errors.New("read replicas are not supported for the CockroachDB datastore engine") + } + + maxRetries, err := safecast.Convert[uint8](opts.MaxRetries) + if err != nil { + return nil, errors.New("max-retries could not be cast to uint8") + } + + watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) + if err != nil { + return nil, err + } + + return NewCRDBDatastore( + ctx, + opts.URI, + GCWindow(opts.GCWindow), + RevisionQuantization(opts.RevisionQuantization), + MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), + ReadConnsMaxOpen(opts.ReadConnPool.MaxOpenConns), + ReadConnsMinOpen(opts.ReadConnPool.MinOpenConns), + ReadConnMaxIdleTime(opts.ReadConnPool.MaxIdleTime), + ReadConnMaxLifetime(opts.ReadConnPool.MaxLifetime), + ReadConnMaxLifetimeJitter(opts.ReadConnPool.MaxLifetimeJitter), + ReadConnHealthCheckInterval(opts.ReadConnPool.HealthCheckInterval), + ReadConnPingTimeout(opts.ReadConnPool.PingTimeout), + WithAcquireTimeout(opts.WriteAcquisitionTimeout), + WriteConnsMaxOpen(opts.WriteConnPool.MaxOpenConns), + WriteConnsMinOpen(opts.WriteConnPool.MinOpenConns), + WriteConnMaxIdleTime(opts.WriteConnPool.MaxIdleTime), + WriteConnMaxLifetime(opts.WriteConnPool.MaxLifetime), + WriteConnMaxLifetimeJitter(opts.WriteConnPool.MaxLifetimeJitter), + WriteConnHealthCheckInterval(opts.WriteConnPool.HealthCheckInterval), + WriteConnPingTimeout(opts.WriteConnPool.PingTimeout), + FollowerReadDelay(opts.FollowerReadDelay), + MaxRetries(maxRetries), + OverlapKey(opts.OverlapKey), + OverlapStrategy(opts.OverlapStrategy), + WatchBufferLength(opts.WatchBufferLength), + WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), + WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), + WatchConnectTimeout(opts.WatchConnectTimeout), + WithEnablePrometheusStats(opts.EnableDatastoreMetrics), + WithEnableConnectionBalancing(opts.EnableConnectionBalancing), + ConnectRate(opts.ConnectRate), + FilterMaximumIDCount(opts.FilterMaximumIDCount), + WithIntegrity(opts.RelationshipIntegrityEnabled), + AllowedMigrations(opts.AllowedMigrations), + WithColumnOptimization(opts.ExperimentalColumnOptimization), + IncludeQueryParametersInTraces(opts.IncludeQueryParametersInTraces), + WithWatchDisabled(opts.DisableWatchSupport), + ) +} diff --git a/internal/datastore/engines/engines.go b/internal/datastore/engines/engines.go new file mode 100644 index 0000000000..ea2bf1ac0e --- /dev/null +++ b/internal/datastore/engines/engines.go @@ -0,0 +1,13 @@ +// Package engines registers, via side effect of import, every datastore +// engine defined in this repository with pkg/cmd/datastore's engine registry. +// Blank-import it from any binary or test that constructs datastores by engine +// name through pkg/cmd/datastore.NewDatastore. +package engines + +import ( + _ "github.com/authzed/spicedb/internal/datastore/crdb" + _ "github.com/authzed/spicedb/internal/datastore/memdb" + _ "github.com/authzed/spicedb/internal/datastore/mysql" + _ "github.com/authzed/spicedb/internal/datastore/postgres" + _ "github.com/authzed/spicedb/internal/datastore/spanner" +) diff --git a/internal/datastore/memdb/enginebuilder.go b/internal/datastore/memdb/enginebuilder.go new file mode 100644 index 0000000000..248be8d7e6 --- /dev/null +++ b/internal/datastore/memdb/enginebuilder.go @@ -0,0 +1,23 @@ +package memdb + +import ( + "context" + "errors" + + log "github.com/authzed/spicedb/internal/logging" + datastorecfg "github.com/authzed/spicedb/pkg/cmd/datastore/dsconfig" + "github.com/authzed/spicedb/pkg/datastore" +) + +func init() { + datastorecfg.RegisterEngine(Engine, newDatastoreFromConfig) +} + +func newDatastoreFromConfig(_ context.Context, opts datastorecfg.Config) (datastore.Datastore, error) { + if len(opts.ReadReplicaURIs) > 0 { + return nil, errors.New("read replicas are not supported for the in-memory datastore engine") + } + + log.Warn().Msg("in-memory datastore is not persistent and not feasible to run in a high availability fashion") + return NewMemdbDatastore(opts.WatchBufferLength, opts.RevisionQuantization, opts.GCWindow) +} diff --git a/internal/datastore/memdb/memdb.go b/internal/datastore/memdb/memdb.go index a5fb74e8c5..b2d12cafd7 100644 --- a/internal/datastore/memdb/memdb.go +++ b/internal/datastore/memdb/memdb.go @@ -112,6 +112,10 @@ func (mdb *memdbDatastore) MetricsID() (string, error) { return "memdb", nil } +func (mdb *memdbDatastore) EngineName() string { + return Engine +} + func (mdb *memdbDatastore) UniqueID(_ context.Context) (string, error) { return mdb.uniqueID, nil } diff --git a/internal/datastore/memdb/memdb_test.go b/internal/datastore/memdb/memdb_test.go index b16298dc55..0c57e5ae91 100644 --- a/internal/datastore/memdb/memdb_test.go +++ b/internal/datastore/memdb/memdb_test.go @@ -1,4 +1,4 @@ -package memdb +package memdb_test import ( "context" @@ -11,6 +11,7 @@ import ( "github.com/stretchr/testify/require" "golang.org/x/sync/errgroup" + "github.com/authzed/spicedb/internal/datastore/memdb" "github.com/authzed/spicedb/pkg/datastore" "github.com/authzed/spicedb/pkg/datastore/options" test "github.com/authzed/spicedb/pkg/datastore/test" @@ -19,24 +20,25 @@ import ( "github.com/authzed/spicedb/pkg/tuple" ) -var memdbFactory = test.NewTesterFactory(ErrSerialization) +var memdbFactory = test.NewTesterFactory(memdb.ErrSerialization) type memDBTest struct{} func (memDBTest) New(_ testing.TB, revisionQuantization, _, gcWindow time.Duration, watchBufferLength uint16) (datastore.Datastore, error) { - return NewMemdbDatastore(watchBufferLength, revisionQuantization, gcWindow) + return memdb.NewMemdbDatastore(watchBufferLength, revisionQuantization, gcWindow) } func TestMemdbDatastore(t *testing.T) { // ConcurrentWrite tests require row-level locking; memdb uses a global write lock // and would deadlock if two write transactions were opened concurrently. - test.AllWithExceptions(t, memdbFactory.NewTester(memDBTest{}), test.WithCategories(test.ConcurrentWriteCategory)) + // Migration tests are excluded because memdb has no schema migrations. + test.AllWithExceptions(t, memdbFactory.NewTester(memDBTest{}), test.WithCategories(test.ConcurrentWriteCategory, test.MigrationCategory)) } func TestConcurrentWritePanic(t *testing.T) { require := require.New(t) - ds, err := NewMemdbDatastore(0, 1*time.Hour, 1*time.Hour) + ds, err := memdb.NewMemdbDatastore(0, 1*time.Hour, 1*time.Hour) require.NoError(err) ctx := t.Context() @@ -87,7 +89,7 @@ func TestConcurrentWritePanic(t *testing.T) { func TestConcurrentWriteRelsSucceed(t *testing.T) { require := require.New(t) - ds, err := NewMemdbDatastore(0, 1*time.Hour, 1*time.Hour) + ds, err := memdb.NewMemdbDatastore(0, 1*time.Hour, 1*time.Hour) require.NoError(err) ctx := t.Context() @@ -116,7 +118,7 @@ func TestConcurrentWriteRelsSucceed(t *testing.T) { func TestAnythingAfterCloseDoesNotPanic(t *testing.T) { require := require.New(t) - ds, err := NewMemdbDatastore(0, 1*time.Hour, 1*time.Hour) + ds, err := memdb.NewMemdbDatastore(0, 1*time.Hour, 1*time.Hour) require.NoError(err) lowestRevision, err := ds.HeadRevision(t.Context()) @@ -129,21 +131,21 @@ func TestAnythingAfterCloseDoesNotPanic(t *testing.T) { select { case err := <-errChan: - require.ErrorIs(err, ErrMemDBIsClosed) + require.ErrorIs(err, memdb.ErrMemDBIsClosed) case <-time.After(time.Second): require.Fail("expected an error but waited too long") } _, err = ds.Statistics(t.Context()) - require.ErrorIs(err, ErrMemDBIsClosed) + require.ErrorIs(err, memdb.ErrMemDBIsClosed) err = ds.CheckRevision(t.Context(), lowestRevision.Revision) - require.ErrorIs(err, ErrMemDBIsClosed) + require.ErrorIs(err, memdb.ErrMemDBIsClosed) _, err = ds.OptimizedRevision(t.Context()) - require.ErrorIs(err, ErrMemDBIsClosed) + require.ErrorIs(err, memdb.ErrMemDBIsClosed) reader := ds.SnapshotReader(datastore.NoRevision) _, err = reader.CountRelationships(t.Context(), "blah") - require.ErrorIs(err, ErrMemDBIsClosed) + require.ErrorIs(err, memdb.ErrMemDBIsClosed) } diff --git a/internal/datastore/mysql/datastore.go b/internal/datastore/mysql/datastore.go index 85e2b52a53..e72144466f 100644 --- a/internal/datastore/mysql/datastore.go +++ b/internal/datastore/mysql/datastore.go @@ -312,6 +312,10 @@ func (mds *mysqlDatastore) MetricsID() (string, error) { return common.MetricsIDFromURL(mds.url) } +func (mds *mysqlDatastore) EngineName() string { + return Engine +} + func (mds *mysqlDatastore) SnapshotReader(rev datastore.Revision) datastore.Reader { createTxFunc := func(ctx context.Context) (*sql.Tx, txCleanupFunc, error) { tx, err := mds.db.BeginTx(ctx, mds.readTxOptions) diff --git a/internal/datastore/mysql/enginebuilder.go b/internal/datastore/mysql/enginebuilder.go new file mode 100644 index 0000000000..413842a7e8 --- /dev/null +++ b/internal/datastore/mysql/enginebuilder.go @@ -0,0 +1,129 @@ +package mysql + +import ( + "context" + "errors" + "fmt" + + "github.com/ccoveille/go-safecast/v2" + sqlDriver "github.com/go-sql-driver/mysql" + + "github.com/authzed/spicedb/internal/datastore/common" + "github.com/authzed/spicedb/internal/datastore/mysql/migrations" + "github.com/authzed/spicedb/internal/datastore/proxy" + log "github.com/authzed/spicedb/internal/logging" + datastorecfg "github.com/authzed/spicedb/pkg/cmd/datastore/dsconfig" + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/migration" +) + +func init() { + datastorecfg.RegisterEngine(Engine, newDatastoreFromConfig) + migration.RegisterMigratableEngine(Engine, migrations.Manager, newMigrationDriverFromConfig, "add_schema_tables") +} + +func newMigrationDriverFromConfig(ctx context.Context, cfg *migration.Config) (*migrations.MySQLDriver, error) { + credentialsProvider, err := cfg.CredentialsProvider(ctx) + if err != nil { + return nil, err + } + + // Do this outside NewMySQLDriverFromDSN to avoid races on MySQL datastore tests + if err := sqlDriver.SetLogger(&log.Logger); err != nil { + return nil, fmt.Errorf("unable to set logging to mysql driver: %w", err) + } + + return migrations.NewMySQLDriverFromDSN(cfg.DatastoreURI, cfg.MySQLTablePrefix, credentialsProvider) +} + +func newDatastoreFromConfig(ctx context.Context, opts datastorecfg.Config) (datastore.Datastore, error) { + primary, err := newPrimaryDatastoreFromConfig(ctx, opts) + if err != nil { + return nil, err + } + + if len(opts.ReadReplicaURIs) > datastorecfg.MaxReplicaCount { + return nil, fmt.Errorf("too many read replicas, max is %d", datastorecfg.MaxReplicaCount) + } + + replicas := make([]datastore.ReadOnlyDatastore, 0, len(opts.ReadReplicaURIs)) + for index, replicaURI := range opts.ReadReplicaURIs { + uintIndex, err := safecast.Convert[uint32](index) + if err != nil { + return nil, errors.New("too many replicas") + } + replica, err := newReplicaDatastoreFromConfig(ctx, uintIndex, replicaURI, opts) + if err != nil { + return nil, err + } + replicas = append(replicas, replica) + } + + return proxy.NewCheckingReplicatedDatastore(primary, replicas...) +} + +func commonDatastoreOptionsFromConfig(opts datastorecfg.Config) ([]Option, error) { + maxRetries, err := safecast.Convert[uint8](opts.MaxRetries) + if err != nil { + return nil, errors.New("max-retries could not be cast to uint8") + } + + watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) + if err != nil { + return nil, err + } + + return []Option{ + TablePrefix(opts.TablePrefix), + MaxRetries(maxRetries), + OverrideLockWaitTimeout(1), + WithEnablePrometheusStats(opts.EnableDatastoreMetrics), + WatchBufferLength(opts.WatchBufferLength), + WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), + WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), + MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), + RevisionQuantization(opts.RevisionQuantization), + FilterMaximumIDCount(opts.FilterMaximumIDCount), + AllowedMigrations(opts.AllowedMigrations), + WithColumnOptimization(opts.ExperimentalColumnOptimization), + }, nil +} + +func newReplicaDatastoreFromConfig(ctx context.Context, replicaIndex uint32, replicaURI string, opts datastorecfg.Config) (datastore.ReadOnlyDatastore, error) { + mysqlOpts := []Option{ //nolint: prealloc // we're not concerned about perf here + MaxOpenConns(opts.ReadReplicaConnPool.MaxOpenConns), + ConnMaxIdleTime(opts.ReadReplicaConnPool.MaxIdleTime), + ConnMaxLifetime(opts.ReadReplicaConnPool.MaxLifetime), + CredentialsProviderName(opts.ReadReplicaCredentialsProviderName), + } + + commonOptions, err := commonDatastoreOptionsFromConfig(opts) + if err != nil { + return nil, err + } + mysqlOpts = append(mysqlOpts, commonOptions...) + return NewReadOnlyMySQLDatastore(ctx, replicaURI, replicaIndex, mysqlOpts...) +} + +func newPrimaryDatastoreFromConfig(ctx context.Context, opts datastorecfg.Config) (datastore.Datastore, error) { + mysqlOpts := []Option{ //nolint: prealloc // we're not concerned about perf here + GCInterval(opts.GCInterval), + GCWindow(opts.GCWindow), + GCInterval(opts.GCInterval), + GCEnabled(!opts.ReadOnly), + GCMaxOperationTime(opts.GCMaxOperationTime), + MaxOpenConns(opts.ReadConnPool.MaxOpenConns), + ConnMaxIdleTime(opts.ReadConnPool.MaxIdleTime), + ConnMaxLifetime(opts.ReadConnPool.MaxLifetime), + WithWatchDisabled(opts.DisableWatchSupport), + CredentialsProviderName(opts.CredentialsProviderName), + FollowerReadDelay(opts.FollowerReadDelay), + } + + commonOptions, err := commonDatastoreOptionsFromConfig(opts) + if err != nil { + return nil, err + } + mysqlOpts = append(mysqlOpts, commonOptions...) + return NewMySQLDatastore(ctx, opts.URI, mysqlOpts...) +} diff --git a/internal/datastore/postgres/enginebuilder.go b/internal/datastore/postgres/enginebuilder.go new file mode 100644 index 0000000000..ddc89fd0db --- /dev/null +++ b/internal/datastore/postgres/enginebuilder.go @@ -0,0 +1,139 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + + "github.com/ccoveille/go-safecast/v2" + + "github.com/authzed/spicedb/internal/datastore/common" + "github.com/authzed/spicedb/internal/datastore/postgres/migrations" + "github.com/authzed/spicedb/internal/datastore/proxy" + datastorecfg "github.com/authzed/spicedb/pkg/cmd/datastore/dsconfig" + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/migration" +) + +func init() { + datastorecfg.RegisterEngine(Engine, newDatastoreFromConfig) + migration.RegisterMigratableEngine(Engine, migrations.DatabaseMigrations, newMigrationDriverFromConfig, "add-schema-tables") +} + +func newMigrationDriverFromConfig(ctx context.Context, cfg *migration.Config) (*migrations.AlembicPostgresDriver, error) { + credentialsProvider, err := cfg.CredentialsProvider(ctx) + if err != nil { + return nil, err + } + return migrations.NewAlembicPostgresDriver(ctx, cfg.DatastoreURI, credentialsProvider, false) +} + +func newDatastoreFromConfig(ctx context.Context, opts datastorecfg.Config) (datastore.Datastore, error) { + primary, err := newPrimaryDatastoreFromConfig(ctx, opts) + if err != nil { + return nil, fmt.Errorf("failed to create primary datastore: %w", err) + } + + if len(opts.ReadReplicaURIs) > datastorecfg.MaxReplicaCount { + return nil, fmt.Errorf("too many read replicas, max is %d", datastorecfg.MaxReplicaCount) + } + + replicas := make([]datastore.StrictReadDatastore, 0, len(opts.ReadReplicaURIs)) + for index, replicaURI := range opts.ReadReplicaURIs { + uintIndex, err := safecast.Convert[uint32](index) + if err != nil { + return nil, errors.New("too many replicas") + } + replica, err := newReplicaDatastoreFromConfig(ctx, uintIndex, replicaURI, opts) + if err != nil { + return nil, err + } + replicas = append(replicas, replica) + } + + return proxy.NewStrictReplicatedDatastore(primary, replicas...) +} + +func commonDatastoreOptionsFromConfig(opts datastorecfg.Config) ([]Option, error) { + maxRetries, err := safecast.Convert[uint8](opts.MaxRetries) + if err != nil { + return nil, errors.New("max-retries could not be cast to uint8") + } + + watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) + if err != nil { + return nil, err + } + + return []Option{ + EnableTracing(), + WithEnablePrometheusStats(opts.EnableDatastoreMetrics), + MaxRetries(maxRetries), + FilterMaximumIDCount(opts.FilterMaximumIDCount), + WithColumnOptimization(opts.ExperimentalColumnOptimization), + WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), + IncludeQueryParametersInTraces(opts.IncludeQueryParametersInTraces), + }, nil +} + +func newReplicaDatastoreFromConfig(ctx context.Context, replicaIndex uint32, replicaURI string, opts datastorecfg.Config) (datastore.StrictReadDatastore, error) { + pgOpts := []Option{ //nolint: prealloc // we're not worried about perf here + CredentialsProviderName(opts.ReadReplicaCredentialsProviderName), + ReadConnsMaxOpen(opts.ReadReplicaConnPool.MaxOpenConns), + ReadConnsMinOpen(opts.ReadReplicaConnPool.MinOpenConns), + ReadConnMaxIdleTime(opts.ReadReplicaConnPool.MaxIdleTime), + ReadConnMaxLifetime(opts.ReadReplicaConnPool.MaxLifetime), + ReadConnMaxLifetimeJitter(opts.ReadReplicaConnPool.MaxLifetimeJitter), + ReadConnHealthCheckInterval(opts.ReadReplicaConnPool.HealthCheckInterval), + ReadConnPingTimeout(opts.ReadReplicaConnPool.PingTimeout), + ReadStrictMode( /* strict read mode is required for Postgres read replicas */ true), + } + + commonOptions, err := commonDatastoreOptionsFromConfig(opts) + if err != nil { + return nil, err + } + pgOpts = append(pgOpts, commonOptions...) + return NewReadOnlyPostgresDatastore(ctx, replicaURI, replicaIndex, pgOpts...) +} + +func newPrimaryDatastoreFromConfig(ctx context.Context, opts datastorecfg.Config) (datastore.Datastore, error) { + pgOpts := []Option{ //nolint: prealloc // we're not worried about perf here + CredentialsProviderName(opts.CredentialsProviderName), + GCWindow(opts.GCWindow), + GCEnabled(!opts.ReadOnly), + RevisionQuantization(opts.RevisionQuantization), + MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), + FollowerReadDelay(opts.FollowerReadDelay), + ReadConnsMaxOpen(opts.ReadConnPool.MaxOpenConns), + ReadConnsMinOpen(opts.ReadConnPool.MinOpenConns), + ReadConnMaxIdleTime(opts.ReadConnPool.MaxIdleTime), + ReadConnMaxLifetime(opts.ReadConnPool.MaxLifetime), + ReadConnMaxLifetimeJitter(opts.ReadConnPool.MaxLifetimeJitter), + ReadConnHealthCheckInterval(opts.ReadConnPool.HealthCheckInterval), + ReadConnPingTimeout(opts.ReadConnPool.PingTimeout), + WriteConnsMaxOpen(opts.WriteConnPool.MaxOpenConns), + WriteConnsMinOpen(opts.WriteConnPool.MinOpenConns), + WriteConnMaxIdleTime(opts.WriteConnPool.MaxIdleTime), + WriteConnMaxLifetime(opts.WriteConnPool.MaxLifetime), + WriteConnMaxLifetimeJitter(opts.ReadConnPool.MaxLifetimeJitter), + WriteConnHealthCheckInterval(opts.WriteConnPool.HealthCheckInterval), + WriteConnPingTimeout(opts.WriteConnPool.PingTimeout), + GCInterval(opts.GCInterval), + GCMaxOperationTime(opts.GCMaxOperationTime), + WatchBufferLength(opts.WatchBufferLength), + WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), + WithWatchDisabled(opts.DisableWatchSupport), + MigrationPhase(opts.MigrationPhase), + AllowedMigrations(opts.AllowedMigrations), + WithRevisionHeartbeat(opts.EnableRevisionHeartbeat), + WithRelaxedIsolationLevel(opts.RelaxedIsolationLevel), + } + + commonOptions, err := commonDatastoreOptionsFromConfig(opts) + if err != nil { + return nil, err + } + pgOpts = append(pgOpts, commonOptions...) + return NewPostgresDatastore(ctx, opts.URI, pgOpts...) +} diff --git a/internal/datastore/postgres/postgres.go b/internal/datastore/postgres/postgres.go index 8ac1fbf88e..ea98ef31d5 100644 --- a/internal/datastore/postgres/postgres.go +++ b/internal/datastore/postgres/postgres.go @@ -408,6 +408,10 @@ func (pgd *pgDatastore) MetricsID() (string, error) { return common.MetricsIDFromURL(pgd.dburl) } +func (pgd *pgDatastore) EngineName() string { + return Engine +} + func (pgd *pgDatastore) IsStrictReadModeEnabled() bool { return pgd.inStrictReadMode } diff --git a/internal/datastore/postgres/postgres_shared_test.go b/internal/datastore/postgres/postgres_shared_test.go index e9abdfe985..0bbf5512ed 100644 --- a/internal/datastore/postgres/postgres_shared_test.go +++ b/internal/datastore/postgres/postgres_shared_test.go @@ -334,7 +334,7 @@ func testPostgresDatastoreWithoutCommitTimestamps(t *testing.T, config postgresT return ds }) return ds, nil - })), test.WithCategories(test.WatchCategory, test.GCCategory)) + })), test.WithCategories(test.WatchCategory, test.GCCategory, test.MigrationCategory)) }) t.Run(fmt.Sprintf("postgres-%s-gc", pgVersion), func(t *testing.T) { diff --git a/internal/datastore/proxy/cachedcheckrev_test.go b/internal/datastore/proxy/cachedcheckrev_test.go index 6b19439f82..8f521b0bd2 100644 --- a/internal/datastore/proxy/cachedcheckrev_test.go +++ b/internal/datastore/proxy/cachedcheckrev_test.go @@ -8,36 +8,35 @@ import ( "github.com/stretchr/testify/require" "github.com/authzed/spicedb/pkg/datastore" - "github.com/authzed/spicedb/pkg/datastore/revisionparsing" ) func TestCachedCheckRevision(t *testing.T) { ds := &fakeBrokenDatastore{checkCount: 0} wrapped := newCachedCheckRevision(ds) - err := wrapped.CheckRevision(t.Context(), revisionparsing.MustParseRevisionForTest("10")) + err := wrapped.CheckRevision(t.Context(), mustParseRevisionForTest("10")) require.NoError(t, err) // Check again for the same revision, should not call the underlying datastore. - err = wrapped.CheckRevision(t.Context(), revisionparsing.MustParseRevisionForTest("10")) + err = wrapped.CheckRevision(t.Context(), mustParseRevisionForTest("10")) require.NoError(t, err) // Check again for a lesser revision, should not call the underlying datastore. - err = wrapped.CheckRevision(t.Context(), revisionparsing.MustParseRevisionForTest("10")) + err = wrapped.CheckRevision(t.Context(), mustParseRevisionForTest("10")) require.NoError(t, err) // Check again for a higher revision, which should call the datastore. - err = wrapped.CheckRevision(t.Context(), revisionparsing.MustParseRevisionForTest("11")) + err = wrapped.CheckRevision(t.Context(), mustParseRevisionForTest("11")) require.Error(t, err) - err = wrapped.CheckRevision(t.Context(), revisionparsing.MustParseRevisionForTest("11")) + err = wrapped.CheckRevision(t.Context(), mustParseRevisionForTest("11")) require.Error(t, err) - err = wrapped.CheckRevision(t.Context(), revisionparsing.MustParseRevisionForTest("12")) + err = wrapped.CheckRevision(t.Context(), mustParseRevisionForTest("12")) require.Error(t, err) // Ensure the older revision still works. - err = wrapped.CheckRevision(t.Context(), revisionparsing.MustParseRevisionForTest("10")) + err = wrapped.CheckRevision(t.Context(), mustParseRevisionForTest("10")) require.NoError(t, err) } diff --git a/internal/datastore/proxy/checkingreplicated_test.go b/internal/datastore/proxy/checkingreplicated_test.go index 4c4c5fd047..c8991f6975 100644 --- a/internal/datastore/proxy/checkingreplicated_test.go +++ b/internal/datastore/proxy/checkingreplicated_test.go @@ -11,13 +11,12 @@ import ( "github.com/authzed/spicedb/pkg/datastore" "github.com/authzed/spicedb/pkg/datastore/options" "github.com/authzed/spicedb/pkg/datastore/queryshape" - "github.com/authzed/spicedb/pkg/datastore/revisionparsing" corev1 "github.com/authzed/spicedb/pkg/proto/core/v1" "github.com/authzed/spicedb/pkg/tuple" ) func TestCheckingReplicatedWithNoReplicasReturnsPrimary(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} ds, err := NewCheckingReplicatedDatastore(primary) require.NoError(t, err) @@ -25,9 +24,9 @@ func TestCheckingReplicatedWithNoReplicasReturnsPrimary(t *testing.T) { } func TestCheckingReplicatedRoundRobinsAcrossReplicas(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replicaA := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("2"), nil} - replicaB := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("2"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replicaA := fakeDatastore{"replica", mustParseRevisionForTest("2"), nil} + replicaB := fakeDatastore{"replica", mustParseRevisionForTest("2"), nil} replicated, err := NewCheckingReplicatedDatastore(primary, replicaA, replicaB) require.NoError(t, err) @@ -47,13 +46,13 @@ func TestCheckingReplicatedRoundRobinsAcrossReplicas(t *testing.T) { } func TestCheckingReplicatedReaderWrapsAllReadMethods(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("2"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica", mustParseRevisionForTest("2"), nil} replicated, err := NewCheckingReplicatedDatastore(primary, replica) require.NoError(t, err) - reader := replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("1")) + reader := replicated.SnapshotReader(mustParseRevisionForTest("1")) // fakeSnapshotReader returns "not implemented" for caveats and counters; // the wrapper just needs to invoke them via the chosen reader. @@ -95,14 +94,14 @@ func TestCheckingReplicatedReaderWrapsAllReadMethods(t *testing.T) { } func TestCheckingReplicatedReaderFallsbackToPrimaryOnCheckRevisionFailure(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("1"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica", mustParseRevisionForTest("1"), nil} replicated, err := NewCheckingReplicatedDatastore(primary, replica) require.NoError(t, err) // Try at revision 1, which should use the replica. - reader := replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("1")) + reader := replicated.SnapshotReader(mustParseRevisionForTest("1")) ns, err := reader.LegacyListAllNamespaces(t.Context()) require.NoError(t, err) require.Empty(t, ns) @@ -110,7 +109,7 @@ func TestCheckingReplicatedReaderFallsbackToPrimaryOnCheckRevisionFailure(t *tes require.False(t, reader.(*checkingStableReader).chosePrimaryForTest) // Try at revision 2, which should use the primary. - reader = replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("2")) + reader = replicated.SnapshotReader(mustParseRevisionForTest("2")) ns, err = reader.LegacyListAllNamespaces(t.Context()) require.NoError(t, err) require.Empty(t, ns) @@ -119,13 +118,13 @@ func TestCheckingReplicatedReaderFallsbackToPrimaryOnCheckRevisionFailure(t *tes } func TestCheckingReplicatedReaderFallsbackToPrimaryOnRevisionNotAvailableError(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("1"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica", mustParseRevisionForTest("1"), nil} replicated, err := NewCheckingReplicatedDatastore(primary, replica) require.NoError(t, err) - reader := replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("3")) + reader := replicated.SnapshotReader(mustParseRevisionForTest("3")) ns, err := reader.LegacyLookupNamespacesWithNames(t.Context(), []string{"ns1"}) require.NoError(t, err) require.Len(t, ns, 1) @@ -134,8 +133,8 @@ func TestCheckingReplicatedReaderFallsbackToPrimaryOnRevisionNotAvailableError(t func TestReplicatedReaderReturnsExpectedError(t *testing.T) { for _, requireCheck := range []bool{true, false} { t.Run(fmt.Sprintf("requireCheck=%v", requireCheck), func(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("1"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica", mustParseRevisionForTest("1"), nil} var ds datastore.Datastore if requireCheck { @@ -149,7 +148,7 @@ func TestReplicatedReaderReturnsExpectedError(t *testing.T) { } // Try at revision 1, which should use the replica. - reader := ds.SnapshotReader(revisionparsing.MustParseRevisionForTest("1")) + reader := ds.SnapshotReader(mustParseRevisionForTest("1")) _, _, err := reader.LegacyReadNamespaceByName(t.Context(), "expecterror") require.Error(t, err) require.ErrorContains(t, err, "raising an expected error") @@ -259,12 +258,12 @@ func (fsr fakeSnapshotReader) LegacyLookupNamespacesWithNames(_ context.Context, Definition: &corev1.NamespaceDefinition{ Name: "ns1", }, - LastWrittenRevision: revisionparsing.MustParseRevisionForTest("2"), + LastWrittenRevision: mustParseRevisionForTest("2"), }, }, nil } - if fsr.revision.GreaterThan(revisionparsing.MustParseRevisionForTest("2")) { + if fsr.revision.GreaterThan(mustParseRevisionForTest("2")) { return nil, common.NewRevisionUnavailableError(fmt.Errorf("revision not available")) } @@ -360,7 +359,7 @@ func fakeIterator(fsr fakeSnapshotReader, explainCallback options.SQLExplainCall return } - if fsr.revision.GreaterThan(revisionparsing.MustParseRevisionForTest("2")) { + if fsr.revision.GreaterThan(mustParseRevisionForTest("2")) { yield(tuple.Relationship{}, common.NewRevisionUnavailableError(fmt.Errorf("revision not available"))) return } diff --git a/internal/datastore/proxy/revision_test.go b/internal/datastore/proxy/revision_test.go new file mode 100644 index 0000000000..defcd48f05 --- /dev/null +++ b/internal/datastore/proxy/revision_test.go @@ -0,0 +1,17 @@ +package proxy + +import ( + "github.com/authzed/spicedb/internal/datastore/revisions" + "github.com/authzed/spicedb/pkg/datastore" +) + +// mustParseRevisionForTest parses a test revision string. It mirrors +// revisionparsing.MustParseRevisionForTest, which cannot be used here: that +// package imports the datastore engines, which import this package. +func mustParseRevisionForTest(revisionStr string) datastore.Revision { + rev, err := revisions.RevisionParser(revisions.HybridLogicalClock)(revisionStr) + if err != nil { + panic(err) + } + return rev +} diff --git a/internal/datastore/proxy/strictreplicated_test.go b/internal/datastore/proxy/strictreplicated_test.go index 3891315e5f..5ad05d0120 100644 --- a/internal/datastore/proxy/strictreplicated_test.go +++ b/internal/datastore/proxy/strictreplicated_test.go @@ -6,7 +6,6 @@ import ( "github.com/stretchr/testify/require" "github.com/authzed/spicedb/pkg/datastore" - "github.com/authzed/spicedb/pkg/datastore/revisionparsing" ) // nonStrictDatastore wraps a fakeDatastore but reports strict read mode disabled, @@ -20,7 +19,7 @@ func (nonStrictDatastore) IsStrictReadModeEnabled() bool { } func TestStrictReplicatedReaderWithOnlyPrimary(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} replicated, err := NewStrictReplicatedDatastore(primary) require.NoError(t, err) @@ -29,14 +28,14 @@ func TestStrictReplicatedReaderWithOnlyPrimary(t *testing.T) { } func TestStrictReplicatedQueryFallsbackToPrimaryOnRevisionNotAvailableError(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("1"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica", mustParseRevisionForTest("1"), nil} replicated, err := NewStrictReplicatedDatastore(primary, replica) require.NoError(t, err) // Query the replicated, which should fallback to the primary. - reader := replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("3")) + reader := replicated.SnapshotReader(mustParseRevisionForTest("3")) iter, err := reader.QueryRelationships(t.Context(), datastore.RelationshipsFilter{ OptionalResourceType: "resource", }) @@ -56,7 +55,7 @@ func TestStrictReplicatedQueryFallsbackToPrimaryOnRevisionNotAvailableError(t *t require.Len(t, revfound, 2) // Query the replica directly, which should error. - reader = replica.SnapshotReader(revisionparsing.MustParseRevisionForTest("3")) + reader = replica.SnapshotReader(mustParseRevisionForTest("3")) iter, err = reader.QueryRelationships(t.Context(), datastore.RelationshipsFilter{ OptionalResourceType: "resource", }) @@ -76,7 +75,7 @@ func TestStrictReplicatedQueryFallsbackToPrimaryOnRevisionNotAvailableError(t *t require.ErrorContains(t, err, "revision not available") // Query the replica for a different revision, which should work. - reader = replica.SnapshotReader(revisionparsing.MustParseRevisionForTest("1")) + reader = replica.SnapshotReader(mustParseRevisionForTest("1")) iter, err = reader.QueryRelationships(t.Context(), datastore.RelationshipsFilter{ OptionalResourceType: "resource", }) @@ -98,14 +97,14 @@ func TestStrictReplicatedQueryFallsbackToPrimaryOnRevisionNotAvailableError(t *t } func TestStrictReplicatedQueryNonFallbackError(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica-with-normal-error", revisionparsing.MustParseRevisionForTest("1"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica-with-normal-error", mustParseRevisionForTest("1"), nil} replicated, err := NewStrictReplicatedDatastore(primary, replica) require.NoError(t, err) // Query the replicated, which should return the error. - reader := replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("3")) + reader := replicated.SnapshotReader(mustParseRevisionForTest("3")) _, err = reader.QueryRelationships(t.Context(), datastore.RelationshipsFilter{ OptionalResourceType: "resource", }) @@ -113,8 +112,8 @@ func TestStrictReplicatedQueryNonFallbackError(t *testing.T) { } func TestStrictReplicatedRejectsReplicaWithoutStrictMode(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := nonStrictDatastore{fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("1"), nil}} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := nonStrictDatastore{fakeDatastore{"replica", mustParseRevisionForTest("1"), nil}} _, err := NewStrictReplicatedDatastore(primary, replica) require.Error(t, err) @@ -127,13 +126,13 @@ func TestStrictReplicatedRejectsReplicaWithoutStrictMode(t *testing.T) { // these calls do not trigger a primary fallback; they simply confirm the wrappers // invoke the replica's reader. func TestStrictReplicatedReaderWrapperMethods(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("2"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica", mustParseRevisionForTest("2"), nil} replicated, err := NewStrictReplicatedDatastore(primary, replica) require.NoError(t, err) - reader := replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("1")) + reader := replicated.SnapshotReader(mustParseRevisionForTest("1")) _, _, err = reader.LegacyReadCaveatByName(t.Context(), "is_weekend") require.ErrorContains(t, err, "not implemented") @@ -161,13 +160,13 @@ func TestStrictReplicatedReaderWrapperMethods(t *testing.T) { // RevisionUnavailableError. The fake returns that error when queried beyond // revision 2. func TestStrictReplicatedReaderFallsbackForNamespaceLookups(t *testing.T) { - primary := fakeDatastore{"primary", revisionparsing.MustParseRevisionForTest("2"), nil} - replica := fakeDatastore{"replica", revisionparsing.MustParseRevisionForTest("1"), nil} + primary := fakeDatastore{"primary", mustParseRevisionForTest("2"), nil} + replica := fakeDatastore{"replica", mustParseRevisionForTest("1"), nil} replicated, err := NewStrictReplicatedDatastore(primary, replica) require.NoError(t, err) - reader := replicated.SnapshotReader(revisionparsing.MustParseRevisionForTest("3")) + reader := replicated.SnapshotReader(mustParseRevisionForTest("3")) ns, err := reader.LegacyLookupNamespacesWithNames(t.Context(), []string{"ns1"}) require.NoError(t, err) require.Len(t, ns, 1) diff --git a/internal/datastore/spanner/enginebuilder.go b/internal/datastore/spanner/enginebuilder.go new file mode 100644 index 0000000000..a8f515841e --- /dev/null +++ b/internal/datastore/spanner/enginebuilder.go @@ -0,0 +1,60 @@ +package spanner + +import ( + "context" + "errors" + + "github.com/authzed/spicedb/internal/datastore/common" + "github.com/authzed/spicedb/internal/datastore/spanner/migrations" + datastorecfg "github.com/authzed/spicedb/pkg/cmd/datastore/dsconfig" + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/migration" +) + +func init() { + datastorecfg.RegisterEngine(Engine, newDatastoreFromConfig) + migration.RegisterMigratableEngine(Engine, migrations.SpannerMigrations, newMigrationDriverFromConfig, "add-schema-tables") +} + +func newMigrationDriverFromConfig(ctx context.Context, cfg *migration.Config) (*migrations.SpannerMigrationDriver, error) { + return migrations.NewSpannerDriver(ctx, cfg.DatastoreURI, cfg.SpannerCredentialsFile, cfg.SpannerEmulatorHost) +} + +func newDatastoreFromConfig(ctx context.Context, opts datastorecfg.Config) (datastore.Datastore, error) { + if len(opts.ReadReplicaURIs) > 0 { + return nil, errors.New("read replicas are not supported for the Spanner datastore engine") + } + + metricsOption := DatastoreMetricsOption(opts.SpannerDatastoreMetricsOption) + if !opts.EnableDatastoreMetrics { + metricsOption = DatastoreMetricsOptionNone + } + + watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) + if err != nil { + return nil, err + } + + return NewSpannerDatastore( + ctx, + opts.URI, + FollowerReadDelay(opts.FollowerReadDelay), + RevisionQuantization(opts.RevisionQuantization), + MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), + CredentialsFile(opts.SpannerCredentialsFile), + CredentialsJSON(opts.SpannerCredentialsJSON), + WatchBufferLength(opts.WatchBufferLength), + WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), + WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), + EmulatorHost(opts.SpannerEmulatorHost), + DisableStats(opts.DisableStats), + WithDatastoreMetricsOption(metricsOption), + ReadConnsMaxOpen(opts.ReadConnPool.MaxOpenConns), + WriteConnsMaxOpen(opts.WriteConnPool.MaxOpenConns), + MigrationPhase(opts.MigrationPhase), + AllowedMigrations(opts.AllowedMigrations), + FilterMaximumIDCount(opts.FilterMaximumIDCount), + WithColumnOptimization(opts.ExperimentalColumnOptimization), + WithWatchDisabled(opts.DisableWatchSupport), + ) +} diff --git a/internal/datastore/spanner/spanner.go b/internal/datastore/spanner/spanner.go index ca2594f513..b70308548e 100644 --- a/internal/datastore/spanner/spanner.go +++ b/internal/datastore/spanner/spanner.go @@ -300,6 +300,10 @@ func (sd *spannerDatastore) MetricsID() (string, error) { return sd.database, nil } +func (sd *spannerDatastore) EngineName() string { + return Engine +} + func (sd *spannerDatastore) readTransactionMetadata(ctx context.Context, transactionTag string) (common.TransactionMetadata, error) { row, err := sd.client.Single().ReadRow(ctx, tableTransactionMetadata, spanner.Key{transactionTag}, []string{colMetadata}) if err != nil { diff --git a/internal/fdw/pgserver_e2e_test.go b/internal/fdw/pgserver_e2e_test.go index 5713bd0cb6..302f052676 100644 --- a/internal/fdw/pgserver_e2e_test.go +++ b/internal/fdw/pgserver_e2e_test.go @@ -26,6 +26,8 @@ import ( "github.com/authzed/authzed-go/v1" "github.com/authzed/grpcutil" + // Register all datastore engines defined in this repository. + _ "github.com/authzed/spicedb/internal/datastore/engines" "github.com/authzed/spicedb/internal/fdw" "github.com/authzed/spicedb/pkg/cmd/datastore" spicedbserver "github.com/authzed/spicedb/pkg/cmd/server" diff --git a/internal/testserver/datastore/config/config.go b/internal/testserver/datastore/config/config.go index 84d4c1ab1e..e09bc8511c 100644 --- a/internal/testserver/datastore/config/config.go +++ b/internal/testserver/datastore/config/config.go @@ -5,6 +5,8 @@ import ( "github.com/stretchr/testify/require" + // Register all datastore engines defined in this repository. + _ "github.com/authzed/spicedb/internal/datastore/engines" testdatastore "github.com/authzed/spicedb/internal/testserver/datastore" dsconfig "github.com/authzed/spicedb/pkg/cmd/datastore" "github.com/authzed/spicedb/pkg/datastore" diff --git a/pkg/cmd/datastore/aliases.go b/pkg/cmd/datastore/aliases.go new file mode 100644 index 0000000000..57f942f7aa --- /dev/null +++ b/pkg/cmd/datastore/aliases.go @@ -0,0 +1,122 @@ +package datastore + +import ( + "github.com/authzed/spicedb/pkg/cmd/datastore/dsconfig" +) + +// The datastore Config type, its options, and the engine registry live in the +// leaf package dsconfig so that datastore engine packages can import them to +// register themselves without creating import cycles. Everything is +// re-exported here for backwards compatibility. + +type ( + Config = dsconfig.Config + ConfigOption = dsconfig.ConfigOption + ConnPoolConfig = dsconfig.ConnPoolConfig + ConnPoolConfigOption = dsconfig.ConnPoolConfigOption + RelIntegrityKey = dsconfig.RelIntegrityKey + RelIntegrityKeyOption = dsconfig.RelIntegrityKeyOption + EngineBuilderFunc = dsconfig.EngineBuilderFunc +) + +const ( + MaxReplicaCount = dsconfig.MaxReplicaCount + DefaultFollowerReadDelay = dsconfig.DefaultFollowerReadDelay + + MemoryEngine = dsconfig.MemoryEngine + PostgresEngine = dsconfig.PostgresEngine + CockroachEngine = dsconfig.CockroachEngine + SpannerEngine = dsconfig.SpannerEngine + MySQLEngine = dsconfig.MySQLEngine +) + +// BuilderForEngine holds the builder for each datastore engine, keyed by +// engine name. See dsconfig.BuilderForEngine. +var BuilderForEngine = dsconfig.BuilderForEngine + +var ( + RegisterEngine = dsconfig.RegisterEngine + DefaultDatastoreConfig = dsconfig.DefaultDatastoreConfig + DefaultReadConnPool = dsconfig.DefaultReadConnPool + DefaultWriteConnPool = dsconfig.DefaultWriteConnPool + ConfigWithOptions = dsconfig.ConfigWithOptions + ConnPoolConfigWithOptions = dsconfig.ConnPoolConfigWithOptions + NewConfigWithOptions = dsconfig.NewConfigWithOptions + NewConfigWithOptionsAndDefaults = dsconfig.NewConfigWithOptionsAndDefaults + NewConnPoolConfigWithOptions = dsconfig.NewConnPoolConfigWithOptions + NewConnPoolConfigWithOptionsAndDefaults = dsconfig.NewConnPoolConfigWithOptionsAndDefaults + NewRelIntegrityKeyWithOptions = dsconfig.NewRelIntegrityKeyWithOptions + NewRelIntegrityKeyWithOptionsAndDefaults = dsconfig.NewRelIntegrityKeyWithOptionsAndDefaults + RelIntegrityKeyWithOptions = dsconfig.RelIntegrityKeyWithOptions + SetAllowedMigrations = dsconfig.SetAllowedMigrations + SetBootstrapFileContents = dsconfig.SetBootstrapFileContents + SetBootstrapFiles = dsconfig.SetBootstrapFiles + SetReadReplicaURIs = dsconfig.SetReadReplicaURIs + SetRelationshipIntegrityExpiredKeys = dsconfig.SetRelationshipIntegrityExpiredKeys + SetSpannerCredentialsJSON = dsconfig.SetSpannerCredentialsJSON + WithAllowedMigrations = dsconfig.WithAllowedMigrations + WithBootstrapFileContents = dsconfig.WithBootstrapFileContents + WithBootstrapFiles = dsconfig.WithBootstrapFiles + WithBootstrapOverwrite = dsconfig.WithBootstrapOverwrite + WithBootstrapTimeout = dsconfig.WithBootstrapTimeout + WithCaveatTypeSet = dsconfig.WithCaveatTypeSet + WithConnectRate = dsconfig.WithConnectRate + WithCredentialsProviderName = dsconfig.WithCredentialsProviderName + WithDisableStats = dsconfig.WithDisableStats + WithDisableWatchSupport = dsconfig.WithDisableWatchSupport + WithEnableConnectionBalancing = dsconfig.WithEnableConnectionBalancing + WithEnableDatastoreMetrics = dsconfig.WithEnableDatastoreMetrics + WithEnableRevisionHeartbeat = dsconfig.WithEnableRevisionHeartbeat + WithEngine = dsconfig.WithEngine + WithExperimentalColumnOptimization = dsconfig.WithExperimentalColumnOptimization + WithFilterMaximumIDCount = dsconfig.WithFilterMaximumIDCount + WithFollowerReadDelay = dsconfig.WithFollowerReadDelay + WithGCInterval = dsconfig.WithGCInterval + WithGCMaxOperationTime = dsconfig.WithGCMaxOperationTime + WithGCWindow = dsconfig.WithGCWindow + WithHealthCheckInterval = dsconfig.WithHealthCheckInterval + WithIncludeQueryParametersInTraces = dsconfig.WithIncludeQueryParametersInTraces + WithKeyFilename = dsconfig.WithKeyFilename + WithKeyID = dsconfig.WithKeyID + WithLegacyFuzzing = dsconfig.WithLegacyFuzzing + WithMaxIdleTime = dsconfig.WithMaxIdleTime + WithMaxLifetime = dsconfig.WithMaxLifetime + WithMaxLifetimeJitter = dsconfig.WithMaxLifetimeJitter + WithMaxOpenConns = dsconfig.WithMaxOpenConns + WithMaxRetries = dsconfig.WithMaxRetries + WithMaxRevisionStalenessPercent = dsconfig.WithMaxRevisionStalenessPercent + WithMigrationPhase = dsconfig.WithMigrationPhase + WithMinOpenConns = dsconfig.WithMinOpenConns + WithOldReadReplicaConnPool = dsconfig.WithOldReadReplicaConnPool + WithOverlapKey = dsconfig.WithOverlapKey + WithOverlapStrategy = dsconfig.WithOverlapStrategy + WithPingTimeout = dsconfig.WithPingTimeout + WithReadConnPool = dsconfig.WithReadConnPool + WithReadOnly = dsconfig.WithReadOnly + WithReadReplicaConnPool = dsconfig.WithReadReplicaConnPool + WithReadReplicaCredentialsProviderName = dsconfig.WithReadReplicaCredentialsProviderName + WithReadReplicaURIs = dsconfig.WithReadReplicaURIs + WithRelationshipIntegrityCurrentKey = dsconfig.WithRelationshipIntegrityCurrentKey + WithRelationshipIntegrityEnabled = dsconfig.WithRelationshipIntegrityEnabled + WithRelationshipIntegrityExpiredKeys = dsconfig.WithRelationshipIntegrityExpiredKeys + WithRelaxedIsolationLevel = dsconfig.WithRelaxedIsolationLevel + WithRequestHedgingEnabled = dsconfig.WithRequestHedgingEnabled + WithRequestHedgingInitialSlowValue = dsconfig.WithRequestHedgingInitialSlowValue + WithRequestHedgingMaxRequests = dsconfig.WithRequestHedgingMaxRequests + WithRequestHedgingQuantile = dsconfig.WithRequestHedgingQuantile + WithRevisionQuantization = dsconfig.WithRevisionQuantization + WithSpannerCredentialsFile = dsconfig.WithSpannerCredentialsFile + WithSpannerCredentialsJSON = dsconfig.WithSpannerCredentialsJSON + WithSpannerDatastoreMetricsOption = dsconfig.WithSpannerDatastoreMetricsOption + WithSpannerEmulatorHost = dsconfig.WithSpannerEmulatorHost + WithSpannerMaxSessions = dsconfig.WithSpannerMaxSessions + WithSpannerMinSessions = dsconfig.WithSpannerMinSessions + WithTablePrefix = dsconfig.WithTablePrefix + WithURI = dsconfig.WithURI + WithWatchBufferLength = dsconfig.WithWatchBufferLength + WithWatchBufferWriteTimeout = dsconfig.WithWatchBufferWriteTimeout + WithWatchChangeBufferMaximumSize = dsconfig.WithWatchChangeBufferMaximumSize + WithWatchConnectTimeout = dsconfig.WithWatchConnectTimeout + WithWriteAcquisitionTimeout = dsconfig.WithWriteAcquisitionTimeout + WithWriteConnPool = dsconfig.WithWriteConnPool +) diff --git a/pkg/cmd/datastore/datastore.go b/pkg/cmd/datastore/datastore.go index 6ba677efc9..f34076ecac 100644 --- a/pkg/cmd/datastore/datastore.go +++ b/pkg/cmd/datastore/datastore.go @@ -9,76 +9,16 @@ import ( "strings" "time" - "github.com/ccoveille/go-safecast/v2" "github.com/spf13/pflag" - "github.com/authzed/spicedb/internal/datastore/common" - "github.com/authzed/spicedb/internal/datastore/crdb" - "github.com/authzed/spicedb/internal/datastore/memdb" - "github.com/authzed/spicedb/internal/datastore/mysql" - "github.com/authzed/spicedb/internal/datastore/postgres" "github.com/authzed/spicedb/internal/datastore/proxy" - "github.com/authzed/spicedb/internal/datastore/spanner" log "github.com/authzed/spicedb/internal/logging" "github.com/authzed/spicedb/internal/sharederrors" - caveattypes "github.com/authzed/spicedb/pkg/caveats/types" "github.com/authzed/spicedb/pkg/datalayer" "github.com/authzed/spicedb/pkg/datastore" "github.com/authzed/spicedb/pkg/validationfile" ) -type engineBuilderFunc func(ctx context.Context, options Config) (datastore.Datastore, error) - -const ( - MaxReplicaCount = 16 - DefaultFollowerReadDelay = 4_800 * time.Millisecond -) - -const ( - MemoryEngine = "memory" - PostgresEngine = "postgres" - CockroachEngine = "cockroachdb" - SpannerEngine = "spanner" - MySQLEngine = "mysql" -) - -var BuilderForEngine = map[string]engineBuilderFunc{ - CockroachEngine: newCRDBDatastore, - PostgresEngine: newPostgresDatastore, - MemoryEngine: newMemoryDatstore, - SpannerEngine: newSpannerDatastore, - MySQLEngine: newMySQLDatastore, -} - -//go:generate go run github.com/ecordell/optgen -output zz_generated.connpool.options.go . ConnPoolConfig -type ConnPoolConfig struct { - MaxIdleTime time.Duration `debugmap:"visible" default:"30m"` - MaxLifetime time.Duration `debugmap:"visible" default:"30m"` - MaxLifetimeJitter time.Duration `debugmap:"visible"` - MaxOpenConns int `debugmap:"visible"` - MinOpenConns int `debugmap:"visible"` - HealthCheckInterval time.Duration `debugmap:"visible" default:"30s"` - PingTimeout time.Duration `debugmap:"visible" default:"5s"` -} - -func DefaultReadConnPool() *ConnPoolConfig { - return &ConnPoolConfig{ - MaxLifetime: 30 * time.Minute, - MaxIdleTime: 30 * time.Minute, - MaxOpenConns: 20, - MinOpenConns: 20, - HealthCheckInterval: 30 * time.Second, - PingTimeout: 5 * time.Second, - } -} - -func DefaultWriteConnPool() *ConnPoolConfig { - cfg := DefaultReadConnPool() - cfg.MaxOpenConns /= 2 - cfg.MinOpenConns /= 2 - return cfg -} - func RegisterConnPoolFlagsWithPrefix(flagSet *pflag.FlagSet, prefix string, defaults, opts *ConnPoolConfig) { if prefix != "" { prefix += "-" @@ -106,141 +46,6 @@ func deprecateUnifiedConnFlags(flagSet *pflag.FlagSet) { _ = flagSet.MarkDeprecated("datastore-conn-ping-timeout", warning) } -//go:generate go run github.com/ecordell/optgen -sensitive-field-name-matches uri,secure -output zz_generated.options.go . Config -type Config struct { - Engine string `debugmap:"visible" default:"memory"` - URI string `debugmap:"sensitive"` - GCWindow time.Duration `debugmap:"visible" default:"24h"` - LegacyFuzzing time.Duration `debugmap:"visible" default:"-1ns"` - RevisionQuantization time.Duration `debugmap:"visible" default:"5s"` - MaxRevisionStalenessPercent float64 `debugmap:"visible" default:"0.1"` - CredentialsProviderName string `debugmap:"visible"` - FilterMaximumIDCount uint16 `debugmap:"hidden" default:"100"` - - // Options - ReadConnPool ConnPoolConfig `debugmap:"visible"` - WriteConnPool ConnPoolConfig `debugmap:"visible"` - ReadOnly bool `debugmap:"visible"` - EnableDatastoreMetrics bool `debugmap:"visible" default:"true"` - DisableStats bool `debugmap:"visible"` - IncludeQueryParametersInTraces bool `debugmap:"visible"` - - // Read Replicas - ReadReplicaConnPool ConnPoolConfig `debugmap:"visible"` - // this holds values from the old flag prefix in case they are used - OldReadReplicaConnPool ConnPoolConfig `debugmap:"hidden"` - ReadReplicaURIs []string `debugmap:"sensitive"` - ReadReplicaCredentialsProviderName string `debugmap:"visible"` - - // Bootstrap - BootstrapFiles []string `debugmap:"visible-format"` - BootstrapFileContents map[string][]byte `debugmap:"visible"` - BootstrapOverwrite bool `debugmap:"visible"` - BootstrapTimeout time.Duration `debugmap:"visible" default:"10s"` - CaveatTypeSet *caveattypes.TypeSet `debugmap:"hidden"` - - // Hedging - RequestHedgingEnabled bool `debugmap:"visible"` - RequestHedgingInitialSlowValue time.Duration `debugmap:"visible"` - RequestHedgingMaxRequests uint64 `debugmap:"visible"` - RequestHedgingQuantile float64 `debugmap:"visible"` - - // CRDB - FollowerReadDelay time.Duration `debugmap:"visible" default:"4800ms"` - MaxRetries int `debugmap:"visible" default:"10"` - OverlapKey string `debugmap:"visible" default:"key"` - OverlapStrategy string `debugmap:"visible" default:"static"` - EnableConnectionBalancing bool `debugmap:"visible" default:"true"` - ConnectRate time.Duration `debugmap:"visible" default:"100ms"` - WriteAcquisitionTimeout time.Duration `debugmap:"visible" default:"30ms"` - - // Postgres - GCInterval time.Duration `debugmap:"visible" default:"3m"` - GCMaxOperationTime time.Duration `debugmap:"visible" default:"1m"` - RelaxedIsolationLevel bool `debugmap:"visible"` - - // Spanner - // SpannerCredentialsFile is a filename reference to a file containing - // spanner client credentials. - // - // Deprecated: Prefer Application Default Credentials for Spanner client credentials: - // https://docs.cloud.google.com/docs/authentication/client-libraries#adc - SpannerCredentialsFile string `debugmap:"visible"` - // SpannerCredentialsJSON is a mechanism for providing client configuration as JSON. - // - // Deprecated: Prefer Application Default Credentials for Spanner client credentials: - // https://docs.cloud.google.com/docs/authentication/client-libraries#adc - SpannerCredentialsJSON []byte `debugmap:"sensitive"` - SpannerEmulatorHost string `debugmap:"visible"` - SpannerMinSessions uint64 `debugmap:"visible" default:"100"` - SpannerMaxSessions uint64 `debugmap:"visible" default:"400"` - SpannerDatastoreMetricsOption string `debugmap:"visible" default:"otel"` - - // MySQL - TablePrefix string `debugmap:"visible"` - - // Relationship Integrity - RelationshipIntegrityEnabled bool `debugmap:"visible"` - RelationshipIntegrityCurrentKey RelIntegrityKey `debugmap:"visible"` - RelationshipIntegrityExpiredKeys []string `debugmap:"visible"` - - // Internal - WatchBufferLength uint16 `debugmap:"visible" default:"1024"` - WatchChangeBufferMaximumSize string `debugmap:"visible" default:"15%"` - WatchBufferWriteTimeout time.Duration `debugmap:"visible" default:"1s"` - WatchConnectTimeout time.Duration `debugmap:"visible" default:"1s"` - DisableWatchSupport bool `debugmap:"hidden"` - - // Migrations - MigrationPhase string `debugmap:"visible"` - AllowedMigrations []string `debugmap:"visible"` - - // Experimental - ExperimentalColumnOptimization bool `debugmap:"visible" default:"true"` - EnableRevisionHeartbeat bool `debugmap:"visible"` -} - -// SetDefaults is invoked by github.com/creasty/defaults after struct-tag -// defaults are applied. It fills the four ConnPoolConfig slots from the -// canonical DefaultReadConnPool / DefaultWriteConnPool constructors because -// each slot receives a different default set from RegisterConnPoolFlagsWithPrefix -// in RegisterDatastoreFlagsWithPrefix (Read pools = 20/20 conns, Write = 10/10). -// It also pre-allocates slice fields to empty (non-nil) values so the -// resulting Config matches what RegisterDatastoreFlags writes via -// StringSliceVar/StringArrayVar. -func (c *Config) SetDefaults() { - c.ReadConnPool = *DefaultReadConnPool() - c.WriteConnPool = *DefaultWriteConnPool() - c.ReadReplicaConnPool = *DefaultReadConnPool() - c.OldReadReplicaConnPool = *DefaultReadConnPool() - - // CaveatTypeSet is hidden from DebugMap but RegisterDatastoreFlags - // initializes it from DefaultDatastoreConfig at line 223. Mirror that - // here so library users get the same value as CLI users. - if c.CaveatTypeSet == nil { - c.CaveatTypeSet = caveattypes.Default.TypeSet - } - - if c.BootstrapFiles == nil { - c.BootstrapFiles = []string{} - } - if c.ReadReplicaURIs == nil { - c.ReadReplicaURIs = []string{} - } - if c.AllowedMigrations == nil { - c.AllowedMigrations = []string{} - } - if c.RelationshipIntegrityExpiredKeys == nil { - c.RelationshipIntegrityExpiredKeys = []string{} - } -} - -//go:generate go run github.com/ecordell/optgen -sensitive-field-name-matches uri,secure -output zz_generated.relintegritykey.options.go . RelIntegrityKey -type RelIntegrityKey struct { - KeyID string `debugmap:"visible"` - KeyFilename string `debugmap:"visible"` -} - // RegisterDatastoreFlags adds datastore flags to a cobra command. func RegisterDatastoreFlags(flagset *pflag.FlagSet, opts *Config) error { return RegisterDatastoreFlagsWithPrefix(flagset, "", opts) @@ -409,56 +214,6 @@ func RegisterDatastoreFlagsWithPrefix(flagSet *pflag.FlagSet, prefix string, opt return nil } -func DefaultDatastoreConfig() *Config { - return &Config{ - Engine: MemoryEngine, - GCWindow: 24 * time.Hour, - LegacyFuzzing: -1, - RevisionQuantization: 5 * time.Second, - MaxRevisionStalenessPercent: .1, // 10% - ReadConnPool: *DefaultReadConnPool(), - WriteConnPool: *DefaultWriteConnPool(), - ReadReplicaConnPool: *DefaultReadConnPool(), - OldReadReplicaConnPool: *DefaultReadConnPool(), - ReadReplicaURIs: []string{}, - ReadOnly: false, - MaxRetries: 10, - OverlapKey: "key", - OverlapStrategy: "static", - ConnectRate: 100 * time.Millisecond, - EnableConnectionBalancing: true, - GCInterval: 3 * time.Minute, - GCMaxOperationTime: 1 * time.Minute, - WatchBufferLength: 1024, - WatchChangeBufferMaximumSize: "15%", - WatchBufferWriteTimeout: 1 * time.Second, - WatchConnectTimeout: 1 * time.Second, - DisableWatchSupport: false, - EnableDatastoreMetrics: true, - DisableStats: false, - BootstrapFiles: []string{}, - BootstrapTimeout: 10 * time.Second, - BootstrapOverwrite: false, - SpannerCredentialsFile: "", - SpannerEmulatorHost: "", - TablePrefix: "", - MigrationPhase: "", - FollowerReadDelay: DefaultFollowerReadDelay, - SpannerMinSessions: 100, - SpannerMaxSessions: 400, - FilterMaximumIDCount: 100, - SpannerDatastoreMetricsOption: spanner.DatastoreMetricsOptionOpenTelemetry, - RelationshipIntegrityEnabled: false, - RelationshipIntegrityCurrentKey: RelIntegrityKey{}, - RelationshipIntegrityExpiredKeys: []string{}, - AllowedMigrations: []string{}, - ExperimentalColumnOptimization: true, - IncludeQueryParametersInTraces: false, - WriteAcquisitionTimeout: 30 * time.Millisecond, - CaveatTypeSet: caveattypes.Default.TypeSet, - } -} - // NewDatastore initializes a datastore given the options func NewDatastore(ctx context.Context, options ...ConfigOption) (datastore.Datastore, error) { opts := DefaultDatastoreConfig() @@ -603,309 +358,3 @@ func readExpiredKeys(expiredKeyStrings []string) ([]proxy.KeyConfig, error) { return expiredKeys, nil } - -func newCRDBDatastore(ctx context.Context, opts Config) (datastore.Datastore, error) { - if len(opts.ReadReplicaURIs) > 0 { - return nil, errors.New("read replicas are not supported for the CockroachDB datastore engine") - } - - maxRetries, err := safecast.Convert[uint8](opts.MaxRetries) - if err != nil { - return nil, errors.New("max-retries could not be cast to uint8") - } - - watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) - if err != nil { - return nil, err - } - - return crdb.NewCRDBDatastore( - ctx, - opts.URI, - crdb.GCWindow(opts.GCWindow), - crdb.RevisionQuantization(opts.RevisionQuantization), - crdb.MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), - crdb.ReadConnsMaxOpen(opts.ReadConnPool.MaxOpenConns), - crdb.ReadConnsMinOpen(opts.ReadConnPool.MinOpenConns), - crdb.ReadConnMaxIdleTime(opts.ReadConnPool.MaxIdleTime), - crdb.ReadConnMaxLifetime(opts.ReadConnPool.MaxLifetime), - crdb.ReadConnMaxLifetimeJitter(opts.ReadConnPool.MaxLifetimeJitter), - crdb.ReadConnHealthCheckInterval(opts.ReadConnPool.HealthCheckInterval), - crdb.ReadConnPingTimeout(opts.ReadConnPool.PingTimeout), - crdb.WithAcquireTimeout(opts.WriteAcquisitionTimeout), - crdb.WriteConnsMaxOpen(opts.WriteConnPool.MaxOpenConns), - crdb.WriteConnsMinOpen(opts.WriteConnPool.MinOpenConns), - crdb.WriteConnMaxIdleTime(opts.WriteConnPool.MaxIdleTime), - crdb.WriteConnMaxLifetime(opts.WriteConnPool.MaxLifetime), - crdb.WriteConnMaxLifetimeJitter(opts.WriteConnPool.MaxLifetimeJitter), - crdb.WriteConnHealthCheckInterval(opts.WriteConnPool.HealthCheckInterval), - crdb.WriteConnPingTimeout(opts.WriteConnPool.PingTimeout), - crdb.FollowerReadDelay(opts.FollowerReadDelay), - crdb.MaxRetries(maxRetries), - crdb.OverlapKey(opts.OverlapKey), - crdb.OverlapStrategy(opts.OverlapStrategy), - crdb.WatchBufferLength(opts.WatchBufferLength), - crdb.WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), - crdb.WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), - crdb.WatchConnectTimeout(opts.WatchConnectTimeout), - crdb.WithEnablePrometheusStats(opts.EnableDatastoreMetrics), - crdb.WithEnableConnectionBalancing(opts.EnableConnectionBalancing), - crdb.ConnectRate(opts.ConnectRate), - crdb.FilterMaximumIDCount(opts.FilterMaximumIDCount), - crdb.WithIntegrity(opts.RelationshipIntegrityEnabled), - crdb.AllowedMigrations(opts.AllowedMigrations), - crdb.WithColumnOptimization(opts.ExperimentalColumnOptimization), - crdb.IncludeQueryParametersInTraces(opts.IncludeQueryParametersInTraces), - crdb.WithWatchDisabled(opts.DisableWatchSupport), - ) -} - -func newPostgresDatastore(ctx context.Context, opts Config) (datastore.Datastore, error) { - primary, err := newPostgresPrimaryDatastore(ctx, opts) - if err != nil { - return nil, fmt.Errorf("failed to create primary datastore: %w", err) - } - - if len(opts.ReadReplicaURIs) > MaxReplicaCount { - return nil, fmt.Errorf("too many read replicas, max is %d", MaxReplicaCount) - } - - replicas := make([]datastore.StrictReadDatastore, 0, len(opts.ReadReplicaURIs)) - for index, replicaURI := range opts.ReadReplicaURIs { - uintIndex, err := safecast.Convert[uint32](index) - if err != nil { - return nil, errors.New("too many replicas") - } - replica, err := newPostgresReplicaDatastore(ctx, uintIndex, replicaURI, opts) - if err != nil { - return nil, err - } - replicas = append(replicas, replica) - } - - return proxy.NewStrictReplicatedDatastore(primary, replicas...) -} - -func commonPostgresDatastoreOptions(opts Config) ([]postgres.Option, error) { - maxRetries, err := safecast.Convert[uint8](opts.MaxRetries) - if err != nil { - return nil, errors.New("max-retries could not be cast to uint8") - } - - watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) - if err != nil { - return nil, err - } - - return []postgres.Option{ - postgres.EnableTracing(), - postgres.WithEnablePrometheusStats(opts.EnableDatastoreMetrics), - postgres.MaxRetries(maxRetries), - postgres.FilterMaximumIDCount(opts.FilterMaximumIDCount), - postgres.WithColumnOptimization(opts.ExperimentalColumnOptimization), - postgres.WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), - postgres.IncludeQueryParametersInTraces(opts.IncludeQueryParametersInTraces), - }, nil -} - -func newPostgresReplicaDatastore(ctx context.Context, replicaIndex uint32, replicaURI string, opts Config) (datastore.StrictReadDatastore, error) { - pgOpts := []postgres.Option{ //nolint: prealloc // we're not worried about perf here - postgres.CredentialsProviderName(opts.ReadReplicaCredentialsProviderName), - postgres.ReadConnsMaxOpen(opts.ReadReplicaConnPool.MaxOpenConns), - postgres.ReadConnsMinOpen(opts.ReadReplicaConnPool.MinOpenConns), - postgres.ReadConnMaxIdleTime(opts.ReadReplicaConnPool.MaxIdleTime), - postgres.ReadConnMaxLifetime(opts.ReadReplicaConnPool.MaxLifetime), - postgres.ReadConnMaxLifetimeJitter(opts.ReadReplicaConnPool.MaxLifetimeJitter), - postgres.ReadConnHealthCheckInterval(opts.ReadReplicaConnPool.HealthCheckInterval), - postgres.ReadConnPingTimeout(opts.ReadReplicaConnPool.PingTimeout), - postgres.ReadStrictMode( /* strict read mode is required for Postgres read replicas */ true), - } - - commonOptions, err := commonPostgresDatastoreOptions(opts) - if err != nil { - return nil, err - } - pgOpts = append(pgOpts, commonOptions...) - return postgres.NewReadOnlyPostgresDatastore(ctx, replicaURI, replicaIndex, pgOpts...) -} - -func newPostgresPrimaryDatastore(ctx context.Context, opts Config) (datastore.Datastore, error) { - pgOpts := []postgres.Option{ //nolint: prealloc // we're not worried about perf here - postgres.CredentialsProviderName(opts.CredentialsProviderName), - postgres.GCWindow(opts.GCWindow), - postgres.GCEnabled(!opts.ReadOnly), - postgres.RevisionQuantization(opts.RevisionQuantization), - postgres.MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), - postgres.FollowerReadDelay(opts.FollowerReadDelay), - postgres.ReadConnsMaxOpen(opts.ReadConnPool.MaxOpenConns), - postgres.ReadConnsMinOpen(opts.ReadConnPool.MinOpenConns), - postgres.ReadConnMaxIdleTime(opts.ReadConnPool.MaxIdleTime), - postgres.ReadConnMaxLifetime(opts.ReadConnPool.MaxLifetime), - postgres.ReadConnMaxLifetimeJitter(opts.ReadConnPool.MaxLifetimeJitter), - postgres.ReadConnHealthCheckInterval(opts.ReadConnPool.HealthCheckInterval), - postgres.ReadConnPingTimeout(opts.ReadConnPool.PingTimeout), - postgres.WriteConnsMaxOpen(opts.WriteConnPool.MaxOpenConns), - postgres.WriteConnsMinOpen(opts.WriteConnPool.MinOpenConns), - postgres.WriteConnMaxIdleTime(opts.WriteConnPool.MaxIdleTime), - postgres.WriteConnMaxLifetime(opts.WriteConnPool.MaxLifetime), - postgres.WriteConnMaxLifetimeJitter(opts.ReadConnPool.MaxLifetimeJitter), - postgres.WriteConnHealthCheckInterval(opts.WriteConnPool.HealthCheckInterval), - postgres.WriteConnPingTimeout(opts.WriteConnPool.PingTimeout), - postgres.GCInterval(opts.GCInterval), - postgres.GCMaxOperationTime(opts.GCMaxOperationTime), - postgres.WatchBufferLength(opts.WatchBufferLength), - postgres.WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), - postgres.WithWatchDisabled(opts.DisableWatchSupport), - postgres.MigrationPhase(opts.MigrationPhase), - postgres.AllowedMigrations(opts.AllowedMigrations), - postgres.WithRevisionHeartbeat(opts.EnableRevisionHeartbeat), - postgres.WithRelaxedIsolationLevel(opts.RelaxedIsolationLevel), - } - - commonOptions, err := commonPostgresDatastoreOptions(opts) - if err != nil { - return nil, err - } - pgOpts = append(pgOpts, commonOptions...) - return postgres.NewPostgresDatastore(ctx, opts.URI, pgOpts...) -} - -func newSpannerDatastore(ctx context.Context, opts Config) (datastore.Datastore, error) { - if len(opts.ReadReplicaURIs) > 0 { - return nil, errors.New("read replicas are not supported for the Spanner datastore engine") - } - - metricsOption := spanner.DatastoreMetricsOption(opts.SpannerDatastoreMetricsOption) - if !opts.EnableDatastoreMetrics { - metricsOption = spanner.DatastoreMetricsOptionNone - } - - watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) - if err != nil { - return nil, err - } - - return spanner.NewSpannerDatastore( - ctx, - opts.URI, - spanner.FollowerReadDelay(opts.FollowerReadDelay), - spanner.RevisionQuantization(opts.RevisionQuantization), - spanner.MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), - spanner.CredentialsFile(opts.SpannerCredentialsFile), - spanner.CredentialsJSON(opts.SpannerCredentialsJSON), - spanner.WatchBufferLength(opts.WatchBufferLength), - spanner.WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), - spanner.WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), - spanner.EmulatorHost(opts.SpannerEmulatorHost), - spanner.DisableStats(opts.DisableStats), - spanner.WithDatastoreMetricsOption(metricsOption), - spanner.ReadConnsMaxOpen(opts.ReadConnPool.MaxOpenConns), - spanner.WriteConnsMaxOpen(opts.WriteConnPool.MaxOpenConns), - spanner.MigrationPhase(opts.MigrationPhase), - spanner.AllowedMigrations(opts.AllowedMigrations), - spanner.FilterMaximumIDCount(opts.FilterMaximumIDCount), - spanner.WithColumnOptimization(opts.ExperimentalColumnOptimization), - spanner.WithWatchDisabled(opts.DisableWatchSupport), - ) -} - -func newMySQLDatastore(ctx context.Context, opts Config) (datastore.Datastore, error) { - primary, err := newMySQLPrimaryDatastore(ctx, opts) - if err != nil { - return nil, err - } - - if len(opts.ReadReplicaURIs) > MaxReplicaCount { - return nil, fmt.Errorf("too many read replicas, max is %d", MaxReplicaCount) - } - - replicas := make([]datastore.ReadOnlyDatastore, 0, len(opts.ReadReplicaURIs)) - for index, replicaURI := range opts.ReadReplicaURIs { - uintIndex, err := safecast.Convert[uint32](index) - if err != nil { - return nil, errors.New("too many replicas") - } - replica, err := newMySQLReplicaDatastore(ctx, uintIndex, replicaURI, opts) - if err != nil { - return nil, err - } - replicas = append(replicas, replica) - } - - return proxy.NewCheckingReplicatedDatastore(primary, replicas...) -} - -func commonMySQLDatastoreOptions(opts Config) ([]mysql.Option, error) { - maxRetries, err := safecast.Convert[uint8](opts.MaxRetries) - if err != nil { - return nil, errors.New("max-retries could not be cast to uint8") - } - - watchChangeBufferMaximumSize, err := common.WatchBufferSize(opts.WatchChangeBufferMaximumSize) - if err != nil { - return nil, err - } - - return []mysql.Option{ - mysql.TablePrefix(opts.TablePrefix), - mysql.MaxRetries(maxRetries), - mysql.OverrideLockWaitTimeout(1), - mysql.WithEnablePrometheusStats(opts.EnableDatastoreMetrics), - mysql.WatchBufferLength(opts.WatchBufferLength), - mysql.WatchBufferWriteTimeout(opts.WatchBufferWriteTimeout), - mysql.WatchChangeBufferMaximumSize(watchChangeBufferMaximumSize), - mysql.MaxRevisionStalenessPercent(opts.MaxRevisionStalenessPercent), - mysql.RevisionQuantization(opts.RevisionQuantization), - mysql.FilterMaximumIDCount(opts.FilterMaximumIDCount), - mysql.AllowedMigrations(opts.AllowedMigrations), - mysql.WithColumnOptimization(opts.ExperimentalColumnOptimization), - }, nil -} - -func newMySQLReplicaDatastore(ctx context.Context, replicaIndex uint32, replicaURI string, opts Config) (datastore.ReadOnlyDatastore, error) { - mysqlOpts := []mysql.Option{ //nolint: prealloc // we're not concerned about perf here - mysql.MaxOpenConns(opts.ReadReplicaConnPool.MaxOpenConns), - mysql.ConnMaxIdleTime(opts.ReadReplicaConnPool.MaxIdleTime), - mysql.ConnMaxLifetime(opts.ReadReplicaConnPool.MaxLifetime), - mysql.CredentialsProviderName(opts.ReadReplicaCredentialsProviderName), - } - - commonOptions, err := commonMySQLDatastoreOptions(opts) - if err != nil { - return nil, err - } - mysqlOpts = append(mysqlOpts, commonOptions...) - return mysql.NewReadOnlyMySQLDatastore(ctx, replicaURI, replicaIndex, mysqlOpts...) -} - -func newMySQLPrimaryDatastore(ctx context.Context, opts Config) (datastore.Datastore, error) { - mysqlOpts := []mysql.Option{ //nolint: prealloc // we're not concerned about perf here - mysql.GCInterval(opts.GCInterval), - mysql.GCWindow(opts.GCWindow), - mysql.GCInterval(opts.GCInterval), - mysql.GCEnabled(!opts.ReadOnly), - mysql.GCMaxOperationTime(opts.GCMaxOperationTime), - mysql.MaxOpenConns(opts.ReadConnPool.MaxOpenConns), - mysql.ConnMaxIdleTime(opts.ReadConnPool.MaxIdleTime), - mysql.ConnMaxLifetime(opts.ReadConnPool.MaxLifetime), - mysql.WithWatchDisabled(opts.DisableWatchSupport), - mysql.CredentialsProviderName(opts.CredentialsProviderName), - mysql.FollowerReadDelay(opts.FollowerReadDelay), - } - - commonOptions, err := commonMySQLDatastoreOptions(opts) - if err != nil { - return nil, err - } - mysqlOpts = append(mysqlOpts, commonOptions...) - return mysql.NewMySQLDatastore(ctx, opts.URI, mysqlOpts...) -} - -func newMemoryDatstore(_ context.Context, opts Config) (datastore.Datastore, error) { - if len(opts.ReadReplicaURIs) > 0 { - return nil, errors.New("read replicas are not supported for the in-memory datastore engine") - } - - log.Warn().Msg("in-memory datastore is not persistent and not feasible to run in a high availability fashion") - return memdb.NewMemdbDatastore(opts.WatchBufferLength, opts.RevisionQuantization, opts.GCWindow) -} diff --git a/pkg/cmd/datastore/datastore_test.go b/pkg/cmd/datastore/datastore_test.go index 0a4a0bf376..3155e0e6e7 100644 --- a/pkg/cmd/datastore/datastore_test.go +++ b/pkg/cmd/datastore/datastore_test.go @@ -1,4 +1,4 @@ -package datastore +package datastore_test import ( "os" @@ -6,22 +6,26 @@ import ( "github.com/spf13/pflag" "github.com/stretchr/testify/require" + + // Register all datastore engines defined in this repository. + _ "github.com/authzed/spicedb/internal/datastore/engines" + "github.com/authzed/spicedb/pkg/cmd/datastore" ) func TestDefaults(t *testing.T) { f := pflag.FlagSet{} - expected := NewConfigWithOptionsAndDefaults() - err := RegisterDatastoreFlagsWithPrefix(&f, "", expected) + expected := datastore.NewConfigWithOptionsAndDefaults() + err := datastore.RegisterDatastoreFlagsWithPrefix(&f, "", expected) require.NoError(t, err) - received := DefaultDatastoreConfig() + received := datastore.DefaultDatastoreConfig() require.Equal(t, expected, received) } func TestLoadDatastoreFromFileContents(t *testing.T) { ctx := t.Context() - ds, err := NewDatastore(ctx, - SetBootstrapFileContents(map[string][]byte{"test": []byte("schema: definition user{}")}), - WithEngine(MemoryEngine)) + ds, err := datastore.NewDatastore(ctx, + datastore.SetBootstrapFileContents(map[string][]byte{"test": []byte("schema: definition user{}")}), + datastore.WithEngine(datastore.MemoryEngine)) require.NoError(t, err) t.Cleanup(func() { ds.Close() @@ -43,9 +47,9 @@ func TestLoadDatastoreFromFile(t *testing.T) { require.NoError(t, err) ctx := t.Context() - ds, err := NewDatastore(ctx, - SetBootstrapFiles([]string{file.Name()}), - WithEngine(MemoryEngine)) + ds, err := datastore.NewDatastore(ctx, + datastore.SetBootstrapFiles([]string{file.Name()}), + datastore.WithEngine(datastore.MemoryEngine)) require.NoError(t, err) t.Cleanup(func() { ds.Close() @@ -85,9 +89,9 @@ relationships: |- require.NoError(t, err) ctx := t.Context() - ds, err := NewDatastore(ctx, - SetBootstrapFiles([]string{file.Name()}), - WithEngine(MemoryEngine)) + ds, err := datastore.NewDatastore(ctx, + datastore.SetBootstrapFiles([]string{file.Name()}), + datastore.WithEngine(datastore.MemoryEngine)) require.NoError(t, err) t.Cleanup(func() { ds.Close() @@ -109,10 +113,10 @@ func TestLoadDatastoreFromFileAndContents(t *testing.T) { require.NoError(t, err) ctx := t.Context() - ds, err := NewDatastore(ctx, - SetBootstrapFiles([]string{file.Name()}), - SetBootstrapFileContents(map[string][]byte{"test": []byte("schema: definition user{}")}), - WithEngine(MemoryEngine)) + ds, err := datastore.NewDatastore(ctx, + datastore.SetBootstrapFiles([]string{file.Name()}), + datastore.SetBootstrapFileContents(map[string][]byte{"test": []byte("schema: definition user{}")}), + datastore.WithEngine(datastore.MemoryEngine)) require.NoError(t, err) revisionResult, err := ds.HeadRevision(ctx) diff --git a/pkg/cmd/datastore/dsconfig/config.go b/pkg/cmd/datastore/dsconfig/config.go new file mode 100644 index 0000000000..45e12ac4fb --- /dev/null +++ b/pkg/cmd/datastore/dsconfig/config.go @@ -0,0 +1,259 @@ +// Package dsconfig holds the configuration type for constructing datastores +// by engine name, along with the registry of engine builders. It is a leaf +// package with minimal dependencies so that datastore engine packages can +// import it to register themselves without creating import cycles; +// pkg/cmd/datastore re-exports everything here and provides flag registration +// and NewDatastore on top. +package dsconfig + +import ( + "context" + "time" + + caveattypes "github.com/authzed/spicedb/pkg/caveats/types" + "github.com/authzed/spicedb/pkg/datastore" +) + +// EngineBuilderFunc builds a datastore of a specific engine from the given config. +type EngineBuilderFunc func(ctx context.Context, options Config) (datastore.Datastore, error) + +const ( + MaxReplicaCount = 16 + DefaultFollowerReadDelay = 4_800 * time.Millisecond +) + +const ( + MemoryEngine = "memory" + PostgresEngine = "postgres" + CockroachEngine = "cockroachdb" + SpannerEngine = "spanner" + MySQLEngine = "mysql" +) + +// BuilderForEngine holds the builder for each datastore engine, keyed by +// engine name. Engines register themselves via RegisterEngine from an init +// function; importing an engine package makes it available to +// pkg/cmd/datastore.NewDatastore. internal/datastore/engines registers every +// engine defined in this repository. +var BuilderForEngine = map[string]EngineBuilderFunc{} + +// RegisterEngine makes a datastore engine available to +// pkg/cmd/datastore.NewDatastore under the given name. It is typically called +// from an init function of the package defining the engine. +func RegisterEngine(engineName string, builder EngineBuilderFunc) { + BuilderForEngine[engineName] = builder +} + +//go:generate go run github.com/ecordell/optgen -output zz_generated.connpool.options.go . ConnPoolConfig +type ConnPoolConfig struct { + MaxIdleTime time.Duration `debugmap:"visible" default:"30m"` + MaxLifetime time.Duration `debugmap:"visible" default:"30m"` + MaxLifetimeJitter time.Duration `debugmap:"visible"` + MaxOpenConns int `debugmap:"visible"` + MinOpenConns int `debugmap:"visible"` + HealthCheckInterval time.Duration `debugmap:"visible" default:"30s"` + PingTimeout time.Duration `debugmap:"visible" default:"5s"` +} + +func DefaultReadConnPool() *ConnPoolConfig { + return &ConnPoolConfig{ + MaxLifetime: 30 * time.Minute, + MaxIdleTime: 30 * time.Minute, + MaxOpenConns: 20, + MinOpenConns: 20, + HealthCheckInterval: 30 * time.Second, + PingTimeout: 5 * time.Second, + } +} + +func DefaultWriteConnPool() *ConnPoolConfig { + cfg := DefaultReadConnPool() + cfg.MaxOpenConns /= 2 + cfg.MinOpenConns /= 2 + return cfg +} + +//go:generate go run github.com/ecordell/optgen -sensitive-field-name-matches uri,secure -output zz_generated.options.go . Config +type Config struct { + Engine string `debugmap:"visible" default:"memory"` + URI string `debugmap:"sensitive"` + GCWindow time.Duration `debugmap:"visible" default:"24h"` + LegacyFuzzing time.Duration `debugmap:"visible" default:"-1ns"` + RevisionQuantization time.Duration `debugmap:"visible" default:"5s"` + MaxRevisionStalenessPercent float64 `debugmap:"visible" default:"0.1"` + CredentialsProviderName string `debugmap:"visible"` + FilterMaximumIDCount uint16 `debugmap:"hidden" default:"100"` + + // Options + ReadConnPool ConnPoolConfig `debugmap:"visible"` + WriteConnPool ConnPoolConfig `debugmap:"visible"` + ReadOnly bool `debugmap:"visible"` + EnableDatastoreMetrics bool `debugmap:"visible" default:"true"` + DisableStats bool `debugmap:"visible"` + IncludeQueryParametersInTraces bool `debugmap:"visible"` + + // Read Replicas + ReadReplicaConnPool ConnPoolConfig `debugmap:"visible"` + // this holds values from the old flag prefix in case they are used + OldReadReplicaConnPool ConnPoolConfig `debugmap:"hidden"` + ReadReplicaURIs []string `debugmap:"sensitive"` + ReadReplicaCredentialsProviderName string `debugmap:"visible"` + + // Bootstrap + BootstrapFiles []string `debugmap:"visible-format"` + BootstrapFileContents map[string][]byte `debugmap:"visible"` + BootstrapOverwrite bool `debugmap:"visible"` + BootstrapTimeout time.Duration `debugmap:"visible" default:"10s"` + CaveatTypeSet *caveattypes.TypeSet `debugmap:"hidden"` + + // Hedging + RequestHedgingEnabled bool `debugmap:"visible"` + RequestHedgingInitialSlowValue time.Duration `debugmap:"visible"` + RequestHedgingMaxRequests uint64 `debugmap:"visible"` + RequestHedgingQuantile float64 `debugmap:"visible"` + + // CRDB + FollowerReadDelay time.Duration `debugmap:"visible" default:"4800ms"` + MaxRetries int `debugmap:"visible" default:"10"` + OverlapKey string `debugmap:"visible" default:"key"` + OverlapStrategy string `debugmap:"visible" default:"static"` + EnableConnectionBalancing bool `debugmap:"visible" default:"true"` + ConnectRate time.Duration `debugmap:"visible" default:"100ms"` + WriteAcquisitionTimeout time.Duration `debugmap:"visible" default:"30ms"` + + // Postgres + GCInterval time.Duration `debugmap:"visible" default:"3m"` + GCMaxOperationTime time.Duration `debugmap:"visible" default:"1m"` + RelaxedIsolationLevel bool `debugmap:"visible"` + + // Spanner + // SpannerCredentialsFile is a filename reference to a file containing + // spanner client credentials. + // + // Deprecated: Prefer Application Default Credentials for Spanner client credentials: + // https://docs.cloud.google.com/docs/authentication/client-libraries#adc + SpannerCredentialsFile string `debugmap:"visible"` + // SpannerCredentialsJSON is a mechanism for providing client configuration as JSON. + // + // Deprecated: Prefer Application Default Credentials for Spanner client credentials: + // https://docs.cloud.google.com/docs/authentication/client-libraries#adc + SpannerCredentialsJSON []byte `debugmap:"sensitive"` + SpannerEmulatorHost string `debugmap:"visible"` + SpannerMinSessions uint64 `debugmap:"visible" default:"100"` + SpannerMaxSessions uint64 `debugmap:"visible" default:"400"` + SpannerDatastoreMetricsOption string `debugmap:"visible" default:"otel"` + + // MySQL + TablePrefix string `debugmap:"visible"` + + // Relationship Integrity + RelationshipIntegrityEnabled bool `debugmap:"visible"` + RelationshipIntegrityCurrentKey RelIntegrityKey `debugmap:"visible"` + RelationshipIntegrityExpiredKeys []string `debugmap:"visible"` + + // Internal + WatchBufferLength uint16 `debugmap:"visible" default:"1024"` + WatchChangeBufferMaximumSize string `debugmap:"visible" default:"15%"` + WatchBufferWriteTimeout time.Duration `debugmap:"visible" default:"1s"` + WatchConnectTimeout time.Duration `debugmap:"visible" default:"1s"` + DisableWatchSupport bool `debugmap:"hidden"` + + // Migrations + MigrationPhase string `debugmap:"visible"` + AllowedMigrations []string `debugmap:"visible"` + + // Experimental + ExperimentalColumnOptimization bool `debugmap:"visible" default:"true"` + EnableRevisionHeartbeat bool `debugmap:"visible"` +} + +// SetDefaults is invoked by github.com/creasty/defaults after struct-tag +// defaults are applied. It fills the four ConnPoolConfig slots from the +// canonical DefaultReadConnPool / DefaultWriteConnPool constructors because +// each slot receives a different default set from RegisterConnPoolFlagsWithPrefix +// in RegisterDatastoreFlagsWithPrefix (Read pools = 20/20 conns, Write = 10/10). +// It also pre-allocates slice fields to empty (non-nil) values so the +// resulting Config matches what RegisterDatastoreFlags writes via +// StringSliceVar/StringArrayVar. +func (c *Config) SetDefaults() { + c.ReadConnPool = *DefaultReadConnPool() + c.WriteConnPool = *DefaultWriteConnPool() + c.ReadReplicaConnPool = *DefaultReadConnPool() + c.OldReadReplicaConnPool = *DefaultReadConnPool() + + // CaveatTypeSet is hidden from DebugMap but RegisterDatastoreFlags + // initializes it from DefaultDatastoreConfig at line 223. Mirror that + // here so library users get the same value as CLI users. + if c.CaveatTypeSet == nil { + c.CaveatTypeSet = caveattypes.Default.TypeSet + } + + if c.BootstrapFiles == nil { + c.BootstrapFiles = []string{} + } + if c.ReadReplicaURIs == nil { + c.ReadReplicaURIs = []string{} + } + if c.AllowedMigrations == nil { + c.AllowedMigrations = []string{} + } + if c.RelationshipIntegrityExpiredKeys == nil { + c.RelationshipIntegrityExpiredKeys = []string{} + } +} + +//go:generate go run github.com/ecordell/optgen -sensitive-field-name-matches uri,secure -output zz_generated.relintegritykey.options.go . RelIntegrityKey +type RelIntegrityKey struct { + KeyID string `debugmap:"visible"` + KeyFilename string `debugmap:"visible"` +} + +func DefaultDatastoreConfig() *Config { + return &Config{ + Engine: MemoryEngine, + GCWindow: 24 * time.Hour, + LegacyFuzzing: -1, + RevisionQuantization: 5 * time.Second, + MaxRevisionStalenessPercent: .1, // 10% + ReadConnPool: *DefaultReadConnPool(), + WriteConnPool: *DefaultWriteConnPool(), + ReadReplicaConnPool: *DefaultReadConnPool(), + OldReadReplicaConnPool: *DefaultReadConnPool(), + ReadReplicaURIs: []string{}, + ReadOnly: false, + MaxRetries: 10, + OverlapKey: "key", + OverlapStrategy: "static", + ConnectRate: 100 * time.Millisecond, + EnableConnectionBalancing: true, + GCInterval: 3 * time.Minute, + GCMaxOperationTime: 1 * time.Minute, + WatchBufferLength: 1024, + WatchChangeBufferMaximumSize: "15%", + WatchBufferWriteTimeout: 1 * time.Second, + WatchConnectTimeout: 1 * time.Second, + DisableWatchSupport: false, + EnableDatastoreMetrics: true, + DisableStats: false, + BootstrapFiles: []string{}, + BootstrapTimeout: 10 * time.Second, + BootstrapOverwrite: false, + SpannerCredentialsFile: "", + SpannerEmulatorHost: "", + TablePrefix: "", + MigrationPhase: "", + FollowerReadDelay: DefaultFollowerReadDelay, + SpannerMinSessions: 100, + SpannerMaxSessions: 400, + FilterMaximumIDCount: 100, + SpannerDatastoreMetricsOption: "otel", + RelationshipIntegrityEnabled: false, + RelationshipIntegrityCurrentKey: RelIntegrityKey{}, + RelationshipIntegrityExpiredKeys: []string{}, + AllowedMigrations: []string{}, + ExperimentalColumnOptimization: true, + IncludeQueryParametersInTraces: false, + WriteAcquisitionTimeout: 30 * time.Millisecond, + CaveatTypeSet: caveattypes.Default.TypeSet, + } +} diff --git a/pkg/cmd/datastore/zz_generated.connpool.options.go b/pkg/cmd/datastore/dsconfig/zz_generated.connpool.options.go similarity index 99% rename from pkg/cmd/datastore/zz_generated.connpool.options.go rename to pkg/cmd/datastore/dsconfig/zz_generated.connpool.options.go index 19e7e84be0..266a344f3d 100644 --- a/pkg/cmd/datastore/zz_generated.connpool.options.go +++ b/pkg/cmd/datastore/dsconfig/zz_generated.connpool.options.go @@ -1,5 +1,5 @@ // Code generated by github.com/ecordell/optgen. DO NOT EDIT. -package datastore +package dsconfig import ( defaults "github.com/creasty/defaults" diff --git a/pkg/cmd/datastore/zz_generated.options.go b/pkg/cmd/datastore/dsconfig/zz_generated.options.go similarity index 99% rename from pkg/cmd/datastore/zz_generated.options.go rename to pkg/cmd/datastore/dsconfig/zz_generated.options.go index 9d814ebf72..261381f4bc 100644 --- a/pkg/cmd/datastore/zz_generated.options.go +++ b/pkg/cmd/datastore/dsconfig/zz_generated.options.go @@ -1,5 +1,5 @@ // Code generated by github.com/ecordell/optgen. DO NOT EDIT. -package datastore +package dsconfig import ( "fmt" diff --git a/pkg/cmd/datastore/zz_generated.relintegritykey.options.go b/pkg/cmd/datastore/dsconfig/zz_generated.relintegritykey.options.go similarity index 99% rename from pkg/cmd/datastore/zz_generated.relintegritykey.options.go rename to pkg/cmd/datastore/dsconfig/zz_generated.relintegritykey.options.go index 35870a3a21..a56994e035 100644 --- a/pkg/cmd/datastore/zz_generated.relintegritykey.options.go +++ b/pkg/cmd/datastore/dsconfig/zz_generated.relintegritykey.options.go @@ -1,5 +1,5 @@ // Code generated by github.com/ecordell/optgen. DO NOT EDIT. -package datastore +package dsconfig import defaults "github.com/creasty/defaults" diff --git a/pkg/cmd/migrate.go b/pkg/cmd/migrate.go index 1ea14d4fe7..472750252b 100644 --- a/pkg/cmd/migrate.go +++ b/pkg/cmd/migrate.go @@ -7,32 +7,34 @@ import ( "time" "github.com/fatih/color" - sqlDriver "github.com/go-sql-driver/mysql" "github.com/jzelinskie/cobrautil/v2" "github.com/spf13/cobra" + "github.com/spf13/pflag" - crdbmigrations "github.com/authzed/spicedb/internal/datastore/crdb/migrations" - mysqlmigrations "github.com/authzed/spicedb/internal/datastore/mysql/migrations" - "github.com/authzed/spicedb/internal/datastore/postgres/migrations" - spannermigrations "github.com/authzed/spicedb/internal/datastore/spanner/migrations" - log "github.com/authzed/spicedb/internal/logging" "github.com/authzed/spicedb/pkg/cmd/server" "github.com/authzed/spicedb/pkg/cmd/termination" "github.com/authzed/spicedb/pkg/cmd/util" "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/migration" "github.com/authzed/spicedb/pkg/migrate" ) // MigrateConfig holds configuration for running database migrations. -type MigrateConfig struct { - DatastoreEngine string - DatastoreURI string - CredentialsProviderName string - SpannerCredentialsFile string - SpannerEmulatorHost string - MySQLTablePrefix string - Timeout time.Duration - BatchSize uint64 +type MigrateConfig = migration.Config + +// RegisterMigratableEngine makes a datastore engine defined outside this +// package available to the migrate and head commands. Engine-specific +// command-line flags are available to newDriver via MigrateConfig.ExtraConfig. +// verifiableMigrationName is the earliest migration at which the engine's +// current datastore code can open the datastore and read and write data. +// It must be called before command execution, typically from an init function. +func RegisterMigratableEngine[D migrate.Driver[C, T], C any, T any]( + engineName string, + manager *migrate.Manager[D, C, T], + newDriver func(ctx context.Context, cfg *MigrateConfig) (D, error), + verifiableMigrationName string, +) { + migration.RegisterMigratableEngine(engineName, manager, newDriver, verifiableMigrationName) } func RegisterMigrateFlags(cmd *cobra.Command) { @@ -63,6 +65,11 @@ func migrateRun(cmd *cobra.Command, args []string) error { return errors.New("missing required argument: 'revision'") } + extraConfig := make(map[string]string) + cmd.Flags().VisitAll(func(f *pflag.Flag) { + extraConfig[f.Name] = f.Value.String() + }) + cfg := &MigrateConfig{ DatastoreEngine: cobrautil.MustGetStringExpanded(cmd, "datastore-engine"), DatastoreURI: cobrautil.MustGetStringExpanded(cmd, "datastore-conn-uri"), @@ -74,100 +81,7 @@ func migrateRun(cmd *cobra.Command, args []string) error { BatchSize: cobrautil.MustGetUint64(cmd, "migration-backfill-batch-size"), } - return executeMigrate(cmd.Context(), cfg, args[0]) -} - -// executeMigrate runs the migration with the given configuration. -// This function is extracted to enable testing without cobra command dependencies. -func executeMigrate(ctx context.Context, cfg *MigrateConfig, revision string) error { - if revision == "" { - return errors.New("missing required revision") - } - - switch cfg.DatastoreEngine { - case "cockroachdb": - log.Ctx(ctx).Info().Msg("migrating cockroachdb datastore") - - migrationDriver, err := crdbmigrations.NewCRDBDriver(cfg.DatastoreURI) - if err != nil { - return fmt.Errorf("unable to create migration driver for %s: %w", cfg.DatastoreEngine, err) - } - return runMigration(ctx, migrationDriver, crdbmigrations.CRDBMigrations, revision, cfg.Timeout, cfg.BatchSize) - - case "postgres": - log.Ctx(ctx).Info().Msg("migrating postgres datastore") - - var credentialsProvider datastore.CredentialsProvider - if cfg.CredentialsProviderName != "" { - var err error - credentialsProvider, err = datastore.NewCredentialsProvider(ctx, cfg.CredentialsProviderName) - if err != nil { - return err - } - } - - migrationDriver, err := migrations.NewAlembicPostgresDriver(ctx, cfg.DatastoreURI, credentialsProvider, false) - if err != nil { - return fmt.Errorf("unable to create migration driver for %s: %w", cfg.DatastoreEngine, err) - } - return runMigration(ctx, migrationDriver, migrations.DatabaseMigrations, revision, cfg.Timeout, cfg.BatchSize) - - case "spanner": - log.Ctx(ctx).Info().Msg("migrating spanner datastore") - - migrationDriver, err := spannermigrations.NewSpannerDriver(ctx, cfg.DatastoreURI, cfg.SpannerCredentialsFile, cfg.SpannerEmulatorHost) - if err != nil { - return fmt.Errorf("unable to create migration driver for %s: %w", cfg.DatastoreEngine, err) - } - return runMigration(ctx, migrationDriver, spannermigrations.SpannerMigrations, revision, cfg.Timeout, cfg.BatchSize) - - case "mysql": - log.Ctx(ctx).Info().Msg("migrating mysql datastore") - - var credentialsProvider datastore.CredentialsProvider - if cfg.CredentialsProviderName != "" { - var err error - credentialsProvider, err = datastore.NewCredentialsProvider(ctx, cfg.CredentialsProviderName) - if err != nil { - return err - } - } - - // Do this outside NewMySQLDriverFromDSN to avoid races on MySQL datastore tests - if err := sqlDriver.SetLogger(&log.Logger); err != nil { - return fmt.Errorf("unable to set logging to mysql driver: %w", err) - } - - migrationDriver, err := mysqlmigrations.NewMySQLDriverFromDSN(cfg.DatastoreURI, cfg.MySQLTablePrefix, credentialsProvider) - if err != nil { - return fmt.Errorf("unable to create migration driver for %s: %w", cfg.DatastoreEngine, err) - } - return runMigration(ctx, migrationDriver, mysqlmigrations.Manager, revision, cfg.Timeout, cfg.BatchSize) - } - - return fmt.Errorf("cannot migrate datastore engine type: %s", cfg.DatastoreEngine) -} - -func runMigration[D migrate.Driver[C, T], C any, T any]( - ctx context.Context, - driver D, - manager *migrate.Manager[D, C, T], - targetRevision string, - timeout time.Duration, - backfillBatchSize uint64, -) error { - log.Ctx(ctx).Info().Str("targetRevision", targetRevision).Msg("running migrations") - ctxWithBatch := context.WithValue(ctx, migrate.BackfillBatchSize, backfillBatchSize) - ctx, cancel := context.WithTimeout(ctxWithBatch, timeout) - defer cancel() - if err := manager.Run(ctx, driver, targetRevision, migrate.LiveRun); err != nil { - return fmt.Errorf("unable to migrate to `%s` revision: %w", targetRevision, err) - } - - if err := driver.Close(ctx); err != nil { - return fmt.Errorf("unable to close migration driver: %w", err) - } - return nil + return migration.Run(cmd.Context(), cfg, args[0]) } func RegisterHeadFlags(cmd *cobra.Command) { @@ -181,7 +95,8 @@ func NewHeadCommand(programName string) *cobra.Command { Short: "compute the head (latest) database migration revision available", PreRunE: server.DefaultPreRunE(programName), RunE: func(cmd *cobra.Command, args []string) error { - headRevision, err := HeadRevision(cobrautil.MustGetStringExpanded(cmd, "datastore-engine")) + engine := cobrautil.MustGetStringExpanded(cmd, "datastore-engine") + headRevision, err := migration.HeadRevision(engine) if err != nil { return fmt.Errorf("unable to compute head revision: %w", err) } @@ -191,19 +106,3 @@ func NewHeadCommand(programName string) *cobra.Command { Args: cobra.ExactArgs(0), } } - -// HeadRevision returns the latest migration revision for a given engine -func HeadRevision(engine string) (string, error) { - switch engine { - case "cockroachdb": - return crdbmigrations.CRDBMigrations.HeadRevision() - case "postgres": - return migrations.DatabaseMigrations.HeadRevision() - case "mysql": - return mysqlmigrations.Manager.HeadRevision() - case "spanner": - return spannermigrations.SpannerMigrations.HeadRevision() - default: - return "", fmt.Errorf("cannot migrate datastore engine type: %s", engine) - } -} diff --git a/pkg/cmd/migrate_test.go b/pkg/cmd/migrate_test.go index ee5a107e32..b9b47b1dc9 100644 --- a/pkg/cmd/migrate_test.go +++ b/pkg/cmd/migrate_test.go @@ -1,28 +1,15 @@ package cmd import ( - "io" "testing" "time" "github.com/spf13/cobra" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/testcontainers/testcontainers-go" - "github.com/testcontainers/testcontainers-go/log" - "github.com/testcontainers/testcontainers-go/network" - "github.com/testcontainers/testcontainers-go/wait" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - - v1 "github.com/authzed/authzed-go/proto/authzed/api/v1" - "github.com/authzed/grpcutil" datastoreTest "github.com/authzed/spicedb/internal/testserver/datastore" - "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/migration" "github.com/authzed/spicedb/pkg/migrate" - "github.com/authzed/spicedb/pkg/testutil" - "github.com/authzed/spicedb/pkg/testutil/sdbtestcontainer" ) func TestExecuteMigrateErrorsOut(t *testing.T) { @@ -146,7 +133,7 @@ func TestExecuteMigrateErrorsOut(t *testing.T) { t.Run(tt.name, func(t *testing.T) { cfg := tt.cfgBuilder(t) - err := executeMigrate(t.Context(), cfg, tt.revision) + err := migration.Run(t.Context(), cfg, tt.revision) if tt.expectedError == "" { require.NoError(t, err) return @@ -157,11 +144,7 @@ func TestExecuteMigrateErrorsOut(t *testing.T) { } func TestExecuteMigrateWithNoDataSucceeds(t *testing.T) { - for _, engineKey := range datastore.Engines { - if engineKey == "memory" { - continue - } - + for _, engineKey := range migration.Engines() { t.Run(engineKey, func(t *testing.T) { r := datastoreTest.RunDatastoreEngine(t, engineKey) db := r.NewDatabase(t) @@ -172,129 +155,11 @@ func TestExecuteMigrateWithNoDataSucceeds(t *testing.T) { Timeout: 1 * time.Hour, BatchSize: 1000, } - require.NoError(t, executeMigrate(t.Context(), cfg, migrate.Head)) - }) - } -} - -// TestExecuteMigrateWithDataSucceeds verifies that migrations introduced on v1.53.0 -// apply on top of a database that has data written by v1.52.0. -func TestExecuteMigrateWithDataSucceeds(t *testing.T) { - for _, engineKey := range datastore.Engines { - if engineKey == "memory" { - continue - } - - ctx := t.Context() - - // Create an internal network - net, err := network.New(ctx) - testcontainers.CleanupNetwork(t, net) - require.NoError(t, err) - - t.Run(engineKey, func(t *testing.T) { - r := datastoreTest.RunDatastoreEngine(t, engineKey, network.WithNetwork([]string{engineKey}, net)) - db := r.NewDatabase(t) - - // 1. Migrate using SpiceDB v1.52.0. - runMigrateHeadWithContainer(t, "v1.52.0", engineKey, db, net) - - // 2. Run v1.52.0 serve and write a schema. - serveContainer := runServe(t, "v1.52.0", engineKey, db, net) - - conn, err := grpc.NewClient( - serveContainer.GRPCEndpoint(), - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpcutil.WithInsecureBearerToken(serveContainer.PresharedKey()), - ) - require.NoError(t, err) - t.Cleanup(func() { - _ = conn.Close() - }) - - require.EventuallyWithT(t, func(collect *assert.CollectT) { - _, err := v1.NewSchemaServiceClient(conn).WriteSchema(t.Context(), &v1.WriteSchemaRequest{ - Schema: ` - caveat is_public(public bool) { - public - } - - definition user {} - definition document { - relation viewer: user with is_public - permission view = viewer - } - `, - }) - assert.NoError(collect, err) - }, 30*time.Second, 1*time.Second) - - // 3. Migrate using the current branch's code, in-process, - // so the migration code is included in the coverage profile. - cfg := &MigrateConfig{ - DatastoreEngine: engineKey, - DatastoreURI: db, - Timeout: 5 * time.Minute, - BatchSize: 1000, - } - require.NoError(t, executeMigrate(t.Context(), cfg, migrate.Head)) + require.NoError(t, migration.Run(t.Context(), cfg, migrate.Head)) }) } } -// runMigrateHeadWithContainer launches a docker container that runs `spicedb migrate head` -// Use this when you need to exercise a released SpiceDB binary. -func runMigrateHeadWithContainer(t *testing.T, spiceDBImageTag, engineKey, db string, net *testcontainers.DockerNetwork) { - t.Helper() - - ctx := t.Context() - - connectionVars, err := testutil.InternalConnectionEnvVars(db, engineKey) - require.NoError(t, err) - - migrateContainer, err := testcontainers.Run(ctx, - "authzed/spicedb:"+spiceDBImageTag, - network.WithNetwork([]string{"migrate"}, net), - testcontainers.WithLogger(log.TestLogger(t)), - testcontainers.WithCmd("migrate", "head"), - testcontainers.WithEnv(connectionVars), - testcontainers.WithWaitStrategy(wait.ForExit().WithExitTimeout(time.Minute)), - ) - require.NoError(t, err) - testcontainers.CleanupContainer(t, migrateContainer) - - // Ensure the command completed successfully. - containerState, err := migrateContainer.State(ctx) - if containerState.ExitCode != 0 { - logReader, err := migrateContainer.Logs(t.Context()) - require.NoError(t, err) - out, err := io.ReadAll(logReader) - require.NoError(t, err) - t.Log("Container logs:") - t.Log(string(out)) - } - require.NoError(t, err) - require.Equal(t, 0, containerState.ExitCode) -} - -func runServe(t *testing.T, spiceDBImageTag, engineKey, dbConnection string, net *testcontainers.DockerNetwork) *sdbtestcontainer.Container { - t.Helper() - - connectionVars, err := testutil.InternalConnectionEnvVars(dbConnection, engineKey) - require.NoError(t, err) - - container, err := sdbtestcontainer.Run( - t.Context(), - "authzed/spicedb:"+spiceDBImageTag, - network.WithNetwork([]string{"spicedb"}, net), - testcontainers.WithEnv(connectionVars), - ) - require.NoError(t, err) - testcontainers.CleanupContainer(t, container) - - return container -} - func TestMigrateRun(t *testing.T) { tests := []struct { name string diff --git a/pkg/cmd/root.go b/pkg/cmd/root.go index 7c088a9dab..69aea46f78 100644 --- a/pkg/cmd/root.go +++ b/pkg/cmd/root.go @@ -11,6 +11,8 @@ import ( "github.com/muesli/roff" "github.com/spf13/cobra" + // Register all datastore engines defined in this repository. + _ "github.com/authzed/spicedb/internal/datastore/engines" log "github.com/authzed/spicedb/internal/logging" "github.com/authzed/spicedb/pkg/cmd/server" "github.com/authzed/spicedb/pkg/cmd/testserver" diff --git a/pkg/datastore/datastore.go b/pkg/datastore/datastore.go index b3c8ebe744..7503346099 100644 --- a/pkg/datastore/datastore.go +++ b/pkg/datastore/datastore.go @@ -876,6 +876,14 @@ type UnwrappableDatastore interface { Unwrap() Datastore } +// EngineIdentifiable is optionally implemented by datastores that can report +// the name of the engine that backs them. Use UnwrapAs to find it behind +// wrapping proxies. +type EngineIdentifiable interface { + // EngineName returns the datastore engine name, e.g. "postgres". + EngineName() string +} + // UnwrapAs recursively attempts to unwrap the datastore into the specified type // In none of the layers of the datastore implement the specified type, nil is returned. func UnwrapAs[T any](datastore Datastore) T { diff --git a/pkg/datastore/migration/migration.go b/pkg/datastore/migration/migration.go new file mode 100644 index 0000000000..2b1e8466be --- /dev/null +++ b/pkg/datastore/migration/migration.go @@ -0,0 +1,172 @@ +// Package migration provides a registry of the datastore engines that can run +// schema migrations. It is shared by the `migrate` and `head` commands and by +// the generic datastore test suite. Engines register themselves via +// RegisterMigratableEngine from an init function; importing an engine package +// makes it migratable. +package migration + +import ( + "context" + "errors" + "fmt" + "slices" + "time" + + log "github.com/authzed/spicedb/internal/logging" + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/migrate" +) + +// Config holds configuration for running database migrations. +type Config struct { + DatastoreEngine string + DatastoreURI string + CredentialsProviderName string + SpannerCredentialsFile string + SpannerEmulatorHost string + MySQLTablePrefix string + Timeout time.Duration + BatchSize uint64 +} + +// CredentialsProvider returns the configured datastore credentials provider, +// or nil if none is configured. +func (cfg *Config) CredentialsProvider(ctx context.Context) (datastore.CredentialsProvider, error) { + if cfg.CredentialsProviderName == "" { + return nil, nil + } + return datastore.NewCredentialsProvider(ctx, cfg.CredentialsProviderName) +} + +type migratableEngine struct { + // verifiableMigrationName is the earliest migration at which the current + // code can open the datastore and read and write data. If the current code + // starts depending on a later migration, this must be advanced to it. + verifiableMigrationName string + + headRevision func() (string, error) + migrationNames func() ([]string, error) + version func(ctx context.Context, cfg *Config) (string, error) + migrate func(ctx context.Context, cfg *Config, revision string) error +} + +// migratableEngines holds the registered migratable engines, keyed by engine +// name. Engines register themselves via RegisterMigratableEngine from an init +// function; importing an engine package makes it migratable. +var migratableEngines = map[string]migratableEngine{} + +// RegisterMigratableEngine makes a datastore engine migratable. It is +// typically called from an init function of the package defining the engine. +// verifiableMigrationName is the earliest migration at which the engine's +// current datastore code can open the datastore and read and write data. +func RegisterMigratableEngine[D migrate.Driver[C, T], C any, T any]( + engineName string, + manager *migrate.Manager[D, C, T], + newDriver func(ctx context.Context, cfg *Config) (D, error), + verifiableMigrationName string, +) { + migratableEngines[engineName] = newMigratableEngine(manager, newDriver, verifiableMigrationName) +} + +// newMigratableEngine pairs a datastore engine's migration manager with the constructor for its migration driver +func newMigratableEngine[D migrate.Driver[C, T], C any, T any]( + manager *migrate.Manager[D, C, T], + newDriver func(ctx context.Context, cfg *Config) (D, error), + verifiableMigrationName string, +) migratableEngine { + return migratableEngine{ + verifiableMigrationName: verifiableMigrationName, + headRevision: manager.HeadRevision, + migrationNames: manager.MigrationNames, + version: func(ctx context.Context, cfg *Config) (string, error) { + driver, err := newDriver(ctx, cfg) + if err != nil { + return "", fmt.Errorf("unable to create migration driver for %s: %w", cfg.DatastoreEngine, err) + } + defer func() { _ = driver.Close(ctx) }() + return driver.Version(ctx) + }, + migrate: func(ctx context.Context, cfg *Config, revision string) error { + driver, err := newDriver(ctx, cfg) + if err != nil { + return fmt.Errorf("unable to create migration driver for %s: %w", cfg.DatastoreEngine, err) + } + log.Ctx(ctx).Info().Str("targetRevision", revision).Msg("running migrations") + ctxWithBatch := context.WithValue(ctx, migrate.BackfillBatchSize, cfg.BatchSize) + ctx, cancel := context.WithTimeout(ctxWithBatch, cfg.Timeout) + defer cancel() + if err := manager.Run(ctx, driver, revision, migrate.LiveRun); err != nil { + return fmt.Errorf("unable to migrate to `%s` revision: %w", revision, err) + } + + if err := driver.Close(ctx); err != nil { + return fmt.Errorf("unable to close migration driver: %w", err) + } + return nil + }, + } +} + +// Run runs the migrations for the configured datastore engine up to the given +// revision. +func Run(ctx context.Context, cfg *Config, revision string) error { + if revision == "" { + return errors.New("missing required revision") + } + + e, ok := migratableEngines[cfg.DatastoreEngine] + if !ok { + return fmt.Errorf("cannot migrate datastore engine type: %s", cfg.DatastoreEngine) + } + + log.Ctx(ctx).Info().Str("engine", cfg.DatastoreEngine).Msg("migrating datastore") + return e.migrate(ctx, cfg, revision) +} + +// HeadRevision returns the latest migration revision for the given engine. +func HeadRevision(engine string) (string, error) { + e, ok := migratableEngines[engine] + if !ok { + return "", fmt.Errorf("cannot migrate datastore engine type: %s", engine) + } + return e.headRevision() +} + +// VerifiableMigrationName returns the earliest migration at which the given +// engine's current datastore code can open the datastore and read and write data. +func VerifiableMigrationName(engine string) (string, error) { + e, ok := migratableEngines[engine] + if !ok { + return "", fmt.Errorf("cannot migrate datastore engine type: %s", engine) + } + return e.verifiableMigrationName, nil +} + +// MigrationNames returns the names of every migration supported by the given +// engine, ordered from oldest to newest (head). +func MigrationNames(engine string) ([]string, error) { + e, ok := migratableEngines[engine] + if !ok { + return nil, fmt.Errorf("cannot migrate datastore engine type: %s", engine) + } + return e.migrationNames() +} + +// Version returns the migration revision the configured datastore is currently at. +func Version(ctx context.Context, cfg *Config) (string, error) { + e, ok := migratableEngines[cfg.DatastoreEngine] + if !ok { + return "", fmt.Errorf("cannot migrate datastore engine type: %s", cfg.DatastoreEngine) + } + return e.version(ctx, cfg) +} + +// Engines returns the names of all registered migratable engines, sorted. +func Engines() []string { + engines := make([]string, 0, len(migratableEngines)) + for engine := range migratableEngines { + engines = append(engines, engine) + } + slices.Sort(engines) + return engines +} diff --git a/pkg/datastore/mocks/mock_datastore.go b/pkg/datastore/mocks/mock_datastore.go index 5f91bc62bc..2a98fcdeae 100644 --- a/pkg/datastore/mocks/mock_datastore.go +++ b/pkg/datastore/mocks/mock_datastore.go @@ -2324,6 +2324,44 @@ func (mr *MockUnwrappableDatastoreMockRecorder) Unwrap() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Unwrap", reflect.TypeOf((*MockUnwrappableDatastore)(nil).Unwrap)) } +// MockEngineIdentifiable is a mock of EngineIdentifiable interface. +type MockEngineIdentifiable struct { + ctrl *gomock.Controller + recorder *MockEngineIdentifiableMockRecorder + isgomock struct{} +} + +// MockEngineIdentifiableMockRecorder is the mock recorder for MockEngineIdentifiable. +type MockEngineIdentifiableMockRecorder struct { + mock *MockEngineIdentifiable +} + +// NewMockEngineIdentifiable creates a new mock instance. +func NewMockEngineIdentifiable(ctrl *gomock.Controller) *MockEngineIdentifiable { + mock := &MockEngineIdentifiable{ctrl: ctrl} + mock.recorder = &MockEngineIdentifiableMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockEngineIdentifiable) EXPECT() *MockEngineIdentifiableMockRecorder { + return m.recorder +} + +// EngineName mocks base method. +func (m *MockEngineIdentifiable) EngineName() string { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "EngineName") + ret0, _ := ret[0].(string) + return ret0 +} + +// EngineName indicates an expected call of EngineName. +func (mr *MockEngineIdentifiableMockRecorder) EngineName() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EngineName", reflect.TypeOf((*MockEngineIdentifiable)(nil).EngineName)) +} + // MockRevision is a mock of Revision interface. type MockRevision struct { ctrl *gomock.Controller diff --git a/pkg/datastore/test/datastore.go b/pkg/datastore/test/datastore.go index 6fecd070fd..5659711099 100644 --- a/pkg/datastore/test/datastore.go +++ b/pkg/datastore/test/datastore.go @@ -95,6 +95,11 @@ func (c Categories) ConcurrentWrite() bool { return ok } +func (c Categories) Migration() bool { + _, ok := c[MigrationCategory] + return ok +} + var noException = Categories{} const ( @@ -107,6 +112,9 @@ const ( // ConcurrentWriteCategory marks tests that open two write transactions concurrently. // Datastores with a global write lock (e.g. memdb) must exclude this category. ConcurrentWriteCategory = "ConcurrentWrite" + // MigrationCategory marks tests that check that database migrations apply + // successfully and preserve the data written before them. + MigrationCategory = "Migration" ) func WithCategories(cats ...string) Categories { @@ -269,6 +277,10 @@ func AllWithExceptions(t *testing.T, tester DatastoreTester, except Categories) } t.Run("TestHeadRevisionSchemaHash", runner(tester, HeadRevisionSchemaHashTest)) t.Run("TestOptimizedRevisionSchemaHash", runner(tester, OptimizedRevisionSchemaHashTest)) + + if !except.Migration() { + t.Run("TestMigration", runner(tester, MigrationTest)) + } } func OnlyGCTests(t *testing.T, tester DatastoreTester) { diff --git a/pkg/datastore/test/migration.go b/pkg/datastore/test/migration.go new file mode 100644 index 0000000000..b73ce71b78 --- /dev/null +++ b/pkg/datastore/test/migration.go @@ -0,0 +1,169 @@ +package test + +import ( + "fmt" + "slices" + "testing" + "time" + + "github.com/stretchr/testify/require" + + v1 "github.com/authzed/authzed-go/proto/authzed/api/v1" + + "github.com/authzed/spicedb/internal/testserver" + testdatastore "github.com/authzed/spicedb/internal/testserver/datastore" + datastorecfg "github.com/authzed/spicedb/pkg/cmd/datastore" + "github.com/authzed/spicedb/pkg/datastore" + "github.com/authzed/spicedb/pkg/datastore/migration" +) + +var migrationTestConfigs = map[string]func(t *testing.T) string{} + +// RegisterMigrationTestConfig registers a config provider for a datastore +// engine defined outside this repository, enabling MigrationTest for it. The +// provider returns the URI of an empty database that has had no migrations +// applied +// The engine must be migratable, registered in datastore.BuilderForEngine, and its datastore must implement +// datastore.EngineIdentifiable. +func RegisterMigrationTestConfig(engineKey string, provider func(t *testing.T) string) { + migrationTestConfigs[engineKey] = provider +} + +// MigrationTest verifies every migration of the datastore engine, in order, +// against an empty, unmigrated database: +// 1. Each migration is applied one at a time, verifying that it applies +// cleanly and that the recorded version advances to it. +// 2. Once the datastore reaches the engine's verifiable migration (the +// earliest one the current code can operate against), a SpiceDB server is +// stood up against it and a schema is written and read back through its API. +// 3. Before each remaining migration a new schema is written; after the +// migration that schema must still be readable, verifying that each +// migration preserves the data written by the server running before it. +func MigrationTest(t *testing.T, tester DatastoreTester) { + ds, err := tester.New(t, 0, veryLargeGCInterval, veryLargeGCWindow, 16) + require.NoError(t, err) + + identifiable := datastore.UnwrapAs[datastore.EngineIdentifiable](ds) + if identifiable == nil { + t.Skip("datastore does not implement datastore.EngineIdentifiable") + return + } + + engineKey := identifiable.EngineName() + require.NoError(t, ds.Close()) + t.Logf("running migration test for engine %q", engineKey) + + var datastoreURI string + if provider, ok := migrationTestConfigs[engineKey]; ok { + t.Logf("creating an empty test database via the registered migration test config") + datastoreURI = provider(t) + } else { + if !slices.Contains(migration.Engines(), engineKey) { + t.Skipf("engine %q is not migratable; register it via migration.RegisterMigratableEngine and RegisterMigrationTestConfig", engineKey) + } + t.Logf("creating an empty test database via testdatastore") + datastoreURI = testdatastore.RunDatastoreEngine(t, engineKey).NewDatabase(t) + } + t.Logf("test database URI: %s", datastoreURI) + + migrationNames, err := migration.MigrationNames(engineKey) + require.NoError(t, err) + verifiableMigrationName, err := migration.VerifiableMigrationName(engineKey) + require.NoError(t, err) + firstVerifiable := slices.Index(migrationNames, verifiableMigrationName) + require.GreaterOrEqualf(t, firstVerifiable, 0, "verifiable migration %q is not a known migration of engine %q", verifiableMigrationName, engineKey) + t.Logf("engine %q has %d migrations: %q; the earliest the current code can operate against is %q", engineKey, len(migrationNames), migrationNames, verifiableMigrationName) + + migrationCfg := &migration.Config{ + DatastoreEngine: engineKey, + DatastoreURI: datastoreURI, + Timeout: 5 * time.Minute, + BatchSize: 1000, + } + + var schemaClient v1.SchemaServiceClient + step := 0 + for i, migrationName := range migrationNames { + t.Logf("applying migration %d/%d: %q", i+1, len(migrationNames), migrationName) + require.NoErrorf(t, migration.Run(t.Context(), migrationCfg, migrationName), "failed to apply migration %q", migrationName) + version, err := migration.Version(t.Context(), migrationCfg) + require.NoError(t, err) + require.Equalf(t, migrationName, version, "datastore version did not advance to %q", migrationName) + + switch { + case i < firstVerifiable: + continue + case i == firstVerifiable: + t.Logf("reached verifiable migration %q; standing up a SpiceDB server", migrationName) + schemaClient = startMigrationTestServer(t, engineKey, datastoreURI, migrationNames[i:]) + default: + // The schema written before this migration must still be readable. + t.Logf("verifying that the schema written before migration %q is still readable", migrationName) + requireStepSchema(t, schemaClient, step) + } + + step++ + writeStepSchema(t, schemaClient, step) + requireStepSchema(t, schemaClient, step) + } +} + +// startMigrationTestServer stands up a SpiceDB server against the given +// database and returns a schema service client for it. +func startMigrationTestServer(t *testing.T, engineKey, datastoreURI string, allowedMigrations []string) v1.SchemaServiceClient { + dsOpts := []datastorecfg.ConfigOption{ + datastorecfg.WithEngine(engineKey), + datastorecfg.WithURI(datastoreURI), + datastorecfg.WithRevisionQuantization(0), + datastorecfg.WithRequestHedgingEnabled(false), + datastorecfg.SetAllowedMigrations(allowedMigrations), + } + ds, err := datastorecfg.NewDatastore(t.Context(), dsOpts...) + require.NoError(t, err) + + conn, _, _ := testserver.NewTestServerWithConfigAndDatastore(t, false, testserver.DefaultTestServerConfig, ds, + func(_ testing.TB, ds datastore.Datastore) (datastore.Datastore, datastore.Revision) { + return ds, datastore.NoRevision + }) + t.Cleanup(func() { + require.NoError(t, conn.Close()) + require.NoError(t, ds.Close()) + }) + return v1.NewSchemaServiceClient(conn) +} + +// stepSchemaText returns a schema unique to the given step, so that a read +// after each migration can be attributed to the write that preceded it. +func stepSchemaText(step int) string { + return fmt.Sprintf(`caveat is_public_step%d(public bool) { + public +} + +definition user {} + +definition document_step%d { + relation viewer: user with is_public_step%d + permission view = viewer +}`, step, step, step) +} + +// writeStepSchema writes the schema for the given step, replacing any previously written schema. +func writeStepSchema(t *testing.T, schemaClient v1.SchemaServiceClient, step int) { + t.Helper() + + t.Logf("writing schema for step %d", step) + _, err := schemaClient.WriteSchema(t.Context(), &v1.WriteSchemaRequest{ + Schema: stepSchemaText(step), + }) + require.NoErrorf(t, err, "failed to write schema for step %d", step) +} + +// requireStepSchema verifies that the schema written for the given step is readable. +func requireStepSchema(t *testing.T, schemaClient v1.SchemaServiceClient, step int) { + t.Helper() + + resp, err := schemaClient.ReadSchema(t.Context(), &v1.ReadSchemaRequest{}) + require.NoErrorf(t, err, "failed to read schema at step %d", step) + require.Containsf(t, resp.SchemaText, fmt.Sprintf("document_step%d", step), "read schema does not contain the definitions written for step %d", step) + t.Logf("schema for step %d is readable", step) +} diff --git a/pkg/migrate/migrate.go b/pkg/migrate/migrate.go index 67af14edbd..a7ea2de467 100644 --- a/pkg/migrate/migrate.go +++ b/pkg/migrate/migrate.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "slices" "strings" log "github.com/authzed/spicedb/internal/logging" @@ -187,6 +188,28 @@ func (m *Manager[D, C, T]) HeadRevision() (string, error) { return allHeads[0], nil } +// MigrationNames returns the names of every registered migration, ordered +// from oldest to newest (head). +func (m *Manager[D, C, T]) MigrationNames() ([]string, error) { + head, err := m.HeadRevision() + if err != nil { + return nil, err + } + + names := make([]string, 0, len(m.migrations)) + for revision := head; revision != None; { + registered, ok := m.migrations[revision] + if !ok { + // The oldest registered migration replaces one that is no longer registered. + break + } + names = append(names, revision) + revision = registered.replaces + } + slices.Reverse(names) + return names, nil +} + func (m *Manager[D, C, T]) IsHeadCompatible(revision string) (bool, error) { headRevision, err := m.HeadRevision() if err != nil { diff --git a/pkg/migrate/migrate_test.go b/pkg/migrate/migrate_test.go index 9c6ba05b40..48226d2db0 100644 --- a/pkg/migrate/migrate_test.go +++ b/pkg/migrate/migrate_test.go @@ -152,6 +152,28 @@ func TestComputeHeadRevision(t *testing.T) { } } +func TestMigrationNames(t *testing.T) { + testCases := []struct { + migrations map[string]migration[fakeConnPool, fakeTx] + names []string + expectError bool + }{ + {noMigrations, nil, true}, + {simpleMigrations, []string{"123"}, false}, + {singleHeadedChain, []string{"123", "456", "789"}, false}, + {multiHeadedChain, nil, true}, + {missingEarlyMigrations, []string{"456", "789", "10"}, false}, + } + + req := require.New(t) + for _, tc := range testCases { + m := Manager[Driver[fakeConnPool, fakeTx], fakeConnPool, fakeTx]{migrations: tc.migrations} + names, err := m.MigrationNames() + req.Equal(tc.expectError, err != nil, err) + req.Equal(tc.names, names) + } +} + func TestIsHeadCompatible(t *testing.T) { testCases := []struct { migrations map[string]migration[fakeConnPool, fakeTx]