Goprometheusmetrics条目自动回收与清理方法

// GaugeVec is a Collector that bundles a set of Gauges that all share the same
// Desc, but have different values for their variable labels. This is used if
// you want to count the same thing partitioned by various dimensions
// (e.g. number of operations queued, partitioned by user and operation
// type). Create instances with NewGaugeVec.
type GaugeVec struct {
	*MetricVec
}
type MetricVec struct {
	*metricMap
	curry []curriedLabelValue
	// hashAdd and hashAddByte can be replaced for testing collision handling.
	hashAdd     func(h uint64, s string) uint64
	hashAddByte func(h uint64, b byte) uint64
}
// metricMap is a helper for metricVec and shared between differently curried
// metricVecs.
type metricMap struct {
	mtx       sync.RWMutex // Protects metrics.
	metrics   map[uint64][]metricWithLabelValues  // 真正的数据存储位置
	desc      *Desc
	newMetric func(labelValues ...string) Metric
}

// Collect implements Collector.
func (m *metricMap) Collect(ch chan<- Metric) {
	m.mtx.RLock()
	defer m.mtx.RUnlock()
	// 遍历 map
	for _, metrics := range m.metrics {
		for _, metric := range metrics {
			ch <- metric.metric // 读取数据到通道
		}
	}
}

// To create Gauge instances, use NewGauge.
type Gauge interface {
	Metric
	Collector
	// Set sets the Gauge to an arbitrary value.
	Set(float64)
	// Inc increments the Gauge by 1. Use Add to increment it by arbitrary
	// values.
	Inc()
	// Dec decrements the Gauge by 1. Use Sub to decrement it by arbitrary
	// values.
	Dec()
	// Add adds the given value to the Gauge. (The value can be negative,
	// resulting in a decrease of the Gauge.)
	Add(float64)
	// Sub subtracts the given value from the Gauge. (The value can be
	// negative, resulting in an increase of the Gauge.)
	Sub(float64)
	// SetToCurrentTime sets the Gauge to the current Unix time in seconds.
	SetToCurrentTime()
}
func NewGauge(opts GaugeOpts) Gauge {
	desc := NewDesc(
		BuildFQName(opts.Namespace, opts.Subsystem, opts.Name),
		opts.Help,
		nil,
		opts.ConstLabels,
	)
	result := &gauge{desc: desc, labelPairs: desc.constLabelPairs}
	result.init(result) // Init self-collection.
	return result
}
type gauge struct {
	// valBits contains the bits of the represented float64 value. It has
	// to go first in the struct to guarantee alignment for atomic
	// operations.  http://golang.org/pkg/sync/atomic/#pkg-note-BUG
	valBits uint64
	selfCollector
	desc       *Desc
	labelPairs []*dto.LabelPair
}
func (g *gauge) Set(val float64) {
	atomic.StoreUint64(&g.valBits, math.Float64bits(val))  // 写入数据到变量
}

// getOrCreateMetricWithLabelValues retrieves the metric by hash and label value
// or creates it and returns the new one.
//
// This function holds the mutex.
func (m *metricMap) getOrCreateMetricWithLabelValues(hash uint64, lvs []string, curry []curriedLabelValue,) Metric { // 返回了一个接口
	m.mtx.RLock()
	metric, ok := m.getMetricWithHashAndLabelValues(hash, lvs, curry)
	m.mtx.RUnlock()
	if ok {
		return metric
	}
	m.mtx.Lock()
	defer m.mtx.Unlock()
	metric, ok = m.getMetricWithHashAndLabelValues(hash, lvs, curry)
	if !ok {
		inlinedLVs := inlineLabelValues(lvs, curry)
		metric = m.newMetric(inlinedLVs...)
		m.metrics[hash] = append(m.metrics[hash], metricWithLabelValues{values: inlinedLVs, metric: metric})  // 这里写入 metricMap.metrics
	}
	return metric
}
// A Metric models a single sample value with its meta data being exported to
// Prometheus. Implementations of Metric in this package are Gauge, Counter,
// Histogram, Summary, and Untyped.
type Metric interface { // 哦哦哦哦，是接口啊。Gauge 实现这个接口
	// Desc returns the descriptor for the Metric. This method idempotently
	// returns the same descriptor throughout the lifetime of the
	// Metric. The returned descriptor is immutable by contract. A Metric
	// unable to describe itself must return an invalid descriptor (created
	// with NewInvalidDesc).
	Desc() *Desc
	// Write encodes the Metric into a "Metric" Protocol Buffer data
	// transmission object.
	//
	// Metric implementations must observe concurrency safety as reads of
	// this metric may occur at any time, and any blocking occurs at the
	// expense of total perfORMance of rendering all registered
	// metrics. ideally, Metric implementations should support concurrent
	// readers.
	//
	// While populating dto.Metric, it is the responsibility of the
	// implementation to ensure validity of the Metric protobuf (like valid
	// UTF-8 strings or syntactically valid metric and label names). It is
	// recommended to sort labels lexicographically. Callers of Write should
	// still make sure of sorting if they depend on it.
	Write(*dto.Metric) error
	// TODO(beorn7): The original rationale of passing in a pre-allocated
	// dto.Metric protobuf to save allocations has disappeared. The
	// signature of this method should be changed to "Write() (*dto.Metric,
	// error)".
}

package main
import (
	"context"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"net/http"
	"strconv"
	"sync"
	"time"
)
type metricsMetaData struct {
	UpdatedAt int64
	Labels    []string
}
func main() {
	var wg sync.WaitGroup
	var status sync.Map
	vec := prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Namespace: "app",
			Name:      "running_status",
		}, []string{"id"},
	)
	prometheus.MustRegister(vec)
	defer prometheus.Unregister(vec)
	// 写入数据
	for i := 0; i < 10; i++ {
		labels := strconv.Itoa(i)
		vec.WithLabelValues(labels).Set(1)                                                            // 写入 metric 条目
		status.Store(labels, metricsMetaData{UpdatedAt: time.Now().Unix(), Labels: []string{labels}}) // 写入状态
	}
	// 创建退出 ctx
	stopCtx, stopCancel := context.WithCancel(context.Background())
	// 启动清理器
	go func(ctx *context.Context, g *sync.WaitGroup) {
		defer g.Done()
		ticker := time.NewTicker(time.Second * 2)
		for {
			select {
			case <-ticker.C:
				now := time.Now().Unix()
				status.Range(func(key, value interface{}) bool {
					if now-value.(metricsMetaData).UpdatedAt > 5 {
						vec.DeleteLabelValues(value.(metricsMetaData).Labels...) // 删除 metrics 条目
						status.Delete(key)                                       // 删除 map 中的记录
					}
					return true
				})
				break
			case <-(*ctx).Done():
				return
			}
		}
	}(&stopCtx, &wg)
	wg.Add(1)
	// 创建 http
	http.Handle("/metrics", promhttp.Handler())
	srv := http.Server{Addr: "0.0.0.0:8080"}
	// 启动 http server
	go func(srv *http.Server, g *sync.WaitGroup) {
		defer g.Done()
		_ = srv.ListenAndServe()
	}(&srv, &wg)
	wg.Add(1)
	// 退出
	time.Sleep(time.Second * 10)
	stopCancel()
	_ = srv.Shutdown(context.Background())
	wg.Wait()
}