Add metrics indexing_pressure.memory.limit_in_bytes and indexing_pres…

…sure.memory.current.current.all_in_bytes to allow auto-scaling based on how close the cluster nodes are to dropping indexing requests due to the indxing request memory buffer reaching capacity.
prometheus-community · Jun 24, 2024 · 8f425be · 8f425be
1 parent 2f33605
commit 8f425be
Show file tree

Hide file tree

Showing 2 changed files with 80 additions and 17 deletions.
diff --git a/collector/nodes.go b/collector/nodes.go
@@ -96,6 +96,7 @@ var (
 	defaultRoleLabels               = []string{"cluster", "host", "name"}
 	defaultThreadPoolLabels         = append(defaultNodeLabels, "type")
 	defaultBreakerLabels            = append(defaultNodeLabels, "breaker")
+	defaultIndexingPressureLabels   = append(defaultNodeLabels, "indexing_pressure")
 	defaultFilesystemDataLabels     = append(defaultNodeLabels, "mount", "path")
 	defaultFilesystemIODeviceLabels = append(defaultNodeLabels, "device")
 	defaultCacheLabels              = append(defaultNodeLabels, "cache")
@@ -150,6 +151,13 @@ type breakerMetric struct {
 	Labels func(cluster string, node NodeStatsNodeResponse, breaker string) []string
 }
 
+type indexingPressureMetric struct {
+	Type   prometheus.ValueType
+	Desc   *prometheus.Desc
+	Value  func(indexingPressureStats NodeStatsIndexingPressureResponse) float64
+	Labels func(cluster string, node NodeStatsNodeResponse, indexingPressure string) []string
+}
+
 type threadPoolMetric struct {
 	Type   prometheus.ValueType
 	Desc   *prometheus.Desc
@@ -185,6 +193,7 @@ type Nodes struct {
 	nodeMetrics               []*nodeMetric
 	gcCollectionMetrics       []*gcCollectionMetric
 	breakerMetrics            []*breakerMetric
+	indexingPressureMetrics   []*indexingPressureMetric
 	threadPoolMetrics         []*threadPoolMetric
 	filesystemDataMetrics     []*filesystemDataMetric
 	filesystemIODeviceMetrics []*filesystemIODeviceMetric
@@ -1607,6 +1616,36 @@ func NewNodes(logger log.Logger, client *http.Client, url *url.URL, all bool, no
 				},
 			},
 		},
+		indexingPressureMetrics: []*indexingPressureMetric{
+			{
+				Type: prometheus.GaugeValue,
+				Desc: prometheus.NewDesc(
+					prometheus.BuildFQName(namespace, "indexing_pressure", "current_all_in_bytes"),
+					"Memory consumed, in bytes, by indexing requests in the coordinating, primary, or replica stage.",
+					defaultIndexingPressureLabels, nil,
+				),
+				Value: func(indexingPressureMem NodeStatsIndexingPressureResponse) float64 {
+					return float64(indexingPressureMem.Current.AllInBytes)
+				},
+				Labels: func(cluster string, node NodeStatsNodeResponse, indexingPressure string) []string {
+					return append(defaultNodeLabelValues(cluster, node), indexingPressure)
+				},
+			},
+			{
+				Type: prometheus.GaugeValue,
+				Desc: prometheus.NewDesc(
+					prometheus.BuildFQName(namespace, "indexing_pressure", "limit_in_bytes"),
+					"Configured memory limit, in bytes, for the indexing requests",
+					defaultIndexingPressureLabels, nil,
+				),
+				Value: func(indexingPressureStats NodeStatsIndexingPressureResponse) float64 {
+					return float64(indexingPressureStats.LimitInBytes)
+				},
+				Labels: func(cluster string, node NodeStatsNodeResponse, indexingPressure string) []string {
+					return append(defaultNodeLabelValues(cluster, node), indexingPressure)
+				},
+			},
+		},
 		threadPoolMetrics: []*threadPoolMetric{
 			{
 				Type: prometheus.CounterValue,
@@ -1919,6 +1958,18 @@ func (c *Nodes) Collect(ch chan<- prometheus.Metric) {
 			}
 		}
 
+		// Indexing Pressure stats
+		for indexingPressure, ipstats := range node.IndexingPressure {
+			for _, metric := range c.indexingPressureMetrics {
+				ch <- prometheus.MustNewConstMetric(
+					metric.Desc,
+					metric.Type,
+					metric.Value(ipstats),
+					metric.Labels(nodeStatsResp.ClusterName, node, indexingPressure)...,
+				)
+			}
+		}
+
 		// Thread Pool stats
 		for pool, pstats := range node.ThreadPool {
 			for _, metric := range c.threadPoolMetrics {

diff --git a/collector/nodes_response.go b/collector/nodes_response.go
@@ -23,23 +23,24 @@ type nodeStatsResponse struct {
 
 // NodeStatsNodeResponse defines node stats information structure for nodes
 type NodeStatsNodeResponse struct {
-	Name             string                                     `json:"name"`
-	Host             string                                     `json:"host"`
-	Timestamp        int64                                      `json:"timestamp"`
-	TransportAddress string                                     `json:"transport_address"`
-	Hostname         string                                     `json:"hostname"`
-	Roles            []string                                   `json:"roles"`
-	Attributes       map[string]string                          `json:"attributes"`
-	Indices          NodeStatsIndicesResponse                   `json:"indices"`
-	OS               NodeStatsOSResponse                        `json:"os"`
-	Network          NodeStatsNetworkResponse                   `json:"network"`
-	FS               NodeStatsFSResponse                        `json:"fs"`
-	ThreadPool       map[string]NodeStatsThreadPoolPoolResponse `json:"thread_pool"`
-	JVM              NodeStatsJVMResponse                       `json:"jvm"`
-	Breakers         map[string]NodeStatsBreakersResponse       `json:"breakers"`
-	HTTP             map[string]interface{}                     `json:"http"`
-	Transport        NodeStatsTransportResponse                 `json:"transport"`
-	Process          NodeStatsProcessResponse                   `json:"process"`
+	Name             string                                       `json:"name"`
+	Host             string                                       `json:"host"`
+	Timestamp        int64                                        `json:"timestamp"`
+	TransportAddress string                                       `json:"transport_address"`
+	Hostname         string                                       `json:"hostname"`
+	Roles            []string                                     `json:"roles"`
+	Attributes       map[string]string                            `json:"attributes"`
+	Indices          NodeStatsIndicesResponse                     `json:"indices"`
+	OS               NodeStatsOSResponse                          `json:"os"`
+	Network          NodeStatsNetworkResponse                     `json:"network"`
+	FS               NodeStatsFSResponse                          `json:"fs"`
+	ThreadPool       map[string]NodeStatsThreadPoolPoolResponse   `json:"thread_pool"`
+	JVM              NodeStatsJVMResponse                         `json:"jvm"`
+	Breakers         map[string]NodeStatsBreakersResponse         `json:"breakers"`
+	HTTP             map[string]interface{}                       `json:"http"`
+	Transport        NodeStatsTransportResponse                   `json:"transport"`
+	Process          NodeStatsProcessResponse                     `json:"process"`
+	IndexingPressure map[string]NodeStatsIndexingPressureResponse `json:"indexing_pressure"`
 }
 
 // NodeStatsBreakersResponse is a representation of a statistics about the field data circuit breaker
@@ -50,6 +51,17 @@ type NodeStatsBreakersResponse struct {
 	Tripped       int64   `json:"tripped"`
 }
 
+// NodeStatsIndexingPressureResponse is a representation of a elasticsearch indexing pressure
+type NodeStatsIndexingPressureResponse struct {
+	Current      NodeStatsIndexingPressureCurrentResponse `json:"current"`
+	LimitInBytes int64                                    `json:"limit_in_bytes"`
+}
+
+// NodeStatsIndexingPressureMemoryCurrentResponse is a representation of a elasticsearch indexing pressure current memory usage
+type NodeStatsIndexingPressureCurrentResponse struct {
+	AllInBytes int64 `json:"all_in_bytes"`
+}
+
 // NodeStatsJVMResponse is a representation of a JVM stats, memory pool information, garbage collection, buffer pools, number of loaded/unloaded classes
 type NodeStatsJVMResponse struct {
 	BufferPools map[string]NodeStatsJVMBufferPoolResponse `json:"buffer_pools"`