discovery/consul/consul.go
func (d *Discovery) watchServices(ctx context.Context, ch chan<- []*targetgroup.Group, lastIndex *uint64, services map[string]func()) error {
	catalog := d.client.Catalog()
	level.Debug(d.logger).Log("msg", "Watching services", "tags", d.watchedTags)

	t0 := time.Now()
	opts := &consul.QueryOptions{
		WaitIndex:  *lastIndex,
		WaitTime:   watchTimeout,
		AllowStale: d.allowStale,
		NodeMeta:   d.watchedNodeMeta,
	}
	srvs, meta, err := catalog.Services(opts.WithContext(ctx))
	elapsed := time.Since(t0)
	rpcDuration.WithLabelValues("catalog", "services").Observe(elapsed.Seconds())

	if err != nil {
		level.Error(d.logger).Log("msg", "Error refreshing service list", "err", err)
		rpcFailuresCount.Inc()
		time.Sleep(retryInterval)
		return err
	}
	// If the index equals the previous one, the watch timed out with no update.
	if meta.LastIndex == *lastIndex {
		return nil
	}
	*lastIndex = meta.LastIndex

	// Check for new services.
	for name := range srvs {
		// catalog.Service() returns a map of service name to tags, we can use that to watch
		// only the services that have the tag we are looking for (if specified).
		// In the future consul will also support server side for service metadata.
		// https://github.com/hashicorp/consul/issues/1107
		if !d.shouldWatch(name, srvs[name]) {
			continue
		}
		if _, ok := services[name]; ok {
			continue // We are already watching the service.
		}

		wctx, cancel := context.WithCancel(ctx)
		d.watchService(wctx, ch, name)
		services[name] = cancel
	}

	// Check for removed services.
	for name, cancel := range services {
		if _, ok := srvs[name]; !ok {
			// Call the watch cancellation function.
			cancel()
			delete(services, name)

			// Send clearing target group.
			select {
			case <-ctx.Done():
				return ctx.Err()
			case ch <- []*targetgroup.Group{{Source: name}}:
			}
		}
	}
	return nil
}
func (d *Discovery) watchServices(ctx context.Context, ch chan<- []*targetgroup.Group, lastIndex *uint64, services map[string]func()) error {
	catalog := d.client.Catalog()
	level.Debug(d.logger).Log("msg", "Watching services", "tags", d.watchedTags)

	t0 := time.Now()
	opts := &consul.QueryOptions{
		WaitIndex:  *lastIndex,
		WaitTime:   watchTimeout,
		AllowStale: d.allowStale,
		NodeMeta:   d.watchedNodeMeta,
	}
	srvs, meta, err := catalog.Services(opts.WithContext(ctx))
	elapsed := time.Since(t0)
	rpcDuration.WithLabelValues("catalog", "services").Observe(elapsed.Seconds())

	if err != nil {
		level.Error(d.logger).Log("msg", "Error refreshing service list", "err", err)
		rpcFailuresCount.Inc()
		time.Sleep(retryInterval)
		return err
	}
	// If the index equals the previous one, the watch timed out with no update.
	if meta.LastIndex == *lastIndex {
		return nil
	}
	*lastIndex = meta.LastIndex

	// Check for new services.
	for name := range srvs {
		// catalog.Service() returns a map of service name to tags, we can use that to watch
		// only the services that have the tag we are looking for (if specified).
		// In the future consul will also support server side for service metadata.
		// https://github.com/hashicorp/consul/issues/1107
		if !d.shouldWatch(name, srvs[name]) {
			continue
		}
		if _, ok := services[name]; ok {
			continue // We are already watching the service.
		}

		wctx, cancel := context.WithCancel(ctx)
		d.watchService(wctx, ch, name)
		services[name] = cancel
	}

	// Check for removed services.
	for name, cancel := range services {
		if _, ok := srvs[name]; !ok {
			// Call the watch cancellation function.
			cancel()
			delete(services, name)

			// Send clearing target group.
			select {
			case <-ctx.Done():
				return ctx.Err()
			case ch <- []*targetgroup.Group{{Source: name}}:
			}
		}
	}
	return nil
}
func ( d * Discovery ) watchServices ( ctx context . Context , ch chan <- [ ] * targetgroup . Group , lastIndex * uint64 , services map [ string ] func ( ) ) error { catalog := d . client . Catalog ( ) level . Debug ( d . logger ) . Log ( " " , " " , " " , d . watchedTags ) t0 := time . Now ( ) opts := & consul . QueryOptions { WaitIndex : * lastIndex , WaitTime : watchTimeout , AllowStale : d . allowStale , NodeMeta : d . watchedNodeMeta , } srvs , meta , err := catalog . Services ( opts . WithContext ( ctx ) ) elapsed := time . Since ( t0 ) rpcDuration . WithLabelValues ( " " , " " ) . Observe ( elapsed . Seconds ( ) ) if err != nil { level . Error ( d . logger ) . Log ( " " , " " , " " , err ) rpcFailuresCount . Inc ( ) time . Sleep ( retryInterval ) return err } // If the index equals the previous one, the watch timed out with no update. if meta . LastIndex == * lastIndex { return nil } * lastIndex = meta . LastIndex // Check for new services. for name := range srvs { // catalog.Service() returns a map of service name to tags, we can use that to watch // only the services that have the tag we are looking for (if specified). // In the future consul will also support server side for service metadata. // https://github.com/hashicorp/consul/issues/1107 if ! d . shouldWatch ( name , srvs [ name ] ) { continue } if _ , ok := services [ name ] ; ok { continue // We are already watching the service. } wctx , cancel := context . WithCancel ( ctx ) d . watchService ( wctx , ch , name ) services [ name ] = cancel } // Check for removed services. for name , cancel := range services { if _ , ok := srvs [ name ] ; ! ok { // Call the watch cancellation function. cancel ( ) delete ( services , name ) // Send clearing target group. select { case <- ctx . Done ( ) : return ctx . Err ( ) case ch <- [ ] * targetgroup . Group { { Source : name } } : } } } return nil }
// Watch the catalog for new services we would like to watch. This is called only
// when we don't know yet the names of the services and need to ask Consul the
// entire list of services.
Watch the catalog for new services we would like to watch . This is called only when we don t know yet the names of the services and need to ask Consul the entire list of services .