// generateInterMetrics calls the Flush method on each
// counter/gauge/histogram/timer/set in order to
// generate an InterMetric corresponding to that value
509216
func (s *Server) generateInterMetrics(ctx context.Context, percentiles []float64, aggregates samplers.HistogramAggregates, tempMetrics []WorkerMetrics, ms metricsSummary) []samplers.InterMetric {

	span, _ := trace.StartSpanFromContext(ctx, "")
	defer span.ClientFinish(s.TraceClient)

	finalMetrics := make([]samplers.InterMetric, 0, ms.totalLength)
	for _, wm := range tempMetrics {
		for _, c := range wm.counters {
			finalMetrics = append(finalMetrics, c.Flush(s.interval)...)
		}
		for _, g := range wm.gauges {
			finalMetrics = append(finalMetrics, g.Flush()...)
		}
		// if we're a local veneur, then percentiles=nil, and only the local
		// parts (count, min, max) will be flushed
		//
		// if we're a global veneur, aggregates will be nil.
		for _, h := range wm.histograms {
			finalMetrics = append(finalMetrics, h.Flush(s.interval, percentiles, s.HistogramAggregates, false)...)
		}
		for _, t := range wm.timers {
			finalMetrics = append(finalMetrics, t.Flush(s.interval, percentiles, s.HistogramAggregates, false)...)
		}

		// local-only samplers should be flushed in their entirety, since they
		// will not be forwarded
		// we still want percentiles for these, even if we're a local veneur, so
		// we use the original percentile list when flushing them
		for _, h := range wm.localHistograms {
			finalMetrics = append(finalMetrics, h.Flush(s.interval, s.HistogramPercentiles, s.HistogramAggregates, false)...)
		}
		for _, s := range wm.localSets {
			finalMetrics = append(finalMetrics, s.Flush()...)
		}
		for _, t := range wm.localTimers {
			finalMetrics = append(finalMetrics, t.Flush(s.interval, s.HistogramPercentiles, s.HistogramAggregates, false)...)
		}

		for _, status := range wm.localStatusChecks {
			finalMetrics = append(finalMetrics, status.Flush()...)
		}

		// TODO (aditya) refactor this out so we don't
		// have to call IsLocal again
		if !s.IsLocal() {
			// sets have no local parts, so if we're a local veneur, there's
			// nothing to flush at all
			for _, s := range wm.sets {
				finalMetrics = append(finalMetrics, s.Flush()...)
			}

			// also do this for global counters
			// global counters have no local parts, so if we're a local veneur,
			// there's nothing to flush
			for _, gc := range wm.globalCounters {
				finalMetrics = append(finalMetrics, gc.Flush(s.interval)...)
			}

			// and global gauges
			for _, gg := range wm.globalGauges {
				finalMetrics = append(finalMetrics, gg.Flush()...)
			}

			for _, h := range wm.globalHistograms {
				finalMetrics = append(finalMetrics, h.Flush(s.interval, s.HistogramPercentiles, s.HistogramAggregates, true)...)
			}
			for _, h := range wm.globalTimers {
				finalMetrics = append(finalMetrics, h.Flush(s.interval, s.HistogramPercentiles, s.HistogramAggregates, true)...)
			}
		}
	}

	return finalMetrics
}
1040443
func (c *ServicedCli) initMetric() {
	c.app.Commands = append(c.app.Commands, cli.Command{
		Name:        "metric",
		Usage:       "Interact with metrics",
		Description: "",
		Subcommands: []cli.Command{
			{
				Name:         "push",
				Usage:        "Push a metric value",
				Description:  "serviced metric push METRICNAME VALUE",
				Before:       c.cmdMetric,
			},
		},
	})
}
503966
func (h *Histo) Flush(interval time.Duration, percentiles []float64, aggregates HistogramAggregates, global bool) []InterMetric {
	now := time.Now().Unix()
	metrics := make([]InterMetric, 0, aggregates.Count+len(percentiles))
	sinks := routeInfo(h.Tags)

	// The second clause in this if statement can be confusing.
	//
	// The infinity check ensures we don't submit mixed scope histogram summaries. These values
	// are initialized to infinity and reset when a value is sampled locally. This never happens
	// for mixed scope values since we only consume them through the merge path.
	//
	// The global check overrides this for global scope histograms. In this case, we're going to
	// use values from the merged histogram struct.
	//
	// The proliferation of flags here suggests this method is the wrong abstraction, since it
	// requires a lot of forking for different paths. (Read: https://www.sandimetz.com/blog/2016/1/20/the-wrong-abstraction)
	//
	// Think twice before adding to complexity here--it may be worth refactoring how histograms
	// are modeled in mixed scope before continuing with new features.
	if (aggregates.Value&AggregateMax) == AggregateMax && (!math.IsInf(h.LocalMax, 0) || global) {
		// Defensively recopy tags to avoid aliasing bugs in case multiple InterMetrics share the same
		// tag array in the future
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)

		val := float64(h.LocalMax)
		if global {
			val = h.Value.Max()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.max", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}
	if (aggregates.Value&AggregateMin) == AggregateMin && (!math.IsInf(h.LocalMin, 0) || global) {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalMin)
		if global {
			val = h.Value.Min()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.min", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value&AggregateSum) == AggregateSum && (h.LocalSum != 0 || global) {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalSum)
		if global {
			val = h.Value.Sum()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.sum", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value&AggregateAverage) == AggregateAverage && (global || h.LocalSum != 0 && h.LocalWeight != 0) {
		// we need both a rate and a non-zero sum before it will make sense
		// to submit an average
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalSum / h.LocalWeight)
		if global {
			val = h.Value.Sum() / h.Value.Count()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.avg", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value&AggregateCount) == AggregateCount && (h.LocalWeight != 0 || global) {
		// if we haven't received any local samples, then leave this sparse,
		// otherwise it can lead to some misleading zeroes in between the
		// flushes of downstream instances
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalWeight)
		if global {
			val = h.Value.Count()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.count", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      CounterMetric,
			Sinks:     sinks,
		})
	}

	if (aggregates.Value & AggregateMedian) == AggregateMedian {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		metrics = append(
			metrics,
			InterMetric{
				Name:      fmt.Sprintf("%s.median", h.Name),
				Timestamp: now,
				Value:     float64(h.Value.Quantile(0.5)),
				Tags:      tags,
				Type:      GaugeMetric,
				Sinks:     sinks,
			},
		)
	}

	if (aggregates.Value&AggregateHarmonicMean) == AggregateHarmonicMean && (global || h.LocalReciprocalSum != 0 && h.LocalWeight != 0) {
		// we need both a rate and a non-zero sum before it will make sense
		// to submit an average
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		val := float64(h.LocalWeight / h.LocalReciprocalSum)
		if global {
			val = h.Value.Count() / h.Value.ReciprocalSum()
		}
		metrics = append(metrics, InterMetric{
			Name:      fmt.Sprintf("%s.hmean", h.Name),
			Timestamp: now,
			Value:     val,
			Tags:      tags,
			Type:      GaugeMetric,
			Sinks:     sinks,
		})
	}

	for _, p := range percentiles {
		tags := make([]string, len(h.Tags))
		copy(tags, h.Tags)
		metrics = append(
			metrics,
			// TODO Fix to allow for p999, etc
			InterMetric{
				Name:      fmt.Sprintf("%s.%dpercentile", h.Name, int(p*100)),
				Timestamp: now,
				Value:     float64(h.Value.Quantile(p)),
				Tags:      tags,
				Type:      GaugeMetric,
				Sinks:     sinks,
			},
		)
	}

	return metrics
}
952406
func SaveMetrics(p *redis.Pool, name string, metrics MetricsRecord) error {
	for _, agg := range aggregationIntervals {
		if err := SaveMetricsForInterval(p, agg, name, metrics); err != nil {
			return errors.Wrap(err, "save metrics for interval error")
		}
	}

	log.WithFields(log.Fields{
		"name":        name,
		"aggregation": aggregationIntervals,
	}).Info("metrics saved")

	return nil
}
25985
func (s *GetCostForecastInput) SetMetric(v string) *GetCostForecastInput {
	s.Metric = &v
	return s
}
1195457
func (graphite *Graphite) sendMetrics(metrics []Metric) error {
	if graphite.IsNop() {
		for _, metric := range metrics {
			log.Printf("Graphite: %s\n", metric)
		}
		return nil
	}
	zeroed_metric := Metric{} // ignore unintialized metrics
	buf := bytes.NewBufferString("")
	for _, metric := range metrics {
		if metric == zeroed_metric {
			continue // ignore unintialized metrics
		}
		if metric.Timestamp == 0 {
			metric.Timestamp = time.Now().Unix()
		}
		metric_name := ""
		if graphite.Prefix != "" {
			metric_name = fmt.Sprintf("%s.%s", graphite.Prefix, metric.Name)
		} else {
			metric_name = metric.Name
		}
		if graphite.Protocol == "udp" {
			fmt.Fprintf(graphite.conn, "%s %s %d\n", metric_name, metric.Value, metric.Timestamp)
			continue
		}
		buf.WriteString(fmt.Sprintf("%s %s %d\n", metric_name, metric.Value, metric.Timestamp))
	}
	if graphite.Protocol == "tcp" {
		_, err := graphite.conn.Write(buf.Bytes())
		//fmt.Print("Sent msg:", buf.String(), "'")
		if err != nil {
			return err
		}
	}
	return nil
}
404794
func WriteMetrics(r metrics.Registry, w io.Writer) {
	var namedMetrics namedMetricSlice
	r.Each(func(name string, i interface{}) {
		namedMetrics = append(namedMetrics, namedMetric{name, i})
	})

	sort.Sort(namedMetrics)
	for _, namedMetric := range namedMetrics {
		switch metric := namedMetric.m.(type) {
		case metrics.Counter:
			fmt.Fprintf(w, "counter %s\n", namedMetric.name)
			fmt.Fprintf(w, "  count: %9d\n", metric.Count())
		case metrics.Gauge:
			fmt.Fprintf(w, "gauge %s\n", namedMetric.name)
			fmt.Fprintf(w, "  value: %9d\n", metric.Value())
		case metrics.GaugeFloat64:
			fmt.Fprintf(w, "gauge %s\n", namedMetric.name)
			fmt.Fprintf(w, "  value: %f\n", metric.Value())
		case metrics.Healthcheck:
			metric.Check()
			fmt.Fprintf(w, "healthcheck %s\n", namedMetric.name)
			fmt.Fprintf(w, "  error: %v\n", metric.Error())
		case metrics.Histogram:
			h := metric.Snapshot()
			ps := h.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999})
			fmt.Fprintf(w, "histogram %s\n", namedMetric.name)
			fmt.Fprintf(w, "  count=%d, mean=%.2f, stddef=%.2f\n", h.Count(), h.Mean(), h.StdDev())
			fmt.Fprintf(w, "  min=%.2fms median=%.2fms max=%.2fms\n",
				float64(h.Min())/float64(time.Millisecond),
				ps[0]/float64(time.Millisecond),
				float64(h.Max())/float64(time.Millisecond))
			fmt.Fprintf(w, "  %%iles (ms): 75=%.2f 95=%.2f 99=%.2f 99.9=%.2f\n",
				ps[1]/float64(time.Millisecond),
				ps[2]/float64(time.Millisecond),
				ps[3]/float64(time.Millisecond),
				ps[4]/float64(time.Millisecond))
		case metrics.Meter:
			m := metric.Snapshot()
			fmt.Fprintf(w, "meter %s\n", namedMetric.name)
			fmt.Fprintf(w, "  count: %d\n", m.Count())
			fmt.Fprintf(w, "  rates: 1m=%.2f 5m=%.2f 15m=%.2f mean=%.2f\n", m.Rate1(), m.Rate5(), m.Rate15(), m.RateMean())
		case metrics.Timer:
			t := metric.Snapshot()
			ps := t.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999})
			fmt.Fprintf(w, "timer %s\n", namedMetric.name)
			fmt.Fprintf(w, "  count=%d, mean=%.2fms, stddev=%.2fms\n",
				t.Count(), t.Mean()/float64(time.Millisecond), t.StdDev()/float64(time.Millisecond))
			fmt.Fprintf(w, "  min=%.2fms median=%.2fms max=%.2fms\n",
				float64(t.Min())/float64(time.Millisecond),
				ps[0]/float64(time.Millisecond),
				float64(t.Max())/float64(time.Millisecond))
			fmt.Fprintf(w, "  %%iles (ms): 75=%.2f 95=%.2f 99=%.2f 99.9=%.2f\n",
				ps[1]/float64(time.Millisecond),
				ps[2]/float64(time.Millisecond),
				ps[3]/float64(time.Millisecond),
				ps[4]/float64(time.Millisecond))
			fmt.Fprintf(w, "  rates: 1m=%.2f 5m=%.2f 15m=%.2f mean=%.2f\n", t.Rate1(), t.Rate5(), t.Rate15(), t.RateMean())
		}
	}
}
247624
func generateInstanceMetric(vals map[string]interface{}, name string, instances []string, t MetricType, s MetricSemantics, u MetricUnit, desc ...string) (*pcpInstanceMetric, error) {
	indomname := name + ".indom"
	indom, err := NewPCPInstanceDomain(indomname, instances)
	if err != nil {
		return nil, errors.Errorf("cannot create indom, error: %v", err)
	}

	d, err := newpcpMetricDesc(name, t, s, u, desc...)
	if err != nil {
		return nil, err
	}

	return newpcpInstanceMetric(vals, indom, d)
}
199763
func NewMetricSystem(interval time.Duration, sysStats bool) *MetricSystem {
	ms := &MetricSystem{
		percentiles: map[string]float64{
			"%s_min":   0,
			"%s_50":    .5,
			"%s_75":    .75,
			"%s_90":    .9,
			"%s_95":    .95,
			"%s_99":    .99,
			"%s_99.9":  .999,
			"%s_99.99": .9999,
			"%s_max":   1,
		},
		interval:                        interval,
		subscribeToRawMetrics:           make(chan chan *RawMetricSet, 64),
		unsubscribeFromRawMetrics:       make(chan chan *RawMetricSet, 64),
		subscribeToProcessedMetrics:     make(chan chan *ProcessedMetricSet, 64),
		unsubscribeFromProcessedMetrics: make(chan chan *ProcessedMetricSet, 64),
		rawSubscribers:                  make(map[chan *RawMetricSet]struct{}),
		rawBadSubscribers:               make(map[chan *RawMetricSet]int),
		processedSubscribers:            make(map[chan *ProcessedMetricSet]struct{}),
		processedBadSubscribers:         make(map[chan *ProcessedMetricSet]int),
		counterStore:                    make(map[string]*uint64),
		counterCache:                    make(map[string]*uint64),
		histogramCache:                  make(map[string]map[int16]*uint64),
		histogramCountStore:             make(map[string]*uint64),
		gaugeFuncs:                      make(map[string]func() float64),
		shutdownChan:                    make(chan struct{}),
	}
	if sysStats {
		ms.gaugeFuncsMu.Lock()
		ms.gaugeFuncs["sys.Alloc"] = func() float64 {
			memStats := new(runtime.MemStats)
			runtime.ReadMemStats(memStats)
			return float64(memStats.Alloc)
		}
		ms.gaugeFuncs["sys.NumGC"] = func() float64 {
			memStats := new(runtime.MemStats)
			runtime.ReadMemStats(memStats)
			return float64(memStats.NumGC)
		}
		ms.gaugeFuncs["sys.PauseTotalNs"] = func() float64 {
			memStats := new(runtime.MemStats)
			runtime.ReadMemStats(memStats)
			return float64(memStats.PauseTotalNs)
		}
		ms.gaugeFuncs["sys.NumGoroutine"] = func() float64 {
			return float64(runtime.NumGoroutine())
		}
		ms.gaugeFuncsMu.Unlock()
	}
	return ms
}
152669
func NewMetrics() *Metrics {
	return &Metrics{Status: StatusUp, Health: 1, Uptime: 0, lastTs: time.Now()}
}
809108
func (s *Event) SetMetrics(v map[string]*float64) *Event {
	s.Metrics = v
	return s
}
774075
func NewMetrics(conf *metrics.Config, sink metrics.MetricSink) (err error) {
	m, err := metrics.NewGlobal(conf, sink)
	SetMetrics(m)

	return nil
}
667985
func (w *Worker) ProcessMetric(m *samplers.UDPMetric) {
	w.mutex.Lock()
	defer w.mutex.Unlock()
	w.processed++
	w.wm.Upsert(m.MetricKey, m.Scope, m.Tags)

	switch m.Type {
	case counterTypeName:
		if m.Scope == samplers.GlobalOnly {
			w.wm.globalCounters[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		} else {
			w.wm.counters[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		}
	case gaugeTypeName:
		if m.Scope == samplers.GlobalOnly {
			w.wm.globalGauges[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		} else {
			w.wm.gauges[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		}
	case histogramTypeName:
		if m.Scope == samplers.LocalOnly {
			w.wm.localHistograms[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		} else if m.Scope == samplers.GlobalOnly {
			w.wm.globalHistograms[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		} else {
			w.wm.histograms[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		}
	case setTypeName:
		if m.Scope == samplers.LocalOnly {
			w.wm.localSets[m.MetricKey].Sample(m.Value.(string), m.SampleRate)
		} else {
			w.wm.sets[m.MetricKey].Sample(m.Value.(string), m.SampleRate)
		}
	case timerTypeName:
		if m.Scope == samplers.LocalOnly {
			w.wm.localTimers[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		} else if m.Scope == samplers.GlobalOnly {
			w.wm.globalTimers[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		} else {
			w.wm.timers[m.MetricKey].Sample(m.Value.(float64), m.SampleRate)
		}
	case statusTypeName:
		v := float64(m.Value.(ssf.SSFSample_Status))
		w.wm.localStatusChecks[m.MetricKey].Sample(v, m.SampleRate, m.Message, m.HostName)
	default:
		log.WithField("type", m.Type).Error("Unknown metric type for processing")
	}
}
333852
func (i *InmemSink) getInterval() *IntervalMetrics {
	intv := time.Now().Truncate(i.interval)
	if m := i.getExistingInterval(intv); m != nil {
		return m
	}
	return i.createInterval(intv)
}
509293
func (c *Counter) Flush(interval time.Duration) []InterMetric {
	tags := make([]string, len(c.Tags))
	copy(tags, c.Tags)
	return []InterMetric{{
		Name:      c.Name,
		Timestamp: time.Now().Unix(),
		Value:     float64(c.value),
		Tags:      tags,
		Type:      CounterMetric,
		Sinks:     routeInfo(tags),
	}}
}
1072030
func (s *StatsdSink) flushMetrics() {
	var sock net.Conn
	var err error
	var wait <-chan time.Time
	ticker := time.NewTicker(flushInterval)
	defer ticker.Stop()

CONNECT:
	// Create a buffer
	buf := bytes.NewBuffer(nil)

	// Attempt to connect
	sock, err = net.Dial("udp", s.addr)
	if err != nil {
		log.Printf("[ERR] Error connecting to statsd! Err: %s", err)
		goto WAIT
	}

	for {
		select {
		case metric, ok := <-s.metricQueue:
			// Get a metric from the queue
			if !ok {
				goto QUIT
			}

			// Check if this would overflow the packet size
			if len(metric)+buf.Len() > statsdMaxLen {
				_, err := sock.Write(buf.Bytes())
				buf.Reset()
				if err != nil {
					log.Printf("[ERR] Error writing to statsd! Err: %s", err)
					goto WAIT
				}
			}

			// Append to the buffer
			buf.WriteString(metric)

		case <-ticker.C:
			if buf.Len() == 0 {
				continue
			}

			_, err := sock.Write(buf.Bytes())
			buf.Reset()
			if err != nil {
				log.Printf("[ERR] Error flushing to statsd! Err: %s", err)
				goto WAIT
			}
		}
	}

WAIT:
	// Wait for a while
	wait = time.After(time.Duration(5) * time.Second)
	for {
		select {
		// Dequeue the messages to avoid backlog
		case _, ok := <-s.metricQueue:
			if !ok {
				goto QUIT
			}
		case <-wait:
			goto CONNECT
		}
	}
QUIT:
	s.metricQueue = nil
}
481859
func (s *GetReservationCoverageInput) SetMetrics(v []*string) *GetReservationCoverageInput {
	s.Metrics = v
	return s
}