RefreshReplicationCfg is called when the cfg changes. Checks whether replications have been added to or removed from this node
(ctx context.Context)
| 763 | // RefreshReplicationCfg is called when the cfg changes. Checks whether replications |
| 764 | // have been added to or removed from this node |
| 765 | func (m *sgReplicateManager) RefreshReplicationCfg(ctx context.Context) error { |
| 766 | |
| 767 | base.InfofCtx(m.loggingCtx, base.KeyCluster, "Replication definitions changed - refreshing...") |
| 768 | configReplications, err := m.GetReplications() |
| 769 | if err != nil { |
| 770 | return err |
| 771 | } |
| 772 | |
| 773 | m.activeReplicatorsLock.Lock() |
| 774 | defer m.activeReplicatorsLock.Unlock() |
| 775 | |
| 776 | // check for active replications that should be stopped |
| 777 | for replicationID, activeReplicator := range m.activeReplicators { |
| 778 | replicationCfg, ok := configReplications[replicationID] |
| 779 | |
| 780 | // Check for active replications removed from this node |
| 781 | if !ok || replicationCfg.AssignedNode != m.localNodeUUID { |
| 782 | base.InfofCtx(m.loggingCtx, base.KeyReplicate, "Stopping reassigned replication %s", replicationID) |
| 783 | err := activeReplicator.Stop() |
| 784 | if err != nil { |
| 785 | base.WarnfCtx(m.loggingCtx, "Unable to gracefully close active replication: %v", err) |
| 786 | } |
| 787 | if !ok { |
| 788 | activeReplicator.purgeCheckpoints() |
| 789 | } |
| 790 | delete(m.activeReplicators, replicationID) |
| 791 | |
| 792 | } else { |
| 793 | // Check for replications assigned to this node with updated state |
| 794 | base.DebugfCtx(m.loggingCtx, base.KeyReplicate, "Aligning state for existing replication %s", replicationID) |
| 795 | |
| 796 | // If the config has changed, re-initialize the replication |
| 797 | isChanged, err := m.isCfgChanged(replicationCfg, activeReplicator.config) |
| 798 | if err != nil { |
| 799 | base.WarnfCtx(m.loggingCtx, "Error evaluating whether cfg has changed, potential changes not applied: %v", err) |
| 800 | } |
| 801 | if isChanged { |
| 802 | replicator, initError := m.InitializeReplication(replicationCfg) |
| 803 | if initError != nil { |
| 804 | base.WarnfCtx(m.loggingCtx, "Error initializing upserted replication %s: %v", replicationID, initError) |
| 805 | } else { |
| 806 | m.activeReplicators[replicationID] = replicator |
| 807 | activeReplicator = replicator |
| 808 | } |
| 809 | } |
| 810 | |
| 811 | stateErr := activeReplicator.alignState(ctx, replicationCfg.TargetState) |
| 812 | if stateErr != nil { |
| 813 | base.WarnfCtx(m.loggingCtx, "Error updating active replication %s to state %s: %v", replicationID, replicationCfg.TargetState, stateErr) |
| 814 | } |
| 815 | // Reset is synchronous - after completion the replication state should be updated to stopped |
| 816 | if replicationCfg.TargetState == ReplicationStateResetting { |
| 817 | postResetErr := m.UpdateReplicationState(replicationCfg.ID, ReplicationStateStopped) |
| 818 | if postResetErr != nil { |
| 819 | base.WarnfCtx(m.loggingCtx, "Error updating replication state to stopped after successful reset: %v", postResetErr) |
| 820 | } |
| 821 | |
| 822 | } |
no test coverage detected