discovery/consul/consul.go
func (srv *consulService) watch(ctx context.Context, ch chan<- []*targetgroup.Group, catalog *consul.Catalog, lastIndex *uint64) error {
	level.Debug(srv.logger).Log("msg", "Watching service", "service", srv.name, "tags", srv.tags)

	t0 := time.Now()
	opts := &consul.QueryOptions{
		WaitIndex:  *lastIndex,
		WaitTime:   watchTimeout,
		AllowStale: srv.discovery.allowStale,
		NodeMeta:   srv.discovery.watchedNodeMeta,
	}
	nodes, meta, err := catalog.ServiceMultipleTags(srv.name, srv.tags, opts.WithContext(ctx))
	elapsed := time.Since(t0)
	rpcDuration.WithLabelValues("catalog", "service").Observe(elapsed.Seconds())

	// Check the context before in order to exit early.
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
		// Continue.
	}

	if err != nil {
		level.Error(srv.logger).Log("msg", "Error refreshing service", "service", srv.name, "tags", srv.tags, "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

	tgroup := targetgroup.Group{
		Source:  srv.name,
		Labels:  srv.labels,
		Targets: make([]model.LabelSet, 0, len(nodes)),
	}

	for _, node := range nodes {

		// We surround the separated list with the separator as well. This way regular expressions
		// in relabeling rules don't have to consider tag positions.
		var tags = srv.tagSeparator + strings.Join(node.ServiceTags, srv.tagSeparator) + srv.tagSeparator

		// If the service address is not empty it should be used instead of the node address
		// since the service may be registered remotely through a different node.
		var addr string
		if node.ServiceAddress != "" {
			addr = net.JoinHostPort(node.ServiceAddress, fmt.Sprintf("%d", node.ServicePort))
		} else {
			addr = net.JoinHostPort(node.Address, fmt.Sprintf("%d", node.ServicePort))
		}

		labels := model.LabelSet{
			model.AddressLabel:  model.LabelValue(addr),
			addressLabel:        model.LabelValue(node.Address),
			nodeLabel:           model.LabelValue(node.Node),
			tagsLabel:           model.LabelValue(tags),
			serviceAddressLabel: model.LabelValue(node.ServiceAddress),
			servicePortLabel:    model.LabelValue(strconv.Itoa(node.ServicePort)),
			serviceIDLabel:      model.LabelValue(node.ServiceID),
		}

		// Add all key/value pairs from the node's metadata as their own labels.
		for k, v := range node.NodeMeta {
			name := strutil.SanitizeLabelName(k)
			labels[metaDataLabel+model.LabelName(name)] = model.LabelValue(v)
		}

		// Add all key/value pairs from the service's metadata as their own labels.
		for k, v := range node.ServiceMeta {
			name := strutil.SanitizeLabelName(k)
			labels[serviceMetaDataLabel+model.LabelName(name)] = model.LabelValue(v)
		}

		// Add all key/value pairs from the service's tagged addresses as their own labels.
		for k, v := range node.TaggedAddresses {
			name := strutil.SanitizeLabelName(k)
			labels[taggedAddressesLabel+model.LabelName(name)] = model.LabelValue(v)
		}

		tgroup.Targets = append(tgroup.Targets, labels)
	}

	select {
	case <-ctx.Done():
		return ctx.Err()
	case ch <- []*targetgroup.Group{&tgroup}:
	}
	return nil
}
func (srv *consulService) watch(ctx context.Context, ch chan<- []*targetgroup.Group, catalog *consul.Catalog, lastIndex *uint64) error {
	level.Debug(srv.logger).Log("msg", "Watching service", "service", srv.name, "tags", srv.tags)

	t0 := time.Now()
	opts := &consul.QueryOptions{
		WaitIndex:  *lastIndex,
		WaitTime:   watchTimeout,
		AllowStale: srv.discovery.allowStale,
		NodeMeta:   srv.discovery.watchedNodeMeta,
	}
	nodes, meta, err := catalog.ServiceMultipleTags(srv.name, srv.tags, opts.WithContext(ctx))
	elapsed := time.Since(t0)
	rpcDuration.WithLabelValues("catalog", "service").Observe(elapsed.Seconds())

	// Check the context before in order to exit early.
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
		// Continue.
	}

	if err != nil {
		level.Error(srv.logger).Log("msg", "Error refreshing service", "service", srv.name, "tags", srv.tags, "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

	tgroup := targetgroup.Group{
		Source:  srv.name,
		Labels:  srv.labels,
		Targets: make([]model.LabelSet, 0, len(nodes)),
	}

	for _, node := range nodes {

		// We surround the separated list with the separator as well. This way regular expressions
		// in relabeling rules don't have to consider tag positions.
		var tags = srv.tagSeparator + strings.Join(node.ServiceTags, srv.tagSeparator) + srv.tagSeparator

		// If the service address is not empty it should be used instead of the node address
		// since the service may be registered remotely through a different node.
		var addr string
		if node.ServiceAddress != "" {
			addr = net.JoinHostPort(node.ServiceAddress, fmt.Sprintf("%d", node.ServicePort))
		} else {
			addr = net.JoinHostPort(node.Address, fmt.Sprintf("%d", node.ServicePort))
		}

		labels := model.LabelSet{
			model.AddressLabel:  model.LabelValue(addr),
			addressLabel:        model.LabelValue(node.Address),
			nodeLabel:           model.LabelValue(node.Node),
			tagsLabel:           model.LabelValue(tags),
			serviceAddressLabel: model.LabelValue(node.ServiceAddress),
			servicePortLabel:    model.LabelValue(strconv.Itoa(node.ServicePort)),
			serviceIDLabel:      model.LabelValue(node.ServiceID),
		}

		// Add all key/value pairs from the node's metadata as their own labels.
		for k, v := range node.NodeMeta {
			name := strutil.SanitizeLabelName(k)
			labels[metaDataLabel+model.LabelName(name)] = model.LabelValue(v)
		}

		// Add all key/value pairs from the service's metadata as their own labels.
		for k, v := range node.ServiceMeta {
			name := strutil.SanitizeLabelName(k)
			labels[serviceMetaDataLabel+model.LabelName(name)] = model.LabelValue(v)
		}

		// Add all key/value pairs from the service's tagged addresses as their own labels.
		for k, v := range node.TaggedAddresses {
			name := strutil.SanitizeLabelName(k)
			labels[taggedAddressesLabel+model.LabelName(name)] = model.LabelValue(v)
		}

		tgroup.Targets = append(tgroup.Targets, labels)
	}

	select {
	case <-ctx.Done():
		return ctx.Err()
	case ch <- []*targetgroup.Group{&tgroup}:
	}
	return nil
}
func ( srv * consulService ) watch ( ctx context . Context , ch chan <- [ ] * targetgroup . Group , catalog * consul . Catalog , lastIndex * uint64 ) error { level . Debug ( srv . logger ) . Log ( " " , " " , " " , srv . name , " " , srv . tags ) t0 := time . Now ( ) opts := & consul . QueryOptions { WaitIndex : * lastIndex , WaitTime : watchTimeout , AllowStale : srv . discovery . allowStale , NodeMeta : srv . discovery . watchedNodeMeta , } nodes , meta , err := catalog . ServiceMultipleTags ( srv . name , srv . tags , opts . WithContext ( ctx ) ) elapsed := time . Since ( t0 ) rpcDuration . WithLabelValues ( " " , " " ) . Observe ( elapsed . Seconds ( ) ) // Check the context before in order to exit early. select { case <- ctx . Done ( ) : return ctx . Err ( ) default : // Continue. } if err != nil { level . Error ( srv . logger ) . Log ( " " , " " , " " , srv . name , " " , srv . tags , " " , 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 tgroup := targetgroup . Group { Source : srv . name , Labels : srv . labels , Targets : make ( [ ] model . LabelSet , 0 , len ( nodes ) ) , } for _ , node := range nodes { // We surround the separated list with the separator as well. This way regular expressions // in relabeling rules don't have to consider tag positions. var tags = srv . tagSeparator + strings . Join ( node . ServiceTags , srv . tagSeparator ) + srv . tagSeparator // If the service address is not empty it should be used instead of the node address // since the service may be registered remotely through a different node. var addr string if node . ServiceAddress != " " { addr = net . JoinHostPort ( node . ServiceAddress , fmt . Sprintf ( " " , node . ServicePort ) ) } else { addr = net . JoinHostPort ( node . Address , fmt . Sprintf ( " " , node . ServicePort ) ) } labels := model . LabelSet { model . AddressLabel : model . LabelValue ( addr ) , addressLabel : model . LabelValue ( node . Address ) , nodeLabel : model . LabelValue ( node . Node ) , tagsLabel : model . LabelValue ( tags ) , serviceAddressLabel : model . LabelValue ( node . ServiceAddress ) , servicePortLabel : model . LabelValue ( strconv . Itoa ( node . ServicePort ) ) , serviceIDLabel : model . LabelValue ( node . ServiceID ) , } // Add all key/value pairs from the node's metadata as their own labels. for k , v := range node . NodeMeta { name := strutil . SanitizeLabelName ( k ) labels [ metaDataLabel + model . LabelName ( name ) ] = model . LabelValue ( v ) } // Add all key/value pairs from the service's metadata as their own labels. for k , v := range node . ServiceMeta { name := strutil . SanitizeLabelName ( k ) labels [ serviceMetaDataLabel + model . LabelName ( name ) ] = model . LabelValue ( v ) } // Add all key/value pairs from the service's tagged addresses as their own labels. for k , v := range node . TaggedAddresses { name := strutil . SanitizeLabelName ( k ) labels [ taggedAddressesLabel + model . LabelName ( name ) ] = model . LabelValue ( v ) } tgroup . Targets = append ( tgroup . Targets , labels ) } select { case <- ctx . Done ( ) : return ctx . Err ( ) case ch <- [ ] * targetgroup . Group { & tgroup } : } return nil }
// Get updates for a service.
Get updates for a service .