func (c *ClusterInfo) GetNSQDStats(producers Producers, selectedTopic string, sortBy string, leaderOnly bool) ([]*TopicStats, map[string]*ChannelStats, error) {
	return c.getNSQDStats(producers, selectedTopic, sortBy, leaderOnly, false)
}
GetNSQDStats returns aggregate topic and channel stats from the given Producers

if selectedTopic is empty, this will return stats for *all* topic/channels
and the ChannelStats dict will be keyed by topic + ':' + channel
package clusterinfo
import (
	"fmt"
	"net"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"sync"
	"errors"
	"math"
	"sync/atomic"
	"github.com/blang/semver"
	"github.com/youzan/nsq/internal/http_api"
	"github.com/youzan/nsq/internal/stringy"
)
var v1EndpointVersion semver.Version
func init() {
	v1EndpointVersion, _ = semver.Parse("0.2.29-alpha")
}
type PartialErr interface {
	error
	Errors() []error
}
type ErrList []error
func (l ErrList) Error() string {
	var es []string
	for _, e := range l {
		es = append(es, e.Error())
	}
	return strings.Join(es, "\n")
}
func (l ErrList) Errors() []error {
	return l
}
type logger interface {
	Output(maxdepth int, s string) error
}
type LookupdAddressDC struct {
	DC   string
	Addr string
}
type ClusterInfo struct {
	log    logger
	client *http_api.Client
}
func New(log logger, client *http_api.Client) *ClusterInfo {
	return &ClusterInfo{
		log:    log,
		client: client,
	}
}
func (c *ClusterInfo) logf(f string, args ...interface{}) {
	if c.log == nil {
		return
	}
	c.log.Output(2, fmt.Sprintf(f, args...))
}
func (c *ClusterInfo) GetVersion(addr string) (semver.Version, error) {
	endpoint := fmt.Sprintf("http://%s/info", addr)
	var resp struct {
		Version string `json:"version"`
	}
	err := c.client.NegotiateV1(endpoint, &resp)
	if err != nil {
		return semver.Version{}, err
	}
	if resp.Version == "" {
		resp.Version = "unknown"
	}
	v, err := semver.Parse(resp.Version)
	if err != nil {
		c.logf("CI: parse version failed %s: %v", resp.Version, err)
	}
	return v, err
}
func (c *ClusterInfo) IsTopicDisableChanelAutoCreate(lookupdHTTPAddrs []LookupdAddressDC, topic string) (bool, error) {
	var errs []error
	type meta struct {
		DisableChannelAutoCreate bool `json:"disable_channel_auto_create"`
	}
	type respType struct {
		Producers Producers `json:"producers"`
		MetaInfo  *meta     `json:"meta,omitempty"`
	}
	for _, addr := range lookupdHTTPAddrs {
		lookupd := addr
		endpoint := fmt.Sprintf("http://%s/lookup?topic=%s&metainfo=true", lookupd.Addr, topic)
		c.logf("CI: querying nsqlookupd %s", endpoint)
		var resp respType
		err := c.client.NegotiateV1(endpoint, &resp)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		if resp.MetaInfo != nil && len(resp.Producers) > 0 {
			return resp.MetaInfo.DisableChannelAutoCreate, nil
		}
	}
	return false, fmt.Errorf("Failed to query any nsqlookupd: %s for topic %s", ErrList(errs), topic)
}
func (c *ClusterInfo) GetLookupdTopicsMeta(lookupdHTTPAddrs []LookupdAddressDC, metaInfo bool) ([]*TopicInfo, error) {
	var topics []*TopicInfo
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	type respType struct {
		Topics   []string     `json:"topics"`
		MetaInfo []*TopicInfo `json:"meta_info,omitempty"`
	}
	topicContain := make(map[string]bool)
	for _, addr := range lookupdHTTPAddrs {
		lookupd := addr
		wg.Add(1)
		go func(lookupd LookupdAddressDC) {
			defer wg.Done()
			var endpoint string
			if metaInfo {
				endpoint = fmt.Sprintf("http://%s/topics?metaInfo=true", lookupd.Addr)
			} else {
				endpoint = fmt.Sprintf("http://%s/topics", lookupd.Addr)
			}
			c.logf("CI: querying nsqlookupd %s", endpoint)
			var resp respType
			err := c.client.NegotiateV1(endpoint, &resp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			lock.Lock()
			defer lock.Unlock()
			if resp.MetaInfo != nil {
				for _, topicMeta := range resp.MetaInfo {
					if !topicContain[topicMeta.TopicName] {
						topics = append(topics, topicMeta)
						topicContain[topicMeta.TopicName] = true
					}
				}
			} else {
				for _, topic := range resp.Topics {
					if !topicContain[topic] {
						topics = append(topics, &TopicInfo{
							TopicName: topic,
						})
						topicContain[topic] = true
					}
				}
			}
		}(lookupd)
	}
	wg.Wait()
	if len(errs) == len(lookupdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqlookupd: %s", ErrList(errs))
	}
	sort.Sort(TopicInfoSortByName(topics))
	if len(errs) > 0 {
		return topics, ErrList(errs)
	}
	return topics, nil
}
func (c *ClusterInfo) GetLookupdTopics(lookupdHTTPAddrs []string) ([]string, error) {
	var topics []string
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	type respType struct {
		Topics []string `json:"topics"`
	}
	for _, addr := range lookupdHTTPAddrs {
		wg.Add(1)
		go func(addr string) {
			defer wg.Done()
			endpoint := fmt.Sprintf("http://%s/topics", addr)
			c.logf("CI: querying nsqlookupd %s", endpoint)
			var resp respType
			err := c.client.NegotiateV1(endpoint, &resp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			lock.Lock()
			defer lock.Unlock()
			topics = append(topics, resp.Topics...)
		}(addr)
	}
	wg.Wait()
	if len(errs) == len(lookupdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqlookupd: %s", ErrList(errs))
	}
	topics = stringy.Uniq(topics)
	sort.Strings(topics)
	if len(errs) > 0 {
		return topics, ErrList(errs)
	}
	return topics, nil
}
func (c *ClusterInfo) GetLookupdTopicChannels(topic string, lookupdHTTPAddrs []LookupdAddressDC) ([]string, error) {
	var channels []string
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	type respType struct {
		Channels []string `json:"channels"`
	}
	for _, addr := range lookupdHTTPAddrs {
		lookupd := addr
		wg.Add(1)
		go func(lookupd LookupdAddressDC) {
			defer wg.Done()
			endpoint :=
				fmt.Sprintf("http://%s/channels?topic=%s", lookupd.Addr,
					url.QueryEscape(topic))
			c.logf("CI: querying nsqlookupd %s", endpoint)
			var resp respType
			err := c.client.NegotiateV1(endpoint, &resp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			lock.Lock()
			defer lock.Unlock()
			channels = append(channels, resp.Channels...)
		}(lookupd)
	}
	wg.Wait()
	if len(errs) == len(lookupdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqlookupd: %s", ErrList(errs))
	}
	channels = stringy.Uniq(channels)
	sort.Strings(channels)
	if len(errs) > 0 {
		return channels, ErrList(errs)
	}
	return channels, nil
}
func (c *ClusterInfo) GetLookupdProducers(lookupdHTTPAddrs []LookupdAddressDC) (Producers, error) {
	var producers []*Producer
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	producersByAddr := make(map[string]*Producer)
	maxVersion, _ := semver.Parse("0.0.0")
	type respType struct {
		Producers []*Producer `json:"producers"`
	}
	for _, addr := range lookupdHTTPAddrs {
		wg.Add(1)
		lookupd := addr
		go func(lookupd LookupdAddressDC) {
			defer wg.Done()
			endpoint := fmt.Sprintf("http://%s/nodes", lookupd.Addr)
			c.logf("CI: querying nsqlookupd %s", endpoint)
			var resp respType
			err := c.client.NegotiateV1(endpoint, &resp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			lock.Lock()
			defer lock.Unlock()
			for _, producer := range resp.Producers {
				key := producer.TCPAddress()
				p, ok := producersByAddr[key]
				if !ok {
					producer.DC = lookupd.DC
					producersByAddr[key] = producer
					producers = append(producers, producer)
					if maxVersion.LT(producer.VersionObj) {
						maxVersion = producer.VersionObj
					}
					sort.Sort(producer.Topics)
					p = producer
				}
				p.RemoteAddresses = append(p.RemoteAddresses,
					fmt.Sprintf("%s/%s", lookupd, producer.Address()))
			}
		}(lookupd)
	}
	wg.Wait()
	if len(errs) == len(lookupdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqlookupd: %s", ErrList(errs))
	}
	for _, producer := range producersByAddr {
		if producer.VersionObj.LT(maxVersion) {
			producer.OutOfDate = true
		}
	}
	sort.Sort(ProducersByHost{producers})
	if len(errs) > 0 {
		return producers, ErrList(errs)
	}
	return producers, nil
}
func (c *ClusterInfo) GetLookupdTopicProducers(topic string, lookupdHTTPAddrs []LookupdAddressDC) (Producers, map[string]Producers, error) {
	var producers Producers
	partitionProducers := make(map[string]Producers)
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	type respType struct {
		Producers          Producers            `json:"producers"`
		PartitionProducers map[string]*Producer `json:"partitions"`
	}
	for _, addr := range lookupdHTTPAddrs {
		wg.Add(1)
		lookupd := addr
		go func(lookupd LookupdAddressDC) {
			defer wg.Done()
			endpoint := fmt.Sprintf("http://%s/lookup?topic=%s", lookupd.Addr, url.QueryEscape(topic))
			c.logf("CI: querying nsqlookupd %s", endpoint)
			var resp respType
			err := c.client.NegotiateV1(endpoint, &resp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			lock.Lock()
			defer lock.Unlock()
			for _, p := range resp.Producers {
				version, err := semver.Parse(p.Version)
				if err != nil {
					c.logf("CI: parse version failed %s: %v", p.Version, err)
					version, _ = semver.Parse("0.0.0")
				}
				p.VersionObj = version
				for _, pp := range producers {
					if p.HTTPAddress() == pp.HTTPAddress() {
						goto skip
					}
				}
				p.DC = lookupd.DC
				producers = append(producers, p)
			skip:
			}
			for pid, p := range resp.PartitionProducers {
				version, err := semver.Parse(p.Version)
				if err != nil {
					c.logf("CI: parse version failed %s: %v", p.Version, err)
					version, _ = semver.Parse("0.0.0")
				}
				p.VersionObj = version
				partproducers := partitionProducers[pid]
				for _, pp := range partproducers {
					if p.HTTPAddress() == pp.HTTPAddress() {
						goto skip2
					}
				}
				p.DC = lookupd.DC
				partproducers = append(partproducers, p)
				partitionProducers[pid] = partproducers
			skip2:
			}
		}(lookupd)
	}
	wg.Wait()
	if len(errs) == len(lookupdHTTPAddrs) {
		return nil, nil, fmt.Errorf("Failed to query any nsqlookupd: %s", ErrList(errs))
	}
	if len(errs) > 0 {
		return producers, partitionProducers, ErrList(errs)
	}
	return producers, partitionProducers, nil
}
type TopicInfo struct {
	TopicName                string   `json:"topic_name"`
	ExtSupport               bool     `json:"extend_support"`
	Ordered                  bool     `json:"ordered"`
	MultiPart                bool     `json:"multi_part"`
	DisableChannelAutoCreate bool     `json:"disable_channel_auto_create"`
	RegisteredChannels       []string `json:"registered_channels"`
}
type TopicInfoSortByName []*TopicInfo
func (c TopicInfoSortByName) Less(i, j int) bool {
	return c[i].TopicName < c[j].TopicName
}
func (c TopicInfoSortByName) Swap(i, j int) {
	c[i], c[j] = c[j], c[i]
}
func (c TopicInfoSortByName) Len() int {
	return len(c)
}
func (c *ClusterInfo) GetNSQDTopics(nsqdHTTPAddrs []string) ([]*TopicInfo, error) {
	var topics []*TopicInfo
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	type respType struct {
		Topics []struct {
			Name                     string   `json:"topic_name"`
			Ordered                  bool     `json:"is_multi_ordered"`
			ExtSupport               bool     `json:"is_ext"`
			MultiPart                bool     `json:"multi_part"`
			DisableChannelAutoCreate bool     `json:"disable_channel_auto_create"`
			RegisteredChannels       []string `json:"registered_channels"`
		} `json:"topics"`
	}
	for _, addr := range nsqdHTTPAddrs {
		wg.Add(1)
		go func(addr string) {
			defer wg.Done()
			endpoint := fmt.Sprintf("http://%s/stats?format=json", addr)
			c.logf("CI: querying nsqd %s", endpoint)
			var resp respType
			err := c.client.NegotiateV1(endpoint, &resp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			lock.Lock()
			defer lock.Unlock()
			for _, topic := range resp.Topics {
				topics = append(topics, &TopicInfo{
					TopicName:                topic.Name,
					Ordered:                  topic.Ordered,
					MultiPart:                topic.MultiPart,
					ExtSupport:               topic.ExtSupport,
					DisableChannelAutoCreate: topic.DisableChannelAutoCreate,
					RegisteredChannels:       topic.RegisteredChannels,
				})
			}
		}(addr)
	}
	wg.Wait()
	if len(errs) == len(nsqdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqd: %s", ErrList(errs))
	}
	sort.Sort(TopicInfoSortByName(topics))
	if len(errs) > 0 {
		return topics, ErrList(errs)
	}
	return topics, nil
}
func (c *ClusterInfo) GetNSQDProducers(nsqdHTTPAddrs []string) (Producers, error) {
	var producers Producers
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	type infoRespType struct {
		Version          string `json:"version"`
		BroadcastAddress string `json:"broadcast_address"`
		Hostname         string `json:"hostname"`
		HTTPPort         int    `json:"http_port"`
		TCPPort          int    `json:"tcp_port"`
	}
	type statsRespType struct {
		Topics []struct {
			Name string `json:"topic_name"`
		} `json:"topics"`
	}
	for _, addr := range nsqdHTTPAddrs {
		wg.Add(1)
		go func(addr string) {
			defer wg.Done()
			endpoint := fmt.Sprintf("http://%s/info", addr)
			c.logf("CI: querying nsqd %s", endpoint)
			var infoResp infoRespType
			err := c.client.NegotiateV1(endpoint, &infoResp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			endpoint = fmt.Sprintf("http://%s/stats?format=json", addr)
			c.logf("CI: querying nsqd %s", endpoint)
			var statsResp statsRespType
			err = c.client.NegotiateV1(endpoint, &statsResp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			var producerTopics ProducerTopics
			for _, t := range statsResp.Topics {
				producerTopics = append(producerTopics, ProducerTopic{Topic: t.Name})
			}
			version, err := semver.Parse(infoResp.Version)
			if err != nil {
				c.logf("CI: parse version failed %s: %v", infoResp.Version, err)
				version, _ = semver.Parse("0.0.0")
			}
			lock.Lock()
			defer lock.Unlock()
			producers = append(producers, &Producer{
				Version:          infoResp.Version,
				VersionObj:       version,
				BroadcastAddress: infoResp.BroadcastAddress,
				Hostname:         infoResp.Hostname,
				HTTPPort:         infoResp.HTTPPort,
				TCPPort:          infoResp.TCPPort,
				Topics:           producerTopics,
			})
		}(addr)
	}
	wg.Wait()
	if len(errs) == len(nsqdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqd: %s", ErrList(errs))
	}
	if len(errs) > 0 {
		return producers, ErrList(errs)
	}
	return producers, nil
}
func (c *ClusterInfo) GetNSQDTopicProducers(topic string, nsqdHTTPAddrs []string) (Producers, error) {
	var producers Producers
	var lock sync.Mutex
	var wg sync.WaitGroup
	var errs []error
	type infoRespType struct {
		Version          string `json:"version"`
		BroadcastAddress string `json:"broadcast_address"`
		Hostname         string `json:"hostname"`
		HTTPPort         int    `json:"http_port"`
		TCPPort          int    `json:"tcp_port"`
	}
	type statsRespType struct {
		Topics []struct {
			Name string `json:"topic_name"`
		} `json:"topics"`
	}
	for _, addr := range nsqdHTTPAddrs {
		wg.Add(1)
		go func(addr string) {
			defer wg.Done()
			endpoint := fmt.Sprintf("http://%s/stats?format=json", addr)
			c.logf("CI: querying nsqd %s", endpoint)
			var statsResp statsRespType
			err := c.client.NegotiateV1(endpoint, &statsResp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			var producerTopics ProducerTopics
			for _, t := range statsResp.Topics {
				producerTopics = append(producerTopics, ProducerTopic{Topic: t.Name})
			}
			for _, t := range statsResp.Topics {
				if t.Name == topic {
					endpoint := fmt.Sprintf("http://%s/info", addr)
					c.logf("CI: querying nsqd %s", endpoint)
					var infoResp infoRespType
					err := c.client.NegotiateV1(endpoint, &infoResp)
					if err != nil {
						lock.Lock()
						errs = append(errs, err)
						lock.Unlock()
						return
					}
					version, err := semver.Parse(infoResp.Version)
					if err != nil {
						c.logf("CI: parse version failed %s: %v", infoResp.Version, err)
						version, _ = semver.Parse("0.0.0")
					}
					if infoResp.BroadcastAddress == "" {
						var p string
						infoResp.BroadcastAddress, p, _ = net.SplitHostPort(addr)
						infoResp.HTTPPort, _ = strconv.Atoi(p)
					}
					if infoResp.Hostname == "" {
						infoResp.Hostname, _, _ = net.SplitHostPort(addr)
					}
					lock.Lock()
					producers = append(producers, &Producer{
						Version:          infoResp.Version,
						VersionObj:       version,
						BroadcastAddress: infoResp.BroadcastAddress,
						Hostname:         infoResp.Hostname,
						HTTPPort:         infoResp.HTTPPort,
						TCPPort:          infoResp.TCPPort,
						Topics:           producerTopics,
					})
					lock.Unlock()
					return
				}
			}
		}(addr)
	}
	wg.Wait()
	if len(errs) == len(nsqdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqd: %s", ErrList(errs))
	}
	if len(errs) > 0 {
		return producers, ErrList(errs)
	}
	return producers, nil
}
func (c *ClusterInfo) ListAllLookupdNodes(lookupdHTTPAddrs []LookupdAddressDC) ([]*LookupdNodes, error) {
	var errs []error
	var dcLookupdNodes []*LookupdNodes
	dcHit := make(map[string]bool)
	for _, lookupd := range lookupdHTTPAddrs {
		if _, exist := dcHit[lookupd.DC]; !exist {
			dcHit[lookupd.DC] = true
		} else {
			continue
		}
		endpoint := fmt.Sprintf("http://%s/listlookup", lookupd.Addr)
		c.logf("CI: querying nsqlookupd %s", endpoint)
		var resp LookupdNodes
		err := c.client.NegotiateV1(endpoint, &resp)
		if err != nil {
			c.logf("CI: querying nsqlookupd %s err: %v", endpoint, err)
			errs = append(errs, err)
			continue
		}
		resp.DC = lookupd.DC
		dcLookupdNodes = append(dcLookupdNodes, &resp)
	}
	if len(errs) == len(lookupdHTTPAddrs) {
		return nil, fmt.Errorf("Failed to query any nsqlookupd: %s", ErrList(errs))
	}
	return dcLookupdNodes, nil
}
func (c *ClusterInfo) GetNSQDAllMessageHistoryStats(producers Producers) (map[string]int64, error) {
	var errs []error
	nodeHistoryStatsMap := make(map[string]int64)
	var lock sync.Mutex
	var wg sync.WaitGroup
	for _, p := range producers {
		wg.Add(1)
		go func(p *Producer) {
			defer wg.Done()
			addr := p.HTTPAddress()
			endpoint := fmt.Sprintf("http://%s/message/historystats", addr)
			var nodeHistoryStatsResp struct {
				HistoryStats []*NodeHourlyPubsize `json:"node_hourly_pub_size_stats"`
			}
			err := c.client.NegotiateV1(endpoint, &nodeHistoryStatsResp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
			}
			for _, topicMsgStat := range nodeHistoryStatsResp.HistoryStats {
				lock.Lock()
				_, ok := nodeHistoryStatsMap[topicMsgStat.TopicName]
				if !ok {
					nodeHistoryStatsMap[topicMsgStat.TopicName] = topicMsgStat.HourlyPubSize
				} else {
					nodeHistoryStatsMap[topicMsgStat.TopicName] += topicMsgStat.HourlyPubSize
				}
				lock.Unlock()
			}
		}(p)
	}
	wg.Wait()
	if len(errs) == len(producers) {
		return nil, fmt.Errorf("Failed to query any nsqd for node topic message history: %s", ErrList(errs))
	}
	return nodeHistoryStatsMap, nil
}
func (c *ClusterInfo) GetNSQDMessageHistoryStats(nsqdHTTPAddr string, selectedTopic string, par string) ([]int64, error) {
	endpoint := fmt.Sprintf("http://%s/message/historystats?topic=%s&partition=%s", nsqdHTTPAddr, selectedTopic, par)
	var historyStatsResp struct {
		HistoryStat []int64 `json:"hourly_pub_size"`
	}
	err := c.client.NegotiateV1(endpoint, &historyStatsResp)
	if err != nil {
		return nil, err
	}
	c.logf("CI: querying nsqd %s resp: %v", endpoint, historyStatsResp)
	return historyStatsResp.HistoryStat, nil
}
func (c *ClusterInfo) GetNSQDMessageByID(p Producer, selectedTopic string,
	part string, msgID int64) (string, int64, error) {
	if selectedTopic == "" {
		return "", 0, fmt.Errorf("missing topic while get message")
	}
	type msgInfo struct {
		ID        int64  `json:"id"`
		TraceID   uint64 `json:"trace_id"`
		Body      string `json:"body"`
		Timestamp int64  `json:"timestamp"`
		Attempts  uint16 `json:"attempts"`
		Offset        int64 `json:"offset"`
		QueueCntIndex int64 `json:"queue_cnt_index"`
	}
	addr := p.HTTPAddress()
	endpoint := fmt.Sprintf("http://%s/message/get?topic=%s&partition=%s&search_mode=id&search_pos=%d", addr,
		url.QueryEscape(selectedTopic), url.QueryEscape(part), msgID)
	c.logf("CI: querying nsqd %s", endpoint)
	var resp msgInfo
	_, err := c.client.GETV1(endpoint, &resp)
	if err != nil {
		return "", 0, err
	}
	return resp.Body, resp.Offset, nil
}
func (c *ClusterInfo) GetNSQDCoordStats(producers Producers, selectedTopic string, part string) (*CoordStats, error) {
	var lock sync.Mutex
	var wg sync.WaitGroup
	var topicCoordStats CoordStats
	var errs []error
	for _, p := range producers {
		wg.Add(1)
		go func(p *Producer) {
			defer wg.Done()
			addr := p.HTTPAddress()
			endpoint := fmt.Sprintf("http://%s/coordinator/stats?format=json", addr)
			if selectedTopic != "" {
				endpoint = fmt.Sprintf("http://%s/coordinator/stats?format=json&topic=%s", addr, url.QueryEscape(selectedTopic))
			}
			if part != "" {
				endpoint = fmt.Sprintf("http://%s/coordinator/stats?format=json&topic=%s&partition=%s",
					addr, url.QueryEscape(selectedTopic), url.QueryEscape(part))
			}
			c.logf("CI: querying nsqd %s", endpoint)
			var resp CoordStats
			err := c.client.NegotiateV1(endpoint, &resp)
			if err != nil {
				lock.Lock()
				errs = append(errs, err)
				lock.Unlock()
				return
			}
			lock.Lock()
			defer lock.Unlock()
			c.logf("CI: querying nsqd %s resp: %v", endpoint, resp)
			if resp.RpcStats != nil {
				topicCoordStats.RpcStats = resp.RpcStats.Snapshot()
			}
			for _, topicStat := range resp.TopicCoordStats {
				topicStat.DC = p.DC
				topicStat.Node = addr
				if selectedTopic != "" && topicStat.Name != selectedTopic {
					continue
				}
				topicCoordStats.TopicCoordStats = append(topicCoordStats.TopicCoordStats, topicStat)
			}
		}(p)
	}
	wg.Wait()
	if len(errs) == len(producers) {
		return nil, fmt.Errorf("Failed to query any nsqd: %s", ErrList(errs))
	}
	if len(errs) > 0 {
		return &topicCoordStats, ErrList(errs)
	}
	return &topicCoordStats, nil
}
var INDEX = int32(0)
func (c *ClusterInfo) GetClusterInfo(lookupdAdresses []string) (*ClusterNodeInfo, error) {
	INDEX = atomic.AddInt32(&INDEX, 1) & math.MaxInt32
	c.logf("INDEX for picking lookup http address: %d", INDEX)
	lookupdAdress := lookupdAdresses[int(INDEX)%(len(lookupdAdresses))]
	c.logf("lookupd http address %s picked.", lookupdAdress)
	endpoint := fmt.Sprintf("http://%s/cluster/stats", lookupdAdress)
	var resp ClusterNodeInfo
	err := c.client.NegotiateV1(endpoint, &resp)
	if err != nil {
		return nil, err
	}
	return &resp, nil
}
func (c *ClusterInfo) GetClusterInfoDC(lookupdAdresses []LookupdAddressDC) ([]*ClusterNodeInfo, error) {
	var errs []error
	var dcClusterInfo []*ClusterNodeInfo
	dcHit := make(map[string]bool)
	for _, lookupd := range lookupdAdresses {
		if _, exist := dcHit[lookupd.DC]; !exist {
			dcHit[lookupd.DC] = true
		} else {
			continue
		}
		c.logf("lookupd address in dc %v picked. %v", lookupd.DC, lookupd.Addr)
		clusterInfo, err := c.GetClusterInfo([]string{lookupd.Addr})
		if err != nil {
			errs = append(errs, err)
		}
		if clusterInfo != nil {
			clusterInfo.DC = lookupd.DC
			dcClusterInfo = append(dcClusterInfo, clusterInfo)
		}
	}
	if len(errs) > 0 {
		return dcClusterInfo, ErrList(errs)
	}
	return dcClusterInfo, nil
}
func (c *ClusterInfo) GetNSQDStatsWithClients(producers Producers, selectedTopic string, sortBy string, leaderOnly bool) ([]*TopicStats, map[string]*ChannelStats, error) {
	return c.getNSQDStats(producers, selectedTopic, sortBy, leaderOnly, true)
}