DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Moved federation sync to service watchers

Browse Source
This commit is contained in:
Peter Grlica
2020-11-02 18:06:30 +01:00
parent 2f860d591a
commit 6311dfb9cb
12 changed files with 652 additions and 437 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/boot_levels.go
+2
View File
@@ -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)
federation/commands/sync.go
-35
View File
@@ -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)
federation/commands/sync_data.go
+3 -207
View File
@@ -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
}
federation/commands/sync_structure.go
+3 -181
View File
@@ -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
}
federation/service/processer.go
+1 -1
View File
@@ -4,6 +4,6 @@ import "context"
type (
Processer interface {
Process(ctx context.Context, payload []byte) error
Process(ctx context.Context, payload []byte) (int, error)
}
)
federation/service/service.go
+18
View File
@@ -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)
}
federation/service/sync.go
+55 -2
View File
@@ -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)
}
federation/service/sync_worker.go
+18
View File
@@ -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
)
federation/service/sync_worker_data.go
+276
View File
@@ -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
}
federation/service/sync_worker_structure.go
+266
View File
@@ -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
}
federation/service/syncer.go
+10 -6
View File
@@ -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,
}
federation/types/syncer_uri.go
-5
View File
@@ -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
}