package nomad

import (
	"context"
	"crypto/tls"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/rpc"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/hashicorp/consul/agent/consul/autopilot"
	consulapi "github.com/hashicorp/consul/api"
	"github.com/hashicorp/consul/lib"
	multierror "github.com/hashicorp/go-multierror"
	lru "github.com/hashicorp/golang-lru"
	"github.com/hashicorp/nomad/command/agent/consul"
	"github.com/hashicorp/nomad/helper/codec"
	"github.com/hashicorp/nomad/helper/pool"
	"github.com/hashicorp/nomad/helper/stats"
	"github.com/hashicorp/nomad/helper/tlsutil"
	"github.com/hashicorp/nomad/nomad/deploymentwatcher"
	"github.com/hashicorp/nomad/nomad/drainer"
	"github.com/hashicorp/nomad/nomad/state"
	"github.com/hashicorp/nomad/nomad/structs"
	"github.com/hashicorp/nomad/nomad/structs/config"
	"github.com/hashicorp/nomad/scheduler"
	"github.com/hashicorp/raft"
	raftboltdb "github.com/hashicorp/raft-boltdb"
	"github.com/hashicorp/serf/serf"
)

const (
	// datacenterQueryLimit sets the max number of DCs that a Nomad
	// Server will query to find bootstrap_expect servers.
	datacenterQueryLimit = 25

	// maxStaleLeadership is the maximum time we will permit this Nomad
	// Server to go without seeing a valid Raft leader.
	maxStaleLeadership = 15 * time.Second

	// peersPollInterval is used as the polling interval between attempts
	// to query Consul for Nomad Servers.
	peersPollInterval = 45 * time.Second

	// peersPollJitter is used to provide a slight amount of variance to
	// the retry interval when querying Consul Servers
	peersPollJitterFactor = 2

	raftState         = "raft/"
	serfSnapshot      = "serf/snapshot"
	snapshotsRetained = 2

	// serverRPCCache controls how long we keep an idle connection open to a server
	serverRPCCache = 2 * time.Minute

	// serverMaxStreams controls how many idle streams we keep open to a server
	serverMaxStreams = 64

	// raftLogCacheSize is the maximum number of logs to cache in-memory.
	// This is used to reduce disk I/O for the recently committed entries.
	raftLogCacheSize = 512

	// raftRemoveGracePeriod is how long we wait to allow a RemovePeer
	// to replicate to gracefully leave the cluster.
	raftRemoveGracePeriod = 5 * time.Second

	// defaultConsulDiscoveryInterval is how often to poll Consul for new
	// servers if there is no leader.
	defaultConsulDiscoveryInterval time.Duration = 3 * time.Second

	// defaultConsulDiscoveryIntervalRetry is how often to poll Consul for
	// new servers if there is no leader and the last Consul query failed.
	defaultConsulDiscoveryIntervalRetry time.Duration = 9 * time.Second

	// aclCacheSize is the number of ACL objects to keep cached. ACLs have a parsing and
	// construction cost, so we keep the hot objects cached to reduce the ACL token resolution time.
	aclCacheSize = 512
)

// Server is Nomad server which manages the job queues,
// schedulers, and notification bus for agents.
type Server struct {
	config *Config

	logger *log.Logger

	// Connection pool to other Nomad servers
	connPool *pool.ConnPool

	// The raft instance is used among Nomad nodes within the
	// region to protect operations that require strong consistency
	leaderCh      <-chan bool
	raft          *raft.Raft
	raftLayer     *RaftLayer
	raftStore     *raftboltdb.BoltStore
	raftInmem     *raft.InmemStore
	raftTransport *raft.NetworkTransport

	// autopilot is the Autopilot instance for this server.
	autopilot *autopilot.Autopilot

	// fsm is the state machine used with Raft
	fsm *nomadFSM

	// rpcListener is used to listen for incoming connections
	rpcListener net.Listener
	listenerCh  chan struct{}

	// tlsWrap is used to wrap outbound connections using TLS. It should be
	// accessed using the lock.
	tlsWrap     tlsutil.RegionWrapper
	tlsWrapLock sync.RWMutex

	// rpcServer is the static RPC server that is used by the local agent.
	rpcServer *rpc.Server

	// clientRpcAdvertise is the advertised RPC address for Nomad clients to connect
	// to this server
	clientRpcAdvertise net.Addr

	// serverRpcAdvertise is the advertised RPC address for Nomad servers to connect
	// to this server
	serverRpcAdvertise net.Addr

	// rpcTLS is the TLS config for incoming TLS requests
	rpcTLS    *tls.Config
	rpcCancel context.CancelFunc

	// staticEndpoints is the set of static endpoints that can be reused across
	// all RPC connections
	staticEndpoints endpoints

	// streamingRpcs is the registry holding our streaming RPC handlers.
	streamingRpcs *structs.StreamingRpcRegistry

	// nodeConns is the set of multiplexed node connections we have keyed by
	// NodeID
	nodeConns     map[string]*nodeConnState
	nodeConnsLock sync.RWMutex

	// peers is used to track the known Nomad servers. This is
	// used for region forwarding and clustering.
	peers      map[string][]*serverParts
	localPeers map[raft.ServerAddress]*serverParts
	peerLock   sync.RWMutex

	// serf is the Serf cluster containing only Nomad
	// servers. This is used for multi-region federation
	// and automatic clustering within regions.
	serf *serf.Serf

	// reconcileCh is used to pass events from the serf handler
	// into the leader manager. Mostly used to handle when servers
	// join/leave from the region.
	reconcileCh chan serf.Member

	// eventCh is used to receive events from the serf cluster
	eventCh chan serf.Event

	// BlockedEvals is used to manage evaluations that are blocked on node
	// capacity changes.
	blockedEvals *BlockedEvals

	// deploymentWatcher is used to watch deployments and their allocations and
	// make the required calls to continue to transition the deployment.
	deploymentWatcher *deploymentwatcher.Watcher

	// nodeDrainer is used to drain allocations from nodes.
	nodeDrainer *drainer.NodeDrainer

	// evalBroker is used to manage the in-progress evaluations
	// that are waiting to be brokered to a sub-scheduler
	evalBroker *EvalBroker

	// periodicDispatcher is used to track and create evaluations for periodic jobs.
	periodicDispatcher *PeriodicDispatch

	// planQueue is used to manage the submitted allocation
	// plans that are waiting to be assessed by the leader
	planQueue *PlanQueue

	// heartbeatTimers track the expiration time of each heartbeat that has
	// a TTL. On expiration, the node status is updated to be 'down'.
	heartbeatTimers     map[string]*time.Timer
	heartbeatTimersLock sync.Mutex

	// consulCatalog is used for discovering other Nomad Servers via Consul
	consulCatalog consul.CatalogAPI

	// vault is the client for communicating with Vault.
	vault VaultClient

	// Worker used for processing
	workers []*Worker

	// aclCache is used to maintain the parsed ACL objects
	aclCache *lru.TwoQueueCache

	// leaderAcl is the management ACL token that is valid when resolved by the
	// current leader.
	leaderAcl     string
	leaderAclLock sync.Mutex

	// statsFetcher is used by autopilot to check the status of the other
	// Nomad router.
	statsFetcher *StatsFetcher

	// EnterpriseState is used to fill in state for Pro/Ent builds
	EnterpriseState

	left         bool
	shutdown     bool
	shutdownCh   chan struct{}
	shutdownLock sync.Mutex
}

