Add cluster label to SLM metrics

collector/cluster_info.go

@@ -23,6 +23,8 @@ import (
 
 	"github.com/blang/semver/v4"
 	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 func init() {
@@ -35,7 +37,7 @@ type ClusterInfoCollector struct {
 	hc     *http.Client
 }
 
-func NewClusterInfo(logger *slog.Logger, u *url.URL, hc *http.Client) (Collector, error) {
+func NewClusterInfo(logger *slog.Logger, u *url.URL, hc *http.Client, ci *clusterinfo.Retriever) (Collector, error) {
 	return &ClusterInfoCollector{
 		logger: logger,
 		u:      u,

collector/cluster_info_test.go

@@ -21,9 +21,12 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/promslog"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 func TestClusterInfo(t *testing.T) {
@@ -80,7 +83,9 @@ func TestClusterInfo(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			c, err := NewClusterInfo(promslog.NewNopLogger(), u, http.DefaultClient)
+			logger := promslog.NewNopLogger()
+			ci := clusterinfo.New(logger, http.DefaultClient, u, time.Duration(300000000000))
+			c, err := NewClusterInfo(logger, u, http.DefaultClient, ci)
 			if err != nil {
 				t.Fatal(err)
 			}

collector/cluster_settings.go

@@ -26,6 +26,8 @@ import (
 
 	"github.com/imdario/mergo"
 	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 func init() {
@@ -38,7 +40,7 @@ type ClusterSettingsCollector struct {
 	hc     *http.Client
 }
 
-func NewClusterSettings(logger *slog.Logger, u *url.URL, hc *http.Client) (Collector, error) {
+func NewClusterSettings(logger *slog.Logger, u *url.URL, hc *http.Client, ci *clusterinfo.Retriever) (Collector, error) {
 	return &ClusterSettingsCollector{
 		logger: logger,
 		u:      u,

collector/cluster_settings_test.go

@@ -21,9 +21,12 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/promslog"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 func TestClusterSettingsStats(t *testing.T) {
@@ -136,7 +139,9 @@ elasticsearch_clustersettings_allocation_watermark_low_bytes 5.24288e+07
 				t.Fatal(err)
 			}
 
-			c, err := NewClusterSettings(promslog.NewNopLogger(), u, http.DefaultClient)
+			logger := promslog.NewNopLogger()
+			ci := clusterinfo.New(logger, http.DefaultClient, u, time.Duration(300000000000))
+			c, err := NewClusterSettings(logger, u, http.DefaultClient, ci)
 			if err != nil {
 				t.Fatal(err)
 			}

collector/collector.go

@@ -26,6 +26,8 @@ import (
 
 	"github.com/alecthomas/kingpin/v2"
 	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 const (
@@ -36,7 +38,7 @@ const (
 	defaultDisabled = false
 )
 
-type factoryFunc func(logger *slog.Logger, u *url.URL, hc *http.Client) (Collector, error)
+type factoryFunc func(logger *slog.Logger, u *url.URL, hc *http.Client, ci *clusterinfo.Retriever) (Collector, error)
 
 var (
 	factories              = make(map[string]factoryFunc)
@@ -97,7 +99,7 @@ type ElasticsearchCollector struct {
 type Option func(*ElasticsearchCollector) error
 
 // NewElasticsearchCollector creates a new ElasticsearchCollector
-func NewElasticsearchCollector(logger *slog.Logger, filters []string, options ...Option) (*ElasticsearchCollector, error) {
+func NewElasticsearchCollector(logger *slog.Logger, filters []string, clusterInfoRetriever *clusterinfo.Retriever, options ...Option) (*ElasticsearchCollector, error) {
 	e := &ElasticsearchCollector{logger: logger}
 	// Apply options to customize the collector
 	for _, o := range options {
@@ -127,7 +129,7 @@ func NewElasticsearchCollector(logger *slog.Logger, filters []string, options ..
 		if collector, ok := initiatedCollectors[key]; ok {
 			collectors[key] = collector
 		} else {
-			collector, err := factories[key](logger.With("collector", key), e.esURL, e.httpClient)
+			collector, err := factories[key](logger.With("collector", key), e.esURL, e.httpClient, clusterInfoRetriever)
 			if err != nil {
 				return nil, err
 			}

collector/data_stream.go

@@ -22,6 +22,8 @@ import (
 	"net/url"
 
 	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 var (
@@ -51,7 +53,7 @@ type DataStream struct {
 }
 
 // NewDataStream defines DataStream Prometheus metrics
-func NewDataStream(logger *slog.Logger, u *url.URL, hc *http.Client) (Collector, error) {
+func NewDataStream(logger *slog.Logger, u *url.URL, hc *http.Client, ci *clusterinfo.Retriever) (Collector, error) {
 	return &DataStream{
 		logger: logger,
 		hc:     hc,

collector/data_stream_test.go

@@ -21,9 +21,12 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/promslog"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 func TestDataStream(t *testing.T) {
@@ -64,7 +67,9 @@ func TestDataStream(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			c, err := NewDataStream(promslog.NewNopLogger(), u, http.DefaultClient)
+			logger := promslog.NewNopLogger()
+			ci := clusterinfo.New(logger, http.DefaultClient, u, time.Duration(300000000000))
+			c, err := NewDataStream(logger, u, http.DefaultClient, ci)
 			if err != nil {
 				t.Fatal(err)
 			}

collector/health_report.go

@@ -21,6 +21,8 @@ import (
 	"net/url"
 
 	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 var (
@@ -166,7 +168,7 @@ type HealthReport struct {
 	url    *url.URL
 }
 
-func NewHealthReport(logger *slog.Logger, url *url.URL, client *http.Client) (Collector, error) {
+func NewHealthReport(logger *slog.Logger, url *url.URL, client *http.Client, ci *clusterinfo.Retriever) (Collector, error) {
 	return &HealthReport{
 		logger: logger,
 		client: client,

collector/health_report_test.go

@@ -21,9 +21,12 @@ import (
 	"os"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/promslog"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 func TestHealthReport(t *testing.T) {
@@ -156,7 +159,9 @@ func TestHealthReport(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			c, err := NewHealthReport(promslog.NewNopLogger(), u, http.DefaultClient)
+			logger := promslog.NewNopLogger()
+			ci := clusterinfo.New(logger, http.DefaultClient, u, time.Duration(300000000000))
+			c, err := NewHealthReport(logger, u, http.DefaultClient, ci)
 			if err != nil {
 				t.Fatal(err)
 			}

collector/ilm.go

@@ -22,6 +22,8 @@ import (
 	"net/url"
 
 	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 var (
@@ -49,7 +51,7 @@ type ILM struct {
 	u      *url.URL
 }
 
-func NewILM(logger *slog.Logger, u *url.URL, hc *http.Client) (Collector, error) {
+func NewILM(logger *slog.Logger, u *url.URL, hc *http.Client, ci *clusterinfo.Retriever) (Collector, error) {
 	return &ILM{
 		logger: logger,
 		hc:     hc,

collector/ilm_test.go

@@ -22,9 +22,12 @@ import (
 	"path"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/promslog"
+
+	"github.com/prometheus-community/elasticsearch_exporter/pkg/clusterinfo"
 )
 
 func TestILM(t *testing.T) {
@@ -81,7 +84,9 @@ func TestILM(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			c, err := NewILM(promslog.NewNopLogger(), u, http.DefaultClient)
+			logger := promslog.NewNopLogger()
+			ci := clusterinfo.New(logger, http.DefaultClient, u, time.Duration(300000000000))
+			c, err := NewILM(logger, u, http.DefaultClient, ci)
 			if err != nil {
 				t.Fatal(err)
 			}

collector/indices.go

@@ -461,17 +461,6 @@ func NewIndices(logger *slog.Logger, client *http.Client, url *url.URL, shards b
 		},
 	}
 
-	// start go routine to fetch clusterinfo updates and save them to lastClusterinfo
-	go func() {
-		logger.Debug("starting cluster info receive loop")
-		for ci := range indices.clusterInfoCh {
-			if ci != nil {
-				logger.Debug("received cluster info update", "cluster", ci.ClusterName)
-				indices.lastClusterInfo = ci
-			}
-		}
-		logger.Debug("exiting cluster info receive loop")
-	}()
 	return indices
 }
 
@@ -481,6 +470,10 @@ func (i *Indices) ClusterLabelUpdates() *chan *clusterinfo.Response {
 	return &i.clusterInfoCh
 }
 
+func (i *Indices) SetClusterInfo(r *clusterinfo.Response) {
+	i.lastClusterInfo = r
+}
+
 // String implements the stringer interface. It is part of the clusterinfo.consumer interface
 func (i *Indices) String() string {
 	return namespace + "indices"

collector/shards.go

@@ -52,6 +52,10 @@ func (s *Shards) ClusterLabelUpdates() *chan *clusterinfo.Response {
 	return &s.clusterInfoCh
 }
 
+func (s *Shards) SetClusterInfo(r *clusterinfo.Response) {
+	s.lastClusterInfo = r
+}
+
 // String implements the stringer interface. It is part of the clusterinfo.consumer interface
 func (s *Shards) String() string {
 	return namespace + "shards"
@@ -111,18 +115,6 @@ func NewShards(logger *slog.Logger, client *http.Client, url *url.URL) *Shards {
 		}),
 	}
 
-	// start go routine to fetch clusterinfo updates and save them to lastClusterinfo
-	go func() {
-		logger.Debug("starting cluster info receive loop")
-		for ci := range shards.clusterInfoCh {
-			if ci != nil {
-				logger.Debug("received cluster info update", "cluster", ci.ClusterName)
-				shards.lastClusterInfo = ci
-			}
-		}
-		logger.Debug("exiting cluster info receive loop")
-	}()
-
 	return shards
 }