DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Refactor primary DAL connection to be preserved in DB

Browse Source
This commit is contained in:
Tomaž Jerman
2022-05-30 15:43:03 +02:00
parent a2606ea58d
commit a70087ace8
7 changed files with 129 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
app/boot_levels.go
View File

@@ -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 {

8
pkg/dal/service.go
View File

@@ -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]

47
pkg/provision/provision.go
View File

@@ -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)
}

2
store/adapters/rdbms/upgrade_tables.go
View File

@@ -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),

26
system/rest/dal_connection.go
View File

@@ -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)
}

76
system/service/dal_connection.go
View File

@@ -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)
}

6
system/service/service.go
View File

@@ -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 {