// Datapoints returns datapoints for json decoder v2
552847
func (decoder *JSONDecoderV2) Datapoints() []*datapoint.Datapoint {
	return []*datapoint.Datapoint{
		sfxclient.Counter("dropped_points", map[string]string{"protocol": "sfx_json_v2", "reason": "unknown_metric_type"}, atomic.LoadInt64(&decoder.unknownMetricType)),
		sfxclient.Counter("dropped_points", map[string]string{"protocol": "sfx_json_v2", "reason": "invalid_value"}, atomic.LoadInt64(&decoder.invalidValue)),
	}
}
118492
func ListPointers(data []byte) ([]string, error) {
	if len(data) == 0 {
		return nil, fmt.Errorf("Invalid JSON")
	}
	rv := []string{""}

	scan := &json.Scanner{}
	scan.Reset()

	offset := 0
	beganLiteral := 0
	var current []string
	for {
		if offset >= len(data) {
			return rv, nil
		}
		newOp := scan.Step(scan, int(data[offset]))
		offset++

		switch newOp {
		case json.ScanBeginArray:
			current = append(current, "0")
		case json.ScanObjectKey:
			current[len(current)-1] = grokLiteral(data[beganLiteral-1 : offset-1])
		case json.ScanBeginLiteral:
			beganLiteral = offset
		case json.ScanArrayValue:
			n := mustParseInt(current[len(current)-1])
			current[len(current)-1] = strconv.Itoa(n + 1)
		case json.ScanEndArray, json.ScanEndObject:
			current = sliceToEnd(current)
		case json.ScanBeginObject:
			current = append(current, "")
		case json.ScanError:
			return nil, fmt.Errorf("Error reading JSON object at offset %v", offset)
		}

		if newOp == json.ScanBeginArray || newOp == json.ScanArrayValue ||
			newOp == json.ScanObjectKey {
			rv = append(rv, encodePointer(current))
		}
	}
}
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
}
1174029
func (h *HTTPSink) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) (err error) {
	if len(points) == 0 || h.DatapointEndpoint == "" {
		return nil
	}
	return h.doBottom(ctx, func() (io.Reader, bool, error) {
		return h.encodePostBodyProtobufV2(points)
	}, "application/x-protobuf", h.DatapointEndpoint, datapointAndEventResponseValidator)
}
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
}
1317437
func (f *ItemFlagger) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint, next Sink) error {
	f.SetDatapointFlags(ctx, points)
	return next.AddDatapoints(ctx, points)
}
876189
func (streamer *Demultiplexer) Datapoints() []*datapoint.Datapoint {
	var dps []*datapoint.Datapoint
	if streamer.FutureDuration != nil {
		dps = append(dps, []*datapoint.Datapoint{
			sfxclient.Cumulative("future.count", map[string]string{"type": "datapoint"}, atomic.LoadInt64(&streamer.stats.futureDps)),
			sfxclient.Cumulative("future.count", map[string]string{"type": "event"}, atomic.LoadInt64(&streamer.stats.futureEvents)),
			sfxclient.Cumulative("future.count", map[string]string{"type": "spans"}, atomic.LoadInt64(&streamer.stats.futureSpans)),
		}...)
	}
	if streamer.LateDuration != nil {
		dps = append(dps, []*datapoint.Datapoint{
			sfxclient.Cumulative("late.count", map[string]string{"type": "datapoint"}, atomic.LoadInt64(&streamer.stats.lateDps)),
			sfxclient.Cumulative("late.count", map[string]string{"type": "event"}, atomic.LoadInt64(&streamer.stats.lateEvents)),
			sfxclient.Cumulative("late.count", map[string]string{"type": "spans"}, atomic.LoadInt64(&streamer.stats.lateSpans)),
		}...)
	}
	return dps
}
708150
func (f *Forwarder) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) error {
	points = f.FilterDatapoints(points)
	f.mu.Lock()
	defer f.mu.Unlock()
	atomic.AddInt64(&f.stats.totalDatapointsForwarded, int64(len(points)))
	for _, dp := range points {
		if _, err := f.writeString(f.file, dp.String()+"\n"); err != nil {
			return errors.Annotate(err, "cannot write datapoint to string")
		}
	}
	return f.file.Sync()
}
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)
}
594394
func (a *AsyncMultiTokenSink) AddDatapointsWithToken(token string, datapoints []*datapoint.Datapoint) (err error) {
	var channelID int64
	if channelID, err = a.getChannel(token, len(a.dpChannels)); err == nil {
		var worker = a.dpChannels[channelID]
		_ = atomic.AddInt64(&a.dpBuffered, int64(len(datapoints)))
		var m = &dpMsg{
			token: token,
			data:  datapoints,
		}
		select {
		// check if the sink is closing and return if so
		// reading from a.closing will only return a value if the a.closing channel is closed
		case <-a.closing:
			err = fmt.Errorf("unable to add datapoints: the worker has been stopped")
		default:
			select {
			case worker.input <- m:
				atomic.AddInt64(&a.stats.TotalDatapointsBuffered, int64(len(datapoints)))
			default:
				err = fmt.Errorf("unable to add datapoints: the input buffer is full")
			}
		}
	} else {
		err = fmt.Errorf("unable to add datapoints: there was an error while hashing the token to a worker. %v", err)
	}

	return
}
705025
func (o *SpanTagRemoval) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) error {
	return o.next.AddDatapoints(ctx, points)
}
1349059
func (v JSONDatapointV2) MarshalJSON() ([]byte, error) {
	w := jwriter.Writer{}
	easyjson3b0ebf6aEncodeGithubComSignalfxGatewayProtocolSignalfxFormat1(&w, v)
	return w.Buffer.BuildBytes(), w.Error
}
1332741
func (w *WithDimensions) Datapoints() []*datapoint.Datapoint {
	return w.appendDimensions(w.Collector.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
}
22160
func (a *AsyncMultiTokenSink) AddDatapoints(ctx context.Context, datapoints []*datapoint.Datapoint) (err error) {
	if token := ctx.Value(TokenCtxKey); token != nil {
		err = a.AddDatapointsWithToken(token.(string), datapoints)
	} else {
		err = fmt.Errorf("no value was found on the context with key '%s'", TokenCtxKey)
	}
	return
}
163731
func datapointLabelsByName(s store.Store) map[string][]datapointLabels {
	rval := map[string][]datapointLabels{}

	for _, obj := range s.Objects() {
		m, ok := obj.(producers.MetricsMessage)
		if !ok {
			promLog.Warnf("Unsupported message type %T", obj)
			continue
		}

		// For each datapoint, collect its labels and add an entry to rval.
		dims := dimsToMap(m.Dimensions)
		for _, d := range m.Datapoints {
			labels := map[string]string{}
			for k, v := range dims {
				labels[k] = v
			}
			for k, v := range d.Tags {
				labels[k] = v
			}
			rval[d.Name] = append(rval[d.Name], datapointLabels{&d, labels})
		}
	}

	return rval
}
1276592
func NewOnHostDatapoint(metric string, value datapoint.Value, metricType datapoint.MetricType) *datapoint.Datapoint {
	return NewOnHostDatapointDimensions(metric, value, metricType, map[string]string{})
}