// Holds the RPC endpoints
type endpoints struct {
	Status     *Status
	Node       *Node
	Job        *Job
	Eval       *Eval
	Plan       *Plan
	Alloc      *Alloc
	Deployment *Deployment
	Region     *Region
	Search     *Search
	Periodic   *Periodic
	System     *System
	Operator   *Operator
	ACL        *ACL
	Enterprise *EnterpriseEndpoints

	// Client endpoints
	ClientStats       *ClientStats
	FileSystem        *FileSystem
	ClientAllocations *ClientAllocations
}

// NewServer is used to construct a new Nomad server from the
// configuration, potentially returning an error
func NewServer(config *Config, consulCatalog consul.CatalogAPI, logger *log.Logger) (*Server, error) {
	// Check the protocol version
	if err := config.CheckVersion(); err != nil {
		return nil, err
	}

	// Create an eval broker
	evalBroker, err := NewEvalBroker(
		config.EvalNackTimeout,
		config.EvalNackInitialReenqueueDelay,
		config.EvalNackSubsequentReenqueueDelay,
		config.EvalDeliveryLimit)
	if err != nil {
		return nil, err
	}

	// Create a new blocked eval tracker.
	blockedEvals := NewBlockedEvals(evalBroker)

	// Create a plan queue
	planQueue, err := NewPlanQueue()
	if err != nil {
		return nil, err
	}

	// Configure TLS
	tlsConf := config.tlsConfig()
	incomingTLS, tlsWrap, err := getTLSConf(config.TLSConfig.EnableRPC, tlsConf)
	if err != nil {
		return nil, err
	}

	// Create the ACL object cache
	aclCache, err := lru.New2Q(aclCacheSize)
	if err != nil {
		return nil, err
	}

	// Create the server
	s := &Server{
		config:        config,
		consulCatalog: consulCatalog,
		connPool:      pool.NewPool(config.LogOutput, serverRPCCache, serverMaxStreams, tlsWrap),
		logger:        logger,
		tlsWrap:       tlsWrap,
		rpcServer:     rpc.NewServer(),
		streamingRpcs: structs.NewStreamingRpcRegistry(),
		nodeConns:     make(map[string]*nodeConnState),
		peers:         make(map[string][]*serverParts),
		localPeers:    make(map[raft.ServerAddress]*serverParts),
		reconcileCh:   make(chan serf.Member, 32),
		eventCh:       make(chan serf.Event, 256),
		evalBroker:    evalBroker,
		blockedEvals:  blockedEvals,
		planQueue:     planQueue,
		rpcTLS:        incomingTLS,
		aclCache:      aclCache,
		shutdownCh:    make(chan struct{}),
	}

	// Create the periodic dispatcher for launching periodic jobs.
	s.periodicDispatcher = NewPeriodicDispatch(s.logger, s)

	// Initialize the stats fetcher that autopilot will use.
	s.statsFetcher = NewStatsFetcher(logger, s.connPool, s.config.Region)

	// Setup Vault
	if err := s.setupVaultClient(); err != nil {
		s.Shutdown()
		s.logger.Printf("[ERR] nomad: failed to setup Vault client: %v", err)
		return nil, fmt.Errorf("Failed to setup Vault client: %v", err)
	}

	// Initialize the RPC layer
	if err := s.setupRPC(tlsWrap); err != nil {
		s.Shutdown()
		s.logger.Printf("[ERR] nomad: failed to start RPC layer: %s", err)
		return nil, fmt.Errorf("Failed to start RPC layer: %v", err)
	}

	// Initialize the Raft server
	if err := s.setupRaft(); err != nil {
		s.Shutdown()
		s.logger.Printf("[ERR] nomad: failed to start Raft: %s", err)
		return nil, fmt.Errorf("Failed to start Raft: %v", err)
	}

	// Initialize the wan Serf
	s.serf, err = s.setupSerf(config.SerfConfig, s.eventCh, serfSnapshot)
	if err != nil {
		s.Shutdown()
		s.logger.Printf("[ERR] nomad: failed to start serf WAN: %s", err)
		return nil, fmt.Errorf("Failed to start serf: %v", err)
	}

	// Initialize the scheduling workers
	if err := s.setupWorkers(); err != nil {
		s.Shutdown()
		s.logger.Printf("[ERR] nomad: failed to start workers: %s", err)
		return nil, fmt.Errorf("Failed to start workers: %v", err)
	}

	// Setup the Consul syncer
	if err := s.setupConsulSyncer(); err != nil {
		return nil, fmt.Errorf("failed to create server Consul syncer: %v", err)
	}

	// Setup the deployment watcher.
	if err := s.setupDeploymentWatcher(); err != nil {
		return nil, fmt.Errorf("failed to create deployment watcher: %v", err)
	}

	// Setup the node drainer.
	s.setupNodeDrainer()

	// Setup the enterprise state
	if err := s.setupEnterprise(config); err != nil {
		return nil, err
	}

	// Monitor leadership changes
	go s.monitorLeadership()

	// Start ingesting events for Serf
	go s.serfEventHandler()

	// start the RPC listener for the server
	s.startRPCListener()

	// Emit metrics for the eval broker
	go evalBroker.EmitStats(time.Second, s.shutdownCh)

	// Emit metrics for the plan queue
	go planQueue.EmitStats(time.Second, s.shutdownCh)

	// Emit metrics for the blocked eval tracker.
	go blockedEvals.EmitStats(time.Second, s.shutdownCh)

	// Emit metrics for the Vault client.
	go s.vault.EmitStats(time.Second, s.shutdownCh)

	// Emit metrics
	go s.heartbeatStats()

	// Start enterprise background workers
	s.startEnterpriseBackground()

	// Done
	return s, nil
}

