Added default options to sync workers

federation/service/service.go

@@ -41,6 +41,8 @@ var (
 	// CurrentSettings represents current system settings
 	CurrentSettings = &types.AppSettings{}
 
+	DefaultOptions options.FederationOpt
+
 	DefaultActionlog actionlog.Recorder
 
 	DefaultNode          *node
@@ -66,6 +68,8 @@ func Initialize(ctx context.Context, log *zap.Logger, s store.Storer, c Config)
 		hcd = healthcheck.Defaults()
 	)
 
+	DefaultOptions = c.Federation
+
 	// we're doing conversion to avoid having
 	// store interface exposed or generated inside app package
 	DefaultStore = s
@@ -142,11 +146,8 @@ func Watchers(ctx context.Context) {
 	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)
+	go syncStructure.Watch(ctx, time.Second*DefaultOptions.StructureMonitorInterval, DefaultOptions.StructurePageSize)
+	go syncData.Watch(ctx, time.Second*DefaultOptions.DataMonitorInterval, DefaultOptions.DataPageSize)
 }
 
 func AddFederationLabel(entity label.LabeledResource, value string) {

federation/service/sync_worker_data.go

@@ -93,7 +93,12 @@ func (w *syncWorkerData) PrepareForNodes(ctx context.Context, urls chan Url) {
 			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)}
+			z := []zap.Field{
+				zap.Uint64("nodeID", n.ID),
+				zap.String("host", n.BaseURL),
+				zap.String("delay", w.delay.String()),
+				zap.Int("pagesize", w.limit),
+			}
 
 			if lastSync != nil {
 				z = append(z, zap.Time("lastSync", *lastSync))

federation/service/sync_worker_structure.go

@@ -63,7 +63,12 @@ func (w *syncWorkerStructure) PrepareForNodes(ctx context.Context, urls chan Url
 		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), zap.String("host", n.BaseURL)}
+		z := []zap.Field{
+			zap.Uint64("nodeID", n.ID),
+			zap.String("host", n.BaseURL),
+			zap.String("delay", w.delay.String()),
+			zap.Int("pagesize", w.limit),
+		}
 
 		if lastSync != nil {
 			z = append(z, zap.Time("lastSync", *lastSync))

federation/types/syncer_uri.go

@@ -14,8 +14,6 @@ type (
 	SyncerURI struct {
 		Limit    int
 		LastSync *time.Time
-		// todo
-		// LastSync *LastSyncTime
 		Path     string
 		BaseURL  string
 		NextPage string