// consulDiscovery waits for the signal to attempt server discovery via Consul.
// It's intended to be started in a goroutine. See triggerDiscovery() for
// causing consul discovery from other code locations.
421732
func (c *Client) consulDiscovery() {
	for {
		select {
		case <-c.triggerDiscoveryCh:
			if err := c.consulDiscoveryImpl(); err != nil {
				c.logger.Error("error discovering nomad servers", "error", err)
			}
		case <-c.shutdownCh:
			return
		}
	}
}
312116
func (a *Agent) consulConfig() (*consul.Config, error) {
	// Start with the provided config or default config
	base := consul.DefaultConfig()

	// This is set when the agent starts up
	base.NodeID = a.config.NodeID

	// Apply dev mode
	base.DevMode = a.config.DevMode

	// Override with our config
	// todo(fs): these are now always set in the runtime config so we can simplify this
	// todo(fs): or is there a reason to keep it like that?
	base.Datacenter = a.config.Datacenter
	base.PrimaryDatacenter = a.config.PrimaryDatacenter
	base.DataDir = a.config.DataDir
	base.NodeName = a.config.NodeName

	base.CoordinateUpdateBatchSize = a.config.ConsulCoordinateUpdateBatchSize
	base.CoordinateUpdateMaxBatches = a.config.ConsulCoordinateUpdateMaxBatches
	base.CoordinateUpdatePeriod = a.config.ConsulCoordinateUpdatePeriod

	base.RaftConfig.HeartbeatTimeout = a.config.ConsulRaftHeartbeatTimeout
	base.RaftConfig.LeaderLeaseTimeout = a.config.ConsulRaftLeaderLeaseTimeout
	base.RaftConfig.ElectionTimeout = a.config.ConsulRaftElectionTimeout

	base.SerfLANConfig.MemberlistConfig.BindAddr = a.config.SerfBindAddrLAN.IP.String()
	base.SerfLANConfig.MemberlistConfig.BindPort = a.config.SerfBindAddrLAN.Port
	base.SerfLANConfig.MemberlistConfig.AdvertiseAddr = a.config.SerfAdvertiseAddrLAN.IP.String()
	base.SerfLANConfig.MemberlistConfig.AdvertisePort = a.config.SerfAdvertiseAddrLAN.Port
	base.SerfLANConfig.MemberlistConfig.GossipVerifyIncoming = a.config.EncryptVerifyIncoming
	base.SerfLANConfig.MemberlistConfig.GossipVerifyOutgoing = a.config.EncryptVerifyOutgoing
	base.SerfLANConfig.MemberlistConfig.GossipInterval = a.config.GossipLANGossipInterval
	base.SerfLANConfig.MemberlistConfig.GossipNodes = a.config.GossipLANGossipNodes
	base.SerfLANConfig.MemberlistConfig.ProbeInterval = a.config.GossipLANProbeInterval
	base.SerfLANConfig.MemberlistConfig.ProbeTimeout = a.config.GossipLANProbeTimeout
	base.SerfLANConfig.MemberlistConfig.SuspicionMult = a.config.GossipLANSuspicionMult
	base.SerfLANConfig.MemberlistConfig.RetransmitMult = a.config.GossipLANRetransmitMult
	if a.config.ReconnectTimeoutLAN != 0 {
		base.SerfLANConfig.ReconnectTimeout = a.config.ReconnectTimeoutLAN
	}

	if a.config.SerfBindAddrWAN != nil {
		base.SerfWANConfig.MemberlistConfig.BindAddr = a.config.SerfBindAddrWAN.IP.String()
		base.SerfWANConfig.MemberlistConfig.BindPort = a.config.SerfBindAddrWAN.Port
		base.SerfWANConfig.MemberlistConfig.AdvertiseAddr = a.config.SerfAdvertiseAddrWAN.IP.String()
		base.SerfWANConfig.MemberlistConfig.AdvertisePort = a.config.SerfAdvertiseAddrWAN.Port
		base.SerfWANConfig.MemberlistConfig.GossipVerifyIncoming = a.config.EncryptVerifyIncoming
		base.SerfWANConfig.MemberlistConfig.GossipVerifyOutgoing = a.config.EncryptVerifyOutgoing
		base.SerfWANConfig.MemberlistConfig.GossipInterval = a.config.GossipWANGossipInterval
		base.SerfWANConfig.MemberlistConfig.GossipNodes = a.config.GossipWANGossipNodes
		base.SerfWANConfig.MemberlistConfig.ProbeInterval = a.config.GossipWANProbeInterval
		base.SerfWANConfig.MemberlistConfig.ProbeTimeout = a.config.GossipWANProbeTimeout
		base.SerfWANConfig.MemberlistConfig.SuspicionMult = a.config.GossipWANSuspicionMult
		base.SerfWANConfig.MemberlistConfig.RetransmitMult = a.config.GossipWANRetransmitMult
		if a.config.ReconnectTimeoutWAN != 0 {
			base.SerfWANConfig.ReconnectTimeout = a.config.ReconnectTimeoutWAN
		}
	} else {
		// Disable serf WAN federation
		base.SerfWANConfig = nil
	}

	base.RPCAddr = a.config.RPCBindAddr
	base.RPCAdvertise = a.config.RPCAdvertiseAddr

	base.Segment = a.config.SegmentName
	if len(a.config.Segments) > 0 {
		segments, err := a.segmentConfig()
		if err != nil {
			return nil, err
		}
		base.Segments = segments
	}
	if a.config.Bootstrap {
		base.Bootstrap = true
	}
	if a.config.RejoinAfterLeave {
		base.RejoinAfterLeave = true
	}
	if a.config.BootstrapExpect != 0 {
		base.BootstrapExpect = a.config.BootstrapExpect
	}
	if a.config.RPCProtocol > 0 {
		base.ProtocolVersion = uint8(a.config.RPCProtocol)
	}
	if a.config.RaftProtocol != 0 {
		base.RaftConfig.ProtocolVersion = raft.ProtocolVersion(a.config.RaftProtocol)
	}
	if a.config.RaftSnapshotThreshold != 0 {
		base.RaftConfig.SnapshotThreshold = uint64(a.config.RaftSnapshotThreshold)
	}
	if a.config.RaftSnapshotInterval != 0 {
		base.RaftConfig.SnapshotInterval = a.config.RaftSnapshotInterval
	}
	if a.config.ACLMasterToken != "" {
		base.ACLMasterToken = a.config.ACLMasterToken
	}
	if a.config.ACLDatacenter != "" {
		base.ACLDatacenter = a.config.ACLDatacenter
	}
	if a.config.ACLTokenTTL != 0 {
		base.ACLTokenTTL = a.config.ACLTokenTTL
	}
	if a.config.ACLPolicyTTL != 0 {
		base.ACLPolicyTTL = a.config.ACLPolicyTTL
	}
	if a.config.ACLRoleTTL != 0 {
		base.ACLRoleTTL = a.config.ACLRoleTTL
	}
	if a.config.ACLDefaultPolicy != "" {
		base.ACLDefaultPolicy = a.config.ACLDefaultPolicy
	}
	if a.config.ACLDownPolicy != "" {
		base.ACLDownPolicy = a.config.ACLDownPolicy
	}
	base.ACLEnforceVersion8 = a.config.ACLEnforceVersion8
	base.ACLTokenReplication = a.config.ACLTokenReplication
	base.ACLsEnabled = a.config.ACLsEnabled
	if a.config.ACLEnableKeyListPolicy {
		base.ACLEnableKeyListPolicy = a.config.ACLEnableKeyListPolicy
	}
	if a.config.SessionTTLMin != 0 {
		base.SessionTTLMin = a.config.SessionTTLMin
	}
	if a.config.NonVotingServer {
		base.NonVoter = a.config.NonVotingServer
	}

	// These are fully specified in the agent defaults, so we can simply
	// copy them over.
	base.AutopilotConfig.CleanupDeadServers = a.config.AutopilotCleanupDeadServers
	base.AutopilotConfig.LastContactThreshold = a.config.AutopilotLastContactThreshold
	base.AutopilotConfig.MaxTrailingLogs = uint64(a.config.AutopilotMaxTrailingLogs)
	base.AutopilotConfig.ServerStabilizationTime = a.config.AutopilotServerStabilizationTime
	base.AutopilotConfig.RedundancyZoneTag = a.config.AutopilotRedundancyZoneTag
	base.AutopilotConfig.DisableUpgradeMigration = a.config.AutopilotDisableUpgradeMigration
	base.AutopilotConfig.UpgradeVersionTag = a.config.AutopilotUpgradeVersionTag

	// make sure the advertise address is always set
	if base.RPCAdvertise == nil {
		base.RPCAdvertise = base.RPCAddr
	}

	// Rate limiting for RPC calls.
	if a.config.RPCRateLimit > 0 {
		base.RPCRate = a.config.RPCRateLimit
	}
	if a.config.RPCMaxBurst > 0 {
		base.RPCMaxBurst = a.config.RPCMaxBurst
	}

	// RPC-related performance configs.
	if a.config.RPCHoldTimeout > 0 {
		base.RPCHoldTimeout = a.config.RPCHoldTimeout
	}
	if a.config.LeaveDrainTime > 0 {
		base.LeaveDrainTime = a.config.LeaveDrainTime
	}

	// set the src address for outgoing rpc connections
	// Use port 0 so that outgoing connections use a random port.
	if !ipaddr.IsAny(base.RPCAddr.IP) {
		base.RPCSrcAddr = &net.TCPAddr{IP: base.RPCAddr.IP}
	}

	// Format the build string
	revision := a.config.Revision
	if len(revision) > 8 {
		revision = revision[:8]
	}
	base.Build = fmt.Sprintf("%s%s:%s", a.config.Version, a.config.VersionPrerelease, revision)

	// Copy the TLS configuration
	base.VerifyIncoming = a.config.VerifyIncoming || a.config.VerifyIncomingRPC
	if a.config.CAPath != "" || a.config.CAFile != "" {
		base.UseTLS = true
	}
	base.VerifyOutgoing = a.config.VerifyOutgoing
	base.VerifyServerHostname = a.config.VerifyServerHostname
	base.CAFile = a.config.CAFile
	base.CAPath = a.config.CAPath
	base.CertFile = a.config.CertFile
	base.KeyFile = a.config.KeyFile
	base.ServerName = a.config.ServerName
	base.Domain = a.config.DNSDomain
	base.TLSMinVersion = a.config.TLSMinVersion
	base.TLSCipherSuites = a.config.TLSCipherSuites
	base.TLSPreferServerCipherSuites = a.config.TLSPreferServerCipherSuites

	// Copy the Connect CA bootstrap config
	if a.config.ConnectEnabled {
		base.ConnectEnabled = true

		// Allow config to specify cluster_id provided it's a valid UUID. This is
		// meant only for tests where a deterministic ID makes fixtures much simpler
		// to work with but since it's only read on initial cluster bootstrap it's not
		// that much of a liability in production. The worst a user could do is
		// configure logically separate clusters with same ID by mistake but we can
		// avoid documenting this is even an option.
		if clusterID, ok := a.config.ConnectCAConfig["cluster_id"]; ok {
			if cIDStr, ok := clusterID.(string); ok {
				if _, err := uuid.ParseUUID(cIDStr); err == nil {
					// Valid UUID configured, use that
					base.CAConfig.ClusterID = cIDStr
				}
			}
			if base.CAConfig.ClusterID == "" {
				// If the tried to specify an ID but typoed it don't ignore as they will
				// then bootstrap with a new ID and have to throw away the whole cluster
				// and start again.
				a.logger.Println("[ERR] connect CA config cluster_id specified but " +
					"is not a valid UUID, aborting startup")
				return nil, fmt.Errorf("cluster_id was supplied but was not a valid UUID")
			}
		}

		if a.config.ConnectCAProvider != "" {
			base.CAConfig.Provider = a.config.ConnectCAProvider
		}

		// Merge connect CA Config regardless of provider (since there are some
		// common config options valid to all like leaf TTL).
		for k, v := range a.config.ConnectCAConfig {
			base.CAConfig.Config[k] = v
		}
	}

	// Setup the user event callback
	base.UserEventHandler = func(e serf.UserEvent) {
		select {
		case a.eventCh <- e:
		case <-a.shutdownCh:
		}
	}

	// Setup the loggers
	base.LogOutput = a.LogOutput

	// This will set up the LAN keyring, as well as the WAN and any segments
	// for servers.
	if err := a.setupKeyrings(base); err != nil {
		return nil, fmt.Errorf("Failed to configure keyring: %v", err)
	}

	base.ConfigEntryBootstrap = a.config.ConfigEntryBootstrap

	return base, nil
}
813257
func Enable(server *prpc.Server) {
	serviceNames := append(server.ServiceNames(), "discovery.Discovery")
	service, err := New(serviceNames...)
	if err != nil {
		panic(err)
	}
	RegisterDiscoveryServer(server, service)
}
288898
func (s *Server) lanNodeJoin(me serf.MemberEvent) {
	for _, m := range me.Members {
		ok, serverMeta := metadata.IsConsulServer(m)
		if !ok || serverMeta.Segment != "" {
			continue
		}
		s.logger.Printf("[INFO] consul: Adding LAN server %s", serverMeta)

		// Update server lookup
		s.serverLookup.AddServer(serverMeta)

		// If we're still expecting to bootstrap, may need to handle this.
		if s.config.BootstrapExpect != 0 {
			s.maybeBootstrap()
		}

		// Kick the join flooders.
		s.FloodNotify()
	}
}
79620
func (c *Client) Discover() (NATType, *Host, error) {
	if c.serverAddr == "" {
		c.SetServerAddr(DefaultServerAddr)
	}
	serverUDPAddr, err := net.ResolveUDPAddr("udp", c.serverAddr)
	if err != nil {
		return NATError, nil, err
	}
	// Use the connection passed to the client if it is not nil, otherwise
	// create a connection and close it at the end.
	conn := c.conn
	if conn == nil {
		conn, err = net.ListenUDP("udp", nil)
		if err != nil {
			return NATError, nil, err
		}
		defer conn.Close()
	}
	return c.discover(conn, serverUDPAddr)
}
377239
func (daemon *Daemon) reloadClusterDiscovery(conf *config.Config, attributes map[string]string) (err error) {
	defer func() {
		// prepare reload event attributes with updatable configurations
		attributes["cluster-store"] = conf.ClusterStore
		attributes["cluster-advertise"] = conf.ClusterAdvertise

		attributes["cluster-store-opts"] = "{}"
		if daemon.configStore.ClusterOpts != nil {
			opts, err2 := json.Marshal(conf.ClusterOpts)
			if err != nil {
				err = err2
			}
			attributes["cluster-store-opts"] = string(opts)
		}
	}()

	newAdvertise := conf.ClusterAdvertise
	newClusterStore := daemon.configStore.ClusterStore
	if conf.IsValueSet("cluster-advertise") {
		if conf.IsValueSet("cluster-store") {
			newClusterStore = conf.ClusterStore
		}
		newAdvertise, err = config.ParseClusterAdvertiseSettings(newClusterStore, conf.ClusterAdvertise)
		if err != nil && err != discovery.ErrDiscoveryDisabled {
			return err
		}
	}

	if daemon.clusterProvider != nil {
		if err := conf.IsSwarmCompatible(); err != nil {
			return err
		}
	}

	// check discovery modifications
	if !config.ModifiedDiscoverySettings(daemon.configStore, newClusterStore, newAdvertise, conf.ClusterOpts) {
		return nil
	}

	// enable discovery for the first time if it was not previously enabled
	if daemon.discoveryWatcher == nil {
		discoveryWatcher, err := discovery.Init(newClusterStore, newAdvertise, conf.ClusterOpts)
		if err != nil {
			return fmt.Errorf("failed to initialize discovery: %v", err)
		}
		daemon.discoveryWatcher = discoveryWatcher
	} else if err == discovery.ErrDiscoveryDisabled {
		// disable discovery if it was previously enabled and it's disabled now
		daemon.discoveryWatcher.Stop()
	} else if err = daemon.discoveryWatcher.Reload(conf.ClusterStore, newAdvertise, conf.ClusterOpts); err != nil {
		// reload discovery
		return err
	}

	daemon.configStore.ClusterStore = newClusterStore
	daemon.configStore.ClusterOpts = conf.ClusterOpts
	daemon.configStore.ClusterAdvertise = newAdvertise

	if daemon.netController == nil {
		return nil
	}
	netOptions, err := daemon.networkOptions(daemon.configStore, daemon.PluginStore, nil)
	if err != nil {
		logrus.WithError(err).Warn("failed to get options with network controller")
		return nil
	}
	err = daemon.netController.ReloadConfiguration(netOptions...)
	if err != nil {
		logrus.Warnf("Failed to reload configuration with network controller: %v", err)
	}
	return nil
}
1191935
func New(client *consul.Client) *ConsulLocker {
	return &ConsulLocker{
		Client: client,
		locks:  make(map[string]*consul.Lock),
		mutex:  new(sync.RWMutex),
	}
}
107631
func NewDiscoveryService(self NetworkMember, comm CommService, crypt CryptoService, disPol DisclosurePolicy,
	config DiscoveryConfig) Discovery {
	d := &gossipDiscoveryImpl{
		self:             self,
		incTime:          uint64(time.Now().UnixNano()),
		seqNum:           uint64(0),
		deadLastTS:       make(map[string]*timestamp),
		aliveLastTS:      make(map[string]*timestamp),
		id2Member:        make(map[string]*NetworkMember),
		aliveMembership:  util.NewMembershipStore(),
		deadMembership:   util.NewMembershipStore(),
		crypt:            crypt,
		comm:             comm,
		lock:             &sync.RWMutex{},
		toDieChan:        make(chan struct{}, 1),
		toDieFlag:        int32(0),
		logger:           util.GetLogger(util.DiscoveryLogger, self.InternalEndpoint),
		disclosurePolicy: disPol,
		pubsub:           util.NewPubSub(),

		aliveTimeInterval:            config.AliveTimeInterval,
		aliveExpirationTimeout:       config.AliveExpirationTimeout,
		aliveExpirationCheckInterval: config.AliveExpirationCheckInterval,
		reconnectInterval:            config.ReconnectInterval,
	}

	d.validateSelfConfig()
	d.msgStore = newAliveMsgStore(d)

	go d.periodicalSendAlive()
	go d.periodicalCheckAlive()
	go d.handleMessages()
	go d.periodicalReconnectToDead()
	go d.handlePresumedDeadPeers()

	return d
}
277376
func (f *ConfigFlags) ToDiscoveryClient() (discovery.CachedDiscoveryInterface, error) {
	config, err := f.ToRESTConfig()
	if err != nil {
		return nil, err
	}

	// The more groups you have, the more discovery requests you need to make.
	// given 25 groups (our groups + a few custom resources) with one-ish version each, discovery needs to make 50 requests
	// double it just so we don't end up here again for a while.  This config is only used for discovery.
	config.Burst = 100

	// retrieve a user-provided value for the "cache-dir"
	// defaulting to ~/.kube/http-cache if no user-value is given.
	httpCacheDir := defaultCacheDir
	if f.CacheDir != nil {
		httpCacheDir = *f.CacheDir
	}

	discoveryCacheDir := computeDiscoverCacheDir(filepath.Join(homedir.HomeDir(), ".kube", "cache", "discovery"), config.Host)
	return diskcached.NewCachedDiscoveryClientForConfig(config, discoveryCacheDir, httpCacheDir, time.Duration(10*time.Minute))
}
777448
func (d *Disco) discover(hostname svchost.Hostname) (*Host, error) {
	discoURL := &url.URL{
		Scheme: "https",
		Host:   hostname.String(),
		Path:   discoPath,
	}

	client := &http.Client{
		Transport: d.Transport,
		Timeout:   discoTimeout,

		CheckRedirect: func(req *http.Request, via []*http.Request) error {
			log.Printf("[DEBUG] Service discovery redirected to %s", req.URL)
			if len(via) > maxRedirects {
				return errors.New("too many redirects") // this error will never actually be seen
			}
			return nil
		},
	}

	req := &http.Request{
		Header: make(http.Header),
		Method: "GET",
		URL:    discoURL,
	}
	req.Header.Set("Accept", "application/json")
	req.Header.Set("User-Agent", httpclient.UserAgentString())

	creds, err := d.CredentialsForHost(hostname)
	if err != nil {
		log.Printf("[WARN] Failed to get credentials for %s: %s (ignoring)", hostname, err)
	}
	if creds != nil {
		// Update the request to include credentials.
		creds.PrepareRequest(req)
	}

	log.Printf("[DEBUG] Service discovery for %s at %s", hostname, discoURL)

	resp, err := client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("Failed to request discovery document: %v", err)
	}
	defer resp.Body.Close()

	host := &Host{
		// Use the discovery URL from resp.Request in
		// case the client followed any redirects.
		discoURL:  resp.Request.URL,
		hostname:  hostname.ForDisplay(),
		transport: d.Transport,
	}

	// Return the host without any services.
	if resp.StatusCode == 404 {
		return host, nil
	}

	if resp.StatusCode != 200 {
		return nil, fmt.Errorf("Failed to request discovery document: %s", resp.Status)
	}

	contentType := resp.Header.Get("Content-Type")
	mediaType, _, err := mime.ParseMediaType(contentType)
	if err != nil {
		return nil, fmt.Errorf("Discovery URL has a malformed Content-Type %q", contentType)
	}
	if mediaType != "application/json" {
		return nil, fmt.Errorf("Discovery URL returned an unsupported Content-Type %q", mediaType)
	}

	// This doesn't catch chunked encoding, because ContentLength is -1 in that case.
	if resp.ContentLength > maxDiscoDocBytes {
		// Size limit here is not a contractual requirement and so we may
		// adjust it over time if we find a different limit is warranted.
		return nil, fmt.Errorf(
			"Discovery doc response is too large (got %d bytes; limit %d)",
			resp.ContentLength, maxDiscoDocBytes,
		)
	}

	// If the response is using chunked encoding then we can't predict its
	// size, but we'll at least prevent reading the entire thing into memory.
	lr := io.LimitReader(resp.Body, maxDiscoDocBytes)

	servicesBytes, err := ioutil.ReadAll(lr)
	if err != nil {
		return nil, fmt.Errorf("Error reading discovery document body: %v", err)
	}

	var services map[string]interface{}
	err = json.Unmarshal(servicesBytes, &services)
	if err != nil {
		return nil, fmt.Errorf("Failed to decode discovery document as a JSON object: %v", err)
	}
	host.services = services

	return host, nil
}
1291693
func GetDiscovery(ctx ttnlog.Interface) (*grpc.ClientConn, discovery.DiscoveryClient) {
	path := path.Join(GetDataDir(), "/ca.cert")
	cert, err := ioutil.ReadFile(path)
	if err == nil && !pool.RootCAs.AppendCertsFromPEM(cert) {
		ctx.Warnf("Could not add root certificates from %s", path)
	}

	conn, err := api.Dial(viper.GetString("discovery-address"))
	if err != nil {
		ctx.WithError(err).Fatal("Could not connect to Discovery server")
	}
	return conn, discovery.NewDiscoveryClient(conn)
}
109386
func (dp *MockStaticDiscoveryProvider) CreateLocalDiscoveryService(mspID string) (fab.DiscoveryService, error) {
	return &MockStaticDiscoveryService{Error: dp.Error, Peers: dp.Peers}, nil
}
922207
func (cfg *discoveryProcessConfig) NewDiscoveryProcess() *discoveryProcess {
	port := int(atomic.LoadInt32(&discoveryBasePort))
	atomic.AddInt32(&discoveryBasePort, 2)

	copied := *cfg
	copied.webPort = port
	copied.discoveryHost = fmt.Sprintf("http://localhost:%d", copied.webPort)

	return &discoveryProcess{cfg: &copied}
}
1289799
func (c *Component) Discover(serviceName, id string) (*pb_discovery.Announcement, error) {
	res, err := c.Discovery.Get(serviceName, id)
	if err != nil {
		return nil, errors.Wrapf(errors.FromGRPCError(err), "Failed to discover %s/%s", serviceName, id)
	}
	return res, nil
}
180848
func (b *AdapterBase) DiscoveryClient() (discovery.DiscoveryInterface, error) {
	if b.discoveryClient == nil {
		clientConfig, err := b.ClientConfig()
		if err != nil {
			return nil, err
		}
		discoveryClient, err := discovery.NewDiscoveryClientForConfig(clientConfig)
		if err != nil {
			return nil, fmt.Errorf("unable to construct discovery client for dynamic client: %v", err)
		}
		b.discoveryClient = discoveryClient
	}
	return b.discoveryClient, nil
}
513835
func (c *RemoteClient) unlock(id string) error {
	// this doesn't use the lock id, because the lock is tied to the consul client.
	if c.consulLock == nil || c.lockCh == nil {
		return nil
	}

	// cancel our monitoring goroutine
	c.monitorCancel()

	defer func() {
		c.consulLock = nil

		// The consul session is only used for this single lock, so cancel it
		// after we unlock.
		// The session is only created and replaced holding Client.mu, so the
		// CancelFunc must be non-nil.
		c.sessionCancel()
	}()

	select {
	case <-c.lockCh:
		return lostLockErr
	default:
	}

	kv := c.Client.KV()

	var errs error

	if _, err := kv.Delete(c.Path+lockInfoSuffix, nil); err != nil {
		errs = multierror.Append(errs, err)
	}

	if err := c.consulLock.Unlock(); err != nil {
		errs = multierror.Append(errs, err)
	}

	// the monitoring goroutine may be in a select on the lockCh, so we need to
	// wait for it to return before changing the value.
	c.monitorWG.Wait()
	c.lockCh = nil

	// This is only cleanup, and will fail if the lock was immediately taken by
	// another client, so we don't report an error to the user here.
	c.consulLock.Destroy()

	return errs
}
1196610
func (d *ConsulStateDriver) WatchAll(baseKey string, rsps chan [2][]byte) error {
	baseKey = processKey(baseKey)
	consulRsps := make(chan api.KVPairs, 1)
	stop := make(chan bool, 1)
	recvErr := make(chan error, 2)

	// Consul returns all the keys as return value of List(). The following maps helps
	// track the state that has been seen and used to appropriately generate
	// create, modify and delete events
	kvCache := map[string]*api.KVPair{}
	// read with index=0 to fetch all existing keys
	var waitIndex uint64
	kvs, qm, err := d.Client.KV().List(baseKey, &api.QueryOptions{WaitIndex: waitIndex})
	if err != nil {
		log.Errorf("consul read failed for key %q. Error: %s", baseKey, err)
		return err
	}
	// Consul returns success and a nil kv when a key is not found.
	// Treat this as starting with no state.
	// XXX: shall we fail the watch in this case?
	if kvs == nil {
		kvs = api.KVPairs{}
	}
	for _, kv := range kvs {
		kvCache[kv.Key] = kv
	}
	waitIndex = qm.LastIndex

	go d.channelConsulEvents(baseKey, kvCache, consulRsps, rsps, recvErr, stop)

	for {
		select {
		case err := <-recvErr:
			return err
		default:
			kvs, qm, err := d.Client.KV().List(baseKey, &api.QueryOptions{WaitIndex: waitIndex})
			if err != nil {
				if api.IsServerError(err) || strings.Contains(err.Error(), "EOF") || strings.Contains(err.Error(), "connection refused") {
					log.Warnf("Consul watch: server error: %v for %s. Retrying..", err, baseKey)
					time.Sleep(5 * time.Second)
					continue
				} else {
					log.Errorf("consul watch failed for key %q. Error: %s. stopping watch..", baseKey, err)
					stop <- true
					return err
				}
			}
			// Consul returns success and a nil kv when a key is not found.
			// This shall translate into appropriate 'Delete' events or
			// no events (depending on whether some keys were seen before)
			// XXX: shall we stop the watch in this case?
			if kvs == nil {
				kvs = api.KVPairs{}
			}

			waitIndex = qm.LastIndex
			consulRsps <- kvs
		}
	}
}