// startRPCListener starts the server's the RPC listener
func (s *Server) startRPCListener() {
	ctx, cancel := context.WithCancel(context.Background())
	s.rpcCancel = cancel
	go s.listen(ctx)
}

// createRPCListener creates the server's RPC listener
func (s *Server) createRPCListener() (*net.TCPListener, error) {
	s.listenerCh = make(chan struct{})
	listener, err := net.ListenTCP("tcp", s.config.RPCAddr)
	if err != nil {
		s.logger.Printf("[ERR] nomad: error when initializing TLS listener %s", err)
		return listener, err
	}

	s.rpcListener = listener
	return listener, nil
}

// getTLSConf gets the server's TLS configuration based on the config supplied
// by the operator
func getTLSConf(enableRPC bool, tlsConf *tlsutil.Config) (*tls.Config, tlsutil.RegionWrapper, error) {
	var tlsWrap tlsutil.RegionWrapper
	var incomingTLS *tls.Config
	if enableRPC {
		tw, err := tlsConf.OutgoingTLSWrapper()
		if err != nil {
			return nil, nil, err
		}
		tlsWrap = tw

		itls, err := tlsConf.IncomingTLSConfig()
		if err != nil {
			return nil, nil, err
		}
		incomingTLS = itls
	}
	return incomingTLS, tlsWrap, nil
}

// reloadTLSConnections updates a server's TLS configuration and reloads RPC
// connections.
func (s *Server) reloadTLSConnections(newTLSConfig *config.TLSConfig) error {
	s.logger.Printf("[INFO] nomad: reloading server connections due to configuration changes")

	// Check if we can reload the RPC listener
	if s.rpcListener == nil || s.rpcCancel == nil {
		s.logger.Println("[WARN] nomad: Unable to reload configuration due to uninitialized rpc listner")
		return fmt.Errorf("can't reload uninitialized RPC listener")
	}

	tlsConf := tlsutil.NewTLSConfiguration(newTLSConfig)
	incomingTLS, tlsWrap, err := getTLSConf(newTLSConfig.EnableRPC, tlsConf)
	if err != nil {
		s.logger.Printf("[ERR] nomad: unable to reset TLS context %s", err)
		return err
	}

	// Store the new tls wrapper.
	s.tlsWrapLock.Lock()
	s.tlsWrap = tlsWrap
	s.tlsWrapLock.Unlock()

	// Keeping configuration in sync is important for other places that require
	// access to config information, such as rpc.go, where we decide on what kind
	// of network connections to accept depending on the server configuration
	s.config.TLSConfig = newTLSConfig

	// Kill any old listeners
	s.rpcCancel()

	s.rpcTLS = incomingTLS
	s.connPool.ReloadTLS(tlsWrap)

	if err := s.rpcListener.Close(); err != nil {
		s.logger.Printf("[ERR] nomad: Unable to close rpc listener %s", err)
		return err
	}

	// Wait for the old listener to exit
	<-s.listenerCh

	// Create the new listener with the update TLS config
	listener, err := s.createRPCListener()
	if err != nil {
		listener.Close()
		return err
	}

	// Start the new RPC listener
	s.startRPCListener()

	// Close and reload existing Raft connections
	wrapper := tlsutil.RegionSpecificWrapper(s.config.Region, tlsWrap)
	s.raftLayer.ReloadTLS(wrapper)
	s.raftTransport.CloseStreams()

	s.logger.Printf("[DEBUG] nomad: finished reloading server connections")
	return nil
}

// Shutdown is used to shutdown the server
func (s *Server) Shutdown() error {
	s.logger.Printf("[INFO] nomad: shutting down server")
	s.shutdownLock.Lock()
	defer s.shutdownLock.Unlock()

	if s.shutdown {
		return nil
	}

	s.shutdown = true
	close(s.shutdownCh)

	if s.serf != nil {
		s.serf.Shutdown()
	}

	if s.raft != nil {
		s.raftTransport.Close()
		s.raftLayer.Close()
		future := s.raft.Shutdown()
		if err := future.Error(); err != nil {
			s.logger.Printf("[WARN] nomad: Error shutting down raft: %s", err)
		}
		if s.raftStore != nil {
			s.raftStore.Close()
		}
	}

	// Shutdown the RPC listener
	if s.rpcListener != nil {
		s.rpcListener.Close()
	}

	// Close the connection pool
	s.connPool.Shutdown()

	// Close the fsm
	if s.fsm != nil {
		s.fsm.Close()
	}

	// Stop Vault token renewal
	if s.vault != nil {
		s.vault.Stop()
	}

	return nil
}

// IsShutdown checks if the server is shutdown
func (s *Server) IsShutdown() bool {
	select {
	case <-s.shutdownCh:
		return true
	default:
		return false
	}
}

