open-consul/consul/acl_replication.go

package consul

import (
	"fmt"
	"sort"
	"time"

	"github.com/armon/go-metrics"
	"github.com/hashicorp/consul/consul/structs"
	"github.com/hashicorp/consul/lib"
)

// aclIDSorter is used to make sure a given list of ACLs is sorted by token ID.
// This should always be true, but since this is crucial for correctness and we
// are accepting input from another server, we sort to make sure.
type aclIDSorter struct {
	acls structs.ACLs
}

// See sort.Interface.
func (a *aclIDSorter) Len() int {
	return len(a.acls)
}

// See sort.Interface.
func (a *aclIDSorter) Swap(i, j int) {
	a.acls[i], a.acls[j] = a.acls[j], a.acls[i]
}

// See sort.Interface.
func (a *aclIDSorter) Less(i, j int) bool {
	return a.acls[i].ID < a.acls[j].ID
}

// aclIterator simplifies the algorithm below by providing a basic iterator that
// moves through a list of ACLs and returns nil when it's exhausted.
type aclIterator struct {
	acls structs.ACLs

	// index is the current position of the iterator.
	index int
}

// Front returns the item at index position, or nil if the list is exhausted.
func (a *aclIterator) Front() *structs.ACL {
	if a.index < len(a.acls) {
		return a.acls[a.index]
	}

	return nil
}

// Next advances the iterator to the next index.
func (a *aclIterator) Next() {
	a.index++
}

// reconcileACLs takes the local and remote ACL state, and produces a list of
// changes required in order to bring the local ACLs into sync with the remote
// ACLs. You can supply lastRemoteIndex as a hint that replication has succeeded
// up to that remote index and it will make this process more efficient by only
// comparing ACL entries modified after that index. Setting this to 0 will force
// a full compare of all existing ACLs.
func reconcileACLs(local, remote structs.ACLs, lastRemoteIndex uint64) structs.ACLRequests {
	// Since sorting the lists is crucial for correctness, we are depending
	// on data coming from other servers potentially running a different,
	// version of Consul, and sorted-ness is kind of a subtle property of
	// the state store indexing, it's prudent to make sure things are sorted
	// before we begin.
	sort.Sort(&aclIDSorter{local})
	sort.Sort(&aclIDSorter{remote})

	// Run through both lists and reconcile them.
	var changes structs.ACLRequests
	localIter, remoteIter := &aclIterator{local, 0}, &aclIterator{remote, 0}
	for localIter.Front() != nil || remoteIter.Front() != nil {
		// If the local list is exhausted, then process this as a remote
		// add. We know from the loop condition that there's something
		// in the remote list.
		if localIter.Front() == nil {
			changes = append(changes, &structs.ACLRequest{
				Op:  structs.ACLSet,
				ACL: *(remoteIter.Front()),
			})
			remoteIter.Next()
			continue
		}

		// If the remote list is exhausted, then process this as a local
		// delete. We know from the loop condition that there's something
		// in the local list.
		if remoteIter.Front() == nil {
			changes = append(changes, &structs.ACLRequest{
				Op:  structs.ACLDelete,
				ACL: *(localIter.Front()),
			})
			localIter.Next()
			continue
		}

		// At this point we know there's something at the front of each
		// list we need to resolve.

		// If the remote list has something local doesn't, we add it.
		if localIter.Front().ID > remoteIter.Front().ID {
			changes = append(changes, &structs.ACLRequest{
				Op:  structs.ACLSet,
				ACL: *(remoteIter.Front()),
			})
			remoteIter.Next()
			continue
		}

		// If local has something remote doesn't, we delete it.
		if localIter.Front().ID < remoteIter.Front().ID {
			changes = append(changes, &structs.ACLRequest{
				Op:  structs.ACLDelete,
				ACL: *(localIter.Front()),
			})
			localIter.Next()
			continue
		}

		// Local and remote have an ACL with the same ID, so we might
		// need to compare them.
		l, r := localIter.Front(), remoteIter.Front()
		if r.RaftIndex.ModifyIndex > lastRemoteIndex && !r.IsSame(l) {
			changes = append(changes, &structs.ACLRequest{
				Op:  structs.ACLSet,
				ACL: *r,
			})
		}
		localIter.Next()
		remoteIter.Next()
	}
	return changes
}

// FetchLocalACLs returns the ACLs in the local state store.
func (s *Server) fetchLocalACLs() (structs.ACLs, error) {
	_, local, err := s.fsm.State().ACLList()
	if err != nil {
		return nil, err
	}
	return local, nil
}

// FetchRemoteACLs is used to get the remote set of ACLs from the ACL
// datacenter. The lastIndex parameter is a hint about which remote index we
// have replicated to, so this is expected to block until something changes.
func (s *Server) fetchRemoteACLs(lastRemoteIndex uint64) (*structs.IndexedACLs, error) {
	args := structs.DCSpecificRequest{
		Datacenter: s.config.ACLDatacenter,
		QueryOptions: structs.QueryOptions{
			Token:         s.config.ACLReplicationToken,
			MinQueryIndex: lastRemoteIndex,
			AllowStale:    true,
		},
	}
	var remote structs.IndexedACLs
	if err := s.RPC("ACL.List", &args, &remote); err != nil {
		return nil, err
	}
	return &remote, nil
}

