// AddDatapoints forwards all points to each sendTo sink.  Returns the error message of the last
// sink to have an error.
967880
func (streamer *Demultiplexer) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint) error {
	if len(points) == 0 {
		return nil
	}
	streamer.handleLateOrFuturePoints(points)
	var errs []error
	for _, sendTo := range streamer.DatapointSinks {
		if err := sendTo.AddDatapoints(ctx, points); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.NewMultiErr(errs)
}
419656
func (c *Counter) AddEvents(ctx context.Context, events []*event.Event, next Sink) error {
	atomic.AddInt64(&c.TotalEvents, int64(len(events)))
	atomic.AddInt64(&c.TotalProcessCalls, 1)
	atomic.AddInt64(&c.CallsInFlight, 1)
	start := time.Now()
	err := next.AddEvents(ctx, events)
	atomic.AddInt64(&c.TotalProcessTimeNs, time.Since(start).Nanoseconds())
	atomic.AddInt64(&c.CallsInFlight, -1)
	if err != nil {
		atomic.AddInt64(&c.TotalProcessErrors, 1)
		atomic.AddInt64(&c.ProcessErrorEvents, int64(len(events)))
		c.logErrMsg(ctx, err, "Unable to process events")
	}
	return err
}
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
			}
		}
	}
}
1005929
func sendDataPoints(data []dataPoint) error {
	json, err := json.Marshal(data)
	if err != nil {
		return err
	}

	resp, err := http.Post(*openTsdbURI, "application/json", bytes.NewReader(json))
	if err != nil {
		return err
	}
	resp.Body.Close()
	return nil
}
374406
func (connector *Forwarder) AddSpans(ctx context.Context, spans []*trace.Span) error {
	atomic.AddInt64(&connector.stats.pipeline, int64(len(spans)))
	defer atomic.AddInt64(&connector.stats.pipeline, -int64(len(spans)))
	atomic.AddInt64(&connector.stats.totalSpansForwarded, int64(len(spans)))
	// could filter here
	if len(spans) == 0 {
		return nil
	}
	return connector.sampler.AddSpans(ctx, spans, connector.sink)
}
1024979
func (G *global_data) SetDatas(datas map[string]string) *global_data {
	if G.Req.Data == nil {
		G.Req.Data = make(map[string]string)
	}
	for key, value := range datas {
		G.Req.Data[key] = value
	}
	return G
}
1336864
func (c *BarChart) AddDataPair(name string, x, y []float64, style Style) {
	n := imin(len(x), len(y))
	data := make([]Point, n)
	for i := 0; i < n; i++ {
		data[i] = Point{X: x[i], Y: y[i]}
	}
	c.AddData(name, data, style)
}
1330504
func statValuesToDatapoints(vals []IOStatValue, prefix string, tags map[string]string, ts string) []producers.Datapoint {
	var datapoints []producers.Datapoint
	for _, s := range vals {
		op := strings.ToLower(s.Operation)
		d := producers.Datapoint{
			Name:      prefix + "." + op,
			Value:     s.Value,
			Tags:      tags,
			Timestamp: ts,
			Unit:      bytes,
		}
		datapoints = append(datapoints, d)
	}
	return datapoints
}
1002868
func (e *ErrorTrackerHandler) Datapoints() []*datapoint.Datapoint {
	return []*datapoint.Datapoint{
		sfxclient.Cumulative("total_errors", nil, atomic.LoadInt64(&e.TotalErrors)),
	}
}
33652
func (s *MetricKeyDataPoints) SetDataPoints(v []*DataPoint) *MetricKeyDataPoints {
	s.DataPoints = v
	return s
}
1301828
func (c *BoxChart) AddData(name string, data []Box, style Style) {
	c.Data = append(c.Data, BoxChartData{name, style, data})
	ps := PlotStyle(PlotStylePoints | PlotStyleBox)
	c.Key.Entries = append(c.Key.Entries, KeyEntry{Text: name, Style: style, PlotStyle: ps})
	// TODO(vodo) min, max
}
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
}
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
}
1220643
func ToEndpoints(obj interface{}) interface{} {
	end, ok := obj.(*api.Endpoints)
	if !ok {
		return nil
	}

	e := &Endpoints{
		Version:   end.GetResourceVersion(),
		Name:      end.GetName(),
		Namespace: end.GetNamespace(),
		Index:     EndpointsKey(end.GetName(), end.GetNamespace()),
		Subsets:   make([]EndpointSubset, len(end.Subsets)),
	}
	for i, eps := range end.Subsets {
		sub := EndpointSubset{
			Addresses: make([]EndpointAddress, len(eps.Addresses)),
		}
		if len(eps.Ports) == 0 {
			// Add sentinal if there are no ports.
			sub.Ports = []EndpointPort{{Port: -1}}
		} else {
			sub.Ports = make([]EndpointPort, len(eps.Ports))
		}

		for j, a := range eps.Addresses {
			ea := EndpointAddress{IP: a.IP, Hostname: a.Hostname}
			if a.NodeName != nil {
				ea.NodeName = *a.NodeName
			}
			if a.TargetRef != nil {
				ea.TargetRefName = a.TargetRef.Name
			}
			sub.Addresses[j] = ea
		}

		for k, p := range eps.Ports {
			ep := EndpointPort{Port: p.Port, Name: p.Name, Protocol: string(p.Protocol)}
			sub.Ports[k] = ep
		}

		e.Subsets[i] = sub
	}

	for _, eps := range end.Subsets {
		for _, a := range eps.Addresses {
			e.IndexIP = append(e.IndexIP, a.IP)
		}
	}

	*end = api.Endpoints{}

	return e
}
528805
func (c *Counter) Datapoints() []*datapoint.Datapoint {
	var dims map[string]string
	if c.DroppedReason != "" {
		dims = map[string]string{"reason": c.DroppedReason}
	}
	return []*datapoint.Datapoint{
		sfxclient.Cumulative("total_process_errors", nil, atomic.LoadInt64(&c.TotalProcessErrors)),
		sfxclient.Cumulative("total_datapoints", nil, atomic.LoadInt64(&c.TotalDatapoints)),
		sfxclient.Cumulative("total_events", nil, atomic.LoadInt64(&c.TotalEvents)),
		sfxclient.Cumulative("total_spans", nil, atomic.LoadInt64(&c.TotalSpans)),
		sfxclient.Cumulative("total_process_calls", nil, atomic.LoadInt64(&c.TotalProcessCalls)),
		sfxclient.Cumulative("dropped_points", dims, atomic.LoadInt64(&c.ProcessErrorPoints)),
		sfxclient.Cumulative("dropped_events", dims, atomic.LoadInt64(&c.ProcessErrorEvents)),
		sfxclient.Cumulative("dropped_spans", dims, atomic.LoadInt64(&c.ProcessErrorSpans)),
		sfxclient.Cumulative("process_time_ns", nil, atomic.LoadInt64(&c.TotalProcessTimeNs)),
		sfxclient.Gauge("calls_in_flight", nil, atomic.LoadInt64(&c.CallsInFlight)),
	}
}
1127535
func (s *PutMetricAlarmInput) SetDatapointsToAlarm(v int64) *PutMetricAlarmInput {
	s.DatapointsToAlarm = &v
	return s
}
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),
	}
}