Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 prometheus.go // +build go1.9 package prometheus import ( "fmt" "log" "regexp" "strings" "sync" "time" "github.com/armon/go-metrics" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/push" ) var ( // DefaultPrometheusOpts is the default set of options used when creating a // PrometheusSink. DefaultPrometheusOpts = PrometheusOpts{ Expiration: 60 * time.Second, Name: "default_prometheus_sink", } ) // PrometheusOpts is used to configure the Prometheus Sink type PrometheusOpts struct { // Expiration is the duration a metric is valid for, after which it will be // untracked. If the value is zero, a metric is never expired. Expiration time.Duration Registerer prometheus.Registerer // Gauges, Summaries, and Counters allow us to pre-declare metrics by giving // their Name, Help, and ConstLabels to the PrometheusSink when it is created. // Metrics declared in this way will be initialized at zero and will not be // deleted or altered when their expiry is reached. // // Ex: PrometheusOpts{ // Expiration: 10 * time.Second, // Gauges: []GaugeDefinition{ // { // Name: []string{ "application", "component", "measurement"}, // Help: "application_component_measurement provides an example of how to declare static metrics", // ConstLabels: []metrics.Label{ { Name: "my_label", Value: "does_not_change" }, }, // }, // }, // } GaugeDefinitions []GaugeDefinition SummaryDefinitions []SummaryDefinition CounterDefinitions []CounterDefinition Name string } type PrometheusSink struct { // If these will ever be copied, they should be converted to sync.Map values and initialized appropriately gauges sync.Map summaries sync.Map counters sync.Map expiration time.Duration help map[string]string name string } // GaugeDefinition can be provided to PrometheusOpts to declare a constant gauge that is not deleted on expiry. type GaugeDefinition struct { Name []string ConstLabels []metrics.Label Help string } type gauge struct { prometheus.Gauge updatedAt time.Time // canDelete is set if the metric is created during runtime so we know it's ephemeral and can delete it on expiry. canDelete bool } // SummaryDefinition can be provided to PrometheusOpts to declare a constant summary that is not deleted on expiry. type SummaryDefinition struct { Name []string ConstLabels []metrics.Label Help string } type summary struct { prometheus.Summary updatedAt time.Time canDelete bool } // CounterDefinition can be provided to PrometheusOpts to declare a constant counter that is not deleted on expiry. type CounterDefinition struct { Name []string ConstLabels []metrics.Label Help string } type counter struct { prometheus.Counter updatedAt time.Time canDelete bool } // NewPrometheusSink creates a new PrometheusSink using the default options. func NewPrometheusSink() (PrometheusSink, error) { return NewPrometheusSinkFrom(DefaultPrometheusOpts) } // NewPrometheusSinkFrom creates a new PrometheusSink using the passed options. func NewPrometheusSinkFrom(opts PrometheusOpts) (PrometheusSink, error) { name := opts.Name if name == "" { name = "default_prometheus_sink" } sink := &PrometheusSink{ gauges: sync.Map{}, summaries: sync.Map{}, counters: sync.Map{}, expiration: opts.Expiration, help: make(map[string]string), name: name, } initGauges(&sink.gauges, opts.GaugeDefinitions, sink.help) initSummaries(&sink.summaries, opts.SummaryDefinitions, sink.help) initCounters(&sink.counters, opts.CounterDefinitions, sink.help) reg := opts.Registerer if reg == nil { reg = prometheus.DefaultRegisterer } return sink, reg.Register(sink) } // Describe sends a Collector.Describe value from the descriptor created around PrometheusSink.Name // Note that we cannot describe all the metrics (gauges, counters, summaries) in the sink as // metrics can be added at any point during the lifecycle of the sink, which does not respect // the idempotency aspect of the Collector.Describe() interface func (p PrometheusSink) Describe(c chan<- prometheus.Desc) { // dummy value to be able to register and unregister "empty" sinks // Note this is not actually retained in the PrometheusSink so this has no side effects // on the caller's sink. So it shouldn't show up to any of its consumers. prometheus.NewGauge(prometheus.GaugeOpts{Name: p.name, Help: p.name}).Describe(c) } // Collect meets the collection interface and allows us to enforce our expiration // logic to clean up ephemeral metrics if their value haven't been set for a // duration exceeding our allowed expiration time. func (p PrometheusSink) Collect(c chan<- prometheus.Metric) { p.collectAtTime(c, time.Now()) } // collectAtTime allows internal testing of the expiry based logic here without // mocking clocks or making tests timing sensitive. func (p PrometheusSink) collectAtTime(c chan<- prometheus.Metric, t time.Time) { expire := p.expiration != 0 p.gauges.Range(func(k, v interface{}) bool { if v == nil { return true } g := v.(gauge) lastUpdate := g.updatedAt if expire && lastUpdate.Add(p.expiration).Before(t) { if g.canDelete { p.gauges.Delete(k) return true } } g.Collect(c) return true }) p.summaries.Range(func(k, v interface{}) bool { if v == nil { return true } s := v.(summary) lastUpdate := s.updatedAt if expire && lastUpdate.Add(p.expiration).Before(t) { if s.canDelete { p.summaries.Delete(k) return true } } s.Collect(c) return true }) p.counters.Range(func(k, v interface{}) bool { if v == nil { return true } count := v.(counter) lastUpdate := count.updatedAt if expire && lastUpdate.Add(p.expiration).Before(t) { if count.canDelete { p.counters.Delete(k) return true } } count.Collect(c) return true }) } func initGauges(m sync.Map, gauges []GaugeDefinition, help map[string]string) { for _, g := range gauges { key, hash := flattenKey(g.Name, g.ConstLabels) help[fmt.Sprintf("gauge.%s", key)] = g.Help pG := prometheus.NewGauge(prometheus.GaugeOpts{ Name: key, Help: g.Help, ConstLabels: prometheusLabels(g.ConstLabels), }) m.Store(hash, &gauge{Gauge: pG}) } return } func initSummaries(m sync.Map, summaries []SummaryDefinition, help map[string]string) { for _, s := range summaries { key, hash := flattenKey(s.Name, s.ConstLabels) help[fmt.Sprintf("summary.%s", key)] = s.Help pS := prometheus.NewSummary(prometheus.SummaryOpts{ Name: key, Help: s.Help, MaxAge: 10 * time.Second, ConstLabels: prometheusLabels(s.ConstLabels), Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, }) m.Store(hash, &summary{Summary: pS}) } return } func initCounters(m sync.Map, counters []CounterDefinition, help map[string]string) { for _, c := range counters { key, hash := flattenKey(c.Name, c.ConstLabels) help[fmt.Sprintf("counter.%s", key)] = c.Help pC := prometheus.NewCounter(prometheus.CounterOpts{ Name: key, Help: c.Help, ConstLabels: prometheusLabels(c.ConstLabels), }) m.Store(hash, &counter{Counter: pC}) } return } var forbiddenChars = regexp.MustCompile("[ .=\\-/]") func flattenKey(parts []string, labels []metrics.Label) (string, string) { key := strings.Join(parts, "_") key = forbiddenChars.ReplaceAllString(key, "_") hash := key for _, label := range labels { hash += fmt.Sprintf(";%s=%s", label.Name, label.Value) } return key, hash } func prometheusLabels(labels []metrics.Label) prometheus.Labels { l := make(prometheus.Labels) for _, label := range labels { l[label.Name] = label.Value } return l } func (p PrometheusSink) SetGauge(parts []string, val float32) { p.SetGaugeWithLabels(parts, val, nil) } func (p PrometheusSink) SetGaugeWithLabels(parts []string, val float32, labels []metrics.Label) { key, hash := flattenKey(parts, labels) pg, ok := p.gauges.Load(hash) // The sync.Map underlying gauges stores pointers to our structs. If we need to make updates, // rather than modifying the underlying value directly, which would be racy, we make a local // copy by dereferencing the pointer we get back, making the appropriate changes, and then // storing a pointer to our local copy. The underlying Prometheus types are threadsafe, // so there's no issues there. It's possible for racy updates to occur to the updatedAt // value, but since we're always setting it to time.Now(), it doesn't really matter. if ok { localGauge := pg.(gauge) localGauge.Set(float64(val)) localGauge.updatedAt = time.Now() p.gauges.Store(hash, &localGauge) // The gauge does not exist, create the gauge and allow it to be deleted } else { help := key existingHelp, ok := p.help[fmt.Sprintf("gauge.%s", key)] if ok { help = existingHelp } g := prometheus.NewGauge(prometheus.GaugeOpts{ Name: key, Help: help, ConstLabels: prometheusLabels(labels), }) g.Set(float64(val)) pg = &gauge{ Gauge: g, updatedAt: time.Now(), canDelete: true, } p.gauges.Store(hash, pg) } } func (p PrometheusSink) AddSample(parts []string, val float32) { p.AddSampleWithLabels(parts, val, nil) } func (p PrometheusSink) AddSampleWithLabels(parts []string, val float32, labels []metrics.Label) { key, hash := flattenKey(parts, labels) ps, ok := p.summaries.Load(hash) // Does the summary already exist for this sample type? if ok { localSummary := ps.(summary) localSummary.Observe(float64(val)) localSummary.updatedAt = time.Now() p.summaries.Store(hash, &localSummary) // The summary does not exist, create the Summary and allow it to be deleted } else { help := key existingHelp, ok := p.help[fmt.Sprintf("summary.%s", key)] if ok { help = existingHelp } s := prometheus.NewSummary(prometheus.SummaryOpts{ Name: key, Help: help, MaxAge: 10 time.Second, ConstLabels: prometheusLabels(labels), Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, }) s.Observe(float64(val)) ps = &summary{ Summary: s, updatedAt: time.Now(), canDelete: true, } p.summaries.Store(hash, ps) } } // EmitKey is not implemented. Prometheus doesn’t offer a type for which an // arbitrary number of values is retained, as Prometheus works with a pull // model, rather than a push model. func (p PrometheusSink) EmitKey(key []string, val float32) { } func (p PrometheusSink) IncrCounter(parts []string, val float32) { p.IncrCounterWithLabels(parts, val, nil) } func (p PrometheusSink) IncrCounterWithLabels(parts []string, val float32, labels []metrics.Label) { key, hash := flattenKey(parts, labels) pc, ok := p.counters.Load(hash) // Does the counter exist? if ok { localCounter := pc.(counter) localCounter.Add(float64(val)) localCounter.updatedAt = time.Now() p.counters.Store(hash, &localCounter) // The counter does not exist yet, create it and allow it to be deleted } else { help := key existingHelp, ok := p.help[fmt.Sprintf("counter.%s", key)] if ok { help = existingHelp } c := prometheus.NewCounter(prometheus.CounterOpts{ Name: key, Help: help, ConstLabels: prometheusLabels(labels), }) c.Add(float64(val)) pc = &counter{ Counter: c, updatedAt: time.Now(), canDelete: true, } p.counters.Store(hash, pc) } } // PrometheusPushSink wraps a normal prometheus sink and provides an address and facilities to export it to an address // on an interval. type PrometheusPushSink struct { PrometheusSink pusher push.Pusher address string pushInterval time.Duration stopChan chan struct{} } // NewPrometheusPushSink creates a PrometheusPushSink by taking an address, interval, and destination name. func NewPrometheusPushSink(address string, pushInterval time.Duration, name string) (PrometheusPushSink, error) { promSink := &PrometheusSink{ gauges: sync.Map{}, summaries: sync.Map{}, counters: sync.Map{}, expiration: 60 * time.Second, name: "default_prometheus_sink", } pusher := push.New(address, name).Collector(promSink) sink := &PrometheusPushSink{ promSink, pusher, address, pushInterval, make(chan struct{}), } sink.flushMetrics() return sink, nil } func (s PrometheusPushSink) flushMetrics() { ticker := time.NewTicker(s.pushInterval) go func() { for { select { case <-ticker.C: err := s.pusher.Push() if err != nil { log.Printf("[ERR] Error pushing to Prometheus! Err: %s", err) } case <-s.stopChan: ticker.Stop() return } } }() } func (s PrometheusPushSink) Shutdown() { close(s.stopChan) }	12.59 KB	2025-05-25 12:27
📄 prometheus_test.go package prometheus import ( "errors" "fmt" "log" "net/http" "net/http/httptest" "net/url" "reflect" "strings" "testing" "time" "github.com/golang/protobuf/proto" dto "github.com/prometheus/client_model/go" "github.com/armon/go-metrics" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/expfmt" ) const ( TestHostname = "test_hostname" ) func TestNewPrometheusSinkFrom(t testing.T) { reg := prometheus.NewRegistry() sink, err := NewPrometheusSinkFrom(PrometheusOpts{ Registerer: reg, }) if err != nil { t.Fatalf("err = %v, want nil", err) } //check if register has a sink by unregistering it. ok := reg.Unregister(sink) if !ok { t.Fatalf("Unregister(sink) = false, want true") } } func TestNewPrometheusSink(t testing.T) { sink, err := NewPrometheusSink() if err != nil { t.Fatalf("err = %v, want nil", err) } //check if register has a sink by unregistering it. ok := prometheus.Unregister(sink) if !ok { t.Fatalf("Unregister(sink) = false, want true") } } // TestMultiplePrometheusSink tests registering multiple sinks on the same registerer with different descriptors func TestMultiplePrometheusSink(t testing.T) { gaugeDef := GaugeDefinition{ Name: []string{"my", "test", "gauge"}, Help: "A gauge for testing? How helpful!", } cfg := PrometheusOpts{ Expiration: 5 time.Second, GaugeDefinitions: append([]GaugeDefinition{}, gaugeDef), SummaryDefinitions: append([]SummaryDefinition{}), CounterDefinitions: append([]CounterDefinition{}), Name: "sink1", } sink1, err := NewPrometheusSinkFrom(cfg) if err != nil { t.Fatalf("err = %v, want nil", err) } reg := prometheus.DefaultRegisterer if reg == nil { t.Fatalf("Expected default register to be non nil, got nil.") } gaugeDef2 := GaugeDefinition{ Name: []string{"my2", "test", "gauge"}, Help: "A gauge for testing? How helpful!", } cfg2 := PrometheusOpts{ Expiration: 15 * time.Second, GaugeDefinitions: append([]GaugeDefinition{}, gaugeDef2), SummaryDefinitions: append([]SummaryDefinition{}), CounterDefinitions: append([]CounterDefinition{}), // commenting out the name to point out that the default name will be used here instead // Name: "sink2", } sink2, err := NewPrometheusSinkFrom(cfg2) if err != nil { t.Fatalf("err = %v, want nil", err) } //check if register has a sink by unregistering it. ok := reg.Unregister(sink1) if !ok { t.Fatalf("Unregister(sink) = false, want true") } //check if register has a sink by unregistering it. ok = reg.Unregister(sink2) if !ok { t.Fatalf("Unregister(sink) = false, want true") } } func TestDefinitions(t testing.T) { gaugeDef := GaugeDefinition{ Name: []string{"my", "test", "gauge"}, Help: "A gauge for testing? How helpful!", } summaryDef := SummaryDefinition{ Name: []string{"my", "test", "summary"}, Help: "A summary for testing? How helpful!", } counterDef := CounterDefinition{ Name: []string{"my", "test", "counter"}, Help: "A counter for testing? How helpful!", } // PrometheusSink config w/ definitions for each metric type cfg := PrometheusOpts{ Expiration: 5 time.Second, GaugeDefinitions: append([]GaugeDefinition{}, gaugeDef), SummaryDefinitions: append([]SummaryDefinition{}, summaryDef), CounterDefinitions: append([]CounterDefinition{}, counterDef), } sink, err := NewPrometheusSinkFrom(cfg) if err != nil { t.Fatalf("err = %v, want nil", err) } defer prometheus.Unregister(sink) // We can't just len(x) where x is a sync.Map, so we range over the single item and assert the name in our metric // definition matches the key we have for the map entry. Should fail if any metrics exist that aren't defined, or if // the defined metrics don't exist. sink.gauges.Range(func(key, value interface{}) bool { name, _ := flattenKey(gaugeDef.Name, gaugeDef.ConstLabels) if name != key { t.Fatalf("expected my_test_gauge, got #{name}") } return true }) sink.summaries.Range(func(key, value interface{}) bool { name, _ := flattenKey(summaryDef.Name, summaryDef.ConstLabels) if name != key { t.Fatalf("expected my_test_summary, got #{name}") } return true }) sink.counters.Range(func(key, value interface{}) bool { name, _ := flattenKey(counterDef.Name, counterDef.ConstLabels) if name != key { t.Fatalf("expected my_test_counter, got #{name}") } return true }) // Set a bunch of values sink.SetGauge(gaugeDef.Name, 42) sink.AddSample(summaryDef.Name, 42) sink.IncrCounter(counterDef.Name, 1) // Test that the expiry behavior works as expected. First pick a time which // is after all the actual updates above. timeAfterUpdates := time.Now() // Buffer the chan to make sure it doesn't block. We expect only 3 metrics to // be produced but give some extra room as this will hang the test if we don't // have a big enough buffer. ch := make(chan prometheus.Metric, 10) // Collect the metrics as if it's some time in the future, way beyond the 5 // second expiry. sink.collectAtTime(ch, timeAfterUpdates.Add(10time.Second)) // We should see all the metrics desired Expiry behavior expectedNum := 3 for i := 0; i < expectedNum; i++ { select { case m := <-ch: // m is a prometheus.Metric the only thing we can do is Write it to a // protobuf type and read from there. var pb dto.Metric if err := m.Write(&pb); err != nil { t.Fatalf("unexpected error reading metric: %s", err) } desc := m.Desc().String() switch { case pb.Counter != nil: if !strings.Contains(desc, counterDef.Help) { t.Fatalf("expected counter to include correct help=%s, but was %s", counterDef.Help, m.Desc().String()) } // Counters should _not_ reset. We could assert not nil too but that // would be a bug in prometheus client code so assume it's never nil... if pb.Counter.Value != float64(1) { t.Fatalf("expected defined counter to have value 42 after expiring, got %f", pb.Counter.Value) } case pb.Gauge != nil: if !strings.Contains(desc, gaugeDef.Help) { t.Fatalf("expected gauge to include correct help=%s, but was %s", gaugeDef.Help, m.Desc().String()) } // Gauges should _not_ reset. We could assert not nil too but that // would be a bug in prometheus client code so assume it's never nil... if pb.Gauge.Value != float64(42) { t.Fatalf("expected defined gauge to have value 42 after expiring, got %f", pb.Gauge.Value) } case pb.Summary != nil: if !strings.Contains(desc, summaryDef.Help) { t.Fatalf("expected summary to include correct help=%s, but was %s", summaryDef.Help, m.Desc().String()) } // Summaries should not be reset. Previous behavior here did attempt to // reset them by calling Observe(NaN) which results in all values being // set to NaN but doesn't actually clear the time window of data // predictably so future observations could also end up as NaN until the // NaN sample has aged out of the window. Since the summary is already // aging out a fixed time window (we fix it a 10 seconds currently for // all summaries and it's not affected by Expiration option), there's no // point in trying to reset it after "expiry". if pb.Summary.SampleSum != float64(42) { t.Fatalf("expected defined summary sum to have value 42 after expiring, got %f", pb.Summary.SampleSum) } default: t.Fatalf("unexpected metric type %v", pb) } case <-time.After(100 time.Millisecond): t.Fatalf("Timed out waiting to collect expected metric. Got %d, want %d", i, expectedNum) } } } func MockGetHostname() string { return TestHostname } func fakeServer(q chan string) httptest.Server { handler := func(w http.ResponseWriter, r http.Request) { w.WriteHeader(202) w.Header().Set("Content-Type", "application/json") defer r.Body.Close() dec := expfmt.NewDecoder(r.Body, expfmt.FmtProtoDelim) m := &dto.MetricFamily{} dec.Decode(m) expectedm := &dto.MetricFamily{ Name: proto.String("default_one_two"), Help: proto.String("default_one_two"), Type: dto.MetricType_GAUGE.Enum(), Metric: []dto.Metric{ &dto.Metric{ Label: []dto.LabelPair{ &dto.LabelPair{ Name: proto.String("host"), Value: proto.String(MockGetHostname()), }, }, Gauge: &dto.Gauge{ Value: proto.Float64(42), }, }, }, } if !reflect.DeepEqual(m, expectedm) { msg := fmt.Sprintf("Unexpected samples extracted, got: %+v, want: %+v", m, expectedm) q <- errors.New(msg).Error() } else { q <- "ok" } } return httptest.NewServer(http.HandlerFunc(handler)) } func TestSetGauge(t testing.T) { q := make(chan string) server := fakeServer(q) defer server.Close() u, err := url.Parse(server.URL) if err != nil { log.Fatal(err) } host := u.Hostname() + ":" + u.Port() sink, err := NewPrometheusPushSink(host, time.Second, "pushtest") metricsConf := metrics.DefaultConfig("default") metricsConf.HostName = MockGetHostname() metricsConf.EnableHostnameLabel = true metrics.NewGlobal(metricsConf, sink) metrics.SetGauge([]string{"one", "two"}, 42) response := <-q if response != "ok" { t.Fatal(response) } } func TestDefinitionsWithLabels(t testing.T) { gaugeDef := GaugeDefinition{ Name: []string{"my", "test", "gauge"}, Help: "A gauge for testing? How helpful!", } summaryDef := SummaryDefinition{ Name: []string{"my", "test", "summary"}, Help: "A summary for testing? How helpful!", } counterDef := CounterDefinition{ Name: []string{"my", "test", "counter"}, Help: "A counter for testing? How helpful!", } // PrometheusSink config w/ definitions for each metric type cfg := PrometheusOpts{ Expiration: 5 * time.Second, GaugeDefinitions: append([]GaugeDefinition{}, gaugeDef), SummaryDefinitions: append([]SummaryDefinition{}, summaryDef), CounterDefinitions: append([]CounterDefinition{}, counterDef), } sink, err := NewPrometheusSinkFrom(cfg) if err != nil { t.Fatalf("err =%#v, want nil", err) } defer prometheus.Unregister(sink) if len(sink.help) != 3 { t.Fatalf("Expected len(sink.help) to be 3, was %d: %#v", len(sink.help), sink.help) } sink.SetGaugeWithLabels(gaugeDef.Name, 42.0, []metrics.Label{ {Name: "version", Value: "some info"}, }) sink.gauges.Range(func(key, value interface{}) bool { localGauge := value.(gauge) if !strings.Contains(localGauge.Desc().String(), gaugeDef.Help) { t.Fatalf("expected gauge to include correct help=%s, but was %s", gaugeDef.Help, localGauge.Desc().String()) } return true }) sink.AddSampleWithLabels(summaryDef.Name, 42.0, []metrics.Label{ {Name: "version", Value: "some info"}, }) sink.summaries.Range(func(key, value interface{}) bool { metric := value.(summary) if !strings.Contains(metric.Desc().String(), summaryDef.Help) { t.Fatalf("expected gauge to include correct help=%s, but was %s", summaryDef.Help, metric.Desc().String()) } return true }) sink.IncrCounterWithLabels(counterDef.Name, 42.0, []metrics.Label{ {Name: "version", Value: "some info"}, }) sink.counters.Range(func(key, value interface{}) bool { metric := value.(counter) if !strings.Contains(metric.Desc().String(), counterDef.Help) { t.Fatalf("expected gauge to include correct help=%s, but was %s", counterDef.Help, metric.Desc().String()) } return true }) }	11.17 KB	2025-05-25 12:27

Terminal