// Collect is supposed to be called by the metrics collector. It returns the actual state of counters.
341770
func (c *ImportMetricCounter) Collect() (metrics.ImportSuccessCounts, metrics.ImportErrorCounts, error) {
	c.counterMutex.Lock()
	defer c.counterMutex.Unlock()

	success := metrics.ImportSuccessCounts{}
	for registry, count := range c.importSuccessCounts {
		success[registry] = count
	}

	failures := metrics.ImportErrorCounts{}
	for info, count := range c.importErrorCounts {
		failures[info] = count
	}

	return success, failures, nil
}
751028
func (collector *procTableCollector) Collect(ch chan<- prometheus.Metric) {
	stats := collector.procTable.Stats()
	addGauge := func(desc *prometheus.Desc, v float64, lv ...string) {
		ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, v, lv...)
	}
	addGauge(procTableRunning, float64(stats.Running))
	addGauge(procTableSucceeded, float64(stats.Succeeded))
	addGauge(procTableFailed, float64(stats.Failed))
}
1088761
func (c *minioCollector) Collect(ch chan<- prometheus.Metric) {

	// Always expose network stats

	// Network Sent/Received Bytes
	ch <- prometheus.MustNewConstMetric(
		prometheus.NewDesc(
			prometheus.BuildFQName("minio", "network", "sent_bytes_total"),
			"Total number of bytes sent by current MinIO server instance",
			nil, nil),
		prometheus.CounterValue,
		float64(globalConnStats.getTotalOutputBytes()),
	)
	ch <- prometheus.MustNewConstMetric(
		prometheus.NewDesc(
			prometheus.BuildFQName("minio", "network", "received_bytes_total"),
			"Total number of bytes received by current MinIO server instance",
			nil, nil),
		prometheus.CounterValue,
		float64(globalConnStats.getTotalInputBytes()),
	)

	// Expose cache stats only if available
	cacheObjLayer := newCacheObjectsFn()
	if cacheObjLayer != nil {
		cs := cacheObjLayer.StorageInfo(context.Background())
		ch <- prometheus.MustNewConstMetric(
			prometheus.NewDesc(
				prometheus.BuildFQName("minio", "disk", "cache_storage_bytes"),
				"Total cache capacity on current MinIO server instance",
				nil, nil),
			prometheus.GaugeValue,
			float64(cs.Total),
		)
		ch <- prometheus.MustNewConstMetric(
			prometheus.NewDesc(
				prometheus.BuildFQName("minio", "disk", "cache_storage_free_bytes"),
				"Total cache available on current MinIO server instance",
				nil, nil),
			prometheus.GaugeValue,
			float64(cs.Free),
		)
	}

	// Expose disk stats only if applicable

	// Fetch disk space info
	objLayer := newObjectLayerFn()
	// Service not initialized yet
	if objLayer == nil {
		return
	}

	s := objLayer.StorageInfo(context.Background())

	// Gateways don't provide disk info
	if s.Backend.Type == Unknown {
		return
	}

	var totalDisks, offlineDisks int
	// Setting totalDisks to 1 and offlineDisks to 0 in FS mode
	if s.Backend.Type == BackendFS {
		totalDisks = 1
		offlineDisks = 0
	} else {
		offlineDisks = s.Backend.OfflineDisks
		totalDisks = s.Backend.OfflineDisks + s.Backend.OnlineDisks
	}

	// Total disk usage by current MinIO server instance
	ch <- prometheus.MustNewConstMetric(
		prometheus.NewDesc(
			prometheus.BuildFQName("minio", "disk", "storage_used_bytes"),
			"Total disk storage used by current MinIO server instance",
			nil, nil),
		prometheus.GaugeValue,
		float64(s.Used),
	)

	// Total disk available space seen by MinIO server instance
	ch <- prometheus.MustNewConstMetric(
		prometheus.NewDesc(
			prometheus.BuildFQName("minio", "disk", "storage_available_bytes"),
			"Total disk available space seen by MinIO server instance",
			nil, nil),
		prometheus.GaugeValue,
		float64(s.Available),
	)

	// Total disk space seen by MinIO server instance
	ch <- prometheus.MustNewConstMetric(
		prometheus.NewDesc(
			prometheus.BuildFQName("minio", "disk", "storage_total_bytes"),
			"Total disk space seen by MinIO server instance",
			nil, nil),
		prometheus.GaugeValue,
		float64(s.Total),
	)

	// MinIO Total Disk/Offline Disk
	ch <- prometheus.MustNewConstMetric(
		prometheus.NewDesc(
			prometheus.BuildFQName("minio", "total", "disks"),
			"Total number of disks for current MinIO server instance",
			nil, nil),
		prometheus.GaugeValue,
		float64(totalDisks),
	)
	ch <- prometheus.MustNewConstMetric(
		prometheus.NewDesc(
			prometheus.BuildFQName("minio", "offline", "disks"),
			"Total number of offline disks for current MinIO server instance",
			nil, nil),
		prometheus.GaugeValue,
		float64(offlineDisks),
	)
}
136928
func (m *metricMap) Collect(ch chan<- Metric) {
	m.mtx.RLock()
	defer m.mtx.RUnlock()

	for _, metrics := range m.metrics {
		for _, metric := range metrics {
			ch <- metric.metric
		}
	}
}
371829
func (r *Registry) Gather() ([]*dto.MetricFamily, error) {
	var (
		checkedMetricChan   = make(chan Metric, capMetricChan)
		uncheckedMetricChan = make(chan Metric, capMetricChan)
		metricHashes        = map[uint64]struct{}{}
		wg                  sync.WaitGroup
		errs                MultiError          // The collected errors to return in the end.
		registeredDescIDs   map[uint64]struct{} // Only used for pedantic checks
	)

	r.mtx.RLock()
	goroutineBudget := len(r.collectorsByID) + len(r.uncheckedCollectors)
	metricFamiliesByName := make(map[string]*dto.MetricFamily, len(r.dimHashesByName))
	checkedCollectors := make(chan Collector, len(r.collectorsByID))
	uncheckedCollectors := make(chan Collector, len(r.uncheckedCollectors))
	for _, collector := range r.collectorsByID {
		checkedCollectors <- collector
	}
	for _, collector := range r.uncheckedCollectors {
		uncheckedCollectors <- collector
	}
	// In case pedantic checks are enabled, we have to copy the map before
	// giving up the RLock.
	if r.pedanticChecksEnabled {
		registeredDescIDs = make(map[uint64]struct{}, len(r.descIDs))
		for id := range r.descIDs {
			registeredDescIDs[id] = struct{}{}
		}
	}
	r.mtx.RUnlock()

	wg.Add(goroutineBudget)

	collectWorker := func() {
		for {
			select {
			case collector := <-checkedCollectors:
				collector.Collect(checkedMetricChan)
			case collector := <-uncheckedCollectors:
				collector.Collect(uncheckedMetricChan)
			default:
				return
			}
			wg.Done()
		}
	}

	// Start the first worker now to make sure at least one is running.
	go collectWorker()
	goroutineBudget--

	// Close checkedMetricChan and uncheckedMetricChan once all collectors
	// are collected.
	go func() {
		wg.Wait()
		close(checkedMetricChan)
		close(uncheckedMetricChan)
	}()

	// Drain checkedMetricChan and uncheckedMetricChan in case of premature return.
	defer func() {
		if checkedMetricChan != nil {
			for range checkedMetricChan {
			}
		}
		if uncheckedMetricChan != nil {
			for range uncheckedMetricChan {
			}
		}
	}()

	// Copy the channel references so we can nil them out later to remove
	// them from the select statements below.
	cmc := checkedMetricChan
	umc := uncheckedMetricChan

	for {
		select {
		case metric, ok := <-cmc:
			if !ok {
				cmc = nil
				break
			}
			errs.Append(processMetric(
				metric, metricFamiliesByName,
				metricHashes,
				registeredDescIDs,
			))
		case metric, ok := <-umc:
			if !ok {
				umc = nil
				break
			}
			errs.Append(processMetric(
				metric, metricFamiliesByName,
				metricHashes,
				nil,
			))
		default:
			if goroutineBudget <= 0 || len(checkedCollectors)+len(uncheckedCollectors) == 0 {
				// All collectors are already being worked on or
				// we have already as many goroutines started as
				// there are collectors. Do the same as above,
				// just without the default.
				select {
				case metric, ok := <-cmc:
					if !ok {
						cmc = nil
						break
					}
					errs.Append(processMetric(
						metric, metricFamiliesByName,
						metricHashes,
						registeredDescIDs,
					))
				case metric, ok := <-umc:
					if !ok {
						umc = nil
						break
					}
					errs.Append(processMetric(
						metric, metricFamiliesByName,
						metricHashes,
						nil,
					))
				}
				break
			}
			// Start more workers.
			go collectWorker()
			goroutineBudget--
			runtime.Gosched()
		}
		// Once both checkedMetricChan and uncheckdMetricChan are closed
		// and drained, the contraption above will nil out cmc and umc,
		// and then we can leave the collect loop here.
		if cmc == nil && umc == nil {
			break
		}
	}
	return internal.NormalizeMetricFamilies(metricFamiliesByName), errs.MaybeUnwrap()
}
876585
func (c *counters) Counts() map[string]int64 {
	c.mu.RLock()
	defer c.mu.RUnlock()

	counts := make(map[string]int64, len(c.counts))
	for k, a := range c.counts {
		counts[k] = atomic.LoadInt64(a)
	}
	return counts
}
594851
func (c *multiTimingsCollector) Collect(ch chan<- prometheus.Metric) {
	for cat, his := range c.mt.Timings.Histograms() {
		labelValues := strings.Split(cat, ".")
		ch <- prometheus.MustNewConstHistogram(
			c.desc,
			uint64(his.Count()),
			float64(his.Total())/1000000000,
			makeCumulativeBuckets(c.cutoffs, his.Buckets()),
			labelValues...)
	}
}
489710
func (c *Collector) Collect(ch chan<- prometheus.Metric) {
	timer := prometheus.NewTimer(prometheus.ObserverFunc(c.scrapeDuration.Set))
	defer c.scrapeDuration.Collect(ch)
	defer timer.ObserveDuration()
	c.scrapeErrors.Set(0)
	defer c.scrapeErrors.Collect(ch)

	c.models.Reset()
	c.machines.Reset()
	c.applications.Reset()
	c.units.Reset()
	c.users.Reset()

	c.updateMetrics()

	c.controller.metrics.Collect(ch)
	c.models.Collect(ch)
	c.machines.Collect(ch)
	c.applications.Collect(ch)
	c.units.Collect(ch)
	c.users.Collect(ch)
}
554430
func (m *ServerMetrics) Collect(ch chan<- prom.Metric) {
	m.serverStartedCounter.Collect(ch)
	m.serverHandledCounter.Collect(ch)
	m.serverStreamMsgReceived.Collect(ch)
	m.serverStreamMsgSent.Collect(ch)
	if m.serverHandledHistogramEnabled {
		m.serverHandledHistogram.Collect(ch)
	}
}
1207693
func (m *Metric) Collect() (a AggregatedMetrics) {
	for n := range m.collect(m.gen()) {
		value := aggregate(n.Value)

		switch n.Name {
		case "Duration":
			a.Duration = value
		case "Errors":
			a.Errors = value
		case "Invocations":
			a.Invocations = value
		case "Throttles":
			a.Throttles = value
		}
	}

	return
}
1293932
func (m *Metrics) Collect() (a map[string]AggregatedMetrics) {
	a = make(map[string]AggregatedMetrics)

	for _, fnName := range m.FunctionNames {
		metric := Metric{
			Config:       m.Config,
			FunctionName: fnName,
		}

		a[fnName] = metric.Collect()
	}

	return
}
1327715
func (c *metricsFuncWithMultiLabelsCollector) Collect(ch chan<- prometheus.Metric) {
	for lvs, val := range c.cfml.Counts() {
		labelValues := strings.Split(lvs, ".")
		value := float64(val)
		ch <- prometheus.MustNewConstMetric(c.desc, c.vt, value, labelValues...)
	}
}
1316425
func (c *Chan) Collect() []Path {
	acc := make([]Path, 0, 0)
	c.Do(func(ts Path) {
		acc = append(acc, ts)
	})
	return acc
}
23230
func (w *collect) Do(stop <-chan struct{}) (err error) {
	defer func() {
		// See bug https://pad/lv/1733469
		// If this function which is run by a PeriodicWorker
		// exits with an error, we need to call stop() to
		// ensure the listener socket is closed.
		if err != nil {
			w.stop()
		}
	}()

	config := w.agent.CurrentConfig()
	tag := config.Tag()
	unitTag, ok := tag.(names.UnitTag)
	if !ok {
		return errors.Errorf("expected a unit tag, got %v", tag)
	}
	paths := uniter.NewWorkerPaths(config.DataDir(), unitTag, "metrics-collect")

	recorder, err := newRecorder(unitTag, paths, w.metricFactory)
	if errors.Cause(err) == errMetricsNotDefined {
		logger.Tracef("%v", err)
		return nil
	} else if err != nil {
		return errors.Annotate(err, "failed to instantiate metric recorder")
	}

	err = w.charmdir.Visit(func() error {
		return w.runner.do(recorder)
	}, stop)
	if err == fortress.ErrAborted {
		logger.Tracef("cannot execute collect-metrics: %v", err)
		return nil
	}
	if spool.IsMetricsDataError(err) {
		logger.Debugf("cannot record metrics: %v", err)
		return nil
	}
	return err
}
147900
func (m *Metric) collect(in <-chan string) <-chan cloudWatchMetric {
	var wg sync.WaitGroup
	out := make(chan cloudWatchMetric)

	for name := range in {
		wg.Add(1)
		name := name

		go func() {
			defer wg.Done()

			res, err := m.stats(name)
			if err != nil {
				// TODO: refactor so that errors are reported in cmd
				fmt.Println(err.Error())
				return
			}

			out <- cloudWatchMetric{
				Name:  name,
				Value: res.Datapoints,
			}
		}()
	}

	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}
1157413
func (c MultiCollector) Collect(s *zipkincore.Span) error {
	return c.aggregateErrors(func(coll Collector) error { return coll.Collect(s) })
}
67955
func (t *TimeCounter) Collector(metric string) Collector {
	return CollectorFunc(func() []*datapoint.Datapoint {
		dur := time.Duration(atomic.LoadInt64(&t.NsBarrier)).String()
		return []*datapoint.Datapoint{
			CumulativeP(metric, map[string]string{"range": "above", "time": dur}, &t.Above),
			CumulativeP(metric, map[string]string{"range": "below", "time": dur}, &t.Below),
		}
	})
}