// Leave is used to prepare for a graceful shutdown of the server
func (s *Server) Leave() error {
	s.logger.Printf("[INFO] nomad: server starting leave")
	s.left = true

	// Check the number of known peers
	numPeers, err := s.numPeers()
	if err != nil {
		s.logger.Printf("[ERR] nomad: failed to check raft peers: %v", err)
		return err
	}

	addr := s.raftTransport.LocalAddr()

	// If we are the current leader, and we have any other peers (cluster has multiple
	// servers), we should do a RemovePeer to safely reduce the quorum size. If we are
	// not the leader, then we should issue our leave intention and wait to be removed
	// for some sane period of time.
	isLeader := s.IsLeader()
	if isLeader && numPeers > 1 {
		minRaftProtocol, err := s.autopilot.MinRaftProtocol()
		if err != nil {
			return err
		}

		if minRaftProtocol >= 2 && s.config.RaftConfig.ProtocolVersion >= 3 {
			future := s.raft.RemoveServer(raft.ServerID(s.config.NodeID), 0, 0)
			if err := future.Error(); err != nil {
				s.logger.Printf("[ERR] nomad: failed to remove ourself as raft peer: %v", err)
			}
		} else {
			future := s.raft.RemovePeer(addr)
			if err := future.Error(); err != nil {
				s.logger.Printf("[ERR] nomad: failed to remove ourself as raft peer: %v", err)
			}
		}
	}

	// Leave the gossip pool
	if s.serf != nil {
		if err := s.serf.Leave(); err != nil {
			s.logger.Printf("[ERR] nomad: failed to leave Serf cluster: %v", err)
		}
	}

	// If we were not leader, wait to be safely removed from the cluster.
	// We must wait to allow the raft replication to take place, otherwise
	// an immediate shutdown could cause a loss of quorum.
	if !isLeader {
		left := false
		limit := time.Now().Add(raftRemoveGracePeriod)
		for !left && time.Now().Before(limit) {
			// Sleep a while before we check.
			time.Sleep(50 * time.Millisecond)

			// Get the latest configuration.
			future := s.raft.GetConfiguration()
			if err := future.Error(); err != nil {
				s.logger.Printf("[ERR] nomad: failed to get raft configuration: %v", err)
				break
			}

			// See if we are no longer included.
			left = true
			for _, server := range future.Configuration().Servers {
				if server.Address == addr {
					left = false
					break
				}
			}
		}

		// TODO (alexdadgar) With the old Raft library we used to force the
		// peers set to empty when a graceful leave occurred. This would
		// keep voting spam down if the server was restarted, but it was
		// dangerous because the peers was inconsistent with the logs and
		// snapshots, so it wasn't really safe in all cases for the server
		// to become leader. This is now safe, but the log spam is noisy.
		// The next new version of the library will have a "you are not a
		// peer stop it" behavior that should address this. We will have
		// to evaluate during the RC period if this interim situation is
		// not too confusing for operators.

		// TODO (alexdadgar) When we take a later new version of the Raft
		// library it won't try to complete replication, so this peer
		// may not realize that it has been removed. Need to revisit this
		// and the warning here.
		if !left {
			s.logger.Printf("[WARN] nomad: failed to leave raft configuration gracefully, timeout")
		}
	}
	return nil
}

// Reload handles a config reload specific to server-only configuration. Not
// all config fields can handle a reload.
func (s *Server) Reload(newConfig *Config) error {
	if newConfig == nil {
		return fmt.Errorf("Reload given a nil config")
	}

	var mErr multierror.Error

	// Handle the Vault reload. Vault should never be nil but just guard.
	if s.vault != nil {
		if err := s.vault.SetConfig(newConfig.VaultConfig); err != nil {
			multierror.Append(&mErr, err)
		}
	}

	tlsInfoEqual, err := newConfig.TLSConfig.CertificateInfoIsEqual(s.config.TLSConfig)
	if err != nil {
		s.logger.Printf("[ERR] nomad: error parsing server TLS configuration: %s", err)
		return err
	}

	if !tlsInfoEqual || newConfig.TLSConfig.EnableRPC != s.config.TLSConfig.EnableRPC {
		if err := s.reloadTLSConnections(newConfig.TLSConfig); err != nil {
			s.logger.Printf("[ERR] nomad: error reloading server TLS configuration: %s", err)
			multierror.Append(&mErr, err)
		}
	}

	return mErr.ErrorOrNil()
}

