From 6311dfb9cb0f4b0da2c9e0d3e809dfcf8c128086 Mon Sep 17 00:00:00 2001 From: Peter Grlica Date: Mon, 2 Nov 2020 18:06:30 +0100 Subject: [PATCH] Moved federation sync to service watchers --- app/boot_levels.go | 2 + federation/commands/sync.go | 35 --- federation/commands/sync_data.go | 210 +-------------- federation/commands/sync_structure.go | 184 +------------ federation/service/processer.go | 2 +- federation/service/service.go | 18 ++ federation/service/sync.go | 57 +++- federation/service/sync_worker.go | 18 ++ federation/service/sync_worker_data.go | 276 ++++++++++++++++++++ federation/service/sync_worker_structure.go | 266 +++++++++++++++++++ federation/service/syncer.go | 16 +- federation/types/syncer_uri.go | 5 - 12 files changed, 652 insertions(+), 437 deletions(-) create mode 100644 federation/service/sync_worker.go create mode 100644 federation/service/sync_worker_data.go create mode 100644 federation/service/sync_worker_structure.go diff --git a/app/boot_levels.go b/app/boot_levels.go index 89d76ae34..685b8f0ad 100644 --- a/app/boot_levels.go +++ b/app/boot_levels.go @@ -7,6 +7,7 @@ import ( cmpService "github.com/cortezaproject/corteza-server/compose/service" cmpEvent "github.com/cortezaproject/corteza-server/compose/service/event" fdrService "github.com/cortezaproject/corteza-server/federation/service" + fedService "github.com/cortezaproject/corteza-server/federation/service" msgService "github.com/cortezaproject/corteza-server/messaging/service" msgEvent "github.com/cortezaproject/corteza-server/messaging/service/event" "github.com/cortezaproject/corteza-server/messaging/websocket" @@ -311,6 +312,7 @@ func (app *CortezaApp) Activate(ctx context.Context) (err error) { sysService.Watchers(ctx) cmpService.Watchers(ctx) msgService.Watchers(ctx) + fedService.Watchers(ctx) rbac.Global().Watch(ctx) diff --git a/federation/commands/sync.go b/federation/commands/sync.go index 7c587774e..30ee1de72 100644 --- a/federation/commands/sync.go +++ b/federation/commands/sync.go @@ -2,34 +2,12 @@ package commands import ( "context" - "fmt" - "time" - "github.com/cortezaproject/corteza-server/federation/service" - "github.com/cortezaproject/corteza-server/federation/types" "github.com/cortezaproject/corteza-server/pkg/cli" "github.com/spf13/cobra" ) -const ( - limit = 100 - - httpTimeout time.Duration = 10 - baseURL string = "http://localhost:8084" -) - -var ( - sync *service.Sync - mapper *service.Mapper - - surls = make(chan service.Surl, 1) - spayloads = make(chan service.Spayload, 1) - - countProcess = 0 - countPersist = 0 -) - type ( serviceInitializer interface { InitServices(ctx context.Context) error @@ -71,19 +49,6 @@ func Sync(app serviceInitializer) *cobra.Command { return cmd } -func queueUrl(url *types.SyncerURI, urls chan service.Surl, meta service.Processer) { - s, _ := url.String() - - service.DefaultLogger.Info(fmt.Sprintf("Adding %s to queue", s)) - - t := service.Surl{ - Url: *url, - Meta: meta, - } - - sync.QueueUrl(t, urls) -} - func commandPreRunInitService(ctx context.Context, app serviceInitializer) func(*cobra.Command, []string) error { return func(_ *cobra.Command, _ []string) error { return app.InitServices(ctx) diff --git a/federation/commands/sync_data.go b/federation/commands/sync_data.go index 24fc0337a..e513ea1fa 100644 --- a/federation/commands/sync_data.go +++ b/federation/commands/sync_data.go @@ -2,226 +2,22 @@ package commands import ( "context" - "fmt" - "io/ioutil" "time" cs "github.com/cortezaproject/corteza-server/compose/service" - ct "github.com/cortezaproject/corteza-server/compose/types" "github.com/cortezaproject/corteza-server/federation/service" - "github.com/cortezaproject/corteza-server/federation/types" - "github.com/cortezaproject/corteza-server/pkg/decoder" "github.com/spf13/cobra" ) -type ( - dataProcesser struct { - ID uint64 - NodeID uint64 - ComposeModuleID uint64 - ComposeNamespaceID uint64 - ModuleMappingValues *ct.RecordValueSet - } -) - func commandSyncData(ctx context.Context) func(*cobra.Command, []string) { - - const syncDelay = 10 - return func(_ *cobra.Command, _ []string) { - - // todo - get auth from the node - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - ticker := time.NewTicker(time.Second * syncDelay) - - mapper = &service.Mapper{} - - sync = service.NewSync( + syncService := service.NewSync( &service.Syncer{}, &service.Mapper{}, service.DefaultSharedModule, cs.DefaultRecord) - queueDataForNodes(ctx, sync) - - for { - select { - case <-ctx.Done(): - // do any cleanup here - service.DefaultLogger.Info(fmt.Sprintf("Stopping sync [processed: %d, persisted: %d]", countProcess, countPersist)) - return - case <-ticker.C: - // do the whole process again - queueDataForNodes(ctx, sync) - case url := <-surls: - select { - case <-ctx.Done(): - // do any cleanup here - service.DefaultLogger.Info(fmt.Sprintf("Stopping sync [processed: %d, persisted: %d]", countProcess, countPersist)) - return - default: - } - - s, err := url.Url.String() - - if err != nil { - continue - } - - responseBody, err := sync.FetchUrl(ctx, s) - - if err != nil { - continue - } - - spayload := service.Spayload{ - Payload: responseBody, - Meta: url.Meta, - } - - spayloads <- spayload - // payloads <- responseBody - case p := <-spayloads: - body, err := ioutil.ReadAll(p.Payload) - - // handle error - if err != nil { - continue - } - - basePath := fmt.Sprintf("/federation/nodes/%d/modules/%d/records/", p.Meta.(*dataProcesser).NodeID, p.Meta.(*dataProcesser).ID) - - u := types.SyncerURI{ - BaseURL: baseURL, - Path: basePath, - Limit: limit, - } - - err = sync.ProcessPayload(ctx, body, surls, u, p.Meta.(*dataProcesser)) - - if err != nil { - // handle error - service.DefaultLogger.Info(fmt.Sprintf("error on handling payload: %s", err)) - } else { - n, err := service.DefaultNode.FindBySharedNodeID(ctx, p.Meta.(*dataProcesser).NodeID) - - if err != nil { - service.DefaultLogger.Info(fmt.Sprintf("could not update sync status: %s", err)) - } - - // add to db - nodes_sync - new := &types.NodeSync{ - NodeID: (*n).ID, - SyncStatus: types.NodeSyncStatusSuccess, - SyncType: types.NodeSyncTypeData, - TimeOfAction: time.Now().UTC(), - } - - new, err = service.DefaultNodeSync.Create(ctx, new) - - if err != nil { - service.DefaultLogger.Info(fmt.Sprintf("could not update sync status: %s", err)) - } - } - } - } + syncData := service.WorkerData(syncService, service.DefaultLogger) + syncData.Watch(ctx, time.Second*30, 50) } } - -func queueDataForNodes(ctx context.Context, sync *service.Sync) { - nodes, err := sync.GetPairedNodes(ctx) - - if err != nil { - return - } - - // get all shared modules and their module mappings - for _, n := range nodes { - set, err := sync.GetSharedModules(ctx, n.ID) - - if err != nil { - service.DefaultLogger.Warn(fmt.Sprintf("could not get shared modules for node: %d, skipping", n.ID)) - continue - } - - // go through set and prepare module mappings for it - for _, sm := range set { - mappings, _ := sync.GetModuleMappings(ctx, sm.ID) - - if mappings == nil { - service.DefaultLogger.Info(fmt.Sprintf("could not prepare module mappings for shared module: %d, skipping", sm.ID)) - continue - } - - mappingValues, err := sync.PrepareModuleMappings(ctx, mappings) - - if err != nil || mappingValues == nil { - service.DefaultLogger.Info(fmt.Sprintf("could not prepare module mappings for shared module: %d, skipping", sm.ID)) - continue - } - - // get the last sync per-node - lastSync, _ := sync.GetLastSyncTime(ctx, n.ID, types.NodeSyncTypeData) - basePath := fmt.Sprintf("/federation/nodes/%d/modules/%d/records/", n.SharedNodeID, sm.ExternalFederationModuleID) - - ex := "" - if lastSync != nil { - ex = fmt.Sprintf(" from last sync on (%s)", lastSync.Format(time.RFC3339)) - } - - service.DefaultLogger.Info(fmt.Sprintf("starting structure sync%s for node %d, module %d", ex, n.ID, sm.ID)) - - url := types.SyncerURI{ - BaseURL: baseURL, - Path: basePath, - Limit: limit, - LastSync: lastSync, - } - - processer := &dataProcesser{ - ID: sm.ExternalFederationModuleID, - NodeID: n.SharedNodeID, - ComposeModuleID: mappings.ComposeModuleID, - ComposeNamespaceID: mappings.ComposeNamespaceID, - ModuleMappingValues: &mappingValues, - } - - go queueUrl(&url, surls, processer) - } - } -} - -// Process gets the payload from syncer and -// uses the decode package to decode the whole set, depending on -// the filtering that was used (limit) -func (dp *dataProcesser) Process(ctx context.Context, payload []byte) error { - countProcess = countProcess + 1 - - o, err := decoder.DecodeFederationRecordSync([]byte(payload)) - - if err != nil { - return err - } - - if len(o) == 0 { - return nil - } - - service.DefaultLogger.Info(fmt.Sprintf("Adding %d objects", len(o))) - - for _, er := range o { - mapper.Merge(&er.Values, dp.ModuleMappingValues) - - rec := &ct.Record{ - ModuleID: dp.ComposeModuleID, - NamespaceID: dp.ComposeNamespaceID, - Values: *dp.ModuleMappingValues, - } - - sync.CreateRecord(ctx, rec) - } - - return nil -} diff --git a/federation/commands/sync_structure.go b/federation/commands/sync_structure.go index 20558dd80..20c5e69ca 100644 --- a/federation/commands/sync_structure.go +++ b/federation/commands/sync_structure.go @@ -2,200 +2,22 @@ package commands import ( "context" - "fmt" - "io/ioutil" "time" cs "github.com/cortezaproject/corteza-server/compose/service" "github.com/cortezaproject/corteza-server/federation/service" - "github.com/cortezaproject/corteza-server/federation/types" - "github.com/cortezaproject/corteza-server/pkg/decoder" "github.com/spf13/cobra" ) -type ( - structureProcesser struct { - NodeID uint64 - } -) - func commandSyncStructure(ctx context.Context) func(*cobra.Command, []string) { - - const syncDelay = 10 - return func(_ *cobra.Command, _ []string) { - - // todo - get auth from the node - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - ticker := time.NewTicker(time.Second * syncDelay) - - sync = service.NewSync( + syncService := service.NewSync( &service.Syncer{}, &service.Mapper{}, service.DefaultSharedModule, cs.DefaultRecord) - queueStructureForNodes(ctx, sync) - - for { - select { - case <-ctx.Done(): - // do any cleanup here - service.DefaultLogger.Info(fmt.Sprintf("Stopping sync [processed: %d, persisted: %d]", countProcess, countPersist)) - return - case <-ticker.C: - // do the whole process again - queueStructureForNodes(ctx, sync) - case url := <-surls: - select { - case <-ctx.Done(): - // do any cleanup here - service.DefaultLogger.Info(fmt.Sprintf("Stopping sync [processed: %d, persisted: %d]", countProcess, countPersist)) - return - default: - } - - s, err := url.Url.String() - - if err != nil { - continue - } - - responseBody, err := sync.FetchUrl(ctx, s) - - if err != nil { - continue - } - - spayload := service.Spayload{ - Payload: responseBody, - Meta: url.Meta, - } - - spayloads <- spayload - case p := <-spayloads: - body, err := ioutil.ReadAll(p.Payload) - - // handle error - if err != nil { - continue - } - - basePath := fmt.Sprintf("/federation/nodes/%d/modules/exposed/", p.Meta.(*structureProcesser).NodeID) - - u := types.SyncerURI{ - BaseURL: baseURL, - Path: basePath, - Limit: limit, - } - - err = sync.ProcessPayload(ctx, body, surls, u, p.Meta.(*structureProcesser)) - - if err != nil { - // handle error - service.DefaultLogger.Info(fmt.Sprintf("error on handling payload: %s", err)) - } else { - n, err := service.DefaultNode.FindBySharedNodeID(ctx, p.Meta.(*structureProcesser).NodeID) - - if err != nil { - service.DefaultLogger.Info(fmt.Sprintf("could not update sync status: %s", err)) - } - - // add to db - nodes_sync - new := &types.NodeSync{ - NodeID: (*n).ID, - SyncStatus: types.NodeSyncStatusSuccess, - SyncType: types.NodeSyncTypeStructure, - TimeOfAction: time.Now().UTC(), - } - - new, err = service.DefaultNodeSync.Create(ctx, new) - - if err != nil { - service.DefaultLogger.Info(fmt.Sprintf("could not update sync status: %s", err)) - } - } - } - } + syncStructure := service.WorkerStructure(syncService, service.DefaultLogger) + syncStructure.Watch(ctx, time.Second*10, 10) } } - -func queueStructureForNodes(ctx context.Context, sync *service.Sync) { - nodes, err := sync.GetPairedNodes(ctx) - - if err != nil { - return - } - - // get all shared modules and their module mappings - for _, n := range nodes { - // get the last sync per-node - lastSync, _ := sync.GetLastSyncTime(ctx, n.ID, types.NodeSyncTypeStructure) - basePath := fmt.Sprintf("/federation/nodes/%d/modules/exposed/", n.SharedNodeID) - - ex := "" - if lastSync != nil { - ex = fmt.Sprintf(" from last sync on (%s)", lastSync.Format(time.RFC3339)) - } - - service.DefaultLogger.Info(fmt.Sprintf("starting structure sync%s for node %d", ex, n.ID)) - - url := types.SyncerURI{ - BaseURL: baseURL, - Path: basePath, - Limit: limit, - LastSync: lastSync, - } - - processer := &structureProcesser{ - NodeID: n.SharedNodeID, - } - - go queueUrl(&url, surls, processer) - } -} - -// Process gets the payload from syncer and -// uses the decode package to decode the whole set, depending on -// the filtering that was used (limit) -func (dp *structureProcesser) Process(ctx context.Context, payload []byte) error { - countProcess = countProcess + 1 - - now := time.Now() - o, err := decoder.DecodeFederationModuleSync([]byte(payload)) - - if err != nil { - return err - } - - if len(o) == 0 { - return nil - } - - service.DefaultLogger.Info(fmt.Sprintf("Adding %d objects", len(o))) - - for i, em := range o { - n := &types.SharedModule{ - NodeID: dp.NodeID, - ExternalFederationModuleID: em.ID, - Fields: em.Fields, - Handle: fmt.Sprintf("Handle %d %d", i, now.Unix()), - Name: fmt.Sprintf("Name %d %d", i, now.Unix()), - } - - n, err := sync.CreateSharedModule(ctx, n) - - service.DefaultLogger.Info(fmt.Sprintf("Added shared module: %d", n.ID)) - - if err != nil { - service.DefaultLogger.Error(fmt.Sprintf("could not create shared module: %s", err)) - continue - } - - countPersist = countPersist + 1 - } - - return nil -} diff --git a/federation/service/processer.go b/federation/service/processer.go index 95a69dbca..f9c91da7f 100644 --- a/federation/service/processer.go +++ b/federation/service/processer.go @@ -4,6 +4,6 @@ import "context" type ( Processer interface { - Process(ctx context.Context, payload []byte) error + Process(ctx context.Context, payload []byte) (int, error) } ) diff --git a/federation/service/service.go b/federation/service/service.go index 38573c1eb..ff59d1048 100644 --- a/federation/service/service.go +++ b/federation/service/service.go @@ -4,6 +4,7 @@ import ( "context" "time" + cs "github.com/cortezaproject/corteza-server/compose/service" "github.com/cortezaproject/corteza-server/pkg/actionlog" "github.com/cortezaproject/corteza-server/pkg/auth" "github.com/cortezaproject/corteza-server/pkg/healthcheck" @@ -128,3 +129,20 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config) return } + +func Watchers(ctx context.Context) { + syncService := NewSync( + &Syncer{}, + &Mapper{}, + DefaultSharedModule, + cs.DefaultRecord) + + syncStructure := WorkerStructure(syncService, DefaultLogger) + syncData := WorkerData(syncService, service.DefaultLogger) + + // each minute, 10 per page + go syncStructure.Watch(ctx, time.Minute*2, 10) + + // each minute, 100 per page + go syncData.Watch(ctx, time.Second*60, 100) +} diff --git a/federation/service/sync.go b/federation/service/sync.go index 3dc6b4710..a9d92f56b 100644 --- a/federation/service/sync.go +++ b/federation/service/sync.go @@ -2,12 +2,14 @@ package service import ( "context" + "encoding/json" "io" "time" cs "github.com/cortezaproject/corteza-server/compose/service" ct "github.com/cortezaproject/corteza-server/compose/types" "github.com/cortezaproject/corteza-server/federation/types" + "github.com/cortezaproject/corteza-server/system/service" ) type ( @@ -28,22 +30,73 @@ func NewSync(s *Syncer, m *Mapper, sm SharedModuleService, cs cs.RecordService) } } -func (s *Sync) ProcessPayload(ctx context.Context, payload []byte, out chan Surl, url types.SyncerURI, processer Processer) error { +// CanUpdateSharedModule checks the origin and destination module +// compatibility of fields +// It currently checks if all of the fields are exactly the same +// TODO - check if any of the newly missing fields are actually being used so a safe update +// is possible +func (s *Sync) CanUpdateSharedModule(ctx context.Context, new *types.SharedModule, existing *types.SharedModule) (bool, error) { + + // check for mapped fields + fstr, err := json.Marshal(new.Fields) + f2str, err := json.Marshal(existing.Fields) + + if err != nil { + return false, err + } + + if string(fstr) == string(f2str) { + return true, nil + } + + return false, nil +} + +// ProcessPayload passes the payload to the syncer lib +func (s *Sync) ProcessPayload(ctx context.Context, payload []byte, out chan Url, url types.SyncerURI, processer Processer) (int, error) { return s.syncer.Process(ctx, payload, out, url, processer) } -func (s *Sync) QueueUrl(url Surl, out chan Surl) { +// QueueUrl passes the url to the syncer +func (s *Sync) QueueUrl(url Url, out chan Url) { s.syncer.Queue(url, out) } +// FetchUrl passes the url to be fetched to the syncer func (s *Sync) FetchUrl(ctx context.Context, url string) (io.Reader, error) { return s.syncer.Fetch(ctx, url) } +// CreateRecord wraps the compose Record service Create func (s *Sync) CreateRecord(ctx context.Context, rec *ct.Record) (*ct.Record, error) { return s.composeRecordService.With(ctx).Create(rec) } +// LookupSharedModule find the shared module if exists +func (s *Sync) LookupSharedModule(ctx context.Context, new *types.SharedModule) (*types.SharedModule, error) { + var sm *types.SharedModule + + list, _, err := service.DefaultStore.SearchFederationSharedModules(ctx, types.SharedModuleFilter{ + NodeID: new.NodeID, + ExternalFederationModuleID: new.ExternalFederationModuleID}) + + if err != nil { + return nil, err + } + + if len(list) == 0 { + return nil, nil + } + + sm = list[0] + + return sm, nil +} + +func (s *Sync) UpdateSharedModule(ctx context.Context, updated *types.SharedModule) (*types.SharedModule, error) { + return s.sharedModuleService.Update(ctx, updated) +} + func (s *Sync) CreateSharedModule(ctx context.Context, new *types.SharedModule) (*types.SharedModule, error) { return s.sharedModuleService.Create(ctx, new) } diff --git a/federation/service/sync_worker.go b/federation/service/sync_worker.go new file mode 100644 index 000000000..3e60f263d --- /dev/null +++ b/federation/service/sync_worker.go @@ -0,0 +1,18 @@ +package service + +import ( + "context" + "time" +) + +type ( + worker interface { + PrepareForNodes(ctx context.Context, sync *Sync, urls chan Url) + Watch(ctx context.Context, delay time.Duration, limit int) + } +) + +const ( + defaultDelay = time.Second * 10 + defaultPage = 10 +) diff --git a/federation/service/sync_worker_data.go b/federation/service/sync_worker_data.go new file mode 100644 index 000000000..5e743a609 --- /dev/null +++ b/federation/service/sync_worker_data.go @@ -0,0 +1,276 @@ +package service + +import ( + "context" + "fmt" + "io/ioutil" + "time" + + ct "github.com/cortezaproject/corteza-server/compose/types" + "github.com/cortezaproject/corteza-server/federation/types" + "github.com/cortezaproject/corteza-server/pkg/decoder" + "github.com/davecgh/go-spew/spew" + "go.uber.org/zap" +) + +type ( + syncWorkerData struct { + syncService *Sync + logger *zap.Logger + delay time.Duration + limit int + } + + dataProcesser struct { + ID uint64 + NodeID uint64 + ComposeModuleID uint64 + ComposeNamespaceID uint64 + NodeBaseURL string + ModuleMappingValues *ct.RecordValueSet + SyncService *Sync + } +) + +func WorkerData(sync *Sync, logger *zap.Logger) *syncWorkerData { + return &syncWorkerData{ + syncService: sync, + logger: logger, + limit: defaultPage, + delay: defaultDelay, + } +} + +func (w *syncWorkerData) queueUrl(url *types.SyncerURI, urls chan Url, meta Processer) { + // s, _ := url.String() + + // w.logger.Debug(fmt.Sprintf("adding %s to queue", s)) + + t := Url{ + Url: *url, + Meta: meta, + } + + w.syncService.QueueUrl(t, urls) +} + +func (w *syncWorkerData) PrepareForNodes(ctx context.Context, urls chan Url) { + + nodes, err := w.syncService.GetPairedNodes(ctx) + + if err != nil { + w.logger.Info("could not get paired nodes", zap.Error(err)) + return + } + + // get all shared modules and their module mappings + for _, n := range nodes { + set, err := w.syncService.GetSharedModules(ctx, n.ID) + + if err != nil { + w.logger.Info("could not get shared modules, skipping", zap.Uint64("nodeID", n.ID), zap.Error(err)) + continue + } + + if len(set) == 0 { + w.logger.Info("there are no valid shared modules, skipping", zap.Uint64("nodeID", n.ID), zap.Error(err)) + continue + } + + // go through set and prepare module mappings for it + for _, sm := range set { + mappings, _ := w.syncService.GetModuleMappings(ctx, sm.ID) + + if mappings == nil { + w.logger.Info("could not prepare module mappings for shared module, skipping", zap.Uint64("nodeID", n.ID), zap.Uint64("moduleID", sm.ID)) + continue + } + + mappingValues, err := w.syncService.PrepareModuleMappings(ctx, mappings) + + if err != nil || mappingValues == nil { + w.logger.Info("could not prepare module mappings for shared module, skipping", zap.Uint64("nodeID", n.ID), zap.Uint64("moduleID", sm.ID)) + continue + } + + // get the last sync per-node + lastSync, _ := w.syncService.GetLastSyncTime(ctx, n.ID, types.NodeSyncTypeData) + basePath := fmt.Sprintf("/federation/nodes/%d/modules/%d/records/", n.SharedNodeID, sm.ExternalFederationModuleID) + + z := []zap.Field{zap.Uint64("nodeID", n.ID)} + + if lastSync != nil { + z = append(z, zap.Time("lastSync", *lastSync)) + } + + w.logger.Info("starting data sync", z...) + + url := types.SyncerURI{ + BaseURL: n.BaseURL, + Path: basePath, + Limit: w.limit, + LastSync: lastSync, + } + + processer := &dataProcesser{ + ID: sm.ExternalFederationModuleID, + NodeID: n.SharedNodeID, + ComposeModuleID: mappings.ComposeModuleID, + ComposeNamespaceID: mappings.ComposeNamespaceID, + ModuleMappingValues: &mappingValues, + NodeBaseURL: n.BaseURL, + SyncService: w.syncService, + } + + go w.queueUrl(&url, urls, processer) + } + } + +} + +func (w *syncWorkerData) Watch(ctx context.Context, delay time.Duration, limit int) { + var ( + urls = make(chan Url, 100) + payloads = make(chan Payload, 10) + + countProcess = 0 + ) + + w.delay = delay + w.limit = limit + + // todo - get auth from the node + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + ticker := time.NewTicker(delay) + + w.PrepareForNodes(ctx, urls) + + for { + select { + case <-ctx.Done(): + // do any cleanup here + w.logger.Info("stopping sync", zap.Int("processed", countProcess)) + return + case <-ticker.C: + // do the whole process again + w.PrepareForNodes(ctx, urls) + case url := <-urls: + select { + case <-ctx.Done(): + // do any cleanup here + w.logger.Info("stopping sync", zap.Int("processed", countProcess)) + return + default: + } + + s, err := url.Url.String() + + if err != nil { + spew.Dump("ERR", err) + continue + } + + responseBody, err := w.syncService.FetchUrl(ctx, s) + + if err != nil { + spew.Dump("ERR", err) + continue + } + + spayload := Payload{ + Payload: responseBody, + Meta: url.Meta, + } + + payloads <- spayload + case p := <-payloads: + + body, err := ioutil.ReadAll(p.Payload) + + // handle error + if err != nil { + continue + } + + basePath := fmt.Sprintf("/federation/nodes/%d/modules/%d/records/", p.Meta.(*dataProcesser).NodeID, p.Meta.(*dataProcesser).ID) + + u := types.SyncerURI{ + BaseURL: p.Meta.(*dataProcesser).NodeBaseURL, + Path: basePath, + Limit: w.limit, + } + + processed, err := w.syncService.ProcessPayload(ctx, body, urls, u, p.Meta.(*dataProcesser)) + countProcess += processed + + if err != nil { + // handle error + w.logger.Info("error on handling payload", zap.Error(err)) + } else { + // TODO + n, err := DefaultNode.FindBySharedNodeID(ctx, p.Meta.(*dataProcesser).NodeID) + + if err != nil { + w.logger.Info("could not update sync status", zap.Error(err)) + continue + } + + // add to db - nodes_sync + new := &types.NodeSync{ + NodeID: (*n).ID, + SyncStatus: types.NodeSyncStatusSuccess, + SyncType: types.NodeSyncTypeData, + TimeOfAction: time.Now().UTC(), + } + + // todo + new, err = DefaultNodeSync.Create(ctx, new) + + if err != nil { + w.logger.Info("could not update sync status", zap.Error(err)) + } + + w.logger.Info("processed objects", zap.Int("processed", processed), zap.Uint64("nodeID", n.ID)) + } + } + } +} + +// Process gets the payload from syncer and +// uses the decode package to decode the whole set, depending on +// the filtering that was used (limit) +func (dp *dataProcesser) Process(ctx context.Context, payload []byte) (int, error) { + + processed := 0 + o, err := decoder.DecodeFederationRecordSync([]byte(payload)) + + if err != nil { + return processed, err + } + + if len(o) == 0 { + return processed, nil + } + + for _, er := range o { + dp.SyncService.mapper.Merge(&er.Values, dp.ModuleMappingValues) + + rec := &ct.Record{ + ModuleID: dp.ComposeModuleID, + NamespaceID: dp.ComposeNamespaceID, + Values: *dp.ModuleMappingValues, + } + + _, err := dp.SyncService.CreateRecord(ctx, rec) + + if err != nil { + continue + } + + processed++ + } + + return processed, nil +} diff --git a/federation/service/sync_worker_structure.go b/federation/service/sync_worker_structure.go new file mode 100644 index 000000000..9c48b4ea6 --- /dev/null +++ b/federation/service/sync_worker_structure.go @@ -0,0 +1,266 @@ +package service + +import ( + "context" + "fmt" + "io/ioutil" + "time" + + "github.com/cortezaproject/corteza-server/federation/types" + "github.com/cortezaproject/corteza-server/pkg/decoder" + "github.com/davecgh/go-spew/spew" + "go.uber.org/zap" +) + +type ( + syncWorkerStructure struct { + syncService *Sync + logger *zap.Logger + delay time.Duration + limit int + } + + structureProcesser struct { + NodeID uint64 + SharedNodeID uint64 + NodeBaseURL string + SyncService *Sync + } +) + +func WorkerStructure(sync *Sync, logger *zap.Logger) *syncWorkerStructure { + return &syncWorkerStructure{ + syncService: sync, + logger: logger, + limit: defaultPage, + delay: defaultDelay, + } +} + +func (w *syncWorkerStructure) queueUrl(url *types.SyncerURI, urls chan Url, meta Processer) { + // s, _ := url.String() + + // w.logger.Debug(fmt.Sprintf("adding %s to queue", s)) + + t := Url{ + Url: *url, + Meta: meta, + } + + w.syncService.QueueUrl(t, urls) +} + +func (w *syncWorkerStructure) PrepareForNodes(ctx context.Context, urls chan Url) { + nodes, err := w.syncService.GetPairedNodes(ctx) + + if err != nil { + return + } + + // get all shared modules and their module mappings + for _, n := range nodes { + // get the last sync per-node + lastSync, _ := w.syncService.GetLastSyncTime(ctx, n.ID, types.NodeSyncTypeStructure) + basePath := fmt.Sprintf("/federation/nodes/%d/modules/exposed/", n.SharedNodeID) + + z := []zap.Field{zap.Uint64("nodeID", n.ID)} + + if lastSync != nil { + z = append(z, zap.Time("lastSync", *lastSync)) + } + + w.logger.Info("starting structure sync", z...) + + url := types.SyncerURI{ + BaseURL: n.BaseURL, + Path: basePath, + Limit: w.limit, + LastSync: lastSync, + } + + processer := &structureProcesser{ + NodeID: n.ID, + SharedNodeID: n.SharedNodeID, + NodeBaseURL: n.BaseURL, + SyncService: w.syncService, + } + + go w.queueUrl(&url, urls, processer) + } +} + +func (w *syncWorkerStructure) Watch(ctx context.Context, delay time.Duration, limit int) { + var ( + urls = make(chan Url, 100) + payloads = make(chan Payload, 10) + + countProcess = 0 + ) + + w.delay = delay + w.limit = limit + + // todo - get auth from the node + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + ticker := time.NewTicker(delay) + + w.PrepareForNodes(ctx, urls) + + for { + select { + case <-ctx.Done(): + // do any cleanup here + w.logger.Info("stopping sync", zap.Int("processed", countProcess)) + return + case <-ticker.C: + // do the whole process again + w.PrepareForNodes(ctx, urls) + case url := <-urls: + select { + case <-ctx.Done(): + // do any cleanup here + w.logger.Info("stopping sync", zap.Int("processed", countProcess)) + return + default: + } + + s, err := url.Url.String() + + if err != nil { + spew.Dump("ERR", err) + continue + } + + responseBody, err := w.syncService.FetchUrl(ctx, s) + + if err != nil { + spew.Dump("ERR", err) + continue + } + + spayload := Payload{ + Payload: responseBody, + Meta: url.Meta, + } + + payloads <- spayload + case p := <-payloads: + body, err := ioutil.ReadAll(p.Payload) + + // handle error + if err != nil { + spew.Dump("ERR", err) + continue + } + + basePath := fmt.Sprintf("/federation/nodes/%d/modules/exposed/", p.Meta.(*structureProcesser).SharedNodeID) + + u := types.SyncerURI{ + BaseURL: p.Meta.(*structureProcesser).NodeBaseURL, + Path: basePath, + Limit: w.limit, + } + + processed, err := w.syncService.ProcessPayload(ctx, body, urls, u, p.Meta.(*structureProcesser)) + countProcess += processed + + if err != nil { + spew.Dump("ERR", err) + // handle error + w.logger.Info("error on handling payload", zap.Error(err)) + } else { + n, err := DefaultNode.FindBySharedNodeID(ctx, p.Meta.(*structureProcesser).SharedNodeID) + + if err != nil { + spew.Dump("ERR", err) + w.logger.Info("could not update sync status", zap.Error(err)) + continue + } + + // add to db - nodes_sync + new := &types.NodeSync{ + NodeID: (*n).ID, + SyncStatus: types.NodeSyncStatusSuccess, + SyncType: types.NodeSyncTypeStructure, + TimeOfAction: time.Now().UTC(), + } + + new, err = DefaultNodeSync.Create(ctx, new) + + if err != nil { + spew.Dump("ERR", err) + w.logger.Info("could not update sync status", zap.Error(err)) + } + + w.logger.Info("processed objects", zap.Int("processed", processed), zap.Uint64("nodeID", n.ID)) + } + } + } +} + +// Process gets the payload from syncer and +// uses the decode package to decode the whole set, depending on +// the filtering that was used (limit) +func (dp *structureProcesser) Process(ctx context.Context, payload []byte) (int, error) { + processed := 0 + + now := time.Now() + o, err := decoder.DecodeFederationModuleSync([]byte(payload)) + + if err != nil { + spew.Dump("ERR", err) + return processed, err + } + + if len(o) == 0 { + return processed, nil + } + + for i, em := range o { + new := &types.SharedModule{ + NodeID: dp.NodeID, + ExternalFederationModuleID: em.ID, + Fields: em.Fields, + Handle: fmt.Sprintf("Handle %d %d", i, now.Unix()), + Name: fmt.Sprintf("Name %d %d", i, now.Unix()), + } + + existing, err := dp.SyncService.LookupSharedModule(ctx, new) + + if err != nil { + continue + } + + if existing == nil { + _, err := dp.SyncService.CreateSharedModule(ctx, new) + + if err != nil { + continue + } + + processed++ + continue + } + + canUpdate, err := dp.SyncService.CanUpdateSharedModule(ctx, new, existing) + + if err != nil { + continue + } + + if canUpdate { + _, err := dp.SyncService.UpdateSharedModule(ctx, existing) + + if err != nil { + continue + } + + processed++ + continue + } + } + + return processed, nil +} diff --git a/federation/service/syncer.go b/federation/service/syncer.go index 85a306d0e..80fa9f520 100644 --- a/federation/service/syncer.go +++ b/federation/service/syncer.go @@ -9,16 +9,17 @@ import ( "time" "github.com/cortezaproject/corteza-server/federation/types" + "github.com/davecgh/go-spew/spew" ) type ( Syncer struct{} - Surl struct { + Url struct { Url types.SyncerURI Meta Processer } - Spayload struct { + Payload struct { Payload io.Reader Meta Processer } @@ -37,7 +38,7 @@ type ( } ) -func (h *Syncer) Queue(url Surl, out chan Surl) { +func (h *Syncer) Queue(url Url, out chan Url) { out <- url } @@ -48,11 +49,13 @@ func (h *Syncer) Fetch(ctx context.Context, url string) (io.Reader, error) { req, err := http.NewRequest("GET", url, nil) if err != nil { + spew.Dump("ERR", err) return nil, err } resp, err := client.Do(req) if err != nil { + spew.Dump("ERR", err) return nil, err } @@ -63,17 +66,18 @@ func (h *Syncer) Fetch(ctx context.Context, url string) (io.Reader, error) { return resp.Body, nil } -func (h *Syncer) Process(ctx context.Context, payload []byte, out chan Surl, url types.SyncerURI, processer Processer) error { +func (h *Syncer) Process(ctx context.Context, payload []byte, out chan Url, url types.SyncerURI, processer Processer) (int, error) { aux, err := h.ParseHeader(ctx, payload) if err != nil { - return err + spew.Dump("ERR", err) + return 0, err } if aux.Response.Filter.NextPage != "" { url.NextPage = aux.Response.Filter.NextPage - out <- Surl{ + out <- Url{ Url: url, Meta: processer, } diff --git a/federation/types/syncer_uri.go b/federation/types/syncer_uri.go index 4dc9778bf..0b83b49e8 100644 --- a/federation/types/syncer_uri.go +++ b/federation/types/syncer_uri.go @@ -6,8 +6,6 @@ import ( "strconv" "strings" "time" - - "github.com/davecgh/go-spew/spew" ) // SyncerURI serves as a default struct for handling url queues @@ -88,15 +86,12 @@ func (s *SyncerURI) Parse(uri string) error { } func parseLastSync(lastSync string) (*time.Time, error) { - spew.Dump("parse last sync") if i, err := strconv.ParseInt(lastSync, 10, 64); err == nil { - spew.Dump("returning") t := time.Unix(i, 0) return &t, nil } // try different format if above fails - spew.Dump("TRY HERE") if t, err := time.Parse(time.RFC3339, lastSync); err == nil { return &t, nil }