// monitorLeadership is used to monitor if we acquire or lose our role
// as the leader in the Raft cluster. There is some work the leader is
// expected to do, so we must react to changes
335043
func (s *Server) monitorLeadership() {
	var weAreLeaderCh chan struct{}
	var leaderLoop sync.WaitGroup
	for {
		select {
		case isLeader := <-s.leaderCh:
			switch {
			case isLeader:
				if weAreLeaderCh != nil {
					s.logger.Error("attempted to start the leader loop while running")
					continue
				}

				weAreLeaderCh = make(chan struct{})
				leaderLoop.Add(1)
				go func(ch chan struct{}) {
					defer leaderLoop.Done()
					s.leaderLoop(ch)
				}(weAreLeaderCh)
				s.logger.Info("cluster leadership acquired")

			default:
				if weAreLeaderCh == nil {
					s.logger.Error("attempted to stop the leader loop while not running")
					continue
				}

				s.logger.Debug("shutting down leader loop")
				close(weAreLeaderCh)
				leaderLoop.Wait()
				weAreLeaderCh = nil
				s.logger.Info("cluster leadership lost")
			}

		case <-s.shutdownCh:
			return
		}
	}
}
1015653
func (l *lastSeenRole) observe(newRole api.NodeRole) bool {
	changed := l.role != newRole
	l.role = newRole
	return changed
}
28287
func (s *Server) revokeLeadership() error {
	defer metrics.MeasureSince([]string{"nomad", "leader", "revoke_leadership"}, time.Now())

	// Clear the leader token since we are no longer the leader.
	s.setLeaderAcl("")

	// Disable autopilot
	s.autopilot.Stop()

	// Disable the plan queue, since we are no longer leader
	s.planQueue.SetEnabled(false)

	// Disable the eval broker, since it is only useful as a leader
	s.evalBroker.SetEnabled(false)

	// Disable the blocked eval tracker, since it is only useful as a leader
	s.blockedEvals.SetEnabled(false)

	// Disable the periodic dispatcher, since it is only useful as a leader
	s.periodicDispatcher.SetEnabled(false)

	// Disable the Vault client as it is only useful as a leader.
	s.vault.SetActive(false)

	// Disable the deployment watcher as it is only useful as a leader.
	s.deploymentWatcher.SetEnabled(false, nil)

	// Disable the node drainer
	s.nodeDrainer.SetEnabled(false, nil)

	// Disable any enterprise systems required.
	if err := s.revokeEnterpriseLeadership(); err != nil {
		return err
	}

	// Clear the heartbeat timers on either shutdown or step down,
	// since we are no longer responsible for TTL expirations.
	if err := s.clearAllHeartbeatTimers(); err != nil {
		s.logger.Error("clearing heartbeat timers failed", "error", err)
		return err
	}

	// Unpause our worker if we paused previously
	if len(s.workers) > 1 {
		for i := 0; i < len(s.workers)/2; i++ {
			s.workers[i].SetPause(false)
		}
	}
	return nil
}
1108183
func (n *Node) doRaftLeader(conn redcon.Conn, cmd redcon.Command) (interface{}, error) {
	if len(cmd.Args) != 1 {
		return nil, ErrWrongNumberOfArguments
	}
	leader := n.raft.Leader()
	if leader == "" {
		conn.WriteNull()
	} else {
		conn.WriteBulkString(leader)
	}
	return nil, nil
}
1101820
func (lsa *LeadershipSettingsAccessor) WatchLeadershipSettings(bulkArgs params.Entities) (params.NotifyWatchResults, error) {

	requireAppName, err := lsa.callerApplication()
	if err != nil {
		return params.NotifyWatchResults{}, err
	}
	results := make([]params.NotifyWatchResult, len(bulkArgs.Entities))

	for i, arg := range bulkArgs.Entities {
		result := &results[i]

		// TODO(fwereade): we shoudn't assume a ApplicationTag: we should
		// use an actual auth func to determine permissions.
		applicationTag, err := names.ParseApplicationTag(arg.Tag)
		if err != nil {
			result.Error = common.ServerError(err)
			continue
		}

		appName := applicationTag.Id()
		if appName != requireAppName {
			result.Error = common.ServerError(common.ErrPerm)
			continue
		}

		watcherId, err := lsa.registerWatcherFn(appName)
		if err != nil {
			result.Error = common.ServerError(err)
			continue
		}

		result.NotifyWatcherId = watcherId
	}
	return params.NotifyWatchResults{Results: results}, nil
}
454946
func monitor(d *Daemon, c *container.Container, stop chan struct{}, probe probe) {
	probeTimeout := timeoutWithDefault(c.Config.Healthcheck.Timeout, defaultProbeTimeout)
	probeInterval := timeoutWithDefault(c.Config.Healthcheck.Interval, defaultProbeInterval)
	for {
		select {
		case <-stop:
			logrus.Debugf("Stop healthcheck monitoring for container %s (received while idle)", c.ID)
			return
		case <-time.After(probeInterval):
			logrus.Debugf("Running health check for container %s ...", c.ID)
			startTime := time.Now()
			ctx, cancelProbe := context.WithTimeout(context.Background(), probeTimeout)
			results := make(chan *types.HealthcheckResult, 1)
			go func() {
				healthChecksCounter.Inc()
				result, err := probe.run(ctx, d, c)
				if err != nil {
					healthChecksFailedCounter.Inc()
					logrus.Warnf("Health check for container %s error: %v", c.ID, err)
					results <- &types.HealthcheckResult{
						ExitCode: -1,
						Output:   err.Error(),
						Start:    startTime,
						End:      time.Now(),
					}
				} else {
					result.Start = startTime
					logrus.Debugf("Health check for container %s done (exitCode=%d)", c.ID, result.ExitCode)
					results <- result
				}
				close(results)
			}()
			select {
			case <-stop:
				logrus.Debugf("Stop healthcheck monitoring for container %s (received while probing)", c.ID)
				cancelProbe()
				// Wait for probe to exit (it might take a while to respond to the TERM
				// signal and we don't want dying probes to pile up).
				<-results
				return
			case result := <-results:
				handleProbeResult(d, c, result, stop)
				// Stop timeout
				cancelProbe()
			case <-ctx.Done():
				logrus.Debugf("Health check for container %s taking too long", c.ID)
				handleProbeResult(d, c, &types.HealthcheckResult{
					ExitCode: -1,
					Output:   fmt.Sprintf("Health check exceeded timeout (%v)", probeTimeout),
					Start:    startTime,
					End:      time.Now(),
				}, stop)
				cancelProbe()
				// Wait for probe to exit (it might take a while to respond to the TERM
				// signal and we don't want dying probes to pile up).
				<-results
			}
		}
	}
}
1016198
func (m leadershipChecker) LeadershipCheck(applicationName, unitName string) leadership.Token {
	token := m.checker.Token(applicationName, unitName)
	return leadershipToken{
		applicationName: applicationName,
		unitName:        unitName,
		token:           token,
	}
}
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
}
111215
func (c *testCluster) Leader() (*testNode, error) {
	resp, err := c.api.ListNodes(context.Background(), &api.ListNodesRequest{
		Filters: &api.ListNodesRequest_Filters{
			Roles: []api.NodeRole{api.NodeRoleManager},
		},
	})
	if err != nil {
		return nil, err
	}
	for _, n := range resp.Nodes {
		if n.ManagerStatus.Leader {
			tn, ok := c.nodes[n.ID]
			if !ok {
				return nil, fmt.Errorf("leader id is %s, but it isn't found in test cluster object", n.ID)
			}
			return tn, nil
		}
	}
	return nil, fmt.Errorf("cluster leader is not found in api response")
}
1349826
func determineLeader(node string, children []string) (isLeader bool, leaderNode string, err error) {
	err = func() error {
		if len(children) == 0 {
			return errors.New("no child nodes")
		}
		sequences, lookup, err := sortedChildrenSequences(children)
		if err != nil {
			return err
		}
		mySeq, err := sequencePart(node)
		if err != nil {
			return errors.Wrap(err, "invalid owner node")
		}
		leaderSeq := sequences[0]
		isLeader = mySeq == leaderSeq
		leaderNode = lookup[leaderSeq]
		return nil
	}()
	return isLeader, leaderNode, err
}
648408
func (n *Nodes) monitorDrainAllocs(ctx context.Context, nodeID string, ignoreSys bool, allocCh chan<- *MonitorMessage) {
	defer close(allocCh)

	q := QueryOptions{AllowStale: true}
	initial := make(map[string]*Allocation, 4)

	for {
		allocs, meta, err := n.Allocations(nodeID, &q)
		if err != nil {
			msg := Messagef(MonitorMsgLevelError, "Error monitoring allocations: %v", err)
			select {
			case allocCh <- msg:
			case <-ctx.Done():
			}
			return
		}

		q.WaitIndex = meta.LastIndex

		runningAllocs := 0
		for _, a := range allocs {
			// Get previous version of alloc
			orig, existing := initial[a.ID]

			// Update local alloc state
			initial[a.ID] = a

			migrating := a.DesiredTransition.ShouldMigrate()

			var msg string
			switch {
			case !existing:
				// Should only be possible if response
				// from initial Allocations call was
				// stale. No need to output

			case orig.ClientStatus != a.ClientStatus:
				// Alloc status has changed; output
				msg = fmt.Sprintf("status %s -> %s", orig.ClientStatus, a.ClientStatus)

			case migrating && !orig.DesiredTransition.ShouldMigrate():
				// Alloc was marked for migration
				msg = "marked for migration"

			case migrating && (orig.DesiredStatus != a.DesiredStatus) && a.DesiredStatus == AllocDesiredStatusStop:
				// Alloc has already been marked for migration and is now being stopped
				msg = "draining"
			}

			if msg != "" {
				select {
				case allocCh <- Messagef(MonitorMsgLevelNormal, "Alloc %q %s", a.ID, msg):
				case <-ctx.Done():
					return
				}
			}

			// Ignore malformed allocs
			if a.Job == nil || a.Job.Type == nil {
				continue
			}

			// Track how many allocs are still running
			if ignoreSys && a.Job.Type != nil && *a.Job.Type == JobTypeSystem {
				continue
			}

			switch a.ClientStatus {
			case AllocClientStatusPending, AllocClientStatusRunning:
				runningAllocs++
			}
		}

		// Exit if all allocs are terminal
		if runningAllocs == 0 {
			msg := Messagef(MonitorMsgLevelInfo, "All allocations on node %q have stopped.", nodeID)
			select {
			case allocCh <- msg:
			case <-ctx.Done():
			}
			return
		}
	}
}
299878
func (s *Store) Leader() string {
	if s.raft == nil {
		return ""
	}
	return s.raft.Leader()
}
243581
func (s *Server) establishLeadership() error {
	// check for the upgrade here - this helps us transition to new ACLs much
	// quicker if this is a new cluster or this is a test agent
	if canUpgrade := s.canUpgradeToNewACLs(true); canUpgrade {
		if err := s.initializeACLs(true); err != nil {
			return err
		}
		atomic.StoreInt32(&s.useNewACLs, 1)
		s.updateACLAdvertisement()
	} else if err := s.initializeACLs(false); err != nil {
		return err
	}

	// Hint the tombstone expiration timer. When we freshly establish leadership
	// we become the authoritative timer, and so we need to start the clock
	// on any pending GC events.
	s.tombstoneGC.SetEnabled(true)
	lastIndex := s.raft.LastIndex()
	s.tombstoneGC.Hint(lastIndex)

	// Setup the session timers. This is done both when starting up or when
	// a leader fail over happens. Since the timers are maintained by the leader
	// node along, effectively this means all the timers are renewed at the
	// time of failover. The TTL contract is that the session will not be expired
	// before the TTL, so expiring it later is allowable.
	//
	// This MUST be done after the initial barrier to ensure the latest Sessions
	// are available to be initialized. Otherwise initialization may use stale
	// data.
	if err := s.initializeSessionTimers(); err != nil {
		return err
	}

	// attempt to bootstrap config entries
	if err := s.bootstrapConfigEntries(s.config.ConfigEntryBootstrap); err != nil {
		return err
	}

	s.getOrCreateAutopilotConfig()
	s.autopilot.Start()

	// todo(kyhavlov): start a goroutine here for handling periodic CA rotation
	if err := s.initializeCA(); err != nil {
		return err
	}

	s.startConfigReplication()

	s.startEnterpriseLeader()

	s.startCARootPruning()

	s.setConsistentReadReady()
	return nil
}
1020501
func (t leadershipToken) Check(attempt int, out interface{}) error {
	err := t.token.Check(attempt, out)
	if errors.Cause(err) == lease.ErrNotHeld {
		return errors.Errorf("%q is not leader of %q", t.unitName, t.applicationName)
	}
	return errors.Trace(err)
}
1167134
func (ru *RelationUnit) WatchScope() *RelationScopeWatcher {
	role := counterpartRole(ru.endpoint.Role)
	return watchRelationScope(ru.st, ru.scope, role, ru.unitName)
}
1248881
func (m *Monitor) Monitor(n *Node) error {
	m.mtx.Lock()
	m.Nodes = append(m.Nodes, n)
	m.mtx.Unlock()

	blockCh := make(chan tmtypes.Header, 10)
	n.SendBlocksTo(blockCh)
	blockLatencyCh := make(chan float64, 10)
	n.SendBlockLatenciesTo(blockLatencyCh)
	disconnectCh := make(chan bool, 10)
	n.NotifyAboutDisconnects(disconnectCh)

	if err := n.Start(); err != nil {
		return err
	}

	m.Network.NewNode(n.Name)

	m.nodeQuit[n.Name] = make(chan struct{})
	go m.listen(n.Name, blockCh, blockLatencyCh, disconnectCh, m.nodeQuit[n.Name])

	return nil
}
891865
func (s *Status) Leader(args struct{}, reply *string) error {
	leader := string(s.server.raft.Leader())
	if leader != "" {
		*reply = leader
	} else {
		*reply = ""
	}
	return nil
}