Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: implement background syncing for warehouse schema
- Loading branch information
1 parent
18f4bdf
commit f539685
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package router | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| obskit "github.com/rudderlabs/rudder-observability-kit/go/labels" | ||
| "github.com/rudderlabs/rudder-server/warehouse/integrations/manager" | ||
| "github.com/rudderlabs/rudder-server/warehouse/internal/model" | ||
| "github.com/rudderlabs/rudder-server/warehouse/schema" | ||
| warehouseutils "github.com/rudderlabs/rudder-server/warehouse/utils" | ||
| ) | ||
|
|
||
| func (r *Router) sync(ctx context.Context) error { | ||
| for { | ||
| r.configSubscriberLock.RLock() | ||
| warehouses := append([]model.Warehouse{}, r.warehouses...) | ||
| r.configSubscriberLock.RUnlock() | ||
| execTime := time.Now() | ||
| whManager, err := manager.New(r.destType, r.conf, r.logger, r.statsFactory) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create warehouse manager: %w", err) | ||
| } | ||
| for _, warehouse := range warehouses { | ||
| err := whManager.Setup(ctx, warehouse, warehouseutils.NewNoOpUploader()) | ||
| if err != nil { | ||
| r.logger.Errorn("failed to setup WH Manager", obskit.Error(err)) | ||
| continue | ||
| } | ||
| if err := schema.SyncSchema(ctx, whManager, warehouse, r.db, r.logger.Child("syncer")); err != nil { | ||
| r.logger.Errorn("failed to sync schema", obskit.Error(err)) | ||
| continue | ||
| } | ||
| } | ||
| nextExecTime := execTime.Add(r.config.syncSchemaFrequency) | ||
| select { | ||
| case <-ctx.Done(): | ||
| r.logger.Infon("context is cancelled, stopped running schema syncer") | ||
| return nil | ||
| case <-time.After(time.Until(nextExecTime)): | ||
| } | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,154 @@ | ||
| package router | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "reflect" | ||
| "testing" | ||
| "time" | ||
|
|
||
| miniogo "github.com/minio/minio-go/v7" | ||
| "github.com/ory/dockertest/v3" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/rudderlabs/rudder-go-kit/config" | ||
| "github.com/rudderlabs/rudder-go-kit/logger" | ||
| "github.com/rudderlabs/rudder-go-kit/stats" | ||
| "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/minio" | ||
| "github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres" | ||
|
|
||
| backendconfig "github.com/rudderlabs/rudder-server/backend-config" | ||
| migrator "github.com/rudderlabs/rudder-server/services/sql-migrator" | ||
| sqlmiddleware "github.com/rudderlabs/rudder-server/warehouse/integrations/middleware/sqlquerywrapper" | ||
| "github.com/rudderlabs/rudder-server/warehouse/internal/model" | ||
| "github.com/rudderlabs/rudder-server/warehouse/schema" | ||
| warehouseutils "github.com/rudderlabs/rudder-server/warehouse/utils" | ||
| ) | ||
|
|
||
| func TestSync_SyncRemoteSchemaIntegration(t *testing.T) { | ||
| destinationType := warehouseutils.POSTGRES | ||
| bucket := "some-bucket" | ||
| sourceID := "test-source-id" | ||
| destinationID := "test-destination-id" | ||
| workspaceID := "test-workspace-id" | ||
| provider := "MINIO" | ||
| sslMode := "disable" | ||
| testNamespace := "test_namespace" | ||
| testTable := "test_table" | ||
|
|
||
| ctx := context.Background() | ||
| pool, err := dockertest.NewPool("") | ||
| require.NoError(t, err) | ||
| pgResource, err := postgres.Setup(pool, t) | ||
| require.NoError(t, err) | ||
| minioResource, err := minio.Setup(pool, t) | ||
| require.NoError(t, err) | ||
|
|
||
| err = minioResource.Client.MakeBucket(ctx, bucket, miniogo.MakeBucketOptions{ | ||
| Region: "us-east-1", | ||
| }) | ||
| require.NoError(t, err) | ||
| t.Log("db:", pgResource.DBDsn) | ||
| conf := config.New() | ||
| err = (&migrator.Migrator{ | ||
| Handle: pgResource.DB, | ||
| MigrationsTable: "wh_schema_migrations", | ||
| }).Migrate("warehouse") | ||
| require.NoError(t, err) | ||
|
|
||
| db := sqlmiddleware.New(pgResource.DB) | ||
|
|
||
| warehouse := model.Warehouse{ | ||
| WorkspaceID: workspaceID, | ||
| Source: backendconfig.SourceT{ | ||
| ID: sourceID, | ||
| }, | ||
| Destination: backendconfig.DestinationT{ | ||
| ID: destinationID, | ||
| DestinationDefinition: backendconfig.DestinationDefinitionT{ | ||
| Name: destinationType, | ||
| }, | ||
| Config: map[string]interface{}{ | ||
| "host": pgResource.Host, | ||
| "port": pgResource.Port, | ||
| "database": pgResource.Database, | ||
| "user": pgResource.User, | ||
| "password": pgResource.Password, | ||
| "sslMode": sslMode, | ||
| "bucketProvider": provider, | ||
| "bucketName": minioResource.BucketName, | ||
| "accessKeyID": minioResource.AccessKeyID, | ||
| "secretAccessKey": minioResource.AccessKeySecret, | ||
| "endPoint": minioResource.Endpoint, | ||
| }, | ||
| }, | ||
| Namespace: "test_namespace", | ||
| Identifier: "RS:test-source-id:test-destination-id-create-jobs", | ||
| } | ||
| r := Router{ | ||
| logger: logger.NOP, | ||
| conf: conf, | ||
| db: db, | ||
| warehouses: []model.Warehouse{warehouse}, | ||
| destType: warehouseutils.POSTGRES, | ||
| statsFactory: stats.NOP, | ||
| } | ||
| r.config.syncSchemaFrequency = 1 * time.Hour | ||
|
|
||
| setupCh := make(chan struct{}) | ||
| syncCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) | ||
| go func() { | ||
| defer close(setupCh) | ||
| err = r.sync(syncCtx) | ||
| require.NoError(t, err) | ||
| }() | ||
|
|
||
| t.Run("fetching schema from postgres", func(t *testing.T) { | ||
| schemaSql := fmt.Sprintf("CREATE SCHEMA %q;", testNamespace) | ||
| _, err = pgResource.DB.Exec(schemaSql) | ||
| require.NoError(t, err) | ||
|
|
||
| tableSql := fmt.Sprintf(`CREATE TABLE %q.%q ( | ||
| job_id BIGSERIAL PRIMARY KEY, | ||
| workspace_id TEXT NOT NULL DEFAULT '', | ||
| uuid UUID NOT NULL, | ||
| user_id TEXT NOT NULL, | ||
| parameters JSONB NOT NULL, | ||
| custom_val VARCHAR(64) NOT NULL, | ||
| event_payload JSONB NOT NULL, | ||
| event_count INTEGER NOT NULL DEFAULT 1, | ||
| created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), | ||
| expire_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW());`, testNamespace, testTable) | ||
| _, err = pgResource.DB.Exec(tableSql) | ||
| require.NoError(t, err) | ||
|
|
||
| <-setupCh | ||
| r.conf.Set("Warehouse.enableSyncSchema", true) | ||
| sh, err := schema.New( | ||
| context.Background(), | ||
| r.db, | ||
| warehouse, | ||
| r.conf, | ||
| r.logger.Child("syncer"), | ||
| r.statsFactory, | ||
| ) | ||
| require.NoError(t, err) | ||
| require.Eventually(t, func() bool { | ||
| schema, err := sh.GetLocalSchema(ctx) | ||
| require.NoError(t, err) | ||
| return reflect.DeepEqual(schema, model.Schema{ | ||
| "test_table": model.TableSchema{ | ||
| "created_at": "datetime", | ||
| "event_count": "int", | ||
| "event_payload": "json", | ||
| "expire_at": "datetime", | ||
| "job_id": "int", | ||
| "parameters": "json", | ||
| "user_id": "string", | ||
| "workspace_id": "string", | ||
| }, | ||
| }) | ||
| }, 3*time.Second, 100*time.Millisecond) | ||
| cancel() | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.