// UpdateLocalACLs is given a list of changes to apply in order to bring the
// local ACLs in-line with the remote ACLs from the ACL datacenter.
func (s *Server) updateLocalACLs(changes structs.ACLRequests) error {
	var ops int
	start := time.Now()
	for _, change := range changes {
		// Do a very simple rate limit algorithm where we check every N
		// operations and wait out to the second before we continue. If
		// it's going slower than that, the sleep time will be negative
		// so we will just keep going without delay.
		if ops > s.config.ACLReplicationApplyLimit {
			elapsed := time.Now().Sub(start)
			time.Sleep(1*time.Second - elapsed)
			ops, start = 0, time.Now()
		}
		ops++

		// Note that we are using the single ACL interface here and not
		// performing all this inside a single transaction. This is OK
		// for two reasons. First, there's nothing else other than this
		// replication routine that alters the local ACLs, so there's
		// nothing to contend with locally. Second, if an apply fails
		// in the middle (most likely due to losing leadership), the
		// next replication pass will clean up and check everything
		// again.
		var reply string
		if err := aclApplyInternal(s, change, &reply); err != nil {
			return err
		}
	}
	return nil
}

// replicateACLs is a runs one pass of the algorithm for replicating ACLs from
// a remote ACL datacenter to local state. If there's any error, this will return
// 0 for the lastRemoteIndex, which will cause us to immediately do a full sync
// next time.
func (s *Server) replicateACLs(lastRemoteIndex uint64) (uint64, error) {
	remote, err := s.fetchRemoteACLs(lastRemoteIndex)
	if err != nil {
		return 0, fmt.Errorf("failed to retrieve remote ACLs: %v", err)
	}

	// This will be pretty common because we will be blocking for a long time
	// and may have lost leadership, so lets control the message here instead
	// of returning deeper error messages from from Raft.
	if !s.IsLeader() {
		return 0, fmt.Errorf("no longer cluster leader")
	}

	// 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{"consul", "leader", "replicateACLs"}, time.Now())

	local, err := s.fetchLocalACLs()
	if err != nil {
		return 0, fmt.Errorf("failed to retrieve local ACLs: %v", 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.
	if remote.QueryMeta.Index < lastRemoteIndex {
		s.logger.Printf("[WARN] consul: ACL replication remote index moved backwards (%d to %d), forcing a full ACL sync", lastRemoteIndex, remote.QueryMeta.Index)
		lastRemoteIndex = 0
	}

	// Calculate the changes required to bring the state into sync and then
	// apply them.
	changes := reconcileACLs(local, remote.ACLs, lastRemoteIndex)
	if err := s.updateLocalACLs(changes); err != nil {
		return 0, fmt.Errorf("failed to sync ACL changes: %v", err)
	}

	// Return the index we got back from the remote side, since we've synced
	// up with the remote state as of that index.
	return remote.QueryMeta.Index, nil
}

// IsACLReplicationEnabled returns true if ACL replication is enabled.
func (s *Server) IsACLReplicationEnabled() bool {
	authDC := s.config.ACLDatacenter
	return len(authDC) > 0 && (authDC != s.config.Datacenter) &&
		len(s.config.ACLReplicationToken) > 0
}

// runACLReplication is a long-running goroutine that will attempt to replicate
// ACLs while the server is the leader, until the shutdown channel closes.
func (s *Server) runACLReplication() {
	// Give each server's replicator a random initial phase for good
	// measure.
	select {
	case <-time.After(lib.RandomStagger(s.config.ACLReplicationInterval)):
	case <-s.shutdownCh:
	}

	var lastRemoteIndex uint64
	var wasActive bool
	replicate := func() {
		if !wasActive {
			s.logger.Printf("[INFO] consul: ACL replication started")
			wasActive = true
		}

		var err error
		lastRemoteIndex, err = s.replicateACLs(lastRemoteIndex)
		if err != nil {
			s.logger.Printf("[WARN] consul: ACL replication error (will retry if still leader): %v", err)
		} else {
			s.logger.Printf("[DEBUG] consul: ACL replication completed through index %d", lastRemoteIndex)
		}
	}
	pause := func() {
		if wasActive {
			s.logger.Printf("[INFO] consul: ACL replication stopped (no longer leader)")
			wasActive = false
		}
	}

	// This will slowly poll to see if replication should be active. Once it
	// is and we've caught up, the replicate() call will begin to block and
	// only wake up when the query timer expires or there are new ACLs to
	// replicate. We've chosen this design so that the ACLReplicationInterval
	// is the lower bound for how quickly we will replicate, no matter how
	// much ACL churn is happening on the remote side.
	//
	// The blocking query inside replicate() respects the shutdown channel,
	// so we won't get stuck in here as things are torn down.
	for {
		select {
		case <-s.shutdownCh:
			return

		case <-time.After(s.config.ACLReplicationInterval):
			if s.IsLeader() {
				replicate()
			} else {
				pause()
			}
		}
	}
}