// setupBootstrapHandler() creates the closure necessary to support a Consul
// fallback handler.
func (s *Server) setupBootstrapHandler() error {
	// peersTimeout is used to indicate to the Consul Syncer that the
	// current Nomad Server has a stale peer set.  peersTimeout will time
	// out if the Consul Syncer bootstrapFn has not observed a Raft
	// leader in maxStaleLeadership.  If peersTimeout has been triggered,
	// the Consul Syncer will begin querying Consul for other Nomad
	// Servers.
	//
	// NOTE: time.Timer is used vs time.Time in order to handle clock
	// drift because time.Timer is implemented as a monotonic clock.
	var peersTimeout *time.Timer = time.NewTimer(0)

	// consulQueryCount is the number of times the bootstrapFn has been
	// called, regardless of success.
	var consulQueryCount uint64

	// leadershipTimedOut is a helper method that returns true if the
	// peersTimeout timer has expired.
	leadershipTimedOut := func() bool {
		select {
		case <-peersTimeout.C:
			return true
		default:
			return false
		}
	}

	// The bootstrapFn callback handler is used to periodically poll
	// Consul to look up the Nomad Servers in Consul.  In the event the
	// server has been brought up without a `retry-join` configuration
	// and this Server is partitioned from the rest of the cluster,
	// periodically poll Consul to reattach this Server to other servers
	// in the same region and automatically reform a quorum (assuming the
	// correct number of servers required for quorum are present).
	bootstrapFn := func() error {
		// If there is a raft leader, do nothing
		if s.raft.Leader() != "" {
			peersTimeout.Reset(maxStaleLeadership)
			return nil
		}

		// (ab)use serf.go's behavior of setting BootstrapExpect to
		// zero if we have bootstrapped.  If we have bootstrapped
		bootstrapExpect := atomic.LoadInt32(&s.config.BootstrapExpect)
		if bootstrapExpect == 0 {
			// This Nomad Server has been bootstrapped.  Rely on
			// the peersTimeout firing as a guard to prevent
			// aggressive querying of Consul.
			if !leadershipTimedOut() {
				return nil
			}
		} else {
			if consulQueryCount > 0 && !leadershipTimedOut() {
				return nil
			}

			// This Nomad Server has not been bootstrapped, reach
			// out to Consul if our peer list is less than
			// `bootstrap_expect`.
			raftPeers, err := s.numPeers()
			if err != nil {
				peersTimeout.Reset(peersPollInterval + lib.RandomStagger(peersPollInterval/peersPollJitterFactor))
				return nil
			}

			// The necessary number of Nomad Servers required for
			// quorum has been reached, we do not need to poll
			// Consul.  Let the normal timeout-based strategy
			// take over.
			if raftPeers >= int(bootstrapExpect) {
				peersTimeout.Reset(peersPollInterval + lib.RandomStagger(peersPollInterval/peersPollJitterFactor))
				return nil
			}
		}
		consulQueryCount++

		s.logger.Printf("[DEBUG] server.nomad: lost contact with Nomad quorum, falling back to Consul for server list")

		dcs, err := s.consulCatalog.Datacenters()
		if err != nil {
			peersTimeout.Reset(peersPollInterval + lib.RandomStagger(peersPollInterval/peersPollJitterFactor))
			return fmt.Errorf("server.nomad: unable to query Consul datacenters: %v", err)
		}
		if len(dcs) > 2 {
			// Query the local DC first, then shuffle the
			// remaining DCs.  If additional calls to bootstrapFn
			// are necessary, this Nomad Server will eventually
			// walk all datacenter until it finds enough hosts to
			// form a quorum.
			shuffleStrings(dcs[1:])
			dcs = dcs[0:lib.MinInt(len(dcs), datacenterQueryLimit)]
		}

		nomadServerServiceName := s.config.ConsulConfig.ServerServiceName
		var mErr multierror.Error
		const defaultMaxNumNomadServers = 8
		nomadServerServices := make([]string, 0, defaultMaxNumNomadServers)
		localNode := s.serf.Memberlist().LocalNode()
		for _, dc := range dcs {
			consulOpts := &consulapi.QueryOptions{
				AllowStale: true,
				Datacenter: dc,
				Near:       "_agent",
				WaitTime:   consul.DefaultQueryWaitDuration,
			}
			consulServices, _, err := s.consulCatalog.Service(nomadServerServiceName, consul.ServiceTagSerf, consulOpts)
			if err != nil {
				err := fmt.Errorf("failed to query service %q in Consul datacenter %q: %v", nomadServerServiceName, dc, err)
				s.logger.Printf("[WARN] server.nomad: %v", err)
				mErr.Errors = append(mErr.Errors, err)
				continue
			}

			for _, cs := range consulServices {
				port := strconv.FormatInt(int64(cs.ServicePort), 10)
				addr := cs.ServiceAddress
				if addr == "" {
					addr = cs.Address
				}
				if localNode.Addr.String() == addr && int(localNode.Port) == cs.ServicePort {
					continue
				}
				serverAddr := net.JoinHostPort(addr, port)
				nomadServerServices = append(nomadServerServices, serverAddr)
			}
		}

		if len(nomadServerServices) == 0 {
			if len(mErr.Errors) > 0 {
				peersTimeout.Reset(peersPollInterval + lib.RandomStagger(peersPollInterval/peersPollJitterFactor))
				return mErr.ErrorOrNil()
			}

			// Log the error and return nil so future handlers
			// can attempt to register the `nomad` service.
			pollInterval := peersPollInterval + lib.RandomStagger(peersPollInterval/peersPollJitterFactor)
			s.logger.Printf("[TRACE] server.nomad: no Nomad Servers advertising service %+q in Consul datacenters %+q, sleeping for %v", nomadServerServiceName, dcs, pollInterval)
			peersTimeout.Reset(pollInterval)
			return nil
		}

		numServersContacted, err := s.Join(nomadServerServices)
		if err != nil {
			peersTimeout.Reset(peersPollInterval + lib.RandomStagger(peersPollInterval/peersPollJitterFactor))
			return fmt.Errorf("contacted %d Nomad Servers: %v", numServersContacted, err)
		}

		peersTimeout.Reset(maxStaleLeadership)
		s.logger.Printf("[INFO] server.nomad: successfully contacted %d Nomad Servers", numServersContacted)

		return nil
	}

	// Hacky replacement for old ConsulSyncer Periodic Handler.
	go func() {
		lastOk := true
		sync := time.NewTimer(0)
		for {
			select {
			case <-sync.C:
				d := defaultConsulDiscoveryInterval
				if err := bootstrapFn(); err != nil {
					// Only log if it worked last time
					if lastOk {
						lastOk = false
						s.logger.Printf("[ERR] consul: error looking up Nomad servers: %v", err)
					}
					d = defaultConsulDiscoveryIntervalRetry
				}
				sync.Reset(d)
			case <-s.shutdownCh:
				return
			}
		}
	}()
	return nil
}

// setupConsulSyncer creates Server-mode consul.Syncer which periodically
// executes callbacks on a fixed interval.
func (s *Server) setupConsulSyncer() error {
	if s.config.ConsulConfig.ServerAutoJoin != nil && *s.config.ConsulConfig.ServerAutoJoin {
		if err := s.setupBootstrapHandler(); err != nil {
			return err
		}
	}

	return nil
}

// setupDeploymentWatcher creates a deployment watcher that consumes the RPC
// endpoints for state information and makes transitions via Raft through a
// shim that provides the appropriate methods.
func (s *Server) setupDeploymentWatcher() error {

	// Create the raft shim type to restrict the set of raft methods that can be
	// made
	raftShim := &deploymentWatcherRaftShim{
		apply: s.raftApply,
	}

	// Create the deployment watcher
	s.deploymentWatcher = deploymentwatcher.NewDeploymentsWatcher(
		s.logger, raftShim,
		deploymentwatcher.LimitStateQueriesPerSecond,
		deploymentwatcher.CrossDeploymentEvalBatchDuration)

	return nil
}

// setupNodeDrainer creates a node drainer which will be enabled when a server
// becomes a leader.
func (s *Server) setupNodeDrainer() {
	// Create a shim around Raft requests
	shim := drainerShim{s}
	c := &drainer.NodeDrainerConfig{
		Logger:                s.logger,
		Raft:                  shim,
		JobFactory:            drainer.GetDrainingJobWatcher,
		NodeFactory:           drainer.GetNodeWatcherFactory(),
		DrainDeadlineFactory:  drainer.GetDeadlineNotifier,
		StateQueriesPerSecond: drainer.LimitStateQueriesPerSecond,
		BatchUpdateInterval:   drainer.BatchUpdateInterval,
	}
	s.nodeDrainer = drainer.NewNodeDrainer(c)
}

// setupVaultClient is used to set up the Vault API client.
func (s *Server) setupVaultClient() error {
	v, err := NewVaultClient(s.config.VaultConfig, s.logger, s.purgeVaultAccessors)
	if err != nil {
		return err
	}
	s.vault = v
	return nil
}

