From a70087ace82e96f67612e24a2a778e4cca047713 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Mon, 30 May 2022 15:43:03 +0200 Subject: [PATCH] Refactor primary DAL connection to be preserved in DB --- app/boot_levels.go | 61 ++++++++++----------- pkg/dal/service.go | 8 +-- pkg/provision/provision.go | 47 ++++++++++++++++ store/adapters/rdbms/upgrade_tables.go | 2 +- system/rest/dal_connection.go | 26 +-------- system/service/dal_connection.go | 76 ++++++++++++++------------ system/service/service.go | 6 +- 7 files changed, 129 insertions(+), 97 deletions(-) diff --git a/app/boot_levels.go b/app/boot_levels.go index aca15334f..300bc602c 100644 --- a/app/boot_levels.go +++ b/app/boot_levels.go @@ -22,9 +22,7 @@ import ( "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/corredor" "github.com/cortezaproject/corteza-server/pkg/dal" - "github.com/cortezaproject/corteza-server/pkg/dal/capabilities" "github.com/cortezaproject/corteza-server/pkg/eventbus" - "github.com/cortezaproject/corteza-server/pkg/geolocation" "github.com/cortezaproject/corteza-server/pkg/healthcheck" "github.com/cortezaproject/corteza-server/pkg/http" "github.com/cortezaproject/corteza-server/pkg/id" @@ -57,10 +55,6 @@ const ( bootLevelProvisioned bootLevelServicesInitialized bootLevelActivated - - defaultComposeRecordTable = "compose_record" - defaultComposeRecordValueCol = "values" - defaultPartitionFormat = "compose_record_{{namespace}}_{{module}}" ) // Setup configures all required services @@ -298,41 +292,44 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) { return } - var primaryDalConnection = 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, + 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 + } - Location: &geolocation.Full{ - // @todo get from .env - }, - Ownership: "@todo", - // @todo - // SensitivityLevel: , + dalLogger := app.Log.Named("dal") - Config: types.ConnectionConfig{ - DefaultModelIdent: defaultComposeRecordTable, - DefaultAttributeIdent: defaultComposeRecordValueCol, - DefaultPartitionFormat: defaultPartitionFormat, + // 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 + } - Connection: dal.NewDSNConnection(app.Opt.DB.DSN), - }, - Capabilities: types.ConnectionCapabilities{ - Supported: capabilities.FullCapabilities(), - }, - CreatedAt: time.Now(), - CreatedBy: auth.ServiceUser().ID, + 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 - dalLogger := app.Log.Named("dal") if _, err = dal.InitGlobalService( ctx, dalLogger, app.Opt.Environment.IsDevelopment(), + primaryDalConnection.ID, // DB_DSN is the default connection with full capabilities primaryDalConnection.Config.Connection, @@ -341,7 +338,7 @@ func (app *CortezaApp) InitServices(ctx context.Context) (err error) { DefaultAttributeIdent: primaryDalConnection.Config.DefaultAttributeIdent, DefaultPartitionFormat: primaryDalConnection.Config.DefaultPartitionFormat, // @todo make it configurable from env - SensitivityLevel: 0, + SensitivityLevel: primaryDalConnection.SensitivityLevel, Label: primaryDalConnection.Handle, }, primaryDalConnection.ActiveCapabilities()...); err != nil { diff --git a/pkg/dal/service.go b/pkg/dal/service.go index 880201087..c58c43e33 100644 --- a/pkg/dal/service.go +++ b/pkg/dal/service.go @@ -55,7 +55,7 @@ var ( ) // InitGlobalService initializes a fresh DAL where the given primary connection -func InitGlobalService(ctx context.Context, log *zap.Logger, inDev bool, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (*service, error) { +func InitGlobalService(ctx context.Context, log *zap.Logger, inDev bool, connectionID uint64, cp ConnectionParams, cm ConnectionMeta, capabilities ...capabilities.Capability) (*service, error) { if gSvc == nil { log.Debug("initializing DAL service with primary connection", zap.Any("connection params", cp)) @@ -73,6 +73,7 @@ func InitGlobalService(ctx context.Context, log *zap.Logger, inDev bool, cp Conn meta: cm, sensitivityLevel: cm.SensitivityLevel, label: cm.Label, + connectionID: connectionID, } cw.connection, err = connect(ctx, log, inDev, cp, capabilities...) if err != nil { @@ -270,9 +271,8 @@ func (svc *service) ReloadModel(ctx context.Context, models ...*Model) (err erro } func (svc *service) ModelIdentFormatter(connectionID uint64) (f *IdentFormatter, err error) { - // @todo ... c := svc.connections[connectionID] - if connectionID == 0 { + if connectionID == 0 || connectionID == svc.primary.connectionID { c = svc.primary } @@ -391,7 +391,7 @@ func (svc *service) GetModelByResource(connectionID uint64, resType string, reso func (svc *service) getConnection(ctx context.Context, connectionID uint64, cc ...capabilities.Capability) (cw *connectionWrap, can capabilities.Set, err error) { err = func() error { // get the requested connection - if connectionID == DefaultConnectionID { + if connectionID == DefaultConnectionID || connectionID == svc.primary.connectionID { cw = svc.primary } else { cw = svc.connections[connectionID] diff --git a/pkg/provision/provision.go b/pkg/provision/provision.go index da2a66023..20363fc09 100644 --- a/pkg/provision/provision.go +++ b/pkg/provision/provision.go @@ -4,6 +4,8 @@ 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" @@ -21,6 +23,12 @@ 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") @@ -42,6 +50,8 @@ func Run(ctx context.Context, log *zap.Logger, s store.Storer, provisionOpt opti func() error { return authAddExternals(ctx, log.Named("auth.externals"), s) }, 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) }, } for _, fn := range ffn { @@ -105,3 +115,40 @@ 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/upgrade_tables.go b/store/adapters/rdbms/upgrade_tables.go index f7c6bec83..d9ecbc42d 100644 --- a/store/adapters/rdbms/upgrade_tables.go +++ b/store/adapters/rdbms/upgrade_tables.go @@ -104,7 +104,7 @@ func tableDalConnections() *Table { ColumnDef("name", ColumnTypeText), ColumnDef("type", ColumnTypeText), - ColumnDef("location", ColumnTypeJson), + ColumnDef("location", ColumnTypeJson, Null), ColumnDef("ownership", ColumnTypeText), ColumnDef("sensitivity_level", ColumnTypeIdentifier), diff --git a/system/rest/dal_connection.go b/system/rest/dal_connection.go index 3d4e83325..b63d22566 100644 --- a/system/rest/dal_connection.go +++ b/system/rest/dal_connection.go @@ -2,7 +2,6 @@ package rest import ( "context" - "fmt" "io" "net/http" "time" @@ -51,7 +50,6 @@ type ( connectionService interface { FindByID(ctx context.Context, ID uint64) (*types.DalConnection, error) - FindPrimary(ctx context.Context) (*types.DalConnection, error) Create(ctx context.Context, new *types.DalConnection) (*types.DalConnection, error) Update(ctx context.Context, upd *types.DalConnection) (*types.DalConnection, error) DeleteByID(ctx context.Context, ID uint64) error @@ -113,13 +111,7 @@ func (ctrl DalConnection) Create(ctx context.Context, r *request.DalConnectionCr Capabilities: r.Capabilities, } - switch connection.Type { - case types.DalConnectionResourceType: - return ctrl.svc.Create(ctx, connection) - default: - return nil, fmt.Errorf("cannot create connection: unsupported connection type %s", connection.Type) - } - + return ctrl.svc.Create(ctx, connection) } func (ctrl DalConnection) Update(ctx context.Context, r *request.DalConnectionUpdate) (interface{}, error) { @@ -187,19 +179,14 @@ func (ctrl DalConnection) federatedNodeToConnection(f *federationTypes.Node) *ty func (ctrl DalConnection) collectConnections(ctx context.Context, f types.DalConnectionFilter) (out types.DalConnectionSet, err error) { var ( - dalConnections types.DalConnectionSet - primaryDalConnection *types.DalConnection - federatedNodes federationTypes.NodeSet + dalConnections types.DalConnectionSet + federatedNodes federationTypes.NodeSet ) if dalConnections, f, err = ctrl.svc.Search(ctx, f); err != nil { return nil, err } - if primaryDalConnection, err = ctrl.svc.FindPrimary(ctx); err != nil { - return nil, err - } - if !reflect2.IsNil(ctrl.federationSvc) { if federatedNodes, _, err = ctrl.federationSvc.Search(ctx, federationTypes.NodeFilter{ // @todo IDs? @@ -209,7 +196,6 @@ func (ctrl DalConnection) collectConnections(ctx context.Context, f types.DalCon } } - out = append(out, primaryDalConnection) out = append(out, dalConnections...) // We're converting federation nodes to DAL connection structs so that we have @@ -251,12 +237,6 @@ func (ctrl DalConnection) filterConnections(baseConnections types.DalConnectionS } } - if f.Check != nil { - // @todo error handling here? - tmp, _ := f.Check(conn) - include = include && tmp - } - if include { out = append(out, conn) } diff --git a/system/service/dal_connection.go b/system/service/dal_connection.go index 04a69a115..9e77fc84b 100644 --- a/system/service/dal_connection.go +++ b/system/service/dal_connection.go @@ -2,6 +2,8 @@ package service import ( "context" + "fmt" + "reflect" "github.com/cortezaproject/corteza-server/pkg/actionlog" a "github.com/cortezaproject/corteza-server/pkg/auth" @@ -18,8 +20,6 @@ type ( store store.Storer ac connectionAccessController dal dalConnections - - primaryConnection types.DalConnection } connectionAccessController interface { @@ -39,13 +39,12 @@ type ( } ) -func Connection(ctx context.Context, pcOpts types.DalConnection, dal dalConnections) *dalConnection { +func Connection(ctx context.Context, dal dalConnections) *dalConnection { return &dalConnection{ - ac: DefaultAccessControl, - actionlog: DefaultActionlog, - store: DefaultStore, - dal: dal, - primaryConnection: pcOpts, + ac: DefaultAccessControl, + actionlog: DefaultActionlog, + store: DefaultStore, + dal: dal, } } @@ -55,13 +54,6 @@ func (svc *dalConnection) FindByID(ctx context.Context, ID uint64) (q *types.Dal ) err = func() error { - if ID == 0 { - // primary; construct it since it doesn't actually exist - aux := svc.primaryConnection - q = &aux - return nil - } - if ID == 0 { return DalConnectionErrInvalidID() } @@ -82,21 +74,6 @@ func (svc *dalConnection) FindByID(ctx context.Context, ID uint64) (q *types.Dal return q, svc.recordAction(ctx, rProps, DalConnectionActionLookup, err) } -func (svc *dalConnection) FindPrimary(ctx context.Context) (q *types.DalConnection, err error) { - var ( - rProps = &dalConnectionActionProps{} - ) - - err = func() error { - // primary; construct it since it doesn't actually exist - aux := svc.primaryConnection - q = &aux - return nil - }() - - return q, svc.recordAction(ctx, rProps, DalConnectionActionLookup, err) -} - func (svc *dalConnection) Create(ctx context.Context, new *types.DalConnection) (q *types.DalConnection, err error) { var ( qProps = &dalConnectionActionProps{new: new} @@ -111,6 +88,15 @@ func (svc *dalConnection) Create(ctx context.Context, new *types.DalConnection) new.CreatedAt = *now() new.CreatedBy = a.GetIdentityFromContext(ctx).Identity() + // validation + { + if new.Type != types.DalPrimaryConnectionResourceType { + // @todo error + err = fmt.Errorf("cannot create connection: unsupported connection type %s", new.Type) + return + } + } + if err = store.CreateDalConnection(ctx, svc.store, new); err != nil { return err } @@ -132,23 +118,41 @@ 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} - qq *types.DalConnection + old *types.DalConnection e error ) err = func() (err error) { - if qq, e = store.LookupDalConnectionByID(ctx, svc.store, upd.ID); e != nil { + if old, e = store.LookupDalConnectionByID(ctx, svc.store, upd.ID); e != nil { return DalConnectionErrNotFound(qProps) } - if !svc.ac.CanUpdateDalConnection(ctx, qq) { + if !svc.ac.CanUpdateDalConnection(ctx, old) { return DalConnectionErrNotAllowedToUpdate(qProps) } upd.UpdatedAt = now() - upd.CreatedAt = qq.CreatedAt + upd.CreatedAt = old.CreatedAt upd.UpdatedBy = a.GetIdentityFromContext(ctx).Identity() + // validate + { + if old.Type == types.DalPrimaryConnectionResourceType { + if !reflect.DeepEqual(old.Config.Connection, upd.Config.Connection) { + // @todo err + return fmt.Errorf("can not update connection parameters for primary connection") + } + + if old.Handle != upd.Handle { + return fmt.Errorf("can not update handle for primary connection") + } + + if old.Type != upd.Type { + return fmt.Errorf("can not update type for primary connection") + } + } + } + if err = store.UpdateDalConnection(ctx, svc.store, upd); err != nil { return } @@ -182,6 +186,10 @@ func (svc *dalConnection) DeleteByID(ctx context.Context, ID uint64) (err error) return } + if q.Type == types.DalPrimaryConnectionResourceType { + return fmt.Errorf("not allowed to delete primary connections") + } + if !svc.ac.CanDeleteDalConnection(ctx, q) { return DalConnectionErrNotAllowedToDelete(qProps) } diff --git a/system/service/service.go b/system/service/service.go index d0e5d0eab..ae116183d 100644 --- a/system/service/service.go +++ b/system/service/service.go @@ -87,7 +87,7 @@ var ( DefaultApigwFilter *apigwFilter DefaultApigwProfiler *apigwProfiler DefaultReport *report - primaryConnectionConfig types.DalConnection + primaryConnectionConfig *types.DalConnection DefaultStatistics *statistics @@ -103,7 +103,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, primaryConn *types.DalConnection, ws websocketSender, c Config) (err error) { var ( hcd = healthcheck.Defaults() ) @@ -150,7 +150,7 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, primaryCon DefaultSettings = Settings(ctx, DefaultStore, DefaultLogger, DefaultAccessControl, CurrentSettings) primaryConnectionConfig = primaryConn - DefaultDalConnection = Connection(ctx, primaryConnectionConfig, dal.Service()) + DefaultDalConnection = Connection(ctx, dal.Service()) DefaultDalSensitivityLevel = SensitivityLevel(ctx, dal.Service()) if err != nil {