Documentation
¶
Overview ¶
Package schemasync is a runner that synchronize database schemas.
Index ¶
- Constants
- type LineageRevalidator
- type Notifier
- type SyncOperation
- type Syncer
- func (s *Syncer) EnqueueDatabases(ctx context.Context, operation *SyncOperation, ...)
- func (s *Syncer) FinishOperation(ctx context.Context, operation *SyncOperation)
- func (s *Syncer) GetInstanceMeta(ctx context.Context, instance *store.InstanceMessage) (*db.InstanceMetadata, error)
- func (s *Syncer) QueueLineageAnalysis(metaGUID string, metaType storepb.MetaType)
- func (s *Syncer) RecordInstanceResult(ctx context.Context, operation *SyncOperation, err error)
- func (s *Syncer) Run(ctx context.Context, wg *sync.WaitGroup)
- func (s *Syncer) StartOperation(trigger storepb.SyncTrigger, initiatorID int, instance *store.InstanceMessage) *SyncOperation
- func (s *Syncer) SyncAllDatabases(ctx context.Context, operation *SyncOperation, instance *store.InstanceMessage)
- func (s *Syncer) SyncDatabaseForUser(ctx context.Context, database *store.DatabaseMessage, initiatorID int) error
- func (s *Syncer) SyncDatabaseSchema(ctx context.Context, database *store.DatabaseMessage) (err error)
- func (s *Syncer) SyncInstance(ctx context.Context, instance *store.InstanceMessage) (*store.InstanceMessage, []*storepb.DatabaseSchemaMetadata, ...)
Constants ¶
const (
MaximumOutstanding = 100
)
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type LineageRevalidator ¶
type LineageRevalidator interface {
Trigger()
}
LineageRevalidator is signalled when a schema sync has added or changed metadata. Ingested lineage that named a relation the registry did not have is re-checked then, as soon as the relation appears, instead of at the next maintenance pass.
type Notifier ¶
type Notifier interface {
Send(ctx context.Context, message *storepb.Notification) error
SendToWorkspaceAdmins(ctx context.Context, message *storepb.Notification) error
}
Notifier delivers the messages a sync operation produces. *notification.Service is the production implementation; the interface keeps the runner testable without a database.
type SyncOperation ¶
type SyncOperation struct {
// contains filtered or unexported fields
}
SyncOperation is one user-visible sync operation: the instance metadata step plus every database that step queued.
The API registers it, the checker reports each database's outcome, and exactly one caller finishes it and writes the message. It is process state, like the queue it tracks: a restart loses the operations in flight, and so does it lose the syncs themselves.
type Syncer ¶
type Syncer struct {
// contains filtered or unexported fields
}
Syncer is the schema syncer.
func NewSyncer ¶
func NewSyncer(stores *store.Store, dbFactory *dbfactory.DBFactory, stateCfg *state.State, lineageAnalyzer *lineageanalyzer.Analyzer, lineageRevalidator LineageRevalidator, notifier Notifier) *Syncer
NewSyncer creates a schema syncer. notifier reports the outcome of a sync operation to the user who asked for it, and a background failure to the workspace administrators; a nil notifier leaves the runner silent, which is what its own tests want.
func (*Syncer) EnqueueDatabases ¶
func (s *Syncer) EnqueueDatabases(ctx context.Context, operation *SyncOperation, databases []*store.DatabaseMessage)
EnqueueDatabases queues the databases this operation covers and seals it: no database may be added afterwards, so an operation whose databases all report finishes here. An empty list is how a caller says "the instance step was all there was".
func (*Syncer) FinishOperation ¶
func (s *Syncer) FinishOperation(ctx context.Context, operation *SyncOperation)
FinishOperation seals an operation that has no databases to wait for, because the instance step was the whole of it.
func (*Syncer) GetInstanceMeta ¶
func (s *Syncer) GetInstanceMeta(ctx context.Context, instance *store.InstanceMessage) (*db.InstanceMetadata, error)
GetInstanceMeta gets the instance metadata.
func (*Syncer) QueueLineageAnalysis ¶
func (*Syncer) RecordInstanceResult ¶
func (s *Syncer) RecordInstanceResult(ctx context.Context, operation *SyncOperation, err error)
RecordInstanceResult records the instance metadata step's outcome.
func (*Syncer) StartOperation ¶
func (s *Syncer) StartOperation(trigger storepb.SyncTrigger, initiatorID int, instance *store.InstanceMessage) *SyncOperation
StartOperation registers one sync operation for the caller.
func (*Syncer) SyncAllDatabases ¶
func (s *Syncer) SyncAllDatabases(ctx context.Context, operation *SyncOperation, instance *store.InstanceMessage)
SyncAllDatabases queues every database of an instance for the operation that asked for it, and seals the operation: a list it could not read leaves the operation reporting the instance step alone instead of waiting for databases that were never queued.
func (*Syncer) SyncDatabaseForUser ¶
func (s *Syncer) SyncDatabaseForUser(ctx context.Context, database *store.DatabaseMessage, initiatorID int) error
SyncDatabaseForUser syncs one database and reports the outcome to the user who asked for it. It returns the sync error so the caller can still answer the request with it; the message is the durable record of the outcome, which is what the caller has nothing else to leave behind.
func (*Syncer) SyncDatabaseSchema ¶
func (s *Syncer) SyncDatabaseSchema(ctx context.Context, database *store.DatabaseMessage) (err error)
SyncDatabaseSchema will sync the schema for a database.
Two callers can ask for the same database at once (the periodic checker and the API both call this function), so the sync is serialized per database: a caller that arrives while one is running waits for it and reuses its result.
func (*Syncer) SyncInstance ¶
func (s *Syncer) SyncInstance(ctx context.Context, instance *store.InstanceMessage) (*store.InstanceMessage, []*storepb.DatabaseSchemaMetadata, []*store.DatabaseMessage, error)
SyncInstance syncs the schema for all databases in an instance.
Source Files