Merge branch 'main' into deps-update/cloud.google.com-go-bigtable-1.x

cmd/logcli/main.go

@@ -692,7 +692,7 @@ func newVolumeQuery(rangeQuery bool, cmd *kingpin.CmdClause) *volume.Query {
 
 func newDetectedFieldsQuery(cmd *kingpin.CmdClause) *detected.FieldsQuery {
 	// calculate query range from cli params
-	var from, to string
+	var fieldName, from, to string
 	var since time.Duration
 
 	q := &detected.FieldsQuery{}
@@ -705,24 +705,28 @@ func newDetectedFieldsQuery(cmd *kingpin.CmdClause) *detected.FieldsQuery {
 		q.Start = mustParse(from, defaultStart)
 		q.End = mustParse(to, defaultEnd)
 
+		q.FieldName = fieldName
+
 		q.Quiet = *quiet
 
 		return nil
 	})
 
-	cmd.Flag("field-limit", "Limit on number of fields to return.").
+	cmd.Flag("limit", "Limit on number of fields or values to return.").
 		Default("100").
-		IntVar(&q.FieldLimit)
+		IntVar(&q.Limit)
 	cmd.Flag("line-limit", "Limit the number of lines each subquery is allowed to process.").
 		Default("1000").
 		IntVar(&q.LineLimit)
 	cmd.Arg("query", "eg '{foo=\"bar\",baz=~\".*blip\"} |~ \".*error.*\"'").
 		Required().
 		StringVar(&q.QueryString)
+	cmd.Arg("field", "The name of the field.").Default("").StringVar(&fieldName)
 	cmd.Flag("since", "Lookback window.").Default("1h").DurationVar(&since)
 	cmd.Flag("from", "Start looking for logs at this absolute time (inclusive)").StringVar(&from)
 	cmd.Flag("to", "Stop looking for logs at this absolute time (exclusive)").StringVar(&to)
 	cmd.Flag("step", "Query resolution step width, for metric queries. Evaluate the query at the specified step over the time range.").
+		Default("10s").
 		DurationVar(&q.Step)
 
 	return q

docs/sources/send-data/alloy/examples/alloy-kafka-logs.md

@@ -392,7 +392,6 @@ Head back to where you started from to continue with the Loki documentation: [Lo
 
 For more information on Grafana Alloy, refer to the following resources:
 - [Grafana Alloy getting started examples](https://grafana.com/docs/alloy/latest/tutorials/)
-- [Grafana Alloy common task examples](https://grafana.com/docs/alloy/latest/collect/)
 - [Grafana Alloy component reference](https://grafana.com/docs/alloy/latest/reference/components/)
 
 ## Complete metrics, logs, traces, and profiling example

docs/sources/send-data/alloy/examples/alloy-otel-logs.md

@@ -279,7 +279,6 @@ Head back to where you started from to continue with the Loki documentation: [Lo
 
 For more information on Grafana Alloy, refer to the following resources:
 - [Grafana Alloy getting started examples](https://grafana.com/docs/alloy/latest/tutorials/)
-- [Grafana Alloy common task examples](https://grafana.com/docs/alloy/latest/collect/)
 - [Grafana Alloy component reference](https://grafana.com/docs/alloy/latest/reference/components/)
 
 ## Complete metrics, logs, traces, and profiling example

docs/sources/send-data/promtail/_index.md

@@ -3,7 +3,7 @@ title: Promtail agent
 menuTitle:  Promtail
 description: How to use the Promtail agent to ship logs to Loki
 aliases: 
-- ../clients/promtail/
+- ../clients/promtail/ # /docs/loki/latest/clients/promtail/
 weight:  200
 ---
 # Promtail agent

docs/sources/send-data/promtail/configuration.md

@@ -2,6 +2,8 @@
 title: Configure Promtail
 menuTitle:  Configuration reference
 description: Configuration parameters for the Promtail agent.
+aliases: 
+- ../../clients/promtail/configuration/ # /docs/loki/latest/clients/promtail/configuration/
 weight:  200
 ---
 

docs/sources/setup/install/helm/reference.md

@@ -5640,6 +5640,7 @@ null
       "/loki/api/v1/index/volume",
       "/loki/api/v1/index/volume_range",
       "/loki/api/v1/format_query",
+      "/loki/api/v1/detected_field",
       "/loki/api/v1/detected_fields",
       "/loki/api/v1/detected_labels",
       "/loki/api/v1/patterns"
@@ -5702,6 +5703,7 @@ null
   "/loki/api/v1/index/volume",
   "/loki/api/v1/index/volume_range",
   "/loki/api/v1/format_query",
+  "/loki/api/v1/detected_field",
   "/loki/api/v1/detected_fields",
   "/loki/api/v1/detected_labels",
   "/loki/api/v1/patterns"

docs/sources/shared/configuration.md

@@ -1298,18 +1298,9 @@ Experimental: The `bloom_gateway` block configures the Loki bloom gateway server
 client:
   # Configures the behavior of the connection pool.
   pool_config:
-    # How frequently to clean up clients for servers that have gone away or are
-    # unhealthy.
+    # How frequently to update the list of servers.
     # CLI flag: -bloom-gateway-client.pool.check-interval
-    [check_interval: <duration> | default = 10s]
-
-    # Run a health check on each server during periodic cleanup.
-    # CLI flag: -bloom-gateway-client.pool.enable-health-check
-    [enable_health_check: <boolean> | default = true]
-
-    # Timeout for the health check if health check is enabled.
-    # CLI flag: -bloom-gateway-client.pool.health-check-timeout
-    [health_check_timeout: <duration> | default = 1s]
+    [check_interval: <duration> | default = 15s]
 
   # The grpc_client block configures the gRPC client used to communicate between
   # a client and server component in Loki.

go.mod

@@ -136,7 +136,7 @@ require (
 	github.com/prometheus/common/sigv4 v0.1.0
 	github.com/richardartoul/molecule v1.0.0
 	github.com/schollz/progressbar/v3 v3.14.6
-	github.com/shirou/gopsutil/v4 v4.24.0-alpha.1
+	github.com/shirou/gopsutil/v4 v4.24.9
 	github.com/thanos-io/objstore v0.0.0-20240828151628-f06322381f09
 	github.com/twmb/franz-go v1.17.1
 	github.com/twmb/franz-go/pkg/kadm v1.13.0
@@ -166,6 +166,7 @@ require (
 	github.com/coreos/etcd v3.3.27+incompatible // indirect
 	github.com/coreos/pkg v0.0.0-20220810130054-c7d1c02cb6cf // indirect
 	github.com/dlclark/regexp2 v1.4.0 // indirect
+	github.com/ebitengine/purego v0.8.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/go-ini/ini v1.67.0 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
@@ -179,7 +180,6 @@ require (
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
-	github.com/shoenig/go-m1cpu v0.1.6 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect

go.sum

@@ -578,6 +578,8 @@ github.com/eapache/go-xerial-snappy v0.0.0-20230111030713-bf00bc1b83b6 h1:8yY/I9
 github.com/eapache/go-xerial-snappy v0.0.0-20230111030713-bf00bc1b83b6/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
+github.com/ebitengine/purego v0.8.0 h1:JbqvnEzRvPpxhCJzJJ2y0RbiZ8nyjccVUrSM3q+GvvE=
+github.com/ebitengine/purego v0.8.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
 github.com/eclipse/paho.mqtt.golang v1.2.0/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts=
 github.com/edsrzf/mmap-go v1.0.0/go.mod h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
 github.com/edsrzf/mmap-go v1.1.0 h1:6EUwBLQ/Mcr1EYLE4Tn1VdW1A4ckqCQWZBw8Hr0kjpQ=
@@ -1709,12 +1711,8 @@ github.com/sercand/kuberesolver/v5 v5.1.1/go.mod h1:Fs1KbKhVRnB2aDWN12NjKCB+RgYM
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/shirou/gopsutil v2.20.9+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/shirou/gopsutil/v3 v3.22.8/go.mod h1:s648gW4IywYzUfE/KjXxUsqrqx/T2xO5VqOXxONeRfI=
-github.com/shirou/gopsutil/v4 v4.24.0-alpha.1 h1:lLPAdP4TpfgJ5byoc3EFwNSKZj8kCnDFHtuWTktWl0s=
-github.com/shirou/gopsutil/v4 v4.24.0-alpha.1/go.mod h1:GVpYUxBee6CTWux2/JslZ7fYPwqkQ8YDJSXmGAryYy4=
-github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
-github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
-github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
-github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k=
+github.com/shirou/gopsutil/v4 v4.24.9 h1:KIV+/HaHD5ka5f570RZq+2SaeFsb/pq+fp2DGNWYoOI=
+github.com/shirou/gopsutil/v4 v4.24.9/go.mod h1:3fkaHNeYsUFCGZ8+9vZVWtbyM1k2eRnlL+bWO8Bxa/Q=
 github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
 github.com/shopspring/decimal v0.0.0-20200105231215-408a2507e114/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/shopspring/decimal v1.2.0 h1:abSATXmQEYyShuxI4/vyW3tV1MrKAJzCZ/0zLUXYbsQ=

operator/bundle/community-openshift/manifests/loki-operator.clusterserviceversion.yaml

@@ -150,7 +150,7 @@ metadata:
     categories: OpenShift Optional, Logging & Tracing
     certified: "false"
     containerImage: docker.io/grafana/loki-operator:0.6.2
-    createdAt: "2024-10-01T11:39:45Z"
+    createdAt: "2024-10-03T12:34:36Z"
     description: The Community Loki Operator provides Kubernetes native deployment
       and management of Loki and related logging components.
     features.operators.openshift.io/disconnected: "true"

operator/bundle/community/manifests/loki-operator.clusterserviceversion.yaml

@@ -150,7 +150,7 @@ metadata:
     categories: OpenShift Optional, Logging & Tracing
     certified: "false"
     containerImage: docker.io/grafana/loki-operator:0.6.2
-    createdAt: "2024-10-01T11:39:42Z"
+    createdAt: "2024-10-03T12:34:33Z"
     description: The Community Loki Operator provides Kubernetes native deployment
       and management of Loki and related logging components.
     operators.operatorframework.io/builder: operator-sdk-unknown

operator/bundle/openshift/manifests/loki-operator.clusterserviceversion.yaml

@@ -150,7 +150,7 @@ metadata:
     categories: OpenShift Optional, Logging & Tracing
     certified: "false"
     containerImage: quay.io/openshift-logging/loki-operator:0.1.0
-    createdAt: "2024-10-01T11:39:47Z"
+    createdAt: "2024-10-03T12:34:39Z"
     description: |
       The Loki Operator for OCP provides a means for configuring and managing a Loki stack for cluster logging.
       ## Prerequisites and Requirements
@@ -167,7 +167,7 @@ metadata:
     features.operators.openshift.io/token-auth-aws: "true"
     features.operators.openshift.io/token-auth-azure: "true"
     features.operators.openshift.io/token-auth-gcp: "false"
-    olm.skipRange: '>=5.8.0-0 <6.0.0'
+    olm.skipRange: '>=5.9.0-0 <6.1.0'
     operatorframework.io/cluster-monitoring: "true"
     operatorframework.io/suggested-namespace: openshift-operators-redhat
     operators.openshift.io/valid-subscription: '["OpenShift Container Platform", "OpenShift

operator/config/manifests/openshift/bases/loki-operator.clusterserviceversion.yaml

@@ -23,7 +23,7 @@ metadata:
     features.operators.openshift.io/token-auth-aws: "true"
     features.operators.openshift.io/token-auth-azure: "true"
     features.operators.openshift.io/token-auth-gcp: "false"
-    olm.skipRange: '>=5.8.0-0 <6.0.0'
+    olm.skipRange: '>=5.9.0-0 <6.1.0'
     operatorframework.io/cluster-monitoring: "true"
     operatorframework.io/suggested-namespace: openshift-operators-redhat
     operators.openshift.io/valid-subscription: '["OpenShift Container Platform", "OpenShift

pkg/bloomgateway/client.go

@@ -161,7 +161,7 @@ func NewClient(
 		}
 	}
 
-	poolFactory := func(addr string) (ringclient.PoolClient, error) {
+	clientFactory := func(addr string) (ringclient.PoolClient, error) {
 		pool, err := NewBloomGatewayGRPCPool(addr, dialOpts)
 		if err != nil {
 			return nil, errors.Wrap(err, "new bloom gateway grpc pool")
@@ -185,17 +185,10 @@ func NewClient(
 	// Make an attempt to do one DNS lookup so we can start with addresses
 	dnsProvider.RunOnce()
 
-	clientPool := ringclient.NewPool(
-		"bloom-gateway",
-		ringclient.PoolConfig(cfg.PoolConfig),
-		func() ([]string, error) { return dnsProvider.Addresses(), nil },
-		ringclient.PoolAddrFunc(poolFactory),
-		metrics.clients,
-		logger,
-	)
-
-	pool := NewJumpHashClientPool(clientPool, dnsProvider, cfg.PoolConfig.CheckInterval, logger)
-	pool.Start()
+	pool, err := NewJumpHashClientPool(clientFactory, dnsProvider, cfg.PoolConfig.CheckInterval, logger)
+	if err != nil {
+		return nil, err
+	}
 
 	return &GatewayClient{
 		cfg:         cfg,

pkg/bloomgateway/client_pool.go

@@ -3,7 +3,7 @@ package bloomgateway
 import (
 	"context"
 	"flag"
-	"sort"
+	"sync"
 	"time"
 
 	"github.com/go-kit/log"
@@ -15,53 +15,65 @@ import (
 )
 
 // PoolConfig is config for creating a Pool.
-// It has the same fields as "github.com/grafana/dskit/ring/client.PoolConfig" so it can be cast.
 type PoolConfig struct {
-	CheckInterval             time.Duration `yaml:"check_interval"`
-	HealthCheckEnabled        bool          `yaml:"enable_health_check"`
-	HealthCheckTimeout        time.Duration `yaml:"health_check_timeout"`
-	MaxConcurrentHealthChecks int           `yaml:"-"`
+	CheckInterval time.Duration `yaml:"check_interval"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *PoolConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
-	f.DurationVar(&cfg.CheckInterval, prefix+"check-interval", 10*time.Second, "How frequently to clean up clients for servers that have gone away or are unhealthy.")
-	f.BoolVar(&cfg.HealthCheckEnabled, prefix+"enable-health-check", true, "Run a health check on each server during periodic cleanup.")
-	f.DurationVar(&cfg.HealthCheckTimeout, prefix+"health-check-timeout", 1*time.Second, "Timeout for the health check if health check is enabled.")
+	f.DurationVar(&cfg.CheckInterval, prefix+"check-interval", 15*time.Second, "How frequently to update the list of servers.")
 }
 
 func (cfg *PoolConfig) Validate() error {
 	return nil
 }
 
+// compiler check
+var _ clientPool = &JumpHashClientPool{}
+
+type ClientFactory func(addr string) (client.PoolClient, error)
+
+func (f ClientFactory) New(addr string) (client.PoolClient, error) {
+	return f(addr)
+}
+
 type JumpHashClientPool struct {
-	*client.Pool
+	services.Service
 	*jumphash.Selector
+	sync.RWMutex
+
+	provider AddressProvider
+	logger   log.Logger
 
-	done   chan struct{}
-	logger log.Logger
+	clients       map[string]client.PoolClient
+	clientFactory ClientFactory
 }
 
 type AddressProvider interface {
 	Addresses() []string
 }
 
-func NewJumpHashClientPool(pool *client.Pool, dnsProvider AddressProvider, updateInterval time.Duration, logger log.Logger) *JumpHashClientPool {
+func NewJumpHashClientPool(clientFactory ClientFactory, dnsProvider AddressProvider, updateInterval time.Duration, logger log.Logger) (*JumpHashClientPool, error) {
 	selector := jumphash.DefaultSelector()
 	err := selector.SetServers(dnsProvider.Addresses()...)
 	if err != nil {
 		level.Warn(logger).Log("msg", "error updating servers", "err", err)
 	}
 
 	p := &JumpHashClientPool{
-		Pool:     pool,
-		Selector: selector,
-		done:     make(chan struct{}),
-		logger:   logger,
+		Selector:      selector,
+		clientFactory: clientFactory,
+		provider:      dnsProvider,
+		logger:        logger,
+		clients:       make(map[string]client.PoolClient, len(dnsProvider.Addresses())),
 	}
-	go p.updateLoop(dnsProvider, updateInterval)
 
-	return p
+	p.Service = services.NewTimerService(updateInterval, nil, p.updateLoop, nil)
+	return p, services.StartAndAwaitRunning(context.Background(), p.Service)
+}
+
+func (p *JumpHashClientPool) Stop() {
+	_ = services.StopAndAwaitTerminated(context.Background(), p.Service)
 }
 
 func (p *JumpHashClientPool) AddrForFingerprint(fp uint64) (string, error) {
@@ -80,35 +92,42 @@ func (p *JumpHashClientPool) Addr(key string) (string, error) {
 	return addr.String(), nil
 }
 
-func (p *JumpHashClientPool) Start() {
-	ctx := context.Background()
-	_ = services.StartAndAwaitRunning(ctx, p.Pool)
+func (p *JumpHashClientPool) updateLoop(_ context.Context) error {
+	err := p.SetServers(p.provider.Addresses()...)
+	if err != nil {
+		level.Warn(p.logger).Log("msg", "error updating servers", "err", err)
+	}
+	return nil
 }
 
-func (p *JumpHashClientPool) Stop() {
-	ctx := context.Background()
-	_ = services.StopAndAwaitTerminated(ctx, p.Pool)
-	close(p.done)
-}
+// GetClientFor implements clientPool.
+func (p *JumpHashClientPool) GetClientFor(addr string) (client.PoolClient, error) {
+	client, ok := p.fromCache(addr)
+	if ok {
+		return client, nil
+	}
+
+	// No client in cache so create one
+	p.Lock()
+	defer p.Unlock()
 
-func (p *JumpHashClientPool) updateLoop(provider AddressProvider, updateInterval time.Duration) {
-	ticker := time.NewTicker(updateInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-p.done:
-			return
-		case <-ticker.C:
-			servers := provider.Addresses()
-			// ServerList deterministically maps keys to _index_ of the server list.
-			// Since DNS returns records in different order each time, we sort to
-			// guarantee best possible match between nodes.
-			sort.Strings(servers)
-			err := p.SetServers(servers...)
-			if err != nil {
-				level.Warn(p.logger).Log("msg", "error updating servers", "err", err)
-			}
-		}
+	// Check if a client has been created just after checking the cache and before acquiring the lock.
+	client, ok = p.clients[addr]
+	if ok {
+		return client, nil
 	}
+
+	client, err := p.clientFactory.New(addr)
+	if err != nil {
+		return nil, err
+	}
+	p.clients[addr] = client
+	return client, nil
+}
+
+func (p *JumpHashClientPool) fromCache(addr string) (client.PoolClient, bool) {
+	p.RLock()
+	defer p.RUnlock()
+	client, ok := p.clients[addr]
+	return client, ok
 }