From bdd9318f9302fe3d65b6db0cf9165866e769f1df Mon Sep 17 00:00:00 2001 From: Denis Arh Date: Fri, 1 Jul 2022 18:09:38 +0200 Subject: [PATCH] Refactor and improve DAL implementation and init Changes: - Boot initialization follows standard impl - Improved DAL connection management (adding, reloading, removing) - Cleaner and more detailed logging - Primary store connection is now reused when added to DAL --- app/boot_dal.go | 26 ++ app/boot_levels.go | 84 +--- .../templates/gocode/store/interfaces.go.tpl | 3 + compose/rest/data_privacy.go | 14 - compose/service/module.go | 6 +- compose/service/service.go | 9 +- pkg/dal/driver.go | 2 +- pkg/dal/service.go | 429 ++++++++++-------- pkg/provision/dal.go | 52 +++ pkg/provision/provision.go | 47 +- store/adapters/rdbms/dal/connection.go | 7 +- store/adapters/rdbms/dal/iterator.go | 2 - store/adapters/rdbms/drivers/mysql/connect.go | 4 + .../rdbms/drivers/postgres/connect.go | 4 + .../adapters/rdbms/drivers/sqlite/connect.go | 4 + store/adapters/rdbms/rdbms_store.go | 10 +- store/dal.go | 11 + store/interfaces.gen.go | 4 + system/dalutils/connection.go | 110 ----- system/service/dal_connection.go | 144 ++++-- system/service/service.go | 17 +- system/types/dal_connection.go | 3 +- tests/dal/dal_utils.go | 10 +- 23 files changed, 496 insertions(+), 506 deletions(-) create mode 100644 app/boot_dal.go create mode 100644 pkg/provision/dal.go create mode 100644 store/dal.go delete mode 100644 system/dalutils/connection.go diff --git a/app/boot_dal.go b/app/boot_dal.go new file mode 100644 index 000000000..2da9f9c4e --- /dev/null +++ b/app/boot_dal.go @@ -0,0 +1,26 @@ +package app + +import ( + "context" + "fmt" + "github.com/cortezaproject/corteza-server/pkg/dal" + "go.uber.org/zap" +) + +func (app *CortezaApp) initDAL(ctx context.Context, log *zap.Logger) (err error) { + // Verify that primary store is connected + // or return error + if app.Store == nil { + return fmt.Errorf("primary store not connected") + } + + // Init DAL and prepare default connection + svc, err := dal.New(log.Named("dal"), app.Opt.Environment.IsDevelopment()) + if err != nil { + return err + } + + dal.SetGlobal(svc) + + return +} diff --git a/app/boot_levels.go b/app/boot_levels.go index d8978cc87..40a703c0a 100644 --- a/app/boot_levels.go +++ b/app/boot_levels.go @@ -291,59 +291,6 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) { return } - var primaryDalConnection *types.DalConnection - primaryDalConnection, err = store.LookupDalConnectionByHandle(ctx, app.Store, "primary_connection") - if err != nil { - // @note we won't log the error here as this is a bigger issue then - // it not connecting - return - } - - dalLogger := app.Log.Named("dal") - - // Assure primary connection configs - { - if primaryDalConnection.Config.DefaultModelIdent != provision.DefaultComposeRecordTable { - dalLogger.Warn("overwriting DefaultModelIdent value", zap.String("old", primaryDalConnection.Config.DefaultModelIdent), zap.String("new", provision.DefaultComposeRecordTable)) - primaryDalConnection.Config.DefaultModelIdent = provision.DefaultComposeRecordTable - } - - if primaryDalConnection.Config.DefaultAttributeIdent != provision.DefaultComposeRecordValueCol { - dalLogger.Warn("overwriting DefaultAttributeIdent value", zap.String("old", primaryDalConnection.Config.DefaultAttributeIdent), zap.String("new", provision.DefaultComposeRecordValueCol)) - primaryDalConnection.Config.DefaultAttributeIdent = provision.DefaultComposeRecordValueCol - } - - if primaryDalConnection.Config.DefaultPartitionFormat != provision.DefaultPartitionFormat { - dalLogger.Warn("overwriting DefaultPartitionFormat value", zap.String("old", primaryDalConnection.Config.DefaultPartitionFormat), zap.String("new", provision.DefaultPartitionFormat)) - primaryDalConnection.Config.DefaultPartitionFormat = provision.DefaultPartitionFormat - } - - c := dal.NewDSNConnection(app.Opt.DB.DSN) - dalLogger.Warn("overwriting Connection value", zap.Any("old", primaryDalConnection.Config.Connection), zap.Any("new", c)) - primaryDalConnection.Config.Connection = c - } - - // Init DAL and prepare default connection - if _, err = dal.InitGlobalService( - ctx, - dalLogger, - app.Opt.Environment.IsDevelopment(), - primaryDalConnection.ID, - - // DB_DSN is the default connection with full capabilities - primaryDalConnection.Config.Connection, - dal.ConnectionMeta{ - DefaultModelIdent: primaryDalConnection.Config.DefaultModelIdent, - DefaultAttributeIdent: primaryDalConnection.Config.DefaultAttributeIdent, - DefaultPartitionFormat: primaryDalConnection.Config.DefaultPartitionFormat, - // @todo make it configurable from env - SensitivityLevel: primaryDalConnection.SensitivityLevel, - Label: primaryDalConnection.Handle, - }, - primaryDalConnection.ActiveCapabilities()...); err != nil { - return err - } - if app.Opt.Auth.DefaultClient != "" { // default client will help streamline authorization with default clients app.DefaultAuthClient, err = store.LookupAuthClientByHandle(ctx, app.Store, app.Opt.Auth.DefaultClient) @@ -460,11 +407,18 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) { return fmt.Errorf("could not reload resource translations: %w", err) } + { + // Initialize Data Access Layer (DAL) + if err = app.initDAL(ctx, app.Log); err != nil { + return fmt.Errorf("can not initialize DAL: %w", err) + } + } + // Initializes system services // // Note: this is a legacy approach, all services from all 3 apps // will most likely be merged in the future - err = sysService.Initialize(ctx, app.Log, app.Store, primaryDalConnection, app.WsServer, sysService.Config{ + err = sysService.Initialize(ctx, app.Log, app.Store, app.WsServer, sysService.Config{ ActionLog: app.Opt.ActionLog, Discovery: app.Opt.Discovery, Storage: app.Opt.ObjStore, @@ -479,20 +433,6 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) { return } - // Initializing DAL components - // - Sensitivity levels - err = sysService.DefaultDalSensitivityLevel.ReloadSensitivityLevels(ctx, sysService.DefaultStore) - if err != nil { - // @note we're not erroring out here so Corteza can still be used - dalLogger.Error("failed to initialize DAL sensitivity levels", zap.Error(err)) - } - // - DAL connections - err = sysService.DefaultDalConnection.ReloadConnections(ctx) - if err != nil { - // @note we're not erroring out here so Corteza can still be used - dalLogger.Error("failed to initialize DAL connections", zap.Error(err)) - } - if app.Opt.Messagebus.Enabled { // initialize all the queue handlers messagebus.Service().Init(ctx, service.DefaultQueue) @@ -527,14 +467,6 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) { return fmt.Errorf("could not initialize compose services: %w", err) } - // Initializing DAL components - // - Models - err = cmpService.DefaultModule.ReloadDALModels(ctx) - if err != nil { - // @note we're not erroring out here so Corteza can still be used - dalLogger.Error("failed to initialize DAL models", zap.Error(err)) - } - corredor.Service().SetUserFinder(sysService.DefaultUser) corredor.Service().SetRoleFinder(sysService.DefaultRole) diff --git a/codegen/assets/templates/gocode/store/interfaces.go.tpl b/codegen/assets/templates/gocode/store/interfaces.go.tpl index 578ef5830..23b553617 100644 --- a/codegen/assets/templates/gocode/store/interfaces.go.tpl +++ b/codegen/assets/templates/gocode/store/interfaces.go.tpl @@ -32,6 +32,9 @@ type ( // and it's highly unlikely we'll support different/multiple logging "backend" SetLogger(*zap.Logger) + // Returns underlying store as DAL connection + ToDalConn() dal.Connection + // Tx is a transaction handler Tx(context.Context, func(context.Context, Storer) error) error diff --git a/compose/rest/data_privacy.go b/compose/rest/data_privacy.go index 86ed643b6..4176f705b 100644 --- a/compose/rest/data_privacy.go +++ b/compose/rest/data_privacy.go @@ -6,7 +6,6 @@ import ( "github.com/cortezaproject/corteza-server/compose/rest/request" "github.com/cortezaproject/corteza-server/compose/service" "github.com/cortezaproject/corteza-server/compose/types" - "github.com/cortezaproject/corteza-server/pkg/dal" "github.com/cortezaproject/corteza-server/pkg/payload" ) @@ -51,14 +50,9 @@ func (DataPrivacy) New() *DataPrivacy { func (ctrl *DataPrivacy) SensitiveDataList(ctx context.Context, r *request.DataPrivacySensitiveDataList) (out interface{}, err error) { outSet := sensitiveDataSetPayload{} - primaryConnID := dal.Service().PrimaryConnectionID(ctx) - reqConns := make(map[uint64]bool) hasReqConns := len(r.ConnectionID) > 0 for _, connectionID := range payload.ParseUint64s(r.ConnectionID) { - if connectionID == 0 { - connectionID = primaryConnID - } reqConns[connectionID] = true } @@ -78,18 +72,10 @@ func (ctrl *DataPrivacy) SensitiveDataList(ctx context.Context, r *request.DataP } for _, m := range modules { conn := m.ModelConfig.ConnectionID - if conn == 0 { - conn = primaryConnID - } if hasReqConns && !reqConns[conn] { continue } - // @todo ignoring this for now; revert after dev release - // if m.Privacy.SensitivityLevel == 0 { - // continue - // } - sData, err := ctrl.record.FindSensitive(ctx, types.RecordFilter{ModuleID: m.ID, NamespaceID: m.NamespaceID}) if err != nil { return nil, err diff --git a/compose/service/module.go b/compose/service/module.go index 4d76bbacc..7d6938a77 100644 --- a/compose/service/module.go +++ b/compose/service/module.go @@ -82,8 +82,8 @@ var ( }) ) -func Module(ctx context.Context, dal dalService) (*module, error) { - svc := &module{ +func Module(dal dalService) *module { + return &module{ ac: DefaultAccessControl, eventbus: eventbus.Service(), actionlog: DefaultActionlog, @@ -91,8 +91,6 @@ func Module(ctx context.Context, dal dalService) (*module, error) { locale: DefaultResourceTranslation, dal: dal, } - - return svc, svc.ReloadDALModels(ctx) } func (svc module) Find(ctx context.Context, filter types.ModuleFilter) (set types.ModuleSet, f types.ModuleFilter, err error) { diff --git a/compose/service/service.go b/compose/service/service.go index 84b835c79..ac5d3dd70 100644 --- a/compose/service/service.go +++ b/compose/service/service.go @@ -169,9 +169,7 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config) } DefaultNamespace = Namespace() - if DefaultModule, err = Module(ctx, dal.Service()); err != nil { - return - } + DefaultModule = Module(dal.Service()) DefaultImportSession = ImportSession() DefaultRecord = Record(dal.Service()) @@ -217,6 +215,11 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config) } func Activate(ctx context.Context) (err error) { + err = DefaultModule.ReloadDALModels(ctx) + if err != nil { + return err + } + return } diff --git a/pkg/dal/driver.go b/pkg/dal/driver.go index ab2b55c4d..0b7fe8184 100644 --- a/pkg/dal/driver.go +++ b/pkg/dal/driver.go @@ -175,7 +175,7 @@ func RegisterDriver(d Driver) { // connect opens a new StoreConnection for the given CRS func connect(ctx context.Context, log *zap.Logger, isDevelopment bool, cp ConnectionParams, capabilities ...capabilities.Capability) (Connection, error) { if cp.Type != "corteza::dal:connection:dsn" { - return nil, fmt.Errorf("cannot open connection: only DSN connections supported") + return nil, fmt.Errorf("cannot open connection: only DSN connections supported (got: %q)", cp.Type) } dsn := cp.Params["dsn"].(string) diff --git a/pkg/dal/service.go b/pkg/dal/service.go index 207c77d1e..d186694d8 100644 --- a/pkg/dal/service.go +++ b/pkg/dal/service.go @@ -3,7 +3,6 @@ package dal import ( "context" "fmt" - "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" "github.com/cortezaproject/corteza-server/pkg/expr" "github.com/cortezaproject/corteza-server/pkg/filter" @@ -11,30 +10,49 @@ import ( ) type ( - connectionWrap struct { - connectionID uint64 - label string + ConnectionWrap struct { + connectionID uint64 + + // @todo remove it and use value from connection meta + label string + + // @todo remove it and use value from connection meta sensitivityLevel uint64 connection Connection - meta ConnectionMeta + + params ConnectionParams + meta ConnectionMeta + capabilities capabilities.Set } ConnectionMeta struct { SensitivityLevel uint64 Label string - DefaultModelIdent string + // When model does not specifiy the ident (table name for example), fallback to this + // @todo we can lose "Default" prefix + // @todo do we need a separate setting or can we get away with using just PartitionFormat + DefaultModelIdent string + + // If model attribute(s) do not specify + // @todo needs to be more explicit that this is for JSON encode attributes + // @todo we can lose "Default" prefix DefaultAttributeIdent string + // If data is partitioned we fallback to this, + // @todo we can lose "Default" prefix DefaultPartitionFormat string PartitionValidator string } service struct { - connections map[uint64]*connectionWrap - primaryConnectionID uint64 + connections map[uint64]*ConnectionWrap + + // Default connection ID + // Can not be changed in the runtime, only set to value different than zero! + defConnID uint64 // Indexed by corresponding storeID models map[uint64]ModelSet @@ -49,36 +67,21 @@ type ( } ) -const ( - DefaultConnectionID uint64 = 0 -) - var ( gSvc *service ) -// InitGlobalService initializes a fresh DAL where the given primary connection -func InitGlobalService(ctx context.Context, log *zap.Logger, inDev bool, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (_ *service, err error) { - log.Debug("initializing DAL service with primary connection", zap.Any("connection params", cp)) - - // To help prevent awkward issues due to globally shared resources - if gSvc != nil { - panic("cannot initialize global DAL service: already initialized") - } - - if gSvc == nil { - gSvc, err = New(ctx, log, inDev, connectionID, cp, cm, capabilities...) - return gSvc, err - } - - return gSvc, nil +func SetGlobal(svc *service) { + gSvc = svc } -func New(ctx context.Context, log *zap.Logger, inDev bool, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (*service, error) { +// New creates a DAL service with the primary connection +// +// It needs an established and working connection to the primary store +func New(log *zap.Logger, inDev bool) (*service, error) { svc := &service{ - connections: make(map[uint64]*connectionWrap), - models: make(map[uint64]ModelSet), - primaryConnectionID: connectionID, + connections: make(map[uint64]*ConnectionWrap), + models: make(map[uint64]ModelSet), logger: log, inDev: inDev, @@ -87,26 +90,12 @@ func New(ctx context.Context, log *zap.Logger, inDev bool, connectionID uint64, modelIssues: make(dalIssueIndex), } - var err error - cw := &connectionWrap{ - meta: cm, - sensitivityLevel: cm.SensitivityLevel, - label: cm.Label, - connectionID: connectionID, - } - cw.connection, err = connect(ctx, log, inDev, cp, capabilities...) - if err != nil { - return nil, err - } - - svc.connections[connectionID] = cw - return svc, nil } // Service returns the global initialized DAL service // -// If InitGlobalService has not yet been called the function will panic +// Function will panic if DAL service is not set (via SetGlobal) func Service() *service { if gSvc == nil { panic("DAL global service not initialized: call dal.InitGlobalService() first") @@ -194,52 +183,132 @@ func (svc *service) DeleteSensitivityLevel(levels ...SensitivityLevel) (err erro // // // // // // // // // // // // // // // // // // // // // // // // // // Connection management -// CreateConnection adds a new connection to the DAL -func (svc *service) CreateConnection(ctx context.Context, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (err error) { - svc.logger.Debug("creating connection", zap.Uint64("connectionID", connectionID), zap.Any("connection params", cp)) +// MakeConnection makes and returns a new connection (wrap) +func MakeConnection(ID uint64, conn Connection, p ConnectionParams, m ConnectionMeta, cap ...capabilities.Capability) *ConnectionWrap { + return &ConnectionWrap{ + connectionID: ID, + connection: conn, + params: p, + + meta: m, + sensitivityLevel: m.SensitivityLevel, + label: m.Label, + + capabilities: cap, + } +} + +// ReplaceConnection adds new or updates an existing connection +// +// We rely on the user to provide stable connection IDs and +// uses valid relations to these connections in the models. +// +// Is isDefault when adding a default connection. Service will then +// compensate and use proper IDs when models refer to connection with ID=0 +func (svc *service) ReplaceConnection(ctx context.Context, cw *ConnectionWrap, isDefault bool) (err error) { + // @todo lock/unlock var ( - issues = newIssueHelper().addConnection(connectionID) + ID = cw.connectionID + issues = newIssueHelper().addConnection(ID) + oldConn *ConnectionWrap + + log = svc.logger.Named("connection").With( + zap.Uint64("ID", ID), + zap.Any("params", cw.params), + zap.Any("meta", cw.meta), + ) ) + + if isDefault { + if svc.defConnID == 0 { + // default connection not set yet + log.Debug("setting as default connection") + svc.defConnID = ID + } else if svc.defConnID != ID { + // default connection set but ID is different. + // this does not make any sense + return fmt.Errorf("different ID for default connection detecte (old: %d, new: %d)", svc.defConnID, ID) + } + } + defer svc.updateIssues(issues) - // sensitivity levels - if !svc.sensitivityLevels.includes(cm.SensitivityLevel) { - issues.addConnectionIssue(connectionID, errConnectionCreateMissingSensitivityLevel(connectionID, cm.SensitivityLevel)) + // Sensitivity level validations + if !svc.sensitivityLevels.includes(cw.meta.SensitivityLevel) { + issues.addConnectionIssue(ID, errConnectionCreateMissingSensitivityLevel(ID, cw.meta.SensitivityLevel)) } - // Prepare connection bits - cw := &connectionWrap{ - connectionID: connectionID, - meta: cm, - sensitivityLevel: cm.SensitivityLevel, - label: cm.Label, + if oldConn = svc.getConnectionByID(ID); oldConn != nil { + // Connection exists, validate models and sensitivity levels and close and remove connection at the end + log.Debug("found existing") + + // Check already registered models and their capabilities + // + // Defer the return till the end so we can get a nicer report of what all is wrong + errored := false + for _, model := range svc.models[ID] { + log.Debug("validating model before connection is updated", zap.String("ident", model.Ident)) + + // - capabilities + if !model.Capabilities.IsSubset(cw.capabilities...) { + issues.addConnectionIssue(ID, fmt.Errorf("cannot update connection %d: new connection does not support existing models", ID)) + errored = true + } + + // - sensitivity levels + if !svc.sensitivityLevels.isSubset(model.SensitivityLevel, cw.meta.SensitivityLevel) { + issues.addConnectionIssue(ID, fmt.Errorf("cannot update connection %d: new connection sensitivity level does not support model %d", ID, model.ResourceID)) + errored = true + } + } + + // Don't update if meta bits are not ok + if errored { + log.Warn("update failed") + return + } + + // close old connection + if cc, ok := oldConn.connection.(ConnectionCloser); ok { + if err = cc.Close(ctx); err != nil { + issues.addConnectionIssue(ID, err) + return nil + } + + log.Debug("disconnected") + } + + svc.removeConnection(ID) } - if cw.connection, err = connect(ctx, svc.logger, svc.inDev, cp, capabilities...); err != nil { - issues.addConnectionIssue(connectionID, errConnectionCreateConnectionFailed(connectionID, err)) + + if cw.connection == nil { + cw.connection, err = connect(ctx, svc.logger, svc.inDev, cw.params, cw.capabilities...) + if err != nil { + log.Warn("could not connect", zap.Error(err)) + issues.addConnectionIssue(ID, err) + } else { + log.Debug("connected") + } + } else { + log.Debug("using preexisting connection") + } svc.addConnection(cw) - - svc.logger.Debug("created connection") + log.Debug("added") return nil } -func (svc *service) PrimaryConnectionID(ctx context.Context) uint64 { - return svc.primaryConnectionID -} - // DeleteConnection removes the given connection from the DAL -func (svc *service) DeleteConnection(ctx context.Context, connectionID uint64) (err error) { - svc.logger.Debug("deleting connection", zap.Uint64("connectionID", connectionID)) - +func (svc *service) RemoveConnection(ctx context.Context, ID uint64) (err error) { var ( - issues = newIssueHelper().addConnection(connectionID) + issues = newIssueHelper().addConnection(ID) ) - c := svc.getConnectionByID(connectionID) + c := svc.getConnectionByID(ID) if c == nil { - return errConnectionDeleteNotFound(connectionID) + return errConnectionDeleteNotFound(ID) } // Potential cleanups @@ -253,90 +322,15 @@ func (svc *service) DeleteConnection(ctx context.Context, connectionID uint64) ( // // @todo this is temporary until a proper update function is prepared. // The primary connection must not be removable! - svc.removeConnection(connectionID) + svc.removeConnection(ID) // Only if successful should we cleanup the issue registry svc.updateIssues(issues) - svc.logger.Debug("deleted connection") - - return nil -} - -// UpdateConnection updates the given connection -func (svc *service) UpdateConnection(ctx context.Context, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (err error) { - svc.logger.Debug("updating connection", zap.Uint64("connectionID", connectionID)) - - var ( - issues = newIssueHelper().addConnection(connectionID) - oldConn *connectionWrap + svc.logger.Named("connection").Debug("deleted", + zap.Uint64("ID", ID), + zap.Any("meta", c.meta), ) - defer svc.updateIssues(issues) - - // Validation - { - // Check if connection exists - oldConn = svc.getConnectionByID(connectionID) - if oldConn == nil { - issues.addConnectionIssue(connectionID, errConnectionUpdateNotFound(connectionID)) - } - - // sensitivity levels - if !svc.sensitivityLevels.includes(cm.SensitivityLevel) { - issues.addConnectionIssue(connectionID, errConnectionUpdateMissingSensitivityLevel(connectionID, cm.SensitivityLevel)) - } - - // Check already registered models and their capabilities - // - // Defer the return till the end so we can get a nicer report of what all is wrong - errored := false - for _, model := range svc.models[connectionID] { - // - capabilities - if !model.Capabilities.IsSubset(capabilities...) { - issues.addConnectionIssue(connectionID, fmt.Errorf("cannot update connection %d: new connection does not support existing models", connectionID)) - errored = errored || true - } - // - sensitivity levels - if !svc.sensitivityLevels.isSubset(model.SensitivityLevel, cm.SensitivityLevel) { - issues.addConnectionIssue(connectionID, fmt.Errorf("cannot update connection %d: new connection sensitivity level does not support model %d", connectionID, model.ResourceID)) - errored = errored || true - } - } - - // Don't update if meta bits are not ok - if errored { - return - } - } - - // close old connection - { - if cc, ok := oldConn.connection.(ConnectionCloser); ok { - if err = cc.Close(ctx); err != nil { - issues.addConnectionIssue(connectionID, err) - return nil - } - } - svc.removeConnection(connectionID) - } - - // open new connection - { - newConnection, err := connect(ctx, svc.logger, svc.inDev, cp, capabilities...) - if err != nil { - issues.addConnectionIssue(connectionID, err) - } - - svc.addConnection(&connectionWrap{ - meta: cm, - sensitivityLevel: cm.SensitivityLevel, - label: cm.Label, - connectionID: connectionID, - connection: newConnection, - }) - } - - svc.logger.Debug("updated connection") return nil } @@ -436,7 +430,7 @@ func (svc *service) Truncate(ctx context.Context, mf ModelFilter, capabilities c return cw.connection.Truncate(ctx, model) } -func (svc *service) storeOpPrep(ctx context.Context, mf ModelFilter, capabilities capabilities.Set) (model *Model, cw *connectionWrap, err error) { +func (svc *service) storeOpPrep(ctx context.Context, mf ModelFilter, capabilities capabilities.Set) (model *Model, cw *ConnectionWrap, err error) { model = svc.getModelByFilter(mf) if model == nil { err = errModelNotFound(mf.ResourceID) @@ -485,19 +479,25 @@ func (svc *service) SearchModels(ctx context.Context) (out ModelSet, err error) // AddModel adds support for a new model func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err error) { - svc.logger.Debug("creating models", zap.Int("count", len(models))) - var ( + log = svc.logger.Named("models") issues = newIssueHelper() auxIssues = newIssueHelper() ) + + svc.logger.Debug("creating", zap.Int("count", len(models))) + defer svc.updateIssues(issues) // Validate models for _, model := range models { - svc.logger.Debug("validating model", zap.Uint64("ID", model.ResourceID)) issues.addModel(model.ResourceID) + if model.ConnectionID == 0 { + // Replace model's connection ID with default one when zero + model.ConnectionID = svc.defConnID + } + // Assure the connection has no issues if svc.hasConnectionIssues(model.ConnectionID) { issues.addModelIssue(model.ConnectionID, model.ResourceID, errModelCreateProblematicConnection(model.ConnectionID, model.ResourceID)) @@ -536,19 +536,28 @@ func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err erro } } - svc.logger.Debug("validated model") + log.Debug("validated", + zap.Uint64("ID", model.ResourceID), + zap.String("ident", model.Ident), + zap.String("label", model.Label), + ) } // Add models to corresponding connections for connection, models := range svc.modelByConnection(models) { for _, model := range models { - svc.logger.Debug("adding model", zap.Uint64("model", model.ResourceID)) + mLog := log.With( + zap.Uint64("connectionID", model.ConnectionID), + zap.Uint64("ID", model.ResourceID), + zap.String("ident", model.Ident), + zap.String("label", model.Label), + ) connectionIssues := svc.hasConnectionIssues(model.ConnectionID) modelIssues := svc.hasModelIssues(model.ConnectionID, model.ResourceID) if !modelIssues && !connectionIssues { - svc.logger.Debug("adding model to connection", zap.Uint64("connection", model.ConnectionID), zap.Uint64("model", model.ResourceID)) + mLog.Debug("adding model to connection") // Add model to connection auxIssues, err = svc.registerModelToConnection(ctx, connection, model) @@ -558,10 +567,10 @@ func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err erro } } else { if connectionIssues { - svc.logger.Warn("not adding to connection due to connection issues", zap.Uint64("connection", model.ConnectionID)) + mLog.Warn("not adding to connection due to connection issues") } if modelIssues { - svc.logger.Warn("not adding to connection due to model issues", zap.Uint64("model", model.ResourceID)) + mLog.Warn("not adding to connection due to model issues") } } @@ -570,23 +579,31 @@ func (svc *service) CreateModel(ctx context.Context, models ...*Model) (err erro } } - svc.logger.Debug("created models") + log.Debug("done") return } // DeleteModel removes support for the model and deletes it from the connection func (svc *service) DeleteModel(ctx context.Context, models ...*Model) (err error) { - svc.logger.Debug("deleting models", zap.Int("count", len(models))) var ( + log = svc.logger.Named("models") issues = newIssueHelper() ) + + log.Debug("deleting", zap.Int("count", len(models))) + defer svc.updateIssues(issues) // validation skip := make(map[uint64]bool) for _, model := range models { + if model.ConnectionID == 0 { + // Replace model's connection ID with default one when zero + model.ConnectionID = svc.defConnID + } + issues.addModel(model.ResourceID) // Validate existence @@ -608,10 +625,15 @@ func (svc *service) DeleteModel(ctx context.Context, models ...*Model) (err erro // Work for _, model := range models { - svc.logger.Debug("deleting model", zap.Uint64("model", model.ResourceID)) + mLog := log.With( + zap.Uint64("connectionID", model.ConnectionID), + zap.Uint64("ID", model.ResourceID), + zap.String("ident", model.Ident), + zap.String("label", model.Label), + ) if skip[model.ResourceID] { - svc.logger.Debug("model does not exist; skipping") + mLog.Debug("model does not exist; skipping") continue } @@ -629,20 +651,39 @@ func (svc *service) DeleteModel(ctx context.Context, models ...*Model) (err erro // how should this be handled; a straight up delete doesn't sound sane to me // anymore - svc.logger.Debug("deleted model") + mLog.Debug("deleted") } return nil } func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (err error) { - svc.logger.Debug("updating model", zap.Uint64("model", old.ResourceID)) + if old.ConnectionID == 0 { + // Replace old model's connection ID with default one when zero + old.ConnectionID = svc.defConnID + } + + if new.ConnectionID == 0 { + // Replace new model's connection ID with default one when zero + new.ConnectionID = svc.defConnID + } var ( - conn *connectionWrap + log = svc.logger.Named("models").With( + zap.Uint64("connection", new.ConnectionID), + zap.Uint64("model", new.ResourceID), + zap.Uint64("ID", new.ResourceID), + zap.String("ident", new.Ident), + zap.String("label", new.Label), + ) + + conn *ConnectionWrap issues = newIssueHelper().addModel(old.ResourceID) ) + + log.Debug("updating", zap.Uint64("model", old.ResourceID)) + defer svc.updateIssues(issues) // Validation @@ -693,7 +734,7 @@ func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (er modelIssues := svc.hasModelIssues(new.ConnectionID, new.ResourceID) if !modelIssues && !connectionIssues { - svc.logger.Debug("updating connection's model", zap.Uint64("connection", new.ConnectionID), zap.Uint64("model", new.ResourceID)) + log.Debug("updating connection's model") err = conn.connection.UpdateModel(ctx, old, new) if err != nil { @@ -701,10 +742,10 @@ func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (er } } else { if connectionIssues { - svc.logger.Warn("not updating connection's model due to connection issues", zap.Uint64("connection", new.ConnectionID)) + log.Warn("not updating connection's model due to connection issues") } if modelIssues { - svc.logger.Warn("not updating connection's model due to model issues", zap.Uint64("model", new.ResourceID)) + log.Warn("not updating connection's model due to model issues") } } @@ -721,7 +762,7 @@ func (svc *service) UpdateModel(ctx context.Context, old *Model, new *Model) (er svc.models[old.ConnectionID] = append(svc.models[old.ConnectionID], new) } - svc.logger.Debug("updated model") + log.Debug("updated") return } @@ -730,7 +771,7 @@ func (svc *service) UpdateModelAttribute(ctx context.Context, model *Model, old, svc.logger.Debug("updating model attribute", zap.Uint64("model", model.ResourceID)) var ( - conn *connectionWrap + conn *ConnectionWrap issues = newIssueHelper().addModel(model.ResourceID) ) defer svc.updateIssues(issues) @@ -814,6 +855,10 @@ func (svc *service) UpdateModelAttribute(ctx context.Context, model *Model, old, } func (svc *service) ModelIdentFormatter(connectionID uint64) (f *IdentFormatter, err error) { + if connectionID == 0 { + connectionID = svc.defConnID + } + c := svc.getConnectionByID(connectionID) if c == nil { @@ -852,15 +897,19 @@ func (svc *service) FindModelByIdent(connectionID uint64, ident string) *Model { // // // // // // // // // // // // // // // // // // // // // // // // // // Utilities -func (svc *service) getConnectionByID(connectionID uint64) (cw *connectionWrap) { - if connectionID == DefaultConnectionID || connectionID == svc.primaryConnectionID { - return svc.connections[svc.primaryConnectionID] - } +func (svc *service) removeConnection(connectionID uint64) { + delete(svc.connections, connectionID) +} +func (svc *service) addConnection(cw *ConnectionWrap) { + svc.connections[cw.connectionID] = cw +} + +func (svc *service) getConnectionByID(connectionID uint64) (cw *ConnectionWrap) { return svc.connections[connectionID] } -func (svc *service) getConnection(connectionID uint64, cc ...capabilities.Capability) (cw *connectionWrap, can capabilities.Set, err error) { +func (svc *service) getConnection(connectionID uint64, cc ...capabilities.Capability) (cw *ConnectionWrap, can capabilities.Set, err error) { err = func() error { // get the requested connection cw = svc.getConnectionByID(connectionID) @@ -885,8 +934,8 @@ func (svc *service) getConnection(connectionID uint64, cc ...capabilities.Capabi } // modelByConnection maps the given models by their CRS -func (svc *service) modelByConnection(models ModelSet) (out map[*connectionWrap]ModelSet) { - out = make(map[*connectionWrap]ModelSet) +func (svc *service) modelByConnection(models ModelSet) (out map[*ConnectionWrap]ModelSet) { + out = make(map[*ConnectionWrap]ModelSet) for _, model := range models { c := svc.getConnectionByID(model.ConnectionID) @@ -896,7 +945,7 @@ func (svc *service) modelByConnection(models ModelSet) (out map[*connectionWrap] return } -func (svc *service) registerModelToConnection(ctx context.Context, cw *connectionWrap, model *Model) (issues *issueHelper, err error) { +func (svc *service) registerModelToConnection(ctx context.Context, cw *ConnectionWrap, model *Model) (issues *issueHelper, err error) { issues = newIssueHelper() available, err := cw.connection.Models(ctx) @@ -928,6 +977,10 @@ func (svc *service) registerModelToConnection(ctx context.Context, cw *connectio } func (svc *service) getModelByFilter(mf ModelFilter) *Model { + if mf.ConnectionID == 0 { + mf.ConnectionID = svc.defConnID + } + if mf.ResourceID > 0 { return svc.FindModelByResourceID(mf.ConnectionID, mf.ResourceID) } @@ -995,7 +1048,7 @@ func (svc *service) newSensitivityLevelIndex(levels SensitivityLevelSet) (out se func (svc *service) validateNewSensitivityLevels(levels sensitivityLevelIndex) (err error) { err = func() (err error) { - cIndex := make(map[uint64]*connectionWrap) + cIndex := make(map[uint64]*ConnectionWrap) // - connections for _, _c := range svc.connections { @@ -1036,18 +1089,6 @@ func (svc *service) validateNewSensitivityLevels(levels sensitivityLevelIndex) ( return } -func (svc *service) removeConnection(connectionID uint64) { - if connectionID == DefaultConnectionID || connectionID == svc.primaryConnectionID { - connectionID = svc.primaryConnectionID - } - - delete(svc.connections, connectionID) -} - -func (svc *service) addConnection(cw *connectionWrap) { - svc.connections[cw.connectionID] = cw -} - func wrapError(pfx string, err error) error { return fmt.Errorf("%s: %v", pfx, err) } diff --git a/pkg/provision/dal.go b/pkg/provision/dal.go new file mode 100644 index 000000000..888289e2b --- /dev/null +++ b/pkg/provision/dal.go @@ -0,0 +1,52 @@ +package provision + +import ( + "context" + "github.com/cortezaproject/corteza-server/pkg/auth" + "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" + "github.com/cortezaproject/corteza-server/pkg/id" + "github.com/cortezaproject/corteza-server/store" + "github.com/cortezaproject/corteza-server/system/types" +) + +const ( + DefaultComposeRecordTable = "compose_record" + DefaultComposeRecordValueCol = "values" + DefaultPartitionFormat = "compose_record" +) + +// Injects primary connection +func defaultDalConnection(ctx context.Context, s store.DalConnections) (err error) { + cc, err := store.LookupDalConnectionByHandle(ctx, s, types.DalPrimaryConnectionHandle) + if err != nil && err != store.ErrNotFound { + return + } + + // Already exists + if cc != nil { + return + } + + // Create it + var conn = &types.DalConnection{ + // Using id.Next since we dropped "special" ids a while ago. + // If needed, use the handle + ID: id.Next(), + Name: "Primary Database", + Handle: types.DalPrimaryConnectionHandle, + Type: types.DalPrimaryConnectionResourceType, + + Config: types.ConnectionConfig{ + DefaultModelIdent: DefaultComposeRecordTable, + DefaultAttributeIdent: DefaultComposeRecordValueCol, + DefaultPartitionFormat: DefaultPartitionFormat, + }, + Capabilities: types.ConnectionCapabilities{ + Supported: capabilities.FullCapabilities(), + }, + CreatedAt: *now(), + CreatedBy: auth.ServiceUser().ID, + } + + return store.CreateDalConnection(ctx, s, conn) +} diff --git a/pkg/provision/provision.go b/pkg/provision/provision.go index 6868c5b90..07c4d76fd 100644 --- a/pkg/provision/provision.go +++ b/pkg/provision/provision.go @@ -4,8 +4,6 @@ import ( "context" "time" - "github.com/cortezaproject/corteza-server/pkg/auth" - "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" "github.com/cortezaproject/corteza-server/pkg/errors" "github.com/cortezaproject/corteza-server/pkg/id" "github.com/cortezaproject/corteza-server/pkg/options" @@ -23,12 +21,6 @@ var ( } ) -const ( - DefaultComposeRecordTable = "compose_record" - DefaultComposeRecordValueCol = "values" - DefaultPartitionFormat = "compose_record_{{namespace}}_{{module}}" -) - func Run(ctx context.Context, log *zap.Logger, s store.Storer, provisionOpt options.ProvisionOpt, authOpt options.AuthOpt) error { log = log.Named("provision") @@ -54,7 +46,7 @@ func Run(ctx context.Context, log *zap.Logger, s store.Storer, provisionOpt opti func() error { return oidcAutoDiscovery(ctx, log.Named("auth.oidc-auto-discovery"), s, authOpt) }, func() error { return defaultAuthClient(ctx, log.Named("auth.clients"), s, authOpt) }, - func() error { return defaultDalConnection(ctx, log.Named("dal.connections"), s) }, + func() error { return defaultDalConnection(ctx, s) }, } for _, fn := range ffn { @@ -118,40 +110,3 @@ func defaultAuthClient(ctx context.Context, log *zap.Logger, s store.AuthClients return nil } - -func defaultDalConnection(ctx context.Context, log *zap.Logger, s store.DalConnections) (err error) { - cc, err := store.LookupDalConnectionByHandle(ctx, s, "primary_connection") - if err != nil && err != store.ErrNotFound { - return - } - - // Already exists - if cc != nil { - return - } - - // Create it - var conn = &types.DalConnection{ - // Using id.Next since we dropped "special" ids a while ago. - // If needed, use the handle - ID: id.Next(), - Name: "Primary Connection", - Handle: "primary_connection", - Type: types.DalPrimaryConnectionResourceType, - - Config: types.ConnectionConfig{ - DefaultModelIdent: DefaultComposeRecordTable, - DefaultAttributeIdent: DefaultComposeRecordValueCol, - DefaultPartitionFormat: DefaultPartitionFormat, - - // The connection should be taken from the .env on boot_level - }, - Capabilities: types.ConnectionCapabilities{ - Supported: capabilities.FullCapabilities(), - }, - CreatedAt: *now(), - CreatedBy: auth.ServiceUser().ID, - } - - return store.CreateDalConnection(ctx, s, conn) -} diff --git a/store/adapters/rdbms/dal/connection.go b/store/adapters/rdbms/dal/connection.go index c8ac8df21..0a236179b 100644 --- a/store/adapters/rdbms/dal/connection.go +++ b/store/adapters/rdbms/dal/connection.go @@ -12,12 +12,15 @@ import ( ) type ( + // connection provides (pkg/dal.Connection) interface to RDBMS implementation + // + // In other words: this allows Corteza to read Records from the supported SQL databases connection struct { mux sync.RWMutex models map[string]*model capabilities capabilities.Set - db *sqlx.DB + db sqlx.ExtContext dialect drivers.Dialect } ) @@ -30,7 +33,7 @@ func init() { }) } -func Connection(db *sqlx.DB, dialect drivers.Dialect, cc ...capabilities.Capability) *connection { +func Connection(db sqlx.ExtContext, dialect drivers.Dialect, cc ...capabilities.Capability) *connection { return &connection{ db: db, dialect: dialect, diff --git a/store/adapters/rdbms/dal/iterator.go b/store/adapters/rdbms/dal/iterator.go index e9c27558e..2e1faa070 100644 --- a/store/adapters/rdbms/dal/iterator.go +++ b/store/adapters/rdbms/dal/iterator.go @@ -162,8 +162,6 @@ func (i *iterator) fetch(ctx context.Context) (_ *sql.Rows, err error) { } sql, _, _ = query.Prepared(false).ToSQL() - println(sql) - if sql, args, err = query.ToSQL(); err != nil { return nil, err } diff --git a/store/adapters/rdbms/drivers/mysql/connect.go b/store/adapters/rdbms/drivers/mysql/connect.go index 73ba64a66..f1e27c77a 100644 --- a/store/adapters/rdbms/drivers/mysql/connect.go +++ b/store/adapters/rdbms/drivers/mysql/connect.go @@ -4,6 +4,8 @@ import ( "context" "database/sql" "fmt" + "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" + "github.com/cortezaproject/corteza-server/store/adapters/rdbms/dal" "strings" "github.com/cortezaproject/corteza-server/pkg/errors" @@ -43,6 +45,8 @@ func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) { s := &rdbms.Store{ DB: db, + DAL: dal.Connection(db, Dialect(), capabilities.FullCapabilities()...), + Dialect: goquDialectWrapper, TxRetryErrHandler: txRetryErrHandler, ErrorHandler: errorHandler, diff --git a/store/adapters/rdbms/drivers/postgres/connect.go b/store/adapters/rdbms/drivers/postgres/connect.go index e631f78c7..eff1393ff 100644 --- a/store/adapters/rdbms/drivers/postgres/connect.go +++ b/store/adapters/rdbms/drivers/postgres/connect.go @@ -3,6 +3,8 @@ package postgres import ( "context" "database/sql" + "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" + "github.com/cortezaproject/corteza-server/store/adapters/rdbms/dal" "net/url" "strings" @@ -46,6 +48,8 @@ func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) { s := &rdbms.Store{ DB: db, + DAL: dal.Connection(db, Dialect(), capabilities.FullCapabilities()...), + Dialect: goquDialectWrapper, ErrorHandler: errorHandler, diff --git a/store/adapters/rdbms/drivers/sqlite/connect.go b/store/adapters/rdbms/drivers/sqlite/connect.go index 709e555bf..feeeb85e2 100644 --- a/store/adapters/rdbms/drivers/sqlite/connect.go +++ b/store/adapters/rdbms/drivers/sqlite/connect.go @@ -4,6 +4,8 @@ import ( "context" "database/sql" "fmt" + "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" + "github.com/cortezaproject/corteza-server/store/adapters/rdbms/dal" "regexp" "strings" @@ -67,6 +69,8 @@ func Connect(ctx context.Context, dsn string) (_ store.Storer, err error) { s := &rdbms.Store{ DB: db, + DAL: dal.Connection(db, Dialect(), capabilities.FullCapabilities()...), + Dialect: goquDialectWrapper, ErrorHandler: errorHandler, diff --git a/store/adapters/rdbms/rdbms_store.go b/store/adapters/rdbms/rdbms_store.go index 3154bb4e1..27997e3b3 100644 --- a/store/adapters/rdbms/rdbms_store.go +++ b/store/adapters/rdbms/rdbms_store.go @@ -4,7 +4,7 @@ import ( "context" "database/sql" "fmt" - + "github.com/cortezaproject/corteza-server/pkg/dal" "github.com/cortezaproject/corteza-server/store" "github.com/cortezaproject/corteza-server/store/adapters/rdbms/ddl" "github.com/doug-martin/goqu/v9" @@ -42,6 +42,9 @@ type ( Store struct { DB sqlx.ExtContext + // DAL connection + DAL dal.Connection + // Logger for connection Logger *zap.Logger @@ -167,6 +170,11 @@ func (s Store) QueryOne(ctx context.Context, q sqlizer, dst interface{}) (err er return exec.NewScanner(rows).ScanStruct(dst) } +// ToDalConn uses store as DAL connection +func (s Store) ToDalConn() dal.Connection { + return s.DAL +} + func DefaultFunctions() *Functions { f := &Functions{} diff --git a/store/dal.go b/store/dal.go new file mode 100644 index 000000000..57b58a2ec --- /dev/null +++ b/store/dal.go @@ -0,0 +1,11 @@ +package store + +import "github.com/cortezaproject/corteza-server/pkg/dal" + +// DAL uses given store as DAL connection +// +// This is mainly used to wrap primary store connection with DAL connection wrap +// and use it to interact with records in a primary DB +func DAL(s Storer) dal.Connection { + return s.ToDalConn() +} diff --git a/store/interfaces.gen.go b/store/interfaces.gen.go index 02cc8c76f..3f152f19e 100644 --- a/store/interfaces.gen.go +++ b/store/interfaces.gen.go @@ -12,6 +12,7 @@ import ( composeType "github.com/cortezaproject/corteza-server/compose/types" federationType "github.com/cortezaproject/corteza-server/federation/types" actionlogType "github.com/cortezaproject/corteza-server/pkg/actionlog" + "github.com/cortezaproject/corteza-server/pkg/dal" discoveryType "github.com/cortezaproject/corteza-server/pkg/discovery/types" flagType "github.com/cortezaproject/corteza-server/pkg/flag/types" labelsType "github.com/cortezaproject/corteza-server/pkg/label/types" @@ -36,6 +37,9 @@ type ( // Tx is a transaction handler Tx(context.Context, func(context.Context, Storer) error) error + // Returns underlying store as DAL connection + ToDalConn() dal.Connection + // Upgrade store's schema to the latest version Upgrade(context.Context) error Actionlogs diff --git a/system/dalutils/connection.go b/system/dalutils/connection.go deleted file mode 100644 index c7773391d..000000000 --- a/system/dalutils/connection.go +++ /dev/null @@ -1,110 +0,0 @@ -package dalutils - -import ( - "context" - - "github.com/cortezaproject/corteza-server/pkg/dal" - "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" - "github.com/cortezaproject/corteza-server/store" - "github.com/cortezaproject/corteza-server/system/types" -) - -type ( - connectionCreator interface { - CreateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, cm dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error) - } - - connectionDeleter interface { - DeleteConnection(ctx context.Context, connectionID uint64) (err error) - } - - connectionUpdater interface { - UpdateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, cm dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error) - } - - connectionDeleteCreator interface { - connectionDeleter - connectionCreator - } -) - -func DalConnectionReload(ctx context.Context, s store.Storer, dc connectionDeleteCreator) (err error) { - // Get all available connections - cc, _, err := store.SearchDalConnections(ctx, s, types.DalConnectionFilter{ - Type: types.DalConnectionResourceType, - }) - if err != nil { - return - } - - for _, c := range cc { - var cm dal.ConnectionMeta - cm, err = ConnectionMeta(ctx, c) - if err != nil { - return - } - if err = dc.CreateConnection(ctx, c.ID, c.Config.Connection, cm, c.ActiveCapabilities()...); err != nil { - return - } - } - - return -} - -func DalConnectionCreate(ctx context.Context, c connectionCreator, connections ...*types.DalConnection) (err error) { - var cm dal.ConnectionMeta - for _, connection := range connections { - cm, err = ConnectionMeta(ctx, connection) - if err != nil { - return - } - - if err = c.CreateConnection(ctx, connection.ID, connection.Config.Connection, cm, connection.ActiveCapabilities()...); err != nil { - return err - } - } - - return -} - -func DalConnectionUpdate(ctx context.Context, u connectionUpdater, connections ...*types.DalConnection) (err error) { - var cm dal.ConnectionMeta - for _, connection := range connections { - cm, err = ConnectionMeta(ctx, connection) - if err != nil { - return - } - - if err = u.UpdateConnection(ctx, connection.ID, connection.Config.Connection, cm, connection.ActiveCapabilities()...); err != nil { - return err - } - } - - return -} - -func DalConnectionDelete(ctx context.Context, d connectionDeleter, connections ...*types.DalConnection) (err error) { - for _, connection := range connections { - if err = d.DeleteConnection(ctx, connection.ID); err != nil { - return err - } - } - - return -} - -// // // // // // // // // // // // // // // // // // // // // // // // // -// Utils - -func ConnectionMeta(ctx context.Context, c *types.DalConnection) (cm dal.ConnectionMeta, err error) { - // @todo we could probably utilize connection params more here - cm = dal.ConnectionMeta{ - DefaultModelIdent: c.Config.DefaultModelIdent, - DefaultAttributeIdent: c.Config.DefaultAttributeIdent, - DefaultPartitionFormat: c.Config.DefaultPartitionFormat, - SensitivityLevel: c.SensitivityLevel, - Label: c.Handle, - } - - return -} diff --git a/system/service/dal_connection.go b/system/service/dal_connection.go index 2bfd28e80..05dd5fa88 100644 --- a/system/service/dal_connection.go +++ b/system/service/dal_connection.go @@ -8,11 +8,9 @@ import ( "github.com/cortezaproject/corteza-server/pkg/actionlog" a "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/dal" - "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" "github.com/cortezaproject/corteza-server/pkg/options" "github.com/cortezaproject/corteza-server/store" - "github.com/cortezaproject/corteza-server/system/dalutils" "github.com/cortezaproject/corteza-server/system/types" ) @@ -21,7 +19,7 @@ type ( actionlog actionlog.Recorder store store.Storer ac connectionAccessController - dal dalConnections + dal dalConnManager dbConf options.DBOpt } @@ -35,15 +33,15 @@ type ( CanDeleteDalConnection(context.Context, *types.DalConnection) bool } - dalConnections interface { - CreateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, dft dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error) - UpdateConnection(ctx context.Context, connectionID uint64, cp dal.ConnectionParams, dft dal.ConnectionMeta, capabilities ...capabilities.Capability) (err error) - DeleteConnection(ctx context.Context, connectionID uint64) (err error) - SearchConnectionIssues(connectionID uint64) (err []error) + // Connection management on DAL Service + dalConnManager interface { + ReplaceConnection(context.Context, *dal.ConnectionWrap, bool) error + RemoveConnection(context.Context, uint64) error + SearchConnectionIssues(uint64) []error } ) -func Connection(ctx context.Context, dal dalConnections, dbConf options.DBOpt) *dalConnection { +func Connection(ctx context.Context, dal dalConnManager, dbConf options.DBOpt) *dalConnection { return &dalConnection{ ac: DefaultAccessControl, actionlog: DefaultActionlog, @@ -108,7 +106,7 @@ func (svc *dalConnection) Create(ctx context.Context, new *types.DalConnection) q = new - if err = dalutils.DalConnectionCreate(ctx, svc.dal, new); err != nil { + if err = dalConnectionReplace(ctx, svc.store.ToDalConn(), svc.dal, new); err != nil { return err } svc.proc(q) @@ -120,20 +118,19 @@ func (svc *dalConnection) Create(ctx context.Context, new *types.DalConnection) func (svc *dalConnection) Update(ctx context.Context, upd *types.DalConnection) (q *types.DalConnection, err error) { var ( - qProps = &dalConnectionActionProps{update: upd} + cProps = &dalConnectionActionProps{update: upd} old *types.DalConnection - e error ) err = func() (err error) { - if old, e = store.LookupDalConnectionByID(ctx, svc.store, upd.ID); e != nil { - return DalConnectionErrNotFound(qProps) + if old, err = store.LookupDalConnectionByID(ctx, svc.store, upd.ID); err != nil { + return DalConnectionErrNotFound(cProps) } svc.proc(old) if !svc.ac.CanUpdateDalConnection(ctx, old) { - return DalConnectionErrNotAllowedToUpdate(qProps) + return DalConnectionErrNotAllowedToUpdate(cProps) } upd.UpdatedAt = now() @@ -176,19 +173,15 @@ func (svc *dalConnection) Update(ctx context.Context, upd *types.DalConnection) q = upd defer svc.proc(q) - if old.HasIssues() { - return dalutils.DalConnectionCreate(ctx, svc.dal, upd) - } - - return dalutils.DalConnectionUpdate(ctx, svc.dal, upd) + return dalConnectionReplace(ctx, svc.store.ToDalConn(), svc.dal, upd) }() - return q, svc.recordAction(ctx, qProps, DalConnectionActionUpdate, err) + return q, svc.recordAction(ctx, cProps, DalConnectionActionUpdate, err) } func (svc *dalConnection) DeleteByID(ctx context.Context, ID uint64) (err error) { var ( - qProps = &dalConnectionActionProps{} - q *types.DalConnection + cProps = &dalConnectionActionProps{} + c *types.DalConnection ) err = func() (err error) { @@ -196,37 +189,37 @@ func (svc *dalConnection) DeleteByID(ctx context.Context, ID uint64) (err error) return DalConnectionErrInvalidID() } - if q, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil { + if c, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil { return } - if q.Type == types.DalPrimaryConnectionResourceType { + if c.Type == types.DalPrimaryConnectionResourceType { return fmt.Errorf("not allowed to delete primary connections") } - if !svc.ac.CanDeleteDalConnection(ctx, q) { - return DalConnectionErrNotAllowedToDelete(qProps) + if !svc.ac.CanDeleteDalConnection(ctx, c) { + return DalConnectionErrNotAllowedToDelete(cProps) } - qProps.setConnection(q) + cProps.setConnection(c) - q.DeletedAt = now() - q.DeletedBy = a.GetIdentityFromContext(ctx).Identity() + c.DeletedAt = now() + c.DeletedBy = a.GetIdentityFromContext(ctx).Identity() - if err = store.UpdateDalConnection(ctx, svc.store, q); err != nil { + if err = store.UpdateDalConnection(ctx, svc.store, c); err != nil { return } - return dalutils.DalConnectionDelete(ctx, svc.dal, q) + return dalConnectionRemove(ctx, svc.dal, c) }() - return svc.recordAction(ctx, qProps, DalConnectionActionDelete, err) + return svc.recordAction(ctx, cProps, DalConnectionActionDelete, err) } func (svc *dalConnection) UndeleteByID(ctx context.Context, ID uint64) (err error) { var ( - qProps = &dalConnectionActionProps{} - q *types.DalConnection + cProps = &dalConnectionActionProps{} + c *types.DalConnection ) err = func() (err error) { @@ -234,28 +227,28 @@ func (svc *dalConnection) UndeleteByID(ctx context.Context, ID uint64) (err erro return DalConnectionErrInvalidID() } - if q, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil { + if c, err = store.LookupDalConnectionByID(ctx, svc.store, ID); err != nil { return } - if !svc.ac.CanDeleteDalConnection(ctx, q) { - return DalConnectionErrNotAllowedToUndelete(qProps) + if !svc.ac.CanDeleteDalConnection(ctx, c) { + return DalConnectionErrNotAllowedToUndelete(cProps) } - qProps.setConnection(q) + cProps.setConnection(c) - q.DeletedAt = nil - q.UpdatedBy = a.GetIdentityFromContext(ctx).Identity() + c.DeletedAt = nil + c.UpdatedBy = a.GetIdentityFromContext(ctx).Identity() - if err = store.UpdateDalConnection(ctx, svc.store, q); err != nil { + if err = store.UpdateDalConnection(ctx, svc.store, c); err != nil { return } // We're creating it here since it was removed on delete - return dalutils.DalConnectionCreate(ctx, svc.dal, q) + return dalConnectionRemove(ctx, svc.dal, c) }() - return svc.recordAction(ctx, qProps, DalConnectionActionDelete, err) + return svc.recordAction(ctx, cProps, DalConnectionActionDelete, err) } func (svc *dalConnection) Search(ctx context.Context, filter types.DalConnectionFilter) (r types.DalConnectionSet, f types.DalConnectionFilter, err error) { @@ -288,7 +281,7 @@ func (svc *dalConnection) Search(ctx context.Context, filter types.DalConnection } func (svc *dalConnection) ReloadConnections(ctx context.Context) (err error) { - return dalutils.DalConnectionReload(ctx, svc.store, svc.dal) + return dalConnectionReload(ctx, svc.store, svc.dal) } func (svc *dalConnection) proc(connections ...*types.DalConnection) { @@ -322,3 +315,64 @@ func (svc *dalConnection) procDal(c *types.DalConnection) { func (svc *dalConnection) procLocale(c *types.DalConnection) { // @todo... } + +func dalConnectionReload(ctx context.Context, s store.Storer, dcm dalConnManager) (err error) { + // Get all available connections + cc, _, err := store.SearchDalConnections(ctx, s, types.DalConnectionFilter{}) + if err != nil { + return + } + + return dalConnectionReplace(ctx, s.ToDalConn(), dcm, cc...) +} + +// Replaces all given connections +func dalConnectionReplace(ctx context.Context, primary dal.Connection, dcm dalConnManager, cc ...*types.DalConnection) (err error) { + var ( + cw *dal.ConnectionWrap + isPrimary bool + ) + + for _, c := range cc { + isPrimary = c.Type == types.DalPrimaryConnectionResourceType + + cw = dal.MakeConnection( + c.ID, + // When connection is primary (type) we use the primary connection + // passed in to the fn + func() dal.Connection { + if isPrimary { + return primary + } + + return nil + }(), + c.Config.Connection, + dal.ConnectionMeta{ + DefaultModelIdent: c.Config.DefaultModelIdent, + DefaultAttributeIdent: c.Config.DefaultAttributeIdent, + DefaultPartitionFormat: c.Config.DefaultPartitionFormat, + SensitivityLevel: c.SensitivityLevel, + Label: c.Handle, + }, + c.ActiveCapabilities()..., + ) + + if err = dcm.ReplaceConnection(ctx, cw, isPrimary); err != nil { + return err + } + } + + return +} + +// Removes a connection from DAL service +func dalConnectionRemove(ctx context.Context, dcm dalConnManager, cc ...*types.DalConnection) (err error) { + for _, c := range cc { + if err = dcm.RemoveConnection(ctx, c.ID); err != nil { + return err + } + } + + return +} diff --git a/system/service/service.go b/system/service/service.go index 4c81c6306..ceaaa4922 100644 --- a/system/service/service.go +++ b/system/service/service.go @@ -88,7 +88,6 @@ var ( DefaultApigwFilter *apigwFilter DefaultApigwProfiler *apigwProfiler DefaultReport *report - primaryConnectionConfig *types.DalConnection DefaultDataPrivacy *dataPrivacy DefaultStatistics *statistics @@ -105,7 +104,7 @@ var ( } ) -func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, primaryConn *types.DalConnection, ws websocketSender, c Config) (err error) { +func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, ws websocketSender, c Config) (err error) { var ( hcd = healthcheck.Defaults() ) @@ -151,7 +150,6 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, primaryCon DefaultSettings = Settings(ctx, DefaultStore, DefaultLogger, DefaultAccessControl, CurrentSettings) - primaryConnectionConfig = primaryConn DefaultDalConnection = Connection(ctx, dal.Service(), c.DB) DefaultDalSensitivityLevel = SensitivityLevel(ctx, dal.Service()) @@ -277,6 +275,19 @@ func Activate(ctx context.Context) (err error) { return } + // Reload DAL sensitivity levels + err = DefaultDalSensitivityLevel.ReloadSensitivityLevels(ctx, DefaultStore) + if err != nil { + return + } + + // Reload DAL connections + println("reloading connections") + err = DefaultDalConnection.ReloadConnections(ctx) + if err != nil { + return + } + return } diff --git a/system/types/dal_connection.go b/system/types/dal_connection.go index be0ff11e1..c4c8df5ae 100644 --- a/system/types/dal_connection.go +++ b/system/types/dal_connection.go @@ -77,7 +77,8 @@ type ( var ( // Used to identify the primary DAL connection instead of an extra flag - DalPrimaryConnectionResourceType = "corteza::system:primary_dal_connection" + DalPrimaryConnectionResourceType = "corteza::system:primary-dal-connection" + DalPrimaryConnectionHandle = "primary-database" ) func (c DalConnection) ActiveCapabilities() capabilities.Set { diff --git a/tests/dal/dal_utils.go b/tests/dal/dal_utils.go index 33417e450..73526a6db 100644 --- a/tests/dal/dal_utils.go +++ b/tests/dal/dal_utils.go @@ -14,7 +14,6 @@ import ( "github.com/cortezaproject/corteza-server/pkg/filter" "github.com/cortezaproject/corteza-server/pkg/id" "github.com/cortezaproject/corteza-server/store" - "github.com/cortezaproject/corteza-server/system/dalutils" "github.com/cortezaproject/corteza-server/system/service" "github.com/cortezaproject/corteza-server/system/types" "github.com/cortezaproject/corteza-server/tests/dal/setup/mysql" @@ -140,9 +139,12 @@ func (h helper) cleanupDal() { func initSvc(ctx context.Context, d driver) (dalService, error) { c := makeConnectionDefinition(d.dsn) - cm, err := dalutils.ConnectionMeta(ctx, c) - if err != nil { - return nil, err + cm := dal.ConnectionMeta{ + DefaultModelIdent: c.Config.DefaultModelIdent, + DefaultAttributeIdent: c.Config.DefaultAttributeIdent, + DefaultPartitionFormat: c.Config.DefaultPartitionFormat, + SensitivityLevel: c.SensitivityLevel, + Label: c.Handle, } svc, err := dal.New(