More Replication Abstractions (#6689)

Also updated ACL replication to use a function to fill in the desired enterprise meta for all remote listing RPCs.

agent/consul/acl_oss.go

@@ -12,6 +12,8 @@ import (
 // EnterpriseACLResolverDelegate stub
 type EnterpriseACLResolverDelegate interface{}
 
+func (s *Server) fillReplicationEnterpriseMeta(_ *structs.EnterpriseMeta) {}
+
 func newEnterpriseACLConfig(*log.Logger) *acl.EnterpriseACLConfig {
 	return nil
 }

agent/consul/acl_replication.go

@@ -112,6 +112,7 @@ func (s *Server) fetchACLRoles(lastRemoteIndex uint64) (*structs.ACLRoleListResp
 			Token:         s.tokens.ReplicationToken(),
 		},
 	}
+	s.fillReplicationEnterpriseMeta(&req.EnterpriseMeta)
 
 	var response structs.ACLRoleListResponse
 	if err := s.RPC("ACL.RoleList", &req, &response); err != nil {
@@ -149,6 +150,7 @@ func (s *Server) fetchACLPolicies(lastRemoteIndex uint64) (*structs.ACLPolicyLis
 			Token:         s.tokens.ReplicationToken(),
 		},
 	}
+	s.fillReplicationEnterpriseMeta(&req.EnterpriseMeta)
 
 	var response structs.ACLPolicyListResponse
 	if err := s.RPC("ACL.PolicyList", &req, &response); err != nil {
@@ -342,6 +344,7 @@ func (s *Server) fetchACLTokens(lastRemoteIndex uint64) (*structs.ACLTokenListRe
 		IncludeLocal:  false,
 		IncludeGlobal: true,
 	}
+	s.fillReplicationEnterpriseMeta(&req.EnterpriseMeta)
 
 	var response structs.ACLTokenListResponse
 	if err := s.RPC("ACL.TokenList", &req, &response); err != nil {

agent/consul/acl_replication_types.go

@@ -34,7 +34,10 @@ func (r *aclTokenReplicator) FetchRemote(srv *Server, lastRemoteIndex uint64) (i
 func (r *aclTokenReplicator) FetchLocal(srv *Server) (int, uint64, error) {
 	r.local = nil
 
-	idx, local, err := srv.fsm.State().ACLTokenList(nil, false, true, "", "", "", structs.ReplicationEnterpriseMeta())
+	var entMeta structs.EnterpriseMeta
+	srv.fillReplicationEnterpriseMeta(&entMeta)
+
+	idx, local, err := srv.fsm.State().ACLTokenList(nil, false, true, "", "", "", &entMeta)
 	if err != nil {
 		return 0, 0, err
 	}
@@ -155,7 +158,10 @@ func (r *aclPolicyReplicator) FetchRemote(srv *Server, lastRemoteIndex uint64) (
 func (r *aclPolicyReplicator) FetchLocal(srv *Server) (int, uint64, error) {
 	r.local = nil
 
-	idx, local, err := srv.fsm.State().ACLPolicyList(nil, structs.ReplicationEnterpriseMeta())
+	var entMeta structs.EnterpriseMeta
+	srv.fillReplicationEnterpriseMeta(&entMeta)
+
+	idx, local, err := srv.fsm.State().ACLPolicyList(nil, &entMeta)
 	if err != nil {
 		return 0, 0, err
 	}
@@ -265,7 +271,10 @@ func (r *aclRoleReplicator) FetchRemote(srv *Server, lastRemoteIndex uint64) (in
 func (r *aclRoleReplicator) FetchLocal(srv *Server) (int, uint64, error) {
 	r.local = nil
 
-	idx, local, err := srv.fsm.State().ACLRoleList(nil, "", nil)
+	var entMeta structs.EnterpriseMeta
+	srv.fillReplicationEnterpriseMeta(&entMeta)
+
+	idx, local, err := srv.fsm.State().ACLRoleList(nil, "", &entMeta)
 	if err != nil {
 		return 0, 0, err
 	}

agent/consul/replication.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	metrics "github.com/armon/go-metrics"
 	"github.com/hashicorp/consul/lib"
 	"golang.org/x/time/rate"
 )
@@ -20,11 +21,15 @@ const (
 	replicationDefaultRate = 1
 )
 
+type ReplicatorDelegate interface {
+	Replicate(ctx context.Context, lastRemoteIndex uint64) (index uint64, exit bool, err error)
+}
+
 type ReplicatorConfig struct {
 	// Name to be used in various logging
 	Name string
-	// Function to perform the actual replication
-	ReplicateFn ReplicatorFunc
+	// Delegate to perform each round of replication
+	Delegate ReplicatorDelegate
 	// The number of replication rounds per second that are allowed
 	Rate int
 	// The number of replication rounds that can be done in a burst
@@ -37,13 +42,11 @@ type ReplicatorConfig struct {
 	Logger *log.Logger
 }
 
-type ReplicatorFunc func(ctx context.Context, lastRemoteIndex uint64) (index uint64, exit bool, err error)
-
 type Replicator struct {
 	name            string
 	limiter         *rate.Limiter
 	waiter          *lib.RetryWaiter
-	replicateFn     ReplicatorFunc
+	delegate        ReplicatorDelegate
 	logger          *log.Logger
 	lastRemoteIndex uint64
 }
@@ -52,8 +55,8 @@ func NewReplicator(config *ReplicatorConfig) (*Replicator, error) {
 	if config == nil {
 		return nil, fmt.Errorf("Cannot create the Replicator without a config")
 	}
-	if config.ReplicateFn == nil {
-		return nil, fmt.Errorf("Cannot create the Replicator without a ReplicateFn set in the config")
+	if config.Delegate == nil {
+		return nil, fmt.Errorf("Cannot create the Replicator without a Delegate set in the config")
 	}
 	if config.Logger == nil {
 		config.Logger = log.New(os.Stderr, "", log.LstdFlags)
@@ -71,11 +74,11 @@ func NewReplicator(config *ReplicatorConfig) (*Replicator, error) {
 	}
 	waiter := lib.NewRetryWaiter(minFailures, 0*time.Second, maxWait, lib.NewJitterRandomStagger(10))
 	return &Replicator{
-		name:        config.Name,
-		limiter:     limiter,
-		waiter:      waiter,
-		replicateFn: config.ReplicateFn,
-		logger:      config.Logger,
+		name:     config.Name,
+		limiter:  limiter,
+		waiter:   waiter,
+		delegate: config.Delegate,
+		logger:   config.Logger,
 	}, nil
 }
 
@@ -91,7 +94,7 @@ func (r *Replicator) Run(ctx context.Context) error {
 		}
 
 		// Perform a single round of replication
-		index, exit, err := r.replicateFn(ctx, atomic.LoadUint64(&r.lastRemoteIndex))
+		index, exit, err := r.delegate.Replicate(ctx, atomic.LoadUint64(&r.lastRemoteIndex))
 		if exit {
 			// the replication function told us to exit
 			return nil
@@ -120,3 +123,143 @@ func (r *Replicator) Run(ctx context.Context) error {
 func (r *Replicator) Index() uint64 {
 	return atomic.LoadUint64(&r.lastRemoteIndex)
 }
+
+type ReplicatorFunc func(ctx context.Context, lastRemoteIndex uint64) (index uint64, exit bool, err error)
+
+type FunctionReplicator struct {
+	ReplicateFn ReplicatorFunc
+}
+
+func (r *FunctionReplicator) Replicate(ctx context.Context, lastRemoteIndex uint64) (uint64, bool, error) {
+	return r.ReplicateFn(ctx, lastRemoteIndex)
+}
+
+type IndexReplicatorDiff struct {
+	NumUpdates   int
+	Updates      interface{}
+	NumDeletions int
+	Deletions    interface{}
+}
+
+type IndexReplicatorDelegate interface {
+	// SingularNoun is the singular form of the item being replicated.
+	SingularNoun() string
+
+	// PluralNoun is the plural form of the item being replicated.
+	PluralNoun() string
+
+	// Name to use when emitting metrics
+	MetricName() string
+
+	// FetchRemote retrieves items newer than the provided index from the
+	// remote datacenter (for diffing purposes).
+	FetchRemote(lastRemoteIndex uint64) (int, interface{}, uint64, error)
+
+	// FetchLocal retrieves items from the current datacenter (for diffing
+	// purposes).
+	FetchLocal() (int, interface{}, error)
+
+	DiffRemoteAndLocalState(local interface{}, remote interface{}, lastRemoteIndex uint64) (*IndexReplicatorDiff, error)
+
+	PerformDeletions(ctx context.Context, deletions interface{}) (exit bool, err error)
+
+	PerformUpdates(ctx context.Context, updates interface{}) (exit bool, err error)
+}
+
+type IndexReplicator struct {
+	Delegate IndexReplicatorDelegate
+	Logger   *log.Logger
+}
+
+func (r *IndexReplicator) Replicate(ctx context.Context, lastRemoteIndex uint64) (uint64, bool, error) {
+	fetchStart := time.Now()
+	lenRemote, remote, remoteIndex, err := r.Delegate.FetchRemote(lastRemoteIndex)
+	metrics.MeasureSince([]string{"leader", "replication", r.Delegate.MetricName(), "fetch"}, fetchStart)
+
+	if err != nil {
+		return 0, false, fmt.Errorf("failed to retrieve %s: %v", r.Delegate.PluralNoun(), err)
+	}
+
+	r.Logger.Printf("[DEBUG] replication: finished fetching %s: %d", r.Delegate.PluralNoun(), lenRemote)
+
+	// Need to check if we should be stopping. This will be common as the fetching process is a blocking
+	// RPC which could have been hanging around for a long time and during that time leadership could
+	// have been lost.
+	select {
+	case <-ctx.Done():
+		return 0, true, nil
+	default:
+		// do nothing
+	}
+
+	// Measure everything after the remote query, which can block for long
+	// periods of time. This metric is a good measure of how expensive the
+	// replication process is.
+	defer metrics.MeasureSince([]string{"leader", "replication", r.Delegate.MetricName(), "apply"}, time.Now())
+
+	lenLocal, local, err := r.Delegate.FetchLocal()
+	if err != nil {
+		return 0, false, fmt.Errorf("failed to retrieve local %s: %v", r.Delegate.PluralNoun(), err)
+	}
+
+	// If the remote index ever goes backwards, it's a good indication that
+	// the remote side was rebuilt and we should do a full sync since we
+	// can't make any assumptions about what's going on.
+	//
+	// Resetting lastRemoteIndex to 0 will work because we never consider local
+	// raft indices. Instead we compare the raft modify index in the response object
+	// with the lastRemoteIndex (only when we already have a config entry of the same kind/name)
+	// to determine if an update is needed. Resetting lastRemoteIndex to 0 then has the affect
+	// of making us think all the local state is out of date and any matching entries should
+	// still be updated.
+	//
+	// The lastRemoteIndex is not used when the entry exists either only in the local state or
+	// only in the remote state. In those situations we need to either delete it or create it.
+	if remoteIndex < lastRemoteIndex {
+		r.Logger.Printf("[WARN] replication: %[1]s replication remote index moved backwards (%d to %d), forcing a full %[1]s sync", r.Delegate.SingularNoun(), lastRemoteIndex, remoteIndex)
+		lastRemoteIndex = 0
+	}
+
+	r.Logger.Printf("[DEBUG] replication: %s replication - local: %d, remote: %d", r.Delegate.SingularNoun(), lenLocal, lenRemote)
+
+	// Calculate the changes required to bring the state into sync and then
+	// apply them.
+	diff, err := r.Delegate.DiffRemoteAndLocalState(local, remote, lastRemoteIndex)
+	if err != nil {
+		return 0, false, fmt.Errorf("failed to diff %s local and remote states: %v", r.Delegate.SingularNoun(), err)
+	}
+
+	r.Logger.Printf("[DEBUG] replication: %s replication - deletions: %d, updates: %d", r.Delegate.SingularNoun(), diff.NumDeletions, diff.NumUpdates)
+
+	if diff.NumDeletions > 0 {
+		r.Logger.Printf("[DEBUG] replication: %s replication - performing %d deletions", r.Delegate.SingularNoun(), diff.NumDeletions)
+
+		exit, err := r.Delegate.PerformDeletions(ctx, diff.Deletions)
+		if exit {
+			return 0, true, nil
+		}
+
+		if err != nil {
+			return 0, false, fmt.Errorf("failed to apply local %s deletions: %v", r.Delegate.SingularNoun(), err)
+		}
+		r.Logger.Printf("[DEBUG] replication: %s replication - finished deletions", r.Delegate.SingularNoun())
+	}
+
+	if diff.NumUpdates > 0 {
+		r.Logger.Printf("[DEBUG] replication: %s replication - performing %d updates", r.Delegate.SingularNoun(), diff.NumUpdates)
+
+		exit, err := r.Delegate.PerformUpdates(ctx, diff.Updates)
+		if exit {
+			return 0, true, nil
+		}
+
+		if err != nil {
+			return 0, false, fmt.Errorf("failed to apply local %s updates: %v", r.Delegate.SingularNoun(), err)
+		}
+		r.Logger.Printf("[DEBUG] replication: %s replication - finished updates", r.Delegate.SingularNoun())
+	}
+
+	// Return the index we got back from the remote side, since we've synced
+	// up with the remote state as of that index.
+	return remoteIndex, false, nil
+}