// Lead is executed by the "leader" of the control center cluster to handle its management responsibilities of:
//    services
//    snapshots
//    virtual IPs
968756
func Lead(shutdown <-chan interface{}, conn coordclient.Connection, cpClient dao.ControlPlane, facade *facade.Facade, poolID string) {

	// creates a listener for the host registry
	unassignmentHandler := zkservice.NewZKHostUnassignmentHandler(conn)
	hreg := zkservice.NewHostRegistryListener(poolID, unassignmentHandler)

	leader := leader{shutdown, conn, cpClient, facade, poolID, hreg}

	// creates a listener for services
	serviceListener := zkservice.NewServiceListener(poolID, &leader)

	// starts all of the listeners
	zzk.Start(shutdown, conn, serviceListener, hreg)
}
280119
func (i *MySQLLock) becomeLeader() error {
	_, err := i.statements["put"].Exec("leader", i.value)
	if err != nil {
		return err
	}

	return nil
}
1337622
func (c *Core) Leader() (isLeader bool, leaderAddr, clusterAddr string, err error) {
	// Check if HA enabled. We don't need the lock for this check as it's set
	// on startup and never modified
	if c.ha == nil {
		return false, "", "", ErrHANotEnabled
	}

	// Check if sealed
	if c.Sealed() {
		return false, "", "", consts.ErrSealed
	}

	c.stateLock.RLock()

	// Check if we are the leader
	if !c.standby {
		c.stateLock.RUnlock()
		return true, c.redirectAddr, c.clusterAddr, nil
	}

	// Initialize a lock
	lock, err := c.ha.LockWith(CoreLockPath, "read")
	if err != nil {
		c.stateLock.RUnlock()
		return false, "", "", err
	}

	// Read the value
	held, leaderUUID, err := lock.Value()
	if err != nil {
		c.stateLock.RUnlock()
		return false, "", "", err
	}
	if !held {
		c.stateLock.RUnlock()
		return false, "", "", nil
	}

	var localLeaderUUID, localRedirectAddr, localClusterAddr string
	clusterLeaderParams := c.clusterLeaderParams.Load().(*ClusterLeaderParams)
	if clusterLeaderParams != nil {
		localLeaderUUID = clusterLeaderParams.LeaderUUID
		localRedirectAddr = clusterLeaderParams.LeaderRedirectAddr
		localClusterAddr = clusterLeaderParams.LeaderClusterAddr
	}

	// If the leader hasn't changed, return the cached value; nothing changes
	// mid-leadership, and the barrier caches anyways
	if leaderUUID == localLeaderUUID && localRedirectAddr != "" {
		c.stateLock.RUnlock()
		return false, localRedirectAddr, localClusterAddr, nil
	}

	c.logger.Trace("found new active node information, refreshing")

	defer c.stateLock.RUnlock()
	c.leaderParamsLock.Lock()
	defer c.leaderParamsLock.Unlock()

	// Validate base conditions again
	clusterLeaderParams = c.clusterLeaderParams.Load().(*ClusterLeaderParams)
	if clusterLeaderParams != nil {
		localLeaderUUID = clusterLeaderParams.LeaderUUID
		localRedirectAddr = clusterLeaderParams.LeaderRedirectAddr
		localClusterAddr = clusterLeaderParams.LeaderClusterAddr
	} else {
		localLeaderUUID = ""
		localRedirectAddr = ""
		localClusterAddr = ""
	}

	if leaderUUID == localLeaderUUID && localRedirectAddr != "" {
		return false, localRedirectAddr, localClusterAddr, nil
	}

	key := coreLeaderPrefix + leaderUUID
	// Use background because postUnseal isn't run on standby
	entry, err := c.barrier.Get(context.Background(), key)
	if err != nil {
		return false, "", "", err
	}
	if entry == nil {
		return false, "", "", nil
	}

	var oldAdv bool

	var adv activeAdvertisement
	err = jsonutil.DecodeJSON(entry.Value, &adv)
	if err != nil {
		// Fall back to pre-struct handling
		adv.RedirectAddr = string(entry.Value)
		c.logger.Debug("parsed redirect addr for new active node", "redirect_addr", adv.RedirectAddr)
		oldAdv = true
	}

	if !oldAdv {
		c.logger.Debug("parsing information for new active node", "active_cluster_addr", adv.ClusterAddr, "active_redirect_addr", adv.RedirectAddr)

		// Ensure we are using current values
		err = c.loadLocalClusterTLS(adv)
		if err != nil {
			return false, "", "", err
		}

		// This will ensure that we both have a connection at the ready and that
		// the address is the current known value
		// Since this is standby, we don't use the active context. Later we may
		// use a process-scoped context
		err = c.refreshRequestForwardingConnection(context.Background(), adv.ClusterAddr)
		if err != nil {
			return false, "", "", err
		}
	}

	// Don't set these until everything has been parsed successfully or we'll
	// never try again
	c.clusterLeaderParams.Store(&ClusterLeaderParams{
		LeaderUUID:         leaderUUID,
		LeaderRedirectAddr: adv.RedirectAddr,
		LeaderClusterAddr:  adv.ClusterAddr,
	})

	return false, adv.RedirectAddr, adv.ClusterAddr, nil
}
1141276
func (g *Group) HEAD(path string, handler HandlerFunc) {
	g.Handle("HEAD", path, handler)
}
210252
func (r *marathonClient) Leader() (string, error) {
	var leader struct {
		Leader string `json:"leader"`
	}
	if err := r.apiGet(marathonAPILeader, nil, &leader); err != nil {
		return "", err
	}

	return leader.Leader, nil
}
426916
func (s *server) Command(cmd string, args [][]byte) <-chan Result {

	if s.IsLeader() {
		cb := s.state.newCommand()
		cmdBytes := bytesForCommand(cb.originAddr, cb.reqNo, cmd, args)
		s.raft.Apply(cmdBytes, commandTimeout)
		return cb.result
	}
	// couldn't exec as leader, fallback to forwarding
	cb, err := s.dispatchToLeader(cmd, args)
	if err != nil {
		if cb != nil {
			cb.cancel()
		}
		ret := make(chan Result, 1)
		ret <- Result{nil, err}
		return ret
	}
	return cb.result
}
503469
func (s *Server) forward(method string, info structs.RPCInfo, args interface{}, reply interface{}) (bool, error) {
	var firstCheck time.Time

	// Handle DC forwarding
	dc := info.RequestDatacenter()
	if dc != s.config.Datacenter {
		err := s.forwardDC(method, dc, args, reply)
		return true, err
	}

	// Check if we can allow a stale read, ensure our local DB is initialized
	if info.IsRead() && info.AllowStaleRead() && !s.raft.LastContact().IsZero() {
		return false, nil
	}

CHECK_LEADER:
	// Fail fast if we are in the process of leaving
	select {
	case <-s.leaveCh:
		return true, structs.ErrNoLeader
	default:
	}

	// Find the leader
	isLeader, leader := s.getLeader()

	// Handle the case we are the leader
	if isLeader {
		return false, nil
	}

	// Handle the case of a known leader
	rpcErr := structs.ErrNoLeader
	if leader != nil {
		rpcErr = s.connPool.RPC(s.config.Datacenter, leader.Addr,
			leader.Version, method, leader.UseTLS, args, reply)
		if rpcErr != nil && canRetry(info, rpcErr) {
			goto RETRY
		}
		return true, rpcErr
	}

RETRY:
	// Gate the request until there is a leader
	if firstCheck.IsZero() {
		firstCheck = time.Now()
	}
	if time.Since(firstCheck) < s.config.RPCHoldTimeout {
		jitter := lib.RandomStagger(s.config.RPCHoldTimeout / jitterFraction)
		select {
		case <-time.After(jitter):
			goto CHECK_LEADER
		case <-s.leaveCh:
		case <-s.shutdownCh:
		}
	}

	// No leader found and hold time exceeded
	return true, rpcErr
}
271983
func NewLeadershipService(
	claimer leadership.Claimer, authorizer facade.Authorizer,
) (LeadershipService, error) {

	if !authorizer.AuthUnitAgent() && !authorizer.AuthApplicationAgent() {
		return nil, errors.Unauthorizedf("permission denied")
	}

	return &leadershipService{
		claimer:    claimer,
		authorizer: authorizer,
	}, nil
}
1158091
func (r *rpcHandler) forward(method string, info structs.RPCInfo, args interface{}, reply interface{}) (bool, error) {
	var firstCheck time.Time

	region := info.RequestRegion()
	if region == "" {
		return true, fmt.Errorf("missing target RPC")
	}

	// Handle region forwarding
	if region != r.config.Region {
		// Mark that we are forwarding the RPC
		info.SetForwarded()
		err := r.forwardRegion(region, method, args, reply)
		return true, err
	}

	// Check if we can allow a stale read
	if info.IsRead() && info.AllowStaleRead() {
		return false, nil
	}

CHECK_LEADER:
	// Find the leader
	isLeader, remoteServer := r.getLeader()

	// Handle the case we are the leader
	if isLeader {
		return false, nil
	}

	// Handle the case of a known leader
	if remoteServer != nil {
		// Mark that we are forwarding the RPC
		info.SetForwarded()
		err := r.forwardLeader(remoteServer, method, args, reply)
		return true, err
	}

	// Gate the request until there is a leader
	if firstCheck.IsZero() {
		firstCheck = time.Now()
	}
	if time.Now().Sub(firstCheck) < r.config.RPCHoldTimeout {
		jitter := lib.RandomStagger(r.config.RPCHoldTimeout / structs.JitterFraction)
		select {
		case <-time.After(jitter):
			goto CHECK_LEADER
		case <-r.shutdownCh:
		}
	}

	// No leader found and hold time exceeded
	return true, structs.ErrNoLeader
}
208029
func (s *Status) Leader(args *structs.GenericRequest, reply *string) error {
	if args.Region == "" {
		args.Region = s.srv.config.Region
	}
	if done, err := s.srv.forward("Status.Leader", args, args, reply); done {
		return err
	}

	leader := string(s.srv.raft.Leader())
	if leader != "" {
		*reply = leader
	} else {
		*reply = ""
	}
	return nil
}
1243494
func GetWatchLeader(conn zkhelper.Conn, rootPath string) (string, <-chan zk.Event, error) {
	data, _, watcher, err := conn.GetW(getLeaderPath(rootPath))
	if err != nil {
		return "", nil, errors.Trace(err)
	}
	addr, err := getLeader(data)
	if err != nil {
		return "", nil, errors.Trace(err)
	}

	// if err != checkLeaderExists(conn); err != nil {
	//  return "", errors.Trace(err)
	// }

	return addr, watcher, nil
}
1250759
func (n *Node) LeaderConn(ctx context.Context) (*grpc.ClientConn, error) {
	cc, err := n.getLeaderConn()
	if err == nil {
		return cc, nil
	}
	if err == raftselector.ErrIsLeader {
		return nil, err
	}
	if atomic.LoadUint32(&n.ticksWithNoLeader) > lostQuorumTimeout {
		return nil, errLostQuorum
	}

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			cc, err := n.getLeaderConn()
			if err == nil {
				return cc, nil
			}
			if err == raftselector.ErrIsLeader {
				return nil, err
			}
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
}
1151482
func NewLeaderListener(path string) *LeaderListener {
	return &LeaderListener{
		path:     path,
		mu:       &sync.Mutex{},
		watchers: make([]watcher, 0),
	}
}
597623
func (t *Tracker) setLeader(untilTime time.Time) error {
	if t.isMinion {
		// If we were a minion, we're now the leader, so we can record the transition.
		logger.Infof("%s promoted to leadership of %s", t.unitName, t.applicationName)
	}
	logger.Tracef("%s confirmed for %s leadership until %s", t.unitName, t.applicationName, untilTime)
	renewTime := untilTime.Add(-t.duration)
	logger.Tracef("%s will renew %s leadership at %s", t.unitName, t.applicationName, renewTime)
	t.isMinion = false
	t.claimLease = nil
	t.renewLease = t.clock.After(renewTime.Sub(t.clock.Now()))

	for len(t.waitingLeader) > 0 {
		logger.Tracef("notifying %s ticket of impending %s leadership", t.unitName, t.applicationName)
		var ticketCh chan bool
		ticketCh, t.waitingLeader = t.waitingLeader[0], t.waitingLeader[1:]
		defer close(ticketCh)
		if err := t.sendTrue(ticketCh); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}
219873
func (m *Main) waitForLeader(timeout time.Duration) error {
	// Ignore leadership if we are a proxy.
	if m.store == nil {
		return nil
	}
	var timeoutCh <-chan time.Time
	if timeout != IndefiniteTimeout {
		timeoutCh = time.After(timeout)
	}
	for {
		select {
		case <-timeoutCh:
			return errors.New("timed out waiting for leader")
		case <-time.After(100 * time.Millisecond):
			if leader := m.store.Leader(); leader != "" {
				return nil
			}
		}
	}
}
1036161
func (s *scheduler) mainloop(conn coordclient.Connection) {
	// become the leader
	leader, err := conn.NewLeader("/scheduler")
	if err != nil {
		glog.Errorf("Could not initialize leader node for scheduler: %s", err)
		return
	}
	leaderDone := make(chan struct{})
	defer close(leaderDone)
	event, err := leader.TakeLead(&zzk.HostLeader{HostID: s.instance_id, Realm: s.realm}, leaderDone)
	if err != nil {
		glog.Errorf("Could not become the leader: %s", err)
		return
	}
	defer leader.ReleaseLead()

	// did I shut down before I became the leader?
	select {
	case <-s.shutdown:
		return
	default:
	}

	var wg sync.WaitGroup
	defer wg.Wait()

	_shutdown := make(chan interface{})
	defer close(_shutdown)

	stopped := make(chan struct{}, 2)

	// ensure all the services are unlocked
	glog.Infof("Resetting service locks")
	locker := s.facade.DFSLock(datastore.Get())
	locker.Lock("reset service locks")
	if err := s.facade.ResetLocks(datastore.Get()); err != nil {
		glog.Errorf("Could not reset dfs locks: %s", err)
		return
	}
	locker.Unlock()
	glog.Infof("DFS locks are all reset")

	// start the storage server
	wg.Add(1)
	go func() {
		defer glog.Infof("Stopping storage sync")
		defer wg.Done()
		if err := s.storageServer.Run(_shutdown, conn); err != nil {
			glog.Errorf("Could not maintain storage lead: %s", err)
			stopped <- struct{}{}
		}
	}()

	// synchronize locally
	wg.Add(1)
	go func() {
		defer glog.Infof("Stopping local sync")
		defer wg.Done()
		s.localSync(_shutdown, conn)
	}()

	wg.Add(1)
	go func() {
		defer glog.Infof("Stopping pool listeners")
		defer wg.Done()
		zzk.Start(_shutdown, conn, s, s.pushreg)
		stopped <- struct{}{}
	}()

	// kicks off the snapshot cleaning goroutine
	if s.snapshotTTL > 0 {
		wg.Add(1)
		go func() {
			defer glog.Infof("Stopping snapshot ttl")
			defer wg.Done()
			ttl.RunSnapshotTTL(s.cpDao, _shutdown, time.Minute, time.Duration(s.snapshotTTL)*time.Hour)
		}()
	}

	// wait for something to happen
	for {
		select {
		case <-event:
			glog.Warningf("Coup d'etat, re-electing...")
			return
		case <-stopped:
			glog.Warningf("Leader died, re-electing...")
			return
		case <-s.shutdown:
			return
		}
	}
}
1015422
func (ctx *facadeContext) LeadershipChecker() (leadership.Checker, error) {
	if ctx.r.shared.featureEnabled(feature.LegacyLeases) {
		return ctx.State().LeadershipChecker(), nil
	}
	checker, err := ctx.r.shared.leaseManager.Checker(
		lease.ApplicationLeadershipNamespace,
		ctx.State().ModelUUID(),
	)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return leadershipChecker{checker}, nil
}