// Datapoints returns connection pool datapoints
970634
func (f *Forwarder) Datapoints() []*datapoint.Datapoint {
	datapoints := f.pool.Datapoints()
	datapoints = append(datapoints, f.GetFilteredDatapoints()...)
	return datapoints
}
1332920
func (d *decoder) Datapoints() []*datapoint.Datapoint {
	dps := d.Bucket.Datapoints()
	dps = append(dps, d.DrainSize.Datapoints()...)
	dps = append(dps,
		sfxclient.Cumulative("prometheus.invalid_requests", nil, atomic.LoadInt64(&d.TotalErrors)),
		sfxclient.Cumulative("prometheus.total_NAN_samples", nil, atomic.LoadInt64(&d.TotalNaNs)),
		sfxclient.Cumulative("prometheus.total_bad_datapoints", nil, atomic.LoadInt64(&d.TotalBadDatapoints)),
	)
	return dps
}
624015
func (s *xlSets) monitorAndConnectEndpoints(monitorInterval time.Duration) {
	ticker := time.NewTicker(monitorInterval)
	// Stop the timer.
	defer ticker.Stop()

	for {
		select {
		case <-GlobalServiceDoneCh:
			return
		case <-s.disksConnectDoneCh:
			return
		case <-ticker.C:
			s.connectDisks()
		}
	}
}
842463
func (connector *Forwarder) Datapoints() []*datapoint.Datapoint {
	dps := connector.stats.requests.Datapoints()
	dps = append(dps, connector.stats.drainSize.Datapoints()...)
	dps = append(dps, connector.GetFilteredDatapoints()...)
	dps = append(dps, connector.sampler.Datapoints()...)
	return dps
}
1272464
func (s *DataMsgServiceServer) DataResyncs(ctx context.Context, req *msg.DataResyncRequests) (
	*msg.DataResyncReplies, error) {
	resyncs := req.GetDataResyncs()
	if resyncs != nil {
		//TODO propagate event like in Kafka transport

		/*localtxn := syncbase.NewLocalBytesTxn(s.adapter.localtransp.PropagateBytesResync)
		if len(resyncs) == 0 {
			log.Debug("received empty resync => DELETE ALL")
		}
		for _, chReq := range resyncs {
			//TODO chReq.ContentType
			localtxn.Put(chReq.Key, chReq.Content)
		}
		err := localtxn.Commit()*/
		var err error
		if err != nil {
			return &msg.DataResyncReplies{MsgId: replySeq(), Error: &msg.Error{Message: err.Error()} /*TODO all other fields*/}, err
		}

		return &msg.DataResyncReplies{MsgId: replySeq() /*TODO all other fields*/}, nil
	}

	err := errors.New("unexpected place - nil resyncs")
	return &msg.DataResyncReplies{MsgId: replySeq(), Error: &msg.Error{Message: err.Error()} /*TODO all other fields*/}, err
}
548597
func (a *api) GetResourcePools() ([]pool.ResourcePool, error) {
	client, err := a.connectMaster()
	if err != nil {
		return nil, err
	}

	return client.GetResourcePools()
}
1084205
func (t *AvailService) DataCenters() (*AvailDataCentersResponse, error) {
	u := &url.Values{}
	v := AvailDataCentersResponse{}
	if err := t.client.do("avail.datacenters", u, &v.Response); err != nil {
		return nil, err
	}
	v.DataCenters = make([]DataCenter, 5)
	if err := json.Unmarshal(v.RawData, &v.DataCenters); err != nil {
		return nil, err
	}
	return &v, nil
}
1123282
func (s *GetMetricStatisticsOutput) SetDatapoints(v []*Datapoint) *GetMetricStatisticsOutput {
	s.Datapoints = v
	return s
}
1220621
func (h *HistoCounter) AddDatapoints(ctx context.Context, points []*datapoint.Datapoint, next Sink) error {
	h.DatapointBucket.Add(float64(len(points)))
	return h.sink.AddDatapoints(ctx, points, next)
}
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
}
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),
	}
}
235057
func getPools(w *rest.ResponseWriter, r *rest.Request, ctx *requestContext) {
	facade := ctx.getFacade()
	dataCtx := ctx.getDatastoreContext()

	pools, err := facade.GetReadPools(dataCtx)
	if err != nil {
		restServerError(w, err)
		return
	}

	w.WriteJson(pools)
}
1332741
func (w *WithDimensions) Datapoints() []*datapoint.Datapoint {
	return w.appendDimensions(w.Collector.Datapoints())
}
480185
func (s Service) Endpoints() map[string]Endpoint {
	es := map[string]Endpoint{}
	for id := range s.p.Services[s.id].Endpoints {
		es[id] = Endpoint{
			id:        id,
			serviceID: s.id,
			p:         s.p,
		}
	}

	return es
}
899757
func (m *MockConnector) ConnectionStats() go_nats.Statistics {
	ret := m.ctrl.Call(m, "ConnectionStats")
	ret0, _ := ret[0].(go_nats.Statistics)
	return ret0
}
428088
func ParseEndpoints(ep *types.Endpoints) (ServiceID, *Endpoints) {
	endpoints := newEndpoints()

	for _, sub := range ep.Subsets {
		for _, addr := range sub.Addresses {
			backend, ok := endpoints.Backends[addr.IP]
			if !ok {
				backend = service.PortConfiguration{}
				endpoints.Backends[addr.IP] = backend
			}

			for _, port := range sub.Ports {
				lbPort := loadbalancer.NewL4Addr(loadbalancer.L4Type(port.Protocol), uint16(port.Port))
				backend[port.Name] = lbPort
			}
		}
	}

	return ParseEndpointsID(ep), endpoints
}
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
}