// setupRPC is used to setup the RPC listener
func (s *Server) setupRPC(tlsWrap tlsutil.RegionWrapper) error {
	// Populate the static RPC server
	s.setupRpcServer(s.rpcServer, nil)

	listener, err := s.createRPCListener()
	if err != nil {
		listener.Close()
		return err
	}

	if s.config.ClientRPCAdvertise != nil {
		s.clientRpcAdvertise = s.config.ClientRPCAdvertise
	} else {
		s.clientRpcAdvertise = s.rpcListener.Addr()
	}

	// Verify that we have a usable advertise address
	clientAddr, ok := s.clientRpcAdvertise.(*net.TCPAddr)
	if !ok {
		listener.Close()
		return fmt.Errorf("Client RPC advertise address is not a TCP Address: %v", clientAddr)
	}
	if clientAddr.IP.IsUnspecified() {
		listener.Close()
		return fmt.Errorf("Client RPC advertise address is not advertisable: %v", clientAddr)
	}

	if s.config.ServerRPCAdvertise != nil {
		s.serverRpcAdvertise = s.config.ServerRPCAdvertise
	} else {
		// Default to the Serf Advertise + RPC Port
		serfIP := s.config.SerfConfig.MemberlistConfig.AdvertiseAddr
		if serfIP == "" {
			serfIP = s.config.SerfConfig.MemberlistConfig.BindAddr
		}

		addr := net.JoinHostPort(serfIP, fmt.Sprintf("%d", clientAddr.Port))
		resolved, err := net.ResolveTCPAddr("tcp", addr)
		if err != nil {
			return fmt.Errorf("Failed to resolve Server RPC advertise address: %v", err)
		}

		s.serverRpcAdvertise = resolved
	}

	// Verify that we have a usable advertise address
	serverAddr, ok := s.serverRpcAdvertise.(*net.TCPAddr)
	if !ok {
		return fmt.Errorf("Server RPC advertise address is not a TCP Address: %v", serverAddr)
	}
	if serverAddr.IP.IsUnspecified() {
		listener.Close()
		return fmt.Errorf("Server RPC advertise address is not advertisable: %v", serverAddr)
	}

	wrapper := tlsutil.RegionSpecificWrapper(s.config.Region, tlsWrap)
	s.raftLayer = NewRaftLayer(s.serverRpcAdvertise, wrapper)
	return nil
}

// setupRpcServer is used to populate an RPC server with endpoints
func (s *Server) setupRpcServer(server *rpc.Server, ctx *RPCContext) {
	// Add the static endpoints to the RPC server.
	if s.staticEndpoints.Status == nil {
		// Initialize the list just once
		s.staticEndpoints.ACL = &ACL{s}
		s.staticEndpoints.Alloc = &Alloc{s}
		s.staticEndpoints.Eval = &Eval{s}
		s.staticEndpoints.Job = &Job{s}
		s.staticEndpoints.Node = &Node{srv: s} // Add but don't register
		s.staticEndpoints.Deployment = &Deployment{srv: s}
		s.staticEndpoints.Operator = &Operator{s}
		s.staticEndpoints.Periodic = &Periodic{s}
		s.staticEndpoints.Plan = &Plan{s}
		s.staticEndpoints.Region = &Region{s}
		s.staticEndpoints.Status = &Status{s}
		s.staticEndpoints.System = &System{s}
		s.staticEndpoints.Search = &Search{s}
		s.staticEndpoints.Enterprise = NewEnterpriseEndpoints(s)

		// Client endpoints
		s.staticEndpoints.ClientStats = &ClientStats{s}
		s.staticEndpoints.ClientAllocations = &ClientAllocations{s}

		// Streaming endpoints
		s.staticEndpoints.FileSystem = &FileSystem{s}
		s.staticEndpoints.FileSystem.register()
	}

	// Register the static handlers
	server.Register(s.staticEndpoints.ACL)
	server.Register(s.staticEndpoints.Alloc)
	server.Register(s.staticEndpoints.Eval)
	server.Register(s.staticEndpoints.Job)
	server.Register(s.staticEndpoints.Deployment)
	server.Register(s.staticEndpoints.Operator)
	server.Register(s.staticEndpoints.Periodic)
	server.Register(s.staticEndpoints.Plan)
	server.Register(s.staticEndpoints.Region)
	server.Register(s.staticEndpoints.Status)
	server.Register(s.staticEndpoints.System)
	server.Register(s.staticEndpoints.Search)
	s.staticEndpoints.Enterprise.Register(server)
	server.Register(s.staticEndpoints.ClientStats)
	server.Register(s.staticEndpoints.ClientAllocations)
	server.Register(s.staticEndpoints.FileSystem)

	// Create new dynamic endpoints and add them to the RPC server.
	node := &Node{srv: s, ctx: ctx}

	// Register the dynamic endpoints
	server.Register(node)
}

