amassivus/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-07-03 02:53:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

chore!: adopt log/slog, remove go-kit/log

Browse source 
For: #14355

This commit updates Prometheus to adopt stdlib's log/slog package in
favor of go-kit/log. As part of converting to use slog, several other
related changes are required to get prometheus working, including:
- removed unused logging util func `RateLimit()`
- forward ported the util/logging/Deduper logging by implementing a small custom slog.Handler that does the deduping before chaining log calls to the underlying real slog.Logger
- move some of the json file logging functionality to use prom/common package functionality
- refactored some of the new json file logging for scraping
- changes to promql.QueryLogger interface to swap out logging methods for relevant slog sugar wrappers
- updated lots of tests that used/replicated custom logging functionality, attempting to keep the logical goal of the tests consistent after the transition
- added a healthy amount of `if logger == nil { $makeLogger }` type conditional checks amongst various functions where none were provided -- old code that used the go-kit/log.Logger interface had several places where there were nil references when trying to use functions like `With()` to add keyvals on the new *slog.Logger type

Signed-off-by: TJ Hoplock <t.hoplock@gmail.com>
This commit is contained in:
TJ Hoplock 2024-09-09 21:41:53 -04:00
parent 65f6103539
commit 6ebfbd2d54
162 changed files with 1534 additions and 1691 deletions
Download patch file Download diff file Expand all files Collapse all files

25
documentation/examples/custom-sd/adapter-usage/main.go
View file

