forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

metrics: improve testing, integrate informational metrics

Browse source
This commit is contained in:
Martin Holst Swende 2023-08-26 10:48:55 +02:00
parent 44a7ea23f5
commit 8e61797780
No known key found for this signature in database
GPG key ID: 683B438C05A5DDF0
20 changed files with 439 additions and 241 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
cmd/geth/config.go
View file

@ -23,11 +23,9 @@ import (
"os"
"reflect"
"runtime"
"strconv"
"strings"
"unicode"
"github.com/urfave/cli/v2"
"github.com/ethereum/go-ethereum/accounts"
"github.com/ethereum/go-ethereum/accounts/external"
"github.com/ethereum/go-ethereum/accounts/keystore"
@ -37,7 +35,6 @@ import (
"github.com/ethereum/go-ethereum/eth/catalyst"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/eth/ethconfig"
"github.com/ethereum/go-ethereum/eth/protocols/eth"
"github.com/ethereum/go-ethereum/internal/ethapi"
"github.com/ethereum/go-ethereum/internal/flags"
"github.com/ethereum/go-ethereum/internal/version"
@ -46,6 +43,7 @@ import (
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/params"
"github.com/naoina/toml"
"github.com/urfave/cli/v2"
)
var (
@ -164,24 +162,6 @@ func makeConfigNode(ctx *cli.Context) (*node.Node, gethConfig) {
}
applyMetricConfig(ctx, &cfg)
// Create Info Gauge with geth system and build information
gethInfoGauge := metrics.NewRegisteredGaugeInfo("geth/info", nil)
protocolVersions := ""
for idx, val := range eth.ProtocolVersions {
protocolVersions += strconv.FormatUint(uint64(val), 10)
if idx < len(eth.ProtocolVersions)-1 {
protocolVersions += ","
}
}
gethInfo := metrics.GaugeInfoValue{
metrics.NewGaugeInfoEntry("version", params.VersionWithMeta),
metrics.NewGaugeInfoEntry("arch", runtime.GOARCH),
metrics.NewGaugeInfoEntry("os", runtime.GOOS),
metrics.NewGaugeInfoEntry("commit", gitCommit),
metrics.NewGaugeInfoEntry("protocol_versions", protocolVersions),
}
gethInfoGauge.Update(gethInfo)
return stack, cfg
}
@ -198,6 +178,20 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
}
backend, eth := utils.RegisterEthService(stack, &cfg.Eth)
// Create gauge with geth system and build information
if eth != nil { // The 'eth' backend may be nil in light mode
var protos []string
for _, p := range eth.Protocols() {
protos = append(protos, fmt.Sprintf("%v/%d", p.Name, p.Version))
}
metrics.NewRegisteredGaugeInfo("geth/info", nil).Update(metrics.GaugeInfoValue{
"arch": runtime.GOARCH,
"os": runtime.GOOS,
"version": cfg.Node.Version,
"eth_protocols": strings.Join(protos, ","),
})
}
// Configure log filter RPC API.
filterSystem := utils.RegisterFilterAPI(stack, backend, &cfg.Eth)

5
core/blockchain.go
View file

@ -325,10 +325,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
bc.currentSafeBlock.Store(nil)
// Update chain info data metrics
chainInfoGauge.Update(
metrics.GaugeInfoValue{
{Key: "chain_id", Val: bc.chainConfig.ChainID.String()},
})
chainInfoGauge.Update(metrics.GaugeInfoValue{"chain_id": bc.chainConfig.ChainID.String()})
// If Geth is initialized with an external ancient store, re-initialize the
// missing chain indexes and chain flags. This procedure can survive crash

2
metrics/exp/exp.go
View file

@ -128,7 +128,7 @@ func (exp *exp) publishGaugeFloat64(name string, metric metrics.GaugeFloat64) {
}
func (exp *exp) publishGaugeInfo(name string, metric metrics.GaugeInfo) {
exp.getInfo(name).Set(fmt.Sprintf("%s", metric.Value()))
exp.getInfo(name).Set(metric.Value().String())
}
func (exp *exp) publishHistogram(name string, metric metrics.Histogram) {

47
metrics/gauge_info.go
View file

@ -1,6 +1,7 @@
package metrics
import (
"encoding/json"
"sync"
)
@ -9,18 +10,14 @@ type GaugeInfo interface {
Snapshot() GaugeInfo
Update(GaugeInfoValue)
Value() GaugeInfoValue
ValueJsonString() string
}
type GaugeInfoEntry struct {
Key string
Val string
}
// GaugeInfoValue is a mappng of (string) keys to (string) values
type GaugeInfoValue map[string]string
type GaugeInfoValue []GaugeInfoEntry
func NewGaugeInfoEntry(key string, val string) GaugeInfoEntry {
return GaugeInfoEntry{key, val}
func (val GaugeInfoValue) String() string {
data, _ := json.Marshal(val)
return string(data)
}
// GetOrRegisterGaugeInfo returns an existing GaugeInfo or constructs and registers a
@ -84,11 +81,6 @@ func (GaugeInfoSnapshot) Update(GaugeInfoValue) {
// Value returns the value at the time the snapshot was taken.
func (g GaugeInfoSnapshot) Value() GaugeInfoValue { return GaugeInfoValue(g) }
// Value returns the value at the time the snapshot was taken in JSON string format.
func (g GaugeInfoSnapshot) ValueJsonString() string {
return gaugeInfoValueToJsonString(g.Value())
}
// NilGauge is a no-op Gauge.
type NilGaugeInfo struct{}
@ -101,9 +93,6 @@ func (NilGaugeInfo) Update(v GaugeInfoValue) {}
// Value is a no-op.
func (NilGaugeInfo) Value() GaugeInfoValue { return GaugeInfoValue{} }
// Value is a no-op.
func (NilGaugeInfo) ValueJsonString() string { return gaugeInfoValueToJsonString(GaugeInfoValue{}) }
// StandardGaugeInfo is the standard implementation of a GaugeInfo and uses
// sync.Mutex to manage a single string value.
type StandardGaugeInfo struct {
@ -130,13 +119,6 @@ func (g *StandardGaugeInfo) Value() GaugeInfoValue {
return g.value
}
// Value returns the gauge's current value in JSON string format.
func (g *StandardGaugeInfo) ValueJsonString() string {
g.mutex.Lock()
defer g.mutex.Unlock()
return gaugeInfoValueToJsonString(g.value)
}
// FunctionalGaugeInfo returns value from given function
type FunctionalGaugeInfo struct {
value func() GaugeInfoValue
@ -149,7 +131,8 @@ func (g FunctionalGaugeInfo) Value() GaugeInfoValue {
// Value returns the gauge's current value in JSON string format
func (g FunctionalGaugeInfo) ValueJsonString() string {
return gaugeInfoValueToJsonString(g.value())
data, _ := json.Marshal(g.value())
return string(data)
}
// Snapshot returns the snapshot.
@ -159,17 +142,3 @@ func (g FunctionalGaugeInfo) Snapshot() GaugeInfo { return GaugeInfoSnapshot(g.V
func (FunctionalGaugeInfo) Update(GaugeInfoValue) {
panic("Update called on a FunctionalGaugeInfo")
}
// Custom conversion to Json to avoid printing "Key" and "Val"
func gaugeInfoValueToJsonString(g GaugeInfoValue) string {
lastIdx := len(g) - 1
v := "{"
for idx, entry := range g {
v += "\"" + entry.Key + "\":\"" + entry.Val + "\""
if idx != lastIdx {
v += ","
}
}
v += "}"
return v
}

114
metrics/gauge_info_test.go
View file

@ -1,86 +1,62 @@
package metrics
import (
"fmt"
"strconv"
"testing"
)
func BenchmarkGuageInfo(b *testing.B) {
g := NewGaugeInfo()
b.ResetTimer()
for i := 0; i < b.N; i++ {
g.Update(GaugeInfoValue{
{"chain_id", string(rune(i))},
})
}
}
func TestGaugeInfo(t *testing.T) {
func TestGaugeInfoJsonString(t *testing.T) {
g := NewGaugeInfo()
g.Update(GaugeInfoValue{
{"chain_id", "5"},
"chain_id": "5",
"anotherKey": "any_string_value",
"third_key": "anything",
},
)
expected := GaugeInfoValue{
{"chain_id", "5"},
}
for idx, v := range g.Value() {
if v.Key != expected[idx].Key || v.Val != expected[idx].Val {
t.Errorf("g.Value()[%v]: %v != %v\n", idx, v, expected[idx])
}
want := `{"anotherKey":"any_string_value","chain_id":"5","third_key":"anything"}`
if have := g.Value().String(); have != want {
t.Errorf("\nhave: %v\nwant: %v\n", have, want)
}
}
func TestGaugeInfoSnapshot(t *testing.T) {
g := NewGaugeInfo()
g.Update(GaugeInfoValue{
{"chain_id", "5"},
})
snapshot := g.Snapshot()
g.Update(GaugeInfoValue{
{"chain_id", "1"},
})
expected := GaugeInfoValue{
{"chain_id", "5"},
}
for idx, v := range snapshot.Value() {
if v.Key != expected[idx].Key || v.Val != expected[idx].Val {
t.Errorf("g.Value()[%v]: %v != %v\n", idx, v, expected[idx])
g.Update(GaugeInfoValue{"value": "original"})
snapshot := g.Snapshot() // Snapshot @chainid 5
g.Update(GaugeInfoValue{"value": "updated"})
// The 'g' should be updated
if have, want := g.Value().String(), `{"value":"updated"}`; have != want {
t.Errorf("\nhave: %v\nwant: %v\n", have, want)
}
// Snapshot should be unupdated
if have, want := snapshot.Value().String(), `{"value":"original"}`; have != want {
t.Errorf("\nhave: %v\nwant: %v\n", have, want)
}
}
func TestGetOrRegisterGaugeInfo(t *testing.T) {
r := NewRegistry()
NewRegisteredGaugeInfo("foo", r).Update(GaugeInfoValue{
{"chain_id", "5"},
})
expected := GaugeInfoValue{
{"chain_id", "5"},
}
NewRegisteredGaugeInfo("foo", r).Update(
GaugeInfoValue{"chain_id": "5"})
g := GetOrRegisterGaugeInfo("foo", r)
for idx, v := range g.Value() {
if v.Key != expected[idx].Key || v.Val != expected[idx].Val {
t.Fatal(g)
}
if have, want := g.Value().String(), `{"chain_id":"5"}`; have != want {
t.Errorf("have\n%v\nwant\n%v\n", have, want)
}
}
func TestFunctionalGaugeInfo(t *testing.T) {
info := GaugeInfoValue{
{"chain_id", "0"},
}
info := GaugeInfoValue{"chain_id": "0"}
counter := 1
// A "functional" gauge invokes the method to obtain the value
fg := NewFunctionalGaugeInfo(func() GaugeInfoValue {
info[0].Val = strconv.Itoa(counter)
info["chain_id"] = strconv.Itoa(counter)
counter++
return info
})
fg.Value()
fg.Value()
if info[0].Val != "2" {
t.Error("info[0].Val != \"2\" -> ", info[0].Val)
if have, want := info["chain_id"], "2"; have != want {
t.Errorf("have %v want %v", have, want)
}
}
@ -88,42 +64,12 @@ func TestGetOrRegisterFunctionalGaugeInfo(t *testing.T) {
r := NewRegistry()
NewRegisteredFunctionalGaugeInfo("foo", r, func() GaugeInfoValue {
return GaugeInfoValue{
{"chain_id", "5"},
"chain_id": "5",
}
})
expected := GaugeInfoValue{
{"chain_id", "5"},
}
g := GetOrRegisterGaugeInfo("foo", r)
for idx, v := range g.Value() {
if v.Key != expected[idx].Key || v.Val != expected[idx].Val {
t.Fatal(g)
want := `{"chain_id":"5"}`
have := GetOrRegisterGaugeInfo("foo", r).Value().String()
if have != want {
t.Errorf("have\n%v\nwant\n%v\n", have, want)
}
}
}
func TestGaugeInfoValueJsonString(t *testing.T) {
g := NewGaugeInfo()
g.Update(GaugeInfoValue{
{"chain_id", "5"},
{"anotherKey", "any_string_value"},
{"third_key", "anything"},
},
)
expected := `{"chain_id":"5","anotherKey":"any_string_value","third_key":"anything"}`
got := g.ValueJsonString()
if got != expected {
t.Errorf("g.ValueToJsonString(): %s != %s\n", got, expected)
}
}
func ExampleGetOrRegisterGaugeInfo() {
m := "chain/info"
g := GetOrRegisterGaugeInfo(m, nil)
g.Update(GaugeInfoValue{
{"chain_id", "5"},
{"random_value", "10"},
{"chain_data", "356"},
})
fmt.Println(g.Value()) // Output: [{chain_id 5} {random_value 10} {chain_data 356}]
}

2
metrics/graphite.go
View file

@ -74,7 +74,7 @@ func graphite(c *GraphiteConfig) error {
case GaugeFloat64:
fmt.Fprintf(w, "%s.%s.value %f %d\n", c.Prefix, name, metric.Value(), now)
case GaugeInfo:
fmt.Fprintf(w, "%s.%s.value %s %d\n", c.Prefix, name, metric.ValueJsonString(), now)
fmt.Fprintf(w, "%s.%s.value %s %d\n", c.Prefix, name, metric.Value().String(), now)
case Histogram:
h := metric.Snapshot()
ps := h.Percentiles(c.Percentiles)

2
metrics/influxdb/influxdb.go
View file

@ -36,7 +36,7 @@ func readMeter(namespace, name string, i interface{}) (string, map[string]interf
ms := metric.Snapshot()
measurement := fmt.Sprintf("%s%s.gauge", namespace, name)
fields := map[string]interface{}{
"value": ms.ValueJsonString(),
"value": ms.Value().String(),
}
return measurement, fields
case metrics.Histogram:

141
metrics/influxdb/influxdb_test.go Normal file
View file

@ -0,0 +1,141 @@
// Copyright 2023 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package influxdb
import (
"fmt"
"io"
"net/http"
"net/http/httptest"
"net/url"
"os"
"testing"
"time"
"github.com/ethereum/go-ethereum/metrics"
influxdb2 "github.com/influxdata/influxdb-client-go/v2"
)
func TestMain(m *testing.M) {
metrics.Enabled = true
os.Exit(m.Run())
}
func setupSampleRegistry(t *testing.T) metrics.Registry {
t.Helper()
r := metrics.NewOrderedRegistry()
metrics.NewRegisteredGaugeInfo("info", r).Update(metrics.GaugeInfoValue{
"version": "1.10.18-unstable",
"arch": "amd64",
"os": "linux",
"commit": "7caa2d8163ae3132c1c2d6978c76610caee2d949",
"protocol_versions": "64 65 66",
})
metrics.NewRegisteredGaugeFloat64("pi", r).Update(3.14)
metrics.NewRegisteredCounter("months", r).Inc(12)
metrics.NewRegisteredCounterFloat64("tau", r).Inc(1.57)
metrics.NewRegisteredMeter("elite", r).Mark(1337)
metrics.NewRegisteredTimer("second", r).Update(time.Second)
metrics.NewRegisteredCounterFloat64("tau", r).Inc(1.57)
metrics.NewRegisteredCounterFloat64("tau", r).Inc(1.57)
return r
}
func TestExampleV1(t *testing.T) {
r := setupSampleRegistry(t)
var have, want string
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
haveB, _ := io.ReadAll(r.Body)
have = string(haveB)
r.Body.Close()
}))
defer ts.Close()
u, _ := url.Parse(ts.URL)
rep := &reporter{
reg: r,
url: *u,
namespace: "goth.",
}
if err := rep.makeClient(); err != nil {
t.Fatal(err)
}
if err := rep.send(978307200); err != nil {
t.Fatal(err)
}
if wantB, err := os.ReadFile("./testdata/influxdbv1.want"); err != nil {
t.Fatal(err)
} else {
want = string(wantB)
}
if have != want {
t.Errorf("\nhave:\n%v\nwant:\n%v\n", have, want)
t.Logf("have vs want:\n %v", findFirstDiffPos(have, want))
}
}
func TestExampleV2(t *testing.T) {
r := setupSampleRegistry(t)
var have, want string
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
haveB, _ := io.ReadAll(r.Body)
have = string(haveB)
r.Body.Close()
}))
defer ts.Close()
rep := &v2Reporter{
reg: r,
endpoint: ts.URL,
namespace: "goth.",
}
rep.client = influxdb2.NewClient(rep.endpoint, rep.token)
defer rep.client.Close()
rep.write = rep.client.WriteAPI(rep.organization, rep.bucket)
rep.send(978307200)
if wantB, err := os.ReadFile("./testdata/influxdbv2.want"); err != nil {
t.Fatal(err)
} else {
want = string(wantB)
}
if have != want {
t.Errorf("\nhave:\n%v\nwant:\n%v\n", have, want)
t.Logf("have vs want:\n %v", findFirstDiffPos(have, want))
}
}
func findFirstDiffPos(a, b string) string {
x, y := []byte(a), []byte(b)
var res []byte
for i, ch := range x {
if i > len(y) {
res = append(res, ch)
res = append(res, fmt.Sprintf("<-- diff: %#x vs EOF", ch)...)
break
}
if ch != y[i] {
res = append(res, fmt.Sprintf("<-- diff: %#x (%c) vs %#x (%c)", ch, ch, y[i], y[i])...)
break
}
res = append(res, ch)
}
if len(res) > 100 {
res = res[len(res)-100:]
}
return string(res)
}

15
metrics/influxdb/influxdbv1.go
View file

@ -79,7 +79,7 @@ func InfluxDBWithTagsOnce(r metrics.Registry, url, database, username, password,
return fmt.Errorf("unable to make InfluxDB client. err: %v", err)
}
if err := rep.send(); err != nil {
if err := rep.send(0); err != nil {
return fmt.Errorf("unable to send to InfluxDB. err: %v", err)
}
@ -107,7 +107,7 @@ func (r *reporter) run() {
for {
select {
case <-intervalTicker.C:
if err := r.send(); err != nil {
if err := r.send(0); err != nil {
log.Warn("Unable to send to InfluxDB", "err", err)
}
case <-pingTicker.C:
@ -123,7 +123,9 @@ func (r *reporter) run() {
}
}
func (r *reporter) send() error {
// send sends the measurements. If provided tstamp is >0, it is used. Otherwise,
// a 'fresh' timestamp is used.
func (r *reporter) send(tstamp int64) error {
bps, err := client.NewBatchPoints(
client.BatchPointsConfig{
Database: r.database,
@ -132,7 +134,12 @@ func (r *reporter) send() error {
return err
}
r.reg.Each(func(name string, i interface{}) {
now := time.Now()
var now time.Time
if tstamp <= 0 {
now = time.Now()
} else {
now = time.Unix(tstamp, 0)
}
measurement, fields := readMeter(r.namespace, name, i)
if fields == nil {
return

13
metrics/influxdb/influxdbv2.go
View file

@ -64,7 +64,7 @@ func (r *v2Reporter) run() {
for {
select {
case <-intervalTicker.C:
r.send()
r.send(0)
case <-pingTicker.C:
_, err := r.client.Health(context.Background())
if err != nil {
@ -74,9 +74,16 @@ func (r *v2Reporter) run() {
}
}
func (r *v2Reporter) send() {
// send sends the measurements. If provided tstamp is >0, it is used. Otherwise,
// a 'fresh' timestamp is used.
func (r *v2Reporter) send(tstamp int64) {
r.reg.Each(func(name string, i interface{}) {
now := time.Now()
var now time.Time
if tstamp <= 0 {
now = time.Now()
} else {
now = time.Unix(tstamp, 0)
}
measurement, fields := readMeter(r.namespace, name, i)
if fields == nil {
return

6
metrics/influxdb/testdata/influxdbv1.want vendored Normal file
View file

@ -0,0 +1,6 @@
goth.elite.meter count=0i,m1=0,m15=0,m5=0,mean=0 978307200000000000
goth.info.gauge value="{\"arch\":\"amd64\",\"commit\":\"7caa2d8163ae3132c1c2d6978c76610caee2d949\",\"os\":\"linux\",\"protocol_versions\":\"64 65 66\",\"version\":\"1.10.18-unstable\"}" 978307200000000000
goth.months.count value=12i 978307200000000000
goth.pi.gauge value=3.14 978307200000000000
goth.second.timer count=1i,m1=0,m15=0,m5=0,max=1000000000i,mean=1000000000,meanrate=0,min=1000000000i,p50=1000000000,p75=1000000000,p95=1000000000,p99=1000000000,p999=1000000000,p9999=1000000000,stddev=0,variance=0 978307200000000000
goth.tau.count value=1.57 978307200000000000

6
metrics/influxdb/testdata/influxdbv2.want vendored Normal file
View file

@ -0,0 +1,6 @@
goth.elite.meter count=0i,m1=0,m15=0,m5=0,mean=0 978307200000000000
goth.info.gauge value="{\"arch\":\"amd64\",\"commit\":\"7caa2d8163ae3132c1c2d6978c76610caee2d949\",\"os\":\"linux\",\"protocol_versions\":\"64 65 66\",\"version\":\"1.10.18-unstable\"}" 978307200000000000
goth.months.count value=12i 978307200000000000
goth.pi.gauge value=3.14 978307200000000000
goth.second.timer count=1i,m1=0,m15=0,m5=0,max=1000000000i,mean=1000000000,meanrate=0,min=1000000000i,p50=1000000000,p75=1000000000,p95=1000000000,p99=1000000000,p999=1000000000,p9999=1000000000,stddev=0,variance=0 978307200000000000
goth.tau.count value=1.57 978307200000000000

27
metrics/opentsdb.go
View file

@ -3,6 +3,7 @@ package metrics
import (
"bufio"
"fmt"
"io"
"log"
"net"
"os"
@ -57,16 +58,10 @@ func getShortHostname() string {
return shortHostName
}
func openTSDB(c *OpenTSDBConfig) error {
shortHostname := getShortHostname()
now := time.Now().Unix()
// writeRegistry writes the registry-metrics on the opentsb format.
func (c *OpenTSDBConfig) writeRegistry(w io.Writer, now int64, shortHostname string) {
du := float64(c.DurationUnit)
conn, err := net.DialTCP("tcp", nil, c.Addr)
if nil != err {
return err
}
defer conn.Close()
w := bufio.NewWriter(conn)
c.Registry.Each(func(name string, i interface{}) {
switch metric := i.(type) {
case Counter:
@ -78,7 +73,7 @@ func openTSDB(c *OpenTSDBConfig) error {
case GaugeFloat64:
fmt.Fprintf(w, "put %s.%s.value %d %f host=%s\n", c.Prefix, name, now, metric.Value(), shortHostname)
case GaugeInfo:
fmt.Fprintf(w, "put %s.%s.value %d %s host=%s\n", c.Prefix, name, now, metric.ValueJsonString(), shortHostname)
fmt.Fprintf(w, "put %s.%s.value %d %s host=%s\n", c.Prefix, name, now, metric.Value().String(), shortHostname)
case Histogram:
h := metric.Snapshot()
ps := h.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999})
@ -117,7 +112,17 @@ func openTSDB(c *OpenTSDBConfig) error {
fmt.Fprintf(w, "put %s.%s.fifteen-minute %d %.2f host=%s\n", c.Prefix, name, now, t.Rate15(), shortHostname)
fmt.Fprintf(w, "put %s.%s.mean-rate %d %.2f host=%s\n", c.Prefix, name, now, t.RateMean(), shortHostname)
}
w.Flush()
})
}
func openTSDB(c *OpenTSDBConfig) error {
conn, err := net.DialTCP("tcp", nil, c.Addr)
if nil != err {
return err
}
defer conn.Close()
w := bufio.NewWriter(conn)
c.writeRegistry(w, time.Now().Unix(), getShortHostname())
w.Flush()
return nil
}

30
metrics/opentsdb_test.go
View file

@ -2,6 +2,9 @@ package metrics
import (
"net"
"os"
"strings"
"testing"
"time"
)
@ -19,3 +22,30 @@ func ExampleOpenTSDBWithConfig() {
DurationUnit: time.Millisecond,
})
}
func TestExampleOpenTSB(t *testing.T) {
r := NewOrderedRegistry()
NewRegisteredGaugeInfo("foo", r).Update(GaugeInfoValue{"chain_id": "5"})
NewRegisteredGaugeFloat64("pi", r).Update(3.14)
NewRegisteredCounter("months", r).Inc(12)
NewRegisteredCounterFloat64("tau", r).Inc(1.57)
NewRegisteredMeter("elite", r).Mark(1337)
NewRegisteredTimer("second", r).Update(time.Second)
NewRegisteredCounterFloat64("tau", r).Inc(1.57)
NewRegisteredCounterFloat64("tau", r).Inc(1.57)
w := new(strings.Builder)
(&OpenTSDBConfig{
Registry: r,
DurationUnit: time.Millisecond,
Prefix: "pre",
}).writeRegistry(w, 978307200, "hal9000")
wantB, err := os.ReadFile("./testdata/opentsb.want")
if err != nil {
t.Fatal(err)
}
if have, want := w.String(), string(wantB); have != want {
t.Errorf("\nhave:\n%v\nwant:\n%v\n", have, want)
}
}

15
metrics/prometheus/collector.go
View file

@ -19,6 +19,7 @@ package prometheus
import (
"bytes"
"fmt"
"sort"
"strconv"
"strings"
@ -109,14 +110,14 @@ func (c *collector) addResettingTimer(name string, m metrics.ResettingTimer) {
func (c *collector) writeGaugeInfo(name string, value metrics.GaugeInfoValue) {
name = mutateKey(name)
c.buff.WriteString(fmt.Sprintf(typeGaugeTpl, name))
c.buff.WriteString(fmt.Sprintf("%s {", name))
for idx, entry := range value {
c.buff.WriteString(fmt.Sprintf("%s=\"%s\"", entry.Key, entry.Val))
if idx != len(value)-1 {
c.buff.WriteString(", ")
c.buff.WriteString(name)
c.buff.WriteString(" ")
var kvs []string
for k, v := range value {
kvs = append(kvs, fmt.Sprintf("%v=%q", k, v))
}
}
c.buff.WriteString("} 1 \n\n")
sort.Strings(kvs)
c.buff.WriteString(fmt.Sprintf("{%v} 1\n\n", strings.Join(kvs, ", ")))
}
func (c *collector) writeGaugeCounter(name string, value interface{}) {

111
metrics/prometheus/collector_test.go
View file

@ -1,6 +1,23 @@
// Copyright 2023 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package prometheus
import (
"fmt"
"os"
"testing"
"time"
@ -34,11 +51,11 @@ func TestCollector(t *testing.T) {
gaugeInfo := metrics.NewGaugeInfo()
gaugeInfo.Update(metrics.GaugeInfoValue{
metrics.NewGaugeInfoEntry("version", "1.10.18-unstable"),
metrics.NewGaugeInfoEntry("arch", "amd64"),
metrics.NewGaugeInfoEntry("os", "linux"),
metrics.NewGaugeInfoEntry("commit", "7caa2d8163ae3132c1c2d6978c76610caee2d949"),
metrics.NewGaugeInfoEntry("protocol_versions", "64 65 66"),
"version": "1.10.18-unstable",
"arch": "amd64",
"os": "linux",
"commit": "7caa2d8163ae3132c1c2d6978c76610caee2d949",
"protocol_versions": "64 65 66",
})
c.addGaugeInfo("geth/info", gaugeInfo)
@ -72,59 +89,37 @@ func TestCollector(t *testing.T) {
emptyResettingTimer := metrics.NewResettingTimer().Snapshot()
c.addResettingTimer("test/empty_resetting_timer", emptyResettingTimer)
const expectedOutput = `# TYPE test_counter gauge
test_counter 12345
# TYPE test_counter_float64 gauge
test_counter_float64 54321.98
# TYPE test_gauge gauge
test_gauge 23456
# TYPE test_gauge_float64 gauge
test_gauge_float64 34567.89
# TYPE geth_info gauge
geth_info {version="1.10.18-unstable", arch="amd64", os="linux", commit="7caa2d8163ae3132c1c2d6978c76610caee2d949", protocol_versions="64 65 66"} 1
# TYPE test_histogram_count counter
test_histogram_count 0
# TYPE test_histogram summary
test_histogram {quantile="0.5"} 0
test_histogram {quantile="0.75"} 0
test_histogram {quantile="0.95"} 0
test_histogram {quantile="0.99"} 0
test_histogram {quantile="0.999"} 0
test_histogram {quantile="0.9999"} 0
# TYPE test_meter gauge
test_meter 9999999
# TYPE test_timer_count counter
test_timer_count 6
# TYPE test_timer summary
test_timer {quantile="0.5"} 2.25e+07
test_timer {quantile="0.75"} 4.8e+07
test_timer {quantile="0.95"} 1.2e+08
test_timer {quantile="0.99"} 1.2e+08
test_timer {quantile="0.999"} 1.2e+08
test_timer {quantile="0.9999"} 1.2e+08
# TYPE test_resetting_timer_count counter
test_resetting_timer_count 6
# TYPE test_resetting_timer summary
test_resetting_timer {quantile="0.50"} 12000000
test_resetting_timer {quantile="0.95"} 120000000
test_resetting_timer {quantile="0.99"} 120000000
`
exp := c.buff.String()
if exp != expectedOutput {
t.Log("Expected Output:\n", expectedOutput)
t.Log("Actual Output:\n", exp)
var want string
if wantB, err := os.ReadFile("./testdata/prometheus.want"); err != nil {
t.Fatal(err)
} else {
want = string(wantB)
}
have := c.buff.String()
if have != want {
t.Logf("have\n%v", have)
t.Logf("have vs want:\n %v", findFirstDiffPos(have, want))
t.Fatal("unexpected collector output")
}
}
func findFirstDiffPos(a, b string) string {
x, y := []byte(a), []byte(b)
var res []byte
for i, ch := range x {
if i > len(y) {
res = append(res, ch)
res = append(res, fmt.Sprintf("<-- diff: %#x vs EOF", ch)...)
break
}
if ch != y[i] {
res = append(res, fmt.Sprintf("<-- diff: %#x (%c) vs %#x (%c)", ch, ch, y[i], y[i])...)
break
}
res = append(res, ch)
}
if len(res) > 100 {
res = res[len(res)-100:]
}
return string(res)
}

48
metrics/prometheus/testdata/prometheus.want vendored Normal file
View file

@ -0,0 +1,48 @@
# TYPE test_counter gauge
test_counter 12345
# TYPE test_counter_float64 gauge
test_counter_float64 54321.98
# TYPE test_gauge gauge
test_gauge 23456
# TYPE test_gauge_float64 gauge
test_gauge_float64 34567.89
# TYPE geth_info gauge
geth_info {arch="amd64", commit="7caa2d8163ae3132c1c2d6978c76610caee2d949", os="linux", protocol_versions="64 65 66", version="1.10.18-unstable"} 1
# TYPE test_histogram_count counter
test_histogram_count 0
# TYPE test_histogram summary
test_histogram {quantile="0.5"} 0
test_histogram {quantile="0.75"} 0
test_histogram {quantile="0.95"} 0
test_histogram {quantile="0.99"} 0
test_histogram {quantile="0.999"} 0
test_histogram {quantile="0.9999"} 0
# TYPE test_meter gauge
test_meter 9999999
# TYPE test_timer_count counter
test_timer_count 6
# TYPE test_timer summary
test_timer {quantile="0.5"} 2.25e+07
test_timer {quantile="0.75"} 4.8e+07
test_timer {quantile="0.95"} 1.2e+08
test_timer {quantile="0.99"} 1.2e+08
test_timer {quantile="0.999"} 1.2e+08
test_timer {quantile="0.9999"} 1.2e+08
# TYPE test_resetting_timer_count counter
test_resetting_timer_count 6
# TYPE test_resetting_timer summary
test_resetting_timer {quantile="0.50"} 12000000
test_resetting_timer {quantile="0.95"} 120000000
test_resetting_timer {quantile="0.99"} 120000000

33
metrics/registry.go
View file

@ -3,6 +3,7 @@ package metrics
import (
"fmt"
"reflect"
"sort"
"strings"
"sync"
)
@ -47,17 +48,39 @@ type Registry interface {
Unregister(string)
}
type orderedRegistry struct {
StandardRegistry
}
// Call the given function for each registered metric.
func (r *orderedRegistry) Each(f func(string, interface{})) {
var names []string
reg := r.registered()
for name := range reg {
names = append(names, name)
}
sort.Strings(names)
for _, name := range names {
f(name, reg[name])
}
}
// NewRegistry creates a new registry.
func NewRegistry() Registry {
return new(StandardRegistry)
}
// NewOrderedRegistry creates a new ordered registry (for testing).
func NewOrderedRegistry() Registry {
return new(orderedRegistry)
}
// The standard implementation of a Registry uses sync.map
// of names to metrics.
type StandardRegistry struct {
metrics sync.Map
}
// Create a new registry.
func NewRegistry() Registry {
return &StandardRegistry{}
}
// Call the given function for each registered metric.
func (r *StandardRegistry) Each(f func(string, interface{})) {
for name, i := range r.registered() {

23
metrics/testdata/opentsb.want vendored Normal file
View file

@ -0,0 +1,23 @@
put pre.elite.count 978307200 0 host=hal9000
put pre.elite.one-minute 978307200 0.00 host=hal9000
put pre.elite.five-minute 978307200 0.00 host=hal9000
put pre.elite.fifteen-minute 978307200 0.00 host=hal9000
put pre.elite.mean 978307200 0.00 host=hal9000
put pre.foo.value 978307200 {"chain_id":"5"} host=hal9000
put pre.months.count 978307200 12 host=hal9000
put pre.pi.value 978307200 3.140000 host=hal9000
put pre.second.count 978307200 1 host=hal9000
put pre.second.min 978307200 1000 host=hal9000
put pre.second.max 978307200 1000 host=hal9000
put pre.second.mean 978307200 1000.00 host=hal9000
put pre.second.std-dev 978307200 0.00 host=hal9000
put pre.second.50-percentile 978307200 1000.00 host=hal9000
put pre.second.75-percentile 978307200 1000.00 host=hal9000
put pre.second.95-percentile 978307200 1000.00 host=hal9000
put pre.second.99-percentile 978307200 1000.00 host=hal9000
put pre.second.999-percentile 978307200 1000.00 host=hal9000
put pre.second.one-minute 978307200 0.00 host=hal9000
put pre.second.five-minute 978307200 0.00 host=hal9000
put pre.second.fifteen-minute 978307200 0.00 host=hal9000
put pre.second.mean-rate 978307200 0.00 host=hal9000
put pre.tau.count 978307200 1.570000 host=hal9000

2
metrics/writer.go
View file

@ -41,7 +41,7 @@ func WriteOnce(r Registry, w io.Writer) {
fmt.Fprintf(w, " value: %f\n", metric.Value())
case GaugeInfo:
fmt.Fprintf(w, "gauge %s\n", namedMetric.name)
fmt.Fprintf(w, " value: %s\n", metric.ValueJsonString())
fmt.Fprintf(w, " value: %s\n", metric.Value().String())
case Healthcheck:
metric.Check()
fmt.Fprintf(w, "healthcheck %s\n", namedMetric.name)