// setupRaft is used to setup and initialize Raft
func (s *Server) setupRaft() error {
	// If we have an unclean exit then attempt to close the Raft store.
	defer func() {
		if s.raft == nil && s.raftStore != nil {
			if err := s.raftStore.Close(); err != nil {
				s.logger.Printf("[ERR] nomad: failed to close Raft store: %v", err)
			}
		}
	}()

	// Create the FSM
	fsmConfig := &FSMConfig{
		EvalBroker: s.evalBroker,
		Periodic:   s.periodicDispatcher,
		Blocked:    s.blockedEvals,
		LogOutput:  s.config.LogOutput,
		Region:     s.Region(),
	}
	var err error
	s.fsm, err = NewFSM(fsmConfig)
	if err != nil {
		return err
	}

	// Create a transport layer
	trans := raft.NewNetworkTransport(s.raftLayer, 3, s.config.RaftTimeout,
		s.config.LogOutput)
	s.raftTransport = trans

	// Make sure we set the LogOutput.
	s.config.RaftConfig.LogOutput = s.config.LogOutput

	// Our version of Raft protocol requires the LocalID to match the network
	// address of the transport.
	s.config.RaftConfig.LocalID = raft.ServerID(trans.LocalAddr())
	if s.config.RaftConfig.ProtocolVersion >= 3 {
		s.config.RaftConfig.LocalID = raft.ServerID(s.config.NodeID)
	}

	// Build an all in-memory setup for dev mode, otherwise prepare a full
	// disk-based setup.
	var log raft.LogStore
	var stable raft.StableStore
	var snap raft.SnapshotStore
	if s.config.DevMode {
		store := raft.NewInmemStore()
		s.raftInmem = store
		stable = store
		log = store
		snap = raft.NewDiscardSnapshotStore()

	} else {
		// Create the base raft path
		path := filepath.Join(s.config.DataDir, raftState)
		if err := ensurePath(path, true); err != nil {
			return err
		}

		// Create the BoltDB backend
		store, err := raftboltdb.NewBoltStore(filepath.Join(path, "raft.db"))
		if err != nil {
			return err
		}
		s.raftStore = store
		stable = store

		// Wrap the store in a LogCache to improve performance
		cacheStore, err := raft.NewLogCache(raftLogCacheSize, store)
		if err != nil {
			store.Close()
			return err
		}
		log = cacheStore

		// Create the snapshot store
		snapshots, err := raft.NewFileSnapshotStore(path, snapshotsRetained, s.config.LogOutput)
		if err != nil {
			if s.raftStore != nil {
				s.raftStore.Close()
			}
			return err
		}
		snap = snapshots

		// For an existing cluster being upgraded to the new version of
		// Raft, we almost never want to run recovery based on the old
		// peers.json file. We create a peers.info file with a helpful
		// note about where peers.json went, and use that as a sentinel
		// to avoid ingesting the old one that first time (if we have to
		// create the peers.info file because it's not there, we also
		// blow away any existing peers.json file).
		peersFile := filepath.Join(path, "peers.json")
		peersInfoFile := filepath.Join(path, "peers.info")
		if _, err := os.Stat(peersInfoFile); os.IsNotExist(err) {
			if err := ioutil.WriteFile(peersInfoFile, []byte(peersInfoContent), 0755); err != nil {
				return fmt.Errorf("failed to write peers.info file: %v", err)
			}

			// Blow away the peers.json file if present, since the
			// peers.info sentinel wasn't there.
			if _, err := os.Stat(peersFile); err == nil {
				if err := os.Remove(peersFile); err != nil {
					return fmt.Errorf("failed to delete peers.json, please delete manually (see peers.info for details): %v", err)
				}
				s.logger.Printf("[INFO] nomad: deleted peers.json file (see peers.info for details)")
			}
		} else if _, err := os.Stat(peersFile); err == nil {
			s.logger.Printf("[INFO] nomad: found peers.json file, recovering Raft configuration...")
			configuration, err := raft.ReadPeersJSON(peersFile)
			if err != nil {
				return fmt.Errorf("recovery failed to parse peers.json: %v", err)
			}
			tmpFsm, err := NewFSM(fsmConfig)
			if err != nil {
				return fmt.Errorf("recovery failed to make temp FSM: %v", err)
			}
			if err := raft.RecoverCluster(s.config.RaftConfig, tmpFsm,
				log, stable, snap, trans, configuration); err != nil {
				return fmt.Errorf("recovery failed: %v", err)
			}
			if err := os.Remove(peersFile); err != nil {
				return fmt.Errorf("recovery failed to delete peers.json, please delete manually (see peers.info for details): %v", err)
			}
			s.logger.Printf("[INFO] nomad: deleted peers.json file after successful recovery")
		}
	}

	// If we are in bootstrap or dev mode and the state is clean then we can
	// bootstrap now.
	if s.config.Bootstrap || s.config.DevMode {
		hasState, err := raft.HasExistingState(log, stable, snap)
		if err != nil {
			return err
		}
		if !hasState {
			configuration := raft.Configuration{
				Servers: []raft.Server{
					{
						ID:      s.config.RaftConfig.LocalID,
						Address: trans.LocalAddr(),
					},
				},
			}
			if err := raft.BootstrapCluster(s.config.RaftConfig,
				log, stable, snap, trans, configuration); err != nil {
				return err
			}
		}
	}

	// Setup the leader channel
	leaderCh := make(chan bool, 1)
	s.config.RaftConfig.NotifyCh = leaderCh
	s.leaderCh = leaderCh

	// Setup the Raft store
	s.raft, err = raft.NewRaft(s.config.RaftConfig, s.fsm, log, stable, snap, trans)
	if err != nil {
		return err
	}
	return nil
}

// setupSerf is used to setup and initialize a Serf
func (s *Server) setupSerf(conf *serf.Config, ch chan serf.Event, path string) (*serf.Serf, error) {
	conf.Init()
	conf.NodeName = fmt.Sprintf("%s.%s", s.config.NodeName, s.config.Region)
	conf.Tags["role"] = "nomad"
	conf.Tags["region"] = s.config.Region
	conf.Tags["dc"] = s.config.Datacenter
	conf.Tags["vsn"] = fmt.Sprintf("%d", structs.ApiMajorVersion)
	conf.Tags["mvn"] = fmt.Sprintf("%d", structs.ApiMinorVersion)
	conf.Tags["build"] = s.config.Build
	conf.Tags["raft_vsn"] = fmt.Sprintf("%d", s.config.RaftConfig.ProtocolVersion)
	conf.Tags["id"] = s.config.NodeID
	conf.Tags["rpc_addr"] = s.clientRpcAdvertise.(*net.TCPAddr).IP.String()         // Address that clients will use to RPC to servers
	conf.Tags["port"] = fmt.Sprintf("%d", s.serverRpcAdvertise.(*net.TCPAddr).Port) // Port servers use to RPC to one and another
	if s.config.Bootstrap || (s.config.DevMode && !s.config.DevDisableBootstrap) {
		conf.Tags["bootstrap"] = "1"
	}
	bootstrapExpect := atomic.LoadInt32(&s.config.BootstrapExpect)
	if bootstrapExpect != 0 {
		conf.Tags["expect"] = fmt.Sprintf("%d", bootstrapExpect)
	}
	if s.config.NonVoter {
		conf.Tags["nonvoter"] = "1"
	}
	if s.config.RedundancyZone != "" {
		conf.Tags[AutopilotRZTag] = s.config.RedundancyZone
	}
	if s.config.UpgradeVersion != "" {
		conf.Tags[AutopilotVersionTag] = s.config.UpgradeVersion
	}
	conf.MemberlistConfig.LogOutput = s.config.LogOutput
	conf.LogOutput = s.config.LogOutput
	conf.EventCh = ch
	if !s.config.DevMode {
		conf.SnapshotPath = filepath.Join(s.config.DataDir, path)
		if err := ensurePath(conf.SnapshotPath, false); err != nil {
			return nil, err
		}
	}
	conf.ProtocolVersion = protocolVersionMap[s.config.ProtocolVersion]
	conf.RejoinAfterLeave = true
	conf.Merge = &serfMergeDelegate{}

	// Until Nomad supports this fully, we disable automatic resolution.
	// When enabled, the Serf gossip may just turn off if we are the minority
	// node which is rather unexpected.
	conf.EnableNameConflictResolution = false
	return serf.Create(conf)
}

// setupWorkers is used to start the scheduling workers
func (s *Server) setupWorkers() error {
	// Check if all the schedulers are disabled
	if len(s.config.EnabledSchedulers) == 0 || s.config.NumSchedulers == 0 {
		s.logger.Printf("[WARN] nomad: no enabled schedulers")
		return nil
	}

	// Check if the core scheduler is not enabled
	foundCore := false
	for _, sched := range s.config.EnabledSchedulers {
		if sched == structs.JobTypeCore {
			foundCore = true
			continue
		}

		if _, ok := scheduler.BuiltinSchedulers[sched]; !ok {
			return fmt.Errorf("invalid configuration: unknown scheduler %q in enabled schedulers", sched)
		}
	}
	if !foundCore {
		return fmt.Errorf("invalid configuration: %q scheduler not enabled", structs.JobTypeCore)
	}

	// Start the workers
	for i := 0; i < s.config.NumSchedulers; i++ {
		if w, err := NewWorker(s); err != nil {
			return err
		} else {
			s.workers = append(s.workers, w)
		}
	}
	s.logger.Printf("[INFO] nomad: starting %d scheduling worker(s) for %v",
		s.config.NumSchedulers, s.config.EnabledSchedulers)
	return nil
}

// numPeers is used to check on the number of known peers, including the local
// node.
func (s *Server) numPeers() (int, error) {
	future := s.raft.GetConfiguration()
	if err := future.Error(); err != nil {
		return 0, err
	}
	configuration := future.Configuration()
	return len(configuration.Servers), nil
}

// IsLeader checks if this server is the cluster leader
func (s *Server) IsLeader() bool {
	return s.raft.State() == raft.Leader
}

// Join is used to have Nomad join the gossip ring
// The target address should be another node listening on the
// Serf address
func (s *Server) Join(addrs []string) (int, error) {
	return s.serf.Join(addrs, true)
}

// LocalMember is used to return the local node
func (c *Server) LocalMember() serf.Member {
	return c.serf.LocalMember()
}

// Members is used to return the members of the serf cluster
func (s *Server) Members() []serf.Member {
	return s.serf.Members()
}

// RemoveFailedNode is used to remove a failed node from the cluster
func (s *Server) RemoveFailedNode(node string) error {
	return s.serf.RemoveFailedNode(node)
}

// KeyManager returns the Serf keyring manager
func (s *Server) KeyManager() *serf.KeyManager {
	return s.serf.KeyManager()
}

// Encrypted determines if gossip is encrypted
func (s *Server) Encrypted() bool {
	return s.serf.EncryptionEnabled()
}

// State returns the underlying state store. This should *not*
// be used to modify state directly.
func (s *Server) State() *state.StateStore {
	return s.fsm.State()
}

// setLeaderAcl stores the given ACL token as the current leader's ACL token.
func (s *Server) setLeaderAcl(token string) {
	s.leaderAclLock.Lock()
	s.leaderAcl = token
	s.leaderAclLock.Unlock()
}

// getLeaderAcl retrieves the leader's ACL token
func (s *Server) getLeaderAcl() string {
	s.leaderAclLock.Lock()
	defer s.leaderAclLock.Unlock()
	return s.leaderAcl
}

// Regions returns the known regions in the cluster.
func (s *Server) Regions() []string {
	s.peerLock.RLock()
	defer s.peerLock.RUnlock()

	regions := make([]string, 0, len(s.peers))
	for region := range s.peers {
		regions = append(regions, region)
	}
	sort.Strings(regions)
	return regions
}

// RPC is used to make a local RPC call
func (s *Server) RPC(method string, args interface{}, reply interface{}) error {
	codec := &codec.InmemCodec{
		Method: method,
		Args:   args,
		Reply:  reply,
	}
	if err := s.rpcServer.ServeRequest(codec); err != nil {
		return err
	}
	return codec.Err
}

// StreamingRpcHandler is used to make a streaming RPC call.
func (s *Server) StreamingRpcHandler(method string) (structs.StreamingRpcHandler, error) {
	return s.streamingRpcs.GetHandler(method)
}

// Stats is used to return statistics for debugging and insight
// for various sub-systems
func (s *Server) Stats() map[string]map[string]string {
	toString := func(v uint64) string {
		return strconv.FormatUint(v, 10)
	}
	stats := map[string]map[string]string{
		"nomad": {
			"server":        "true",
			"leader":        fmt.Sprintf("%v", s.IsLeader()),
			"leader_addr":   string(s.raft.Leader()),
			"bootstrap":     fmt.Sprintf("%v", s.config.Bootstrap),
			"known_regions": toString(uint64(len(s.peers))),
		},
		"raft":    s.raft.Stats(),
		"serf":    s.serf.Stats(),
		"runtime": stats.RuntimeStats(),
	}

	return stats
}

// Region returns the region of the server
func (s *Server) Region() string {
	return s.config.Region
}

// Datacenter returns the data center of the server
func (s *Server) Datacenter() string {
	return s.config.Datacenter
}

// GetConfig returns the config of the server for testing purposes only
func (s *Server) GetConfig() *Config {
	return s.config
}

// ReplicationToken returns the token used for replication. We use a method to support
// dynamic reloading of this value later.
func (s *Server) ReplicationToken() string {
	return s.config.ReplicationToken
}

// peersInfoContent is used to help operators understand what happened to the
// peers.json file. This is written to a file called peers.info in the same
// location.
const peersInfoContent = `
As of Nomad 0.5.5, the peers.json file is only used for recovery
after an outage. It should be formatted as a JSON array containing the address
and port (RPC) of each Nomad server in the cluster, like this:

["10.1.0.1:4647","10.1.0.2:4647","10.1.0.3:4647"]

Under normal operation, the peers.json file will not be present.

When Nomad starts for the first time, it will create this peers.info file and
delete any existing peers.json file so that recovery doesn't occur on the first
startup.

Once this peers.info file is present, any peers.json file will be ingested at
startup, and will set the Raft peer configuration manually to recover from an
outage. It's crucial that all servers in the cluster are shut down before
creating the peers.json file, and that all servers receive the same
configuration. Once the peers.json file is successfully ingested and applied, it
will be deleted.

Please see https://www.nomadproject.io/guides/outage.html for more information.
`