// Datapoints implements Collector interface and returns metrics
185302
func (z *ReadZipper) Datapoints() []*datapoint.Datapoint {
	return []*datapoint.Datapoint{
		sfxclient.CumulativeP("zipper.hitCount", nil, &z.HitCount),
		sfxclient.CumulativeP("zipper.missCount", nil, &z.MissCount),
		sfxclient.CumulativeP("zipper.newCount", nil, &z.NewCount),
		sfxclient.CumulativeP("zipper.errCount", nil, &z.ErrCount),
	}
}
666188
func DataPoint(obs float64, vars []float64) *dataPoint {
	return &dataPoint{Observed: obs, Variables: vars}
}
8060
func (p *Plugin) Metrics() ([]producers.MetricsMessage, error) {
	var messages []producers.MetricsMessage

	// Fetch node metrics
	nodeMetrics, err := p.getNodeMetrics()
	if err != nil {
		return nil, err
	}
	messages = append(messages, nodeMetrics)

	// Master only collects node metrics; return without checking containers
	if p.Role == dcos.RoleMaster {
		return messages, nil
	}

	// Fetch container metrics
	containerMetrics, err := p.getContainerMetrics()
	if err != nil {
		return nil, err
	}
	messages = append(messages, containerMetrics...)
	return messages, nil
}
809525
func (a *AsyncTokenStatusCounter) Datapoints() (dps []*datapoint.Datapoint) {
	var request bool
	returnDatapoints := make(chan []*datapoint.Datapoint)
	defer close(returnDatapoints)
	for {
		select {
		case <-a.stop:
			return
		case dps = <-returnDatapoints: // listen on the newly created channel for datapoints
			return
		default:
			if !request {
				a.requestDatapoints <- returnDatapoints
				request = true
			}
		}
	}
}
1029313
func toStats(mm producers.MetricsMessage, ddURL string) error {
	c, err := statsd.New(ddURL)
	if err != nil {
		return err
	}

	c.Namespace = "dcos-metrics"

	tags, err := buildTags(mm.Dimensions)
	if err != nil {
		return err
	}

	for _, dp := range mm.Datapoints {
		name := strings.Join([]string{mm.Name, dp.Name}, producers.MetricNamespaceSep)
		val, ok := dp.Value.(float64)
		if !ok {
			return errors.New("Datapoint can not be coerced to float64")
		}

		err = c.Gauge(name, val, tags, 1)
		if err != nil {
			return err
		}
	}

	return nil
}
131311
func (e *EmptyMetricFilter) FilterDatapoints(points []*datapoint.Datapoint) []*datapoint.Datapoint {
	validDatapoints := make([]*datapoint.Datapoint, 0, len(points))
	for _, dat := range points {
		if dat.Metric != "" {
			validDatapoints = append(validDatapoints, dat)
		}
	}
	invalidDatumCount := int64(len(points) - len(validDatapoints))
	if invalidDatumCount != 0 {
		atomic.AddInt64(&e.EmptyMetricFiltered, invalidDatumCount)
	}
	return validDatapoints
}
970634
func (f *Forwarder) Datapoints() []*datapoint.Datapoint {
	datapoints := f.pool.Datapoints()
	datapoints = append(datapoints, f.GetFilteredDatapoints()...)
	return datapoints
}
906059
func (backend *openTSDBBackend) getDataPoints() []dataPoint {
	dataCollector := &dataCollector{
		settings:  backend,
		timestamp: time.Now().Unix(),
	}

	expvar.Do(func(kv expvar.KeyValue) {
		dataCollector.addExpVar(kv)
	})

	return dataCollector.dataPoints
}
1240429
func (a *AsyncMultiTokenSink) Datapoints() (dps []*datapoint.Datapoint) {
	dps = append(dps, Gauge("total_datapoints_buffered", a.stats.DefaultDimensions, atomic.LoadInt64(&a.stats.TotalDatapointsBuffered)))
	dps = append(dps, Gauge("total_events_buffered", a.stats.DefaultDimensions, atomic.LoadInt64(&a.stats.TotalEventsBuffered)))
	dps = append(dps, Gauge("total_spans_buffered", a.stats.DefaultDimensions, atomic.LoadInt64(&a.stats.TotalSpansBuffered)))
	dps = append(dps, a.stats.TotalDatapointsByToken.Datapoints()...)
	dps = append(dps, a.stats.TotalEventsByToken.Datapoints()...)
	dps = append(dps, a.stats.TotalSpansByToken.Datapoints()...)
	dps = append(dps, a.stats.DPBatchSizes.Datapoints()...)
	dps = append(dps, a.stats.EVBatchSizes.Datapoints()...)
	dps = append(dps, a.stats.SpanBatchSizes.Datapoints()...)
	dps = append(dps, Cumulative("total_retries", a.stats.DefaultDimensions, atomic.LoadInt64(&a.stats.NumberOfRetries)))
	return
}
661903
func deviceStatsToDatapoints(stats []IODeviceStats, prefix string, baseTags map[string]string, ts string) []producers.Datapoint {
	var datapoints []producers.Datapoint

	for _, stat := range stats {
		dev := devTotal
		if stat.Device.Major > 0 {
			dev = fmt.Sprintf("%d:%d", stat.Device.Major, stat.Device.Minor)
		}
		// Create a new tags map for these datapoints
		tags := map[string]string{blkioDevice: dev}
		for k, v := range baseTags {
			tags[k] = v
		}
		datapoints = append(datapoints, statValuesToDatapoints(stat.Serviced, prefix+"."+serviced, tags, ts)...)
		datapoints = append(datapoints, statValuesToDatapoints(stat.ServiceBytes, prefix+"."+serviceBytes, tags, ts)...)
		datapoints = append(datapoints, statValuesToDatapoints(stat.ServiceTime, prefix+"."+serviceTime, tags, ts)...)
		datapoints = append(datapoints, statValuesToDatapoints(stat.Merged, prefix+"."+merged, tags, ts)...)
		datapoints = append(datapoints, statValuesToDatapoints(stat.Queued, prefix+"."+queued, tags, ts)...)
		datapoints = append(datapoints, statValuesToDatapoints(stat.WaitTime, prefix+"."+waitTime, tags, ts)...)
	}

	return datapoints
}
643828
func convertDatapointToStatsd(datapoint producers.Datapoint) (string, string, bool) {
	val, err := normalize(datapoint.Value)
	if err != nil {
		// This is only debug-level because we expect many NaNs in regular usage
		log.Debugf("Metric %s failed to convert: %q", datapoint.Name, err)
		return "", "", false
	}
	return datapoint.Name, fmt.Sprintf("%d|g", val), true
}
421650
func (d *Debugger) Breakpoints() []*api.Breakpoint {
	d.processMutex.Lock()
	defer d.processMutex.Unlock()
	return d.breakpoints()
}
712823
func (e *EmptyMetricFilter) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint, next Sink) error {
	validDatapoints := e.FilterDatapoints(points)
	if len(validDatapoints) == 0 {
		return nil
	}
	return next.AddDatapoints(ctx, validDatapoints)
}
732498
func emitDatapointsOverStatsd(datapoints []producers.Datapoint, client *statsd.StatsdClient) error {
	data := make(map[string]string)
	for _, dp := range datapoints {
		name, val, ok := convertDatapointToStatsd(dp)
		// we silently drop metrics which could not be converted
		if ok {
			data[name] = val
		}
	}
	client.Send(data, 1)
	return nil
}
432497
func (v *version) Endpoints() EndpointsInformer {
	return &endpointsInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions}
}
484690
func (m *MetricBatch) Metrics() []Metric {
	result := make([]Metric, len(m.doc.Metrics))
	copy(result, m.doc.Metrics)
	return result
}
34397
func (c *Counter) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint, next Sink) error {
	atomic.AddInt64(&c.TotalDatapoints, int64(len(points)))
	atomic.AddInt64(&c.TotalProcessCalls, 1)
	atomic.AddInt64(&c.CallsInFlight, 1)
	start := time.Now()
	err := next.AddDatapoints(ctx, points)
	atomic.AddInt64(&c.TotalProcessTimeNs, time.Since(start).Nanoseconds())
	atomic.AddInt64(&c.CallsInFlight, -1)
	if err != nil {
		atomic.AddInt64(&c.TotalProcessErrors, 1)
		atomic.AddInt64(&c.ProcessErrorPoints, int64(len(points)))
		c.logErrMsg(ctx, err, "Unable to process datapoints")
	}
	return err
}