@ -18,6 +18,7 @@ import (
"encoding/json"
"fmt"
"io"
"log/slog"
"net"
"net/http"
"os"
@ -26,10 +27,9 @@ import (
"time"
"github.com/alecthomas/kingpin/v2"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/common/model"
"github.com/prometheus/common/promslog"
prom_discovery "github.com/prometheus/prometheus/discovery"
"github.com/prometheus/prometheus/discovery/targetgroup"
@ -41,7 +41,7 @@ var (
a = kingpin.New("sd adapter usage", "Tool to generate file_sd target files for unimplemented SD mechanisms.")
outputFile = a.Flag("output.file", "Output file for file_sd compatible file.").Default("custom_sd.json").String()
listenAddress = a.Flag("listen.address", "The address the Consul HTTP API is listening on for requests.").Default("localhost:8500").String()
logger log.Logger
logger *slog.Logger
// addressLabel is the name for the label containing a target's address.
addressLabel = model.MetaLabelPrefix + "consul_address"
@ -90,7 +90,7 @@ type discovery struct {
address string
refreshInterval int
tagSeparator string
logger log.Logger
logger *slog.Logger
oldSourceList map[string]bool
}
@ -164,7 +164,7 @@ func (d *discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
var srvs map[string][]string
resp, err := http.Get(fmt.Sprintf("http://%s/v1/catalog/services", d.address))
if err != nil {
level.Error(d.logger).Log("msg", "Error getting services list", "err", err)
d.logger.Error("Error getting services list", "err", err)
time.Sleep(time.Duration(d.refreshInterval) * time.Second)
continue
}
@ -173,7 +173,7 @@ func (d *discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
if err != nil {
level.Error(d.logger).Log("msg", "Error reading services list", "err", err)
d.logger.Error("Error reading services list", "err", err)
time.Sleep(time.Duration(d.refreshInterval) * time.Second)
continue
}
@ -181,7 +181,7 @@ func (d *discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
err = json.Unmarshal(b, &srvs)
resp.Body.Close()
if err != nil {
level.Error(d.logger).Log("msg", "Error parsing services list", "err", err)
d.logger.Error("Error parsing services list", "err", err)
time.Sleep(time.Duration(d.refreshInterval) * time.Second)
continue
}
@ -200,13 +200,13 @@ func (d *discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
}
resp, err := http.Get(fmt.Sprintf("http://%s/v1/catalog/service/%s", d.address, name))
if err != nil {
level.Error(d.logger).Log("msg", "Error getting services nodes", "service", name, "err", err)
d.logger.Error("Error getting services nodes", "service", name, "err", err)
break
}
tg, err := d.parseServiceNodes(resp, name)
if err != nil {
level.Error(d.logger).Log("msg", "Error parsing services nodes", "service", name, "err", err)
d.logger.Error("Error parsing services nodes", "service", name, "err", err)
break
}
tgs = append(tgs, tg)
@ -254,8 +254,7 @@ func main() {
fmt.Println("err: ", err)
return
}
logger = log.NewSyncLogger(log.NewLogfmtLogger(os.Stdout))
logger = log.With(logger, "ts", log.DefaultTimestampUTC, "caller", log.DefaultCaller)
logger = promslog.New(&promslog.Config{})
ctx := context.Background()
@ -272,7 +271,7 @@ func main() {
}
if err != nil {
level.Error(logger).Log("msg", "failed to create discovery metrics", "err", err)
logger.Error("failed to create discovery metrics", "err", err)
os.Exit(1)
}
@ -280,7 +279,7 @@ func main() {
refreshMetrics := prom_discovery.NewRefreshMetrics(reg)
metrics, err := prom_discovery.RegisterSDMetrics(reg, refreshMetrics)
if err != nil {
level.Error(logger).Log("msg", "failed to register service discovery metrics", "err", err)
logger.Error("failed to register service discovery metrics", "err", err)
os.Exit(1)
}

9
documentation/examples/custom-sd/adapter/adapter.go
View file

@ -18,13 +18,12 @@ import (
"context"
"encoding/json"
"fmt"
"log/slog"
"os"
"path/filepath"
"reflect"
"sort"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/common/model"
@ -55,7 +54,7 @@ type Adapter struct {
manager *discovery.Manager
output string
name string
logger log.Logger
logger *slog.Logger
}
func mapToArray(m map[string]*customSD) []customSD {
@ -106,7 +105,7 @@ func (a *Adapter) refreshTargetGroups(allTargetGroups map[string][]*targetgroup.
a.groups = tempGroups
err := a.writeOutput()
if err != nil {
level.Error(log.With(a.logger, "component", "sd-adapter")).Log("err", err)
a.logger.With("component", "sd-adapter").Error("failed to write output", "err", err)
}
}
}
@ -163,7 +162,7 @@ func (a *Adapter) Run() {
}
// NewAdapter creates a new instance of Adapter.
func NewAdapter(ctx context.Context, file, name string, d discovery.Discoverer, logger log.Logger, sdMetrics map[string]discovery.DiscovererMetrics, registerer prometheus.Registerer) *Adapter {
func NewAdapter(ctx context.Context, file, name string, d discovery.Discoverer, logger *slog.Logger, sdMetrics map[string]discovery.DiscovererMetrics, registerer prometheus.Registerer) *Adapter {
return &Adapter{
ctx: ctx,
disc: d,

2
documentation/examples/remote_storage/go.mod
View file

@ -4,7 +4,6 @@ go 1.22.0
require (
github.com/alecthomas/kingpin/v2 v2.4.0
github.com/go-kit/log v0.2.1
github.com/gogo/protobuf v1.3.2
github.com/golang/snappy v0.0.4
github.com/influxdata/influxdb v1.11.6
@ -26,6 +25,7 @@ require (
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/dennwc/varint v1.0.0 // indirect
github.com/felixge/httpsnoop v1.0.4 // indirect
github.com/go-kit/log v0.2.1 // indirect
github.com/go-logfmt/logfmt v0.6.0 // indirect
github.com/go-logr/logr v1.4.1 // indirect
github.com/go-logr/stdr v1.2.2 // indirect

12
documentation/examples/remote_storage/remote_storage_adapter/graphite/client.go
View file

@ -16,19 +16,19 @@ package graphite
import (
"bytes"
"fmt"
"log/slog"
"math"
"net"
"sort"
"time"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/prometheus/common/model"
"github.com/prometheus/common/promslog"
)
// Client allows sending batches of Prometheus samples to Graphite.
type Client struct {
logger log.Logger
logger *slog.Logger
address string
transport string
@ -37,9 +37,9 @@ type Client struct {
}
// NewClient creates a new Client.
func NewClient(logger log.Logger, address, transport string, timeout time.Duration, prefix string) *Client {
func NewClient(logger *slog.Logger, address, transport string, timeout time.Duration, prefix string) *Client {
if logger == nil {
logger = log.NewNopLogger()
logger = promslog.NewNopLogger()
}
return &Client{
logger: logger,
@ -93,7 +93,7 @@ func (c *Client) Write(samples model.Samples) error {
t := float64(s.Timestamp.UnixNano()) / 1e9
v := float64(s.Value)
if math.IsNaN(v) || math.IsInf(v, 0) {
level.Debug(c.logger).Log("msg", "Cannot send value to Graphite, skipping sample", "value", v, "sample", s)
c.logger.Debug("Cannot send value to Graphite, skipping sample", "value", v, "sample", s)
continue
}
fmt.Fprintf(&buf, "%s %f %f\n", k, v, t)

14
documentation/examples/remote_storage/remote_storage_adapter/influxdb/client.go
View file

@ -17,22 +17,22 @@ import (
"encoding/json"
"errors"
"fmt"
"log/slog"
"math"
"os"
"strings"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
influx "github.com/influxdata/influxdb/client/v2"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/common/model"
"github.com/prometheus/common/promslog"
"github.com/prometheus/prometheus/prompb"
)
// Client allows sending batches of Prometheus samples to InfluxDB.
type Client struct {
logger log.Logger
logger *slog.Logger
client influx.Client
database string
@ -41,16 +41,16 @@ type Client struct {
}
// NewClient creates a new Client.
func NewClient(logger log.Logger, conf influx.HTTPConfig, db, rp string) *Client {
func NewClient(logger *slog.Logger, conf influx.HTTPConfig, db, rp string) *Client {
c, err := influx.NewHTTPClient(conf)
// Currently influx.NewClient() *should* never return an error.
if err != nil {
level.Error(logger).Log("err", err)
logger.Error("Error creating influx HTTP client", "err", err)
os.Exit(1)
}
if logger == nil {
logger = log.NewNopLogger()
logger = promslog.NewNopLogger()
}
return &Client{
@ -84,7 +84,7 @@ func (c *Client) Write(samples model.Samples) error {
for _, s := range samples {
v := float64(s.Value)
if math.IsNaN(v) || math.IsInf(v, 0) {
level.Debug(c.logger).Log("msg", "Cannot send to InfluxDB, skipping sample", "value", v, "sample", s)
c.logger.Debug("Cannot send to InfluxDB, skipping sample", "value", v, "sample", s)
c.ignoredSamples.Inc()
continue
}

47
documentation/examples/remote_storage/remote_storage_adapter/main.go
View file

@ -17,6 +17,7 @@ package main
import (
"fmt"
"io"
"log/slog"
"net/http"
_ "net/http/pprof"
"net/url"
@ -26,16 +27,14 @@ import (
"time"
"github.com/alecthomas/kingpin/v2"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/gogo/protobuf/proto"
"github.com/golang/snappy"
influx "github.com/influxdata/influxdb/client/v2"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/prometheus/common/model"
"github.com/prometheus/common/promlog"
"github.com/prometheus/common/promlog/flag"
"github.com/prometheus/common/promslog"
"github.com/prometheus/common/promslog/flag"
"github.com/prometheus/prometheus/documentation/examples/remote_storage/remote_storage_adapter/graphite"
"github.com/prometheus/prometheus/documentation/examples/remote_storage/remote_storage_adapter/influxdb"
@ -57,7 +56,7 @@ type config struct {
remoteTimeout time.Duration
listenAddr string
telemetryPath string
promlogConfig promlog.Config
promslogConfig promslog.Config
}
var (
@ -105,11 +104,11 @@ func main() {
cfg := parseFlags()
http.Handle(cfg.telemetryPath, promhttp.Handler())
logger := promlog.New(&cfg.promlogConfig)
logger := promslog.New(&cfg.promslogConfig)
writers, readers := buildClients(logger, cfg)
if err := serve(logger, cfg.listenAddr, writers, readers); err != nil {
level.Error(logger).Log("msg", "Failed to listen", "addr", cfg.listenAddr, "err", err)
logger.Error("Failed to listen", "addr", cfg.listenAddr, "err", err)
os.Exit(1)
}
}
@ -120,7 +119,7 @@ func parseFlags() *config {
cfg := &config{
influxdbPassword: os.Getenv("INFLUXDB_PW"),
promlogConfig: promlog.Config{},
promslogConfig: promslog.Config{},
}
a.Flag("graphite-address", "The host:port of the Graphite server to send samples to. None, if empty.").
@ -146,7 +145,7 @@ func parseFlags() *config {
a.Flag("web.telemetry-path", "Address to listen on for web endpoints.").
Default("/metrics").StringVar(&cfg.telemetryPath)
flag.AddFlags(a, &cfg.promlogConfig)
flag.AddFlags(a, &cfg.promslogConfig)
_, err := a.Parse(os.Args[1:])
if err != nil {
@ -168,19 +167,19 @@ type reader interface {
Name() string
}
func buildClients(logger log.Logger, cfg *config) ([]writer, []reader) {
func buildClients(logger *slog.Logger, cfg *config) ([]writer, []reader) {
var writers []writer
var readers []reader
if cfg.graphiteAddress != "" {
c := graphite.NewClient(
log.With(logger, "storage", "Graphite"),
logger.With("storage", "Graphite"),
cfg.graphiteAddress, cfg.graphiteTransport,
cfg.remoteTimeout, cfg.graphitePrefix)
writers = append(writers, c)
}
if cfg.opentsdbURL != "" {
c := opentsdb.NewClient(
log.With(logger, "storage", "OpenTSDB"),
logger.With("storage", "OpenTSDB"),
cfg.opentsdbURL,
cfg.remoteTimeout,
)
@ -189,7 +188,7 @@ func buildClients(logger log.Logger, cfg *config) ([]writer, []reader) {
if cfg.influxdbURL != "" {
url, err := url.Parse(cfg.influxdbURL)
if err != nil {
level.Error(logger).Log("msg", "Failed to parse InfluxDB URL", "url", cfg.influxdbURL, "err", err)
logger.Error("Failed to parse InfluxDB URL", "url", cfg.influxdbURL, "err", err)
os.Exit(1)
}
conf := influx.HTTPConfig{
@ -199,7 +198,7 @@ func buildClients(logger log.Logger, cfg *config) ([]writer, []reader) {
Timeout: cfg.remoteTimeout,
}
c := influxdb.NewClient(
log.With(logger, "storage", "InfluxDB"),
logger.With("storage", "InfluxDB"),
conf,
cfg.influxdbDatabase,
cfg.influxdbRetentionPolicy,
@ -208,15 +207,15 @@ func buildClients(logger log.Logger, cfg *config) ([]writer, []reader) {
writers = append(writers, c)
readers = append(readers, c)
}
level.Info(logger).Log("msg", "Starting up...")
logger.Info("Starting up...")
return writers, readers
}
func serve(logger log.Logger, addr string, writers []writer, readers []reader) error {
func serve(logger *slog.Logger, addr string, writers []writer, readers []reader) error {
http.HandleFunc("/write", func(w http.ResponseWriter, r *http.Request) {
req, err := remote.DecodeWriteRequest(r.Body)
if err != nil {
level.Error(logger).Log("msg", "Read error", "err", err.Error())
logger.Error("Read error", "err", err.Error())
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
@ -238,21 +237,21 @@ func serve(logger log.Logger, addr string, writers []writer, readers []reader) e
http.HandleFunc("/read", func(w http.ResponseWriter, r *http.Request) {
compressed, err := io.ReadAll(r.Body)
if err != nil {
level.Error(logger).Log("msg", "Read error", "err", err.Error())
logger.Error("Read error", "err", err.Error())
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
reqBuf, err := snappy.Decode(nil, compressed)
if err != nil {
level.Error(logger).Log("msg", "Decode error", "err", err.Error())
logger.Error("Decode error", "err", err.Error())
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
var req prompb.ReadRequest
if err := proto.Unmarshal(reqBuf, &req); err != nil {
level.Error(logger).Log("msg", "Unmarshal error", "err", err.Error())
logger.Error("Unmarshal error", "err", err.Error())
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@ -267,7 +266,7 @@ func serve(logger log.Logger, addr string, writers []writer, readers []reader) e
var resp *prompb.ReadResponse
resp, err = reader.Read(&req)
if err != nil {
level.Warn(logger).Log("msg", "Error executing query", "query", req, "storage", reader.Name(), "err", err)
logger.Warn("Error executing query", "query", req, "storage", reader.Name(), "err", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
@ -283,7 +282,7 @@ func serve(logger log.Logger, addr string, writers []writer, readers []reader) e
compressed = snappy.Encode(nil, data)
if _, err := w.Write(compressed); err != nil {
level.Warn(logger).Log("msg", "Error writing response", "storage", reader.Name(), "err", err)
logger.Warn("Error writing response", "storage", reader.Name(), "err", err)
}
})
@ -309,12 +308,12 @@ func protoToSamples(req *prompb.WriteRequest) model.Samples {
return samples
}
func sendSamples(logger log.Logger, w writer, samples model.Samples) {
func sendSamples(logger *slog.Logger, w writer, samples model.Samples) {
begin := time.Now()
err := w.Write(samples)
duration := time.Since(begin).Seconds()
if err != nil {
level.Warn(logger).Log("msg", "Error sending samples to remote storage", "err", err, "storage", w.Name(), "num_samples", len(samples))
logger.Warn("Error sending samples to remote storage", "err", err, "storage", w.Name(), "num_samples", len(samples))
failedSamples.WithLabelValues(w.Name()).Add(float64(len(samples)))
}
sentSamples.WithLabelValues(w.Name()).Add(float64(len(samples)))

9
documentation/examples/remote_storage/remote_storage_adapter/opentsdb/client.go
View file

@ -19,13 +19,12 @@ import (
"encoding/json"
"fmt"
"io"
"log/slog"
"math"
"net/http"
"net/url"
"time"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
"github.com/prometheus/common/model"
)
@ -36,14 +35,14 @@ const (
// Client allows sending batches of Prometheus samples to OpenTSDB.
type Client struct {
logger log.Logger
logger *slog.Logger
url string
timeout time.Duration
}
// NewClient creates a new Client.
func NewClient(logger log.Logger, url string, timeout time.Duration) *Client {
func NewClient(logger *slog.Logger, url string, timeout time.Duration) *Client {
return &Client{
logger: logger,
url: url,
@ -78,7 +77,7 @@ func (c *Client) Write(samples model.Samples) error {
for _, s := range samples {
v := float64(s.Value)
if math.IsNaN(v) || math.IsInf(v, 0) {
level.Debug(c.logger).Log("msg", "Cannot send value to OpenTSDB, skipping sample", "value", v, "sample", s)
c.logger.Debug("Cannot send value to OpenTSDB, skipping sample", "value", v, "sample", s)
continue
}
metric := TagValue(s.Metric[model.MetricNameLabel])