[receiver/elasticsearch]: add scraping indices (open-telemetry#14748)

.chloggen/elasticsearch-receiver-add-indices.yaml

@@ -0,0 +1,5 @@
+change_type: enhancement
+component: elasticsearchreceiver
+note: Add scraping index stats and generate search metrics
+issues: [14635]
+

receiver/elasticsearchreceiver/README.md

@@ -6,7 +6,7 @@
 | Supported pipeline types | metrics   |
 | Distributions            | [contrib] |
 
-This receiver queries the Elasticsearch [node stats](https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-nodes-stats.html) and [cluster health](https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-health.html) endpoints in order to scrape metrics from a running elasticsearch cluster.
+This receiver queries the Elasticsearch [node stats](https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-nodes-stats.html), [cluster health](https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-health.html) and [index stats](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-stats.html) endpoints in order to scrape metrics from a running elasticsearch cluster.
 
 ## Prerequisites
 
@@ -21,6 +21,7 @@ The following settings are optional:
 - `metrics` (default: see `DefaultMetricsSettings` [here](./internal/metadata/generated_metrics.go): Allows enabling and disabling specific metrics from being collected in this receiver.
 - `nodes` (default: `["_all"]`): Allows specifying node filters that define which nodes are scraped for node-level metrics. See [the Elasticsearch documentation](https://www.elastic.co/guide/en/elasticsearch/reference/7.9/cluster.html#cluster-nodes) for allowed filters. If this option is left explicitly empty, then no node-level metrics will be scraped.
 - `skip_cluster_metrics` (default: `false`): If true, cluster-level metrics will not be scraped.
+- `indices` (default: `["_all"]`): Allows specifying index filters that define which indices are scraped for index-level metrics. See [the Elasticsearch documentation](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-stats.html#index-stats-api-path-params) for allowed filters. If this option is left explicitly empty, then no index-level metrics will be scraped.
 - `endpoint` (default = `http://localhost:9200`): The base URL of the Elasticsearch API for the cluster to monitor.
 - `username` (no default): Specifies the username used to authenticate with Elasticsearch using basic auth. Must be specified if password is specified.
 - `password` (no default): Specifies the password used to authenticate with Elasticsearch using basic auth. Must be specified if username is specified.
@@ -36,6 +37,7 @@ receivers:
         enabled: false
     nodes: ["_local"]
     skip_cluster_metrics: true
+    indices: [".geoip_databases"]
     endpoint: http://localhost:9200
     username: otel
     password: password

receiver/elasticsearchreceiver/client.go

@@ -40,7 +40,8 @@ var (
 type elasticsearchClient interface {
 	NodeStats(ctx context.Context, nodes []string) (*model.NodeStats, error)
 	ClusterHealth(ctx context.Context) (*model.ClusterHealth, error)
-	Version(ctx context.Context) (*model.VersionResponse, error)
+	IndexStats(ctx context.Context, indices []string) (*model.IndexStats, error)
+	ClusterMetadata(ctx context.Context) (*model.ClusterMetadataResponse, error)
 }
 
 // defaultElasticsearchClient is the main implementation of elasticsearchClient.
@@ -88,6 +89,8 @@ const nodeStatsMetrics = "breaker,indices,process,jvm,thread_pool,transport,http
 // nodeStatsIndexMetrics is a comma separated list of index metrics that will be gathered from NodeStats.
 const nodeStatsIndexMetrics = "store,docs,indexing,get,search,merge,refresh,flush,warmer,query_cache,fielddata,translog"
 
+const indexStatsMetrics = "search"
+
 func (c defaultElasticsearchClient) NodeStats(ctx context.Context, nodes []string) (*model.NodeStats, error) {
 	var nodeSpec string
 	if len(nodes) > 0 {
@@ -119,13 +122,34 @@ func (c defaultElasticsearchClient) ClusterHealth(ctx context.Context) (*model.C
 	return &clusterHealth, err
 }
 
-func (c defaultElasticsearchClient) Version(ctx context.Context) (*model.VersionResponse, error) {
+func (c defaultElasticsearchClient) IndexStats(ctx context.Context, indices []string) (*model.IndexStats, error) {
+	var indexSpec string
+	if len(indices) > 0 {
+		indexSpec = strings.Join(indices, ",")
+	} else {
+		indexSpec = "_all"
+	}
+
+	indexStatsPath := fmt.Sprintf("%s/_stats/%s", indexSpec, indexStatsMetrics)
+
+	body, err := c.doRequest(ctx, indexStatsPath)
+	if err != nil {
+		return nil, err
+	}
+
+	indexStats := model.IndexStats{}
+	err = json.Unmarshal(body, &indexStats)
+
+	return &indexStats, err
+}
+
+func (c defaultElasticsearchClient) ClusterMetadata(ctx context.Context) (*model.ClusterMetadataResponse, error) {
 	body, err := c.doRequest(ctx, "")
 	if err != nil {
 		return nil, err
 	}
 
-	versionResponse := model.VersionResponse{}
+	versionResponse := model.ClusterMetadataResponse{}
 	err = json.Unmarshal(body, &versionResponse)
 	return &versionResponse, err
 }

receiver/elasticsearchreceiver/client_test.go

@@ -236,12 +236,12 @@ func TestClusterHealthNoAuthorization(t *testing.T) {
 	require.ErrorIs(t, err, errUnauthorized)
 }
 
-func TestVersionNoPassword(t *testing.T) {
-	versionJSON, err := os.ReadFile("./testdata/sample_payloads/version.json")
+func TestMetadataNoPassword(t *testing.T) {
+	metadataJSON, err := os.ReadFile("./testdata/sample_payloads/metadata.json")
 	require.NoError(t, err)
 
-	actualVersion := model.VersionResponse{}
-	require.NoError(t, json.Unmarshal(versionJSON, &actualVersion))
+	actualMetadata := model.ClusterMetadataResponse{}
+	require.NoError(t, json.Unmarshal(metadataJSON, &actualMetadata))
 
 	elasticsearchMock := mockServer(t, "", "")
 	defer elasticsearchMock.Close()
@@ -254,18 +254,18 @@ func TestVersionNoPassword(t *testing.T) {
 	require.NoError(t, err)
 
 	ctx := context.Background()
-	version, err := client.Version(ctx)
+	metadata, err := client.ClusterMetadata(ctx)
 	require.NoError(t, err)
 
-	require.Equal(t, &actualVersion, version)
+	require.Equal(t, &actualMetadata, metadata)
 }
 
-func TestVersionAuthentication(t *testing.T) {
-	versionJSON, err := os.ReadFile("./testdata/sample_payloads/version.json")
+func TestMetadataAuthentication(t *testing.T) {
+	metadataJSON, err := os.ReadFile("./testdata/sample_payloads/metadata.json")
 	require.NoError(t, err)
 
-	actualVersion := model.VersionResponse{}
-	require.NoError(t, json.Unmarshal(versionJSON, &actualVersion))
+	actualMetadata := model.ClusterMetadataResponse{}
+	require.NoError(t, json.Unmarshal(metadataJSON, &actualMetadata))
 
 	username := "user"
 	password := "pass"
@@ -283,13 +283,13 @@ func TestVersionAuthentication(t *testing.T) {
 	require.NoError(t, err)
 
 	ctx := context.Background()
-	version, err := client.Version(ctx)
+	metadata, err := client.ClusterMetadata(ctx)
 	require.NoError(t, err)
 
-	require.Equal(t, &actualVersion, version)
+	require.Equal(t, &actualMetadata, metadata)
 }
 
-func TestVersionNoAuthentication(t *testing.T) {
+func TestMetadataNoAuthentication(t *testing.T) {
 	elasticsearchMock := mockServer(t, "user", "pass")
 	defer elasticsearchMock.Close()
 
@@ -301,11 +301,11 @@ func TestVersionNoAuthentication(t *testing.T) {
 	require.NoError(t, err)
 
 	ctx := context.Background()
-	_, err = client.Version(ctx)
+	_, err = client.ClusterMetadata(ctx)
 	require.ErrorIs(t, err, errUnauthenticated)
 }
 
-func TestVersionNoAuthorization(t *testing.T) {
+func TestMetadataNoAuthorization(t *testing.T) {
 	elasticsearchMock := mockServer(t, "user", "pass")
 	defer elasticsearchMock.Close()
 
@@ -319,7 +319,7 @@ func TestVersionNoAuthorization(t *testing.T) {
 	require.NoError(t, err)
 
 	ctx := context.Background()
-	_, err = client.Version(ctx)
+	_, err = client.ClusterMetadata(ctx)
 	require.ErrorIs(t, err, errUnauthorized)
 }
 
@@ -366,14 +366,126 @@ func TestDoRequest404(t *testing.T) {
 	require.Contains(t, err.Error(), "404")
 }
 
+func TestIndexStatsNoPassword(t *testing.T) {
+	indexJSON, err := os.ReadFile("./testdata/sample_payloads/indices.json")
+	require.NoError(t, err)
+
+	actualIndexStats := model.IndexStats{}
+	require.NoError(t, json.Unmarshal(indexJSON, &actualIndexStats))
+
+	elasticsearchMock := mockServer(t, "", "")
+	defer elasticsearchMock.Close()
+
+	client, err := newElasticsearchClient(componenttest.NewNopTelemetrySettings(), Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{
+			Endpoint: elasticsearchMock.URL,
+		},
+	}, componenttest.NewNopHost())
+	require.NoError(t, err)
+	ctx := context.Background()
+	indexStats, err := client.IndexStats(ctx, []string{"_all"})
+	require.NoError(t, err)
+
+	require.Equal(t, &actualIndexStats, indexStats)
+}
+
+func TestIndexStatsNilNodes(t *testing.T) {
+	indexJSON, err := os.ReadFile("./testdata/sample_payloads/indices.json")
+	require.NoError(t, err)
+
+	actualIndexStats := model.IndexStats{}
+	require.NoError(t, json.Unmarshal(indexJSON, &actualIndexStats))
+
+	elasticsearchMock := mockServer(t, "", "")
+	defer elasticsearchMock.Close()
+
+	client, err := newElasticsearchClient(componenttest.NewNopTelemetrySettings(), Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{
+			Endpoint: elasticsearchMock.URL,
+		},
+	}, componenttest.NewNopHost())
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	indexStats, err := client.IndexStats(ctx, nil)
+	require.NoError(t, err)
+
+	require.Equal(t, &actualIndexStats, indexStats)
+}
+
+func TestIndexStatsAuthentication(t *testing.T) {
+	indexJSON, err := os.ReadFile("./testdata/sample_payloads/indices.json")
+	require.NoError(t, err)
+
+	actualIndexStats := model.IndexStats{}
+	require.NoError(t, json.Unmarshal(indexJSON, &actualIndexStats))
+
+	username := "user"
+	password := "pass"
+
+	elasticsearchMock := mockServer(t, username, password)
+	defer elasticsearchMock.Close()
+
+	client, err := newElasticsearchClient(componenttest.NewNopTelemetrySettings(), Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{
+			Endpoint: elasticsearchMock.URL,
+		},
+		Username: username,
+		Password: password,
+	}, componenttest.NewNopHost())
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	indexStats, err := client.IndexStats(ctx, []string{"_all"})
+	require.NoError(t, err)
+
+	require.Equal(t, &actualIndexStats, indexStats)
+}
+
+func TestIndexStatsNoAuthentication(t *testing.T) {
+	elasticsearchMock := mockServer(t, "user", "pass")
+	defer elasticsearchMock.Close()
+
+	client, err := newElasticsearchClient(componenttest.NewNopTelemetrySettings(), Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{
+			Endpoint: elasticsearchMock.URL,
+		},
+	}, componenttest.NewNopHost())
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	_, err = client.IndexStats(ctx, []string{"_all"})
+	require.ErrorIs(t, err, errUnauthenticated)
+}
+
+func TestIndexStatsBadAuthentication(t *testing.T) {
+	elasticsearchMock := mockServer(t, "user", "pass")
+	defer elasticsearchMock.Close()
+
+	client, err := newElasticsearchClient(componenttest.NewNopTelemetrySettings(), Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{
+			Endpoint: elasticsearchMock.URL,
+		},
+		Username: "bad_user",
+		Password: "bad_pass",
+	}, componenttest.NewNopHost())
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	_, err = client.IndexStats(ctx, []string{"_all"})
+	require.ErrorIs(t, err, errUnauthorized)
+}
+
 // mockServer gives a mock elasticsearch server for testing; if username or password is included, they will be required for the client.
 // otherwise, authorization is ignored.
 func mockServer(t *testing.T, username, password string) *httptest.Server {
 	nodes, err := os.ReadFile("./testdata/sample_payloads/nodes_linux.json")
 	require.NoError(t, err)
+	indices, err := os.ReadFile("./testdata/sample_payloads/indices.json")
+	require.NoError(t, err)
 	health, err := os.ReadFile("./testdata/sample_payloads/health.json")
 	require.NoError(t, err)
-	version, err := os.ReadFile("./testdata/sample_payloads/version.json")
+	metadata, err := os.ReadFile("./testdata/sample_payloads/metadata.json")
 	require.NoError(t, err)
 
 	elasticsearchMock := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
@@ -395,17 +507,24 @@ func mockServer(t *testing.T, username, password string) *httptest.Server {
 			return
 		}
 
+		if strings.HasPrefix(req.URL.Path, "/_all/_stats") {
+			rw.WriteHeader(200)
+			_, err = rw.Write(indices)
+			require.NoError(t, err)
+			return
+		}
+
 		if strings.HasPrefix(req.URL.Path, "/_cluster/health") {
 			rw.WriteHeader(200)
 			_, err = rw.Write(health)
 			require.NoError(t, err)
 			return
 		}
 
-		// version check
+		// metadata check
 		if req.URL.Path == "/" {
 			rw.WriteHeader(200)
-			_, err = rw.Write(version)
+			_, err = rw.Write(metadata)
 			require.NoError(t, err)
 			return
 		}

receiver/elasticsearchreceiver/config.go

@@ -49,6 +49,11 @@ type Config struct {
 	Nodes []string `mapstructure:"nodes"`
 	// SkipClusterMetrics indicates whether cluster level metrics from /_cluster/health should be scraped or not.
 	SkipClusterMetrics bool `mapstructure:"skip_cluster_metrics"`
+	// Indices defines the indices to scrape.
+	// See https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-stats.html#index-stats-api-path-params
+	// for which names are viable.
+	// If Indices is empty, no indices will be scraped.
+	Indices []string `mapstructure:"indices"`
 	// Username is the username used when making REST calls to elasticsearch. Must be specified if Password is. Not required.
 	Username string `mapstructure:"username"`
 	// Password is the password used when making REST calls to elasticsearch. Must be specified if Username is. Not required.

receiver/elasticsearchreceiver/config_test.go

@@ -168,6 +168,7 @@ func TestLoadConfig(t *testing.T) {
 			expected: &Config{
 				SkipClusterMetrics: true,
 				Nodes:              []string{"_local"},
+				Indices:            []string{".geoip_databases"},
 				ScraperControllerSettings: scraperhelper.ScraperControllerSettings{
 					ReceiverSettings:   config.NewReceiverSettings(config.NewComponentID(typeStr)),
 					CollectionInterval: 2 * time.Minute,

receiver/elasticsearchreceiver/documentation.md

@@ -22,6 +22,8 @@ These are the metrics available for this scraper.
 | **elasticsearch.cluster.state_queue** | Number of cluster states in queue. | 1 | Sum(Int) | <ul> <li>cluster_state_queue_state</li> </ul> |
 | **elasticsearch.cluster.state_update.count** | The number of cluster state update attempts that changed the cluster state since the node started. | 1 | Sum(Int) | <ul> <li>cluster_state_update_state</li> </ul> |
 | **elasticsearch.cluster.state_update.time** | The cumulative amount of time updating the cluster state since the node started. | ms | Sum(Int) | <ul> <li>cluster_state_update_state</li> <li>cluster_state_update_type</li> </ul> |
+| **elasticsearch.index.operations.completed** | The number of operations completed for an index. | {operations} | Sum(Int) | <ul> <li>operation</li> <li>index_aggregation_type</li> </ul> |
+| **elasticsearch.index.operations.time** | Time spent on operations for an index. | ms | Sum(Int) | <ul> <li>operation</li> <li>index_aggregation_type</li> </ul> |
 | **elasticsearch.indexing_pressure.memory.limit** | Configured memory limit, in bytes, for the indexing requests. | By | Gauge(Int) | <ul> </ul> |
 | **elasticsearch.indexing_pressure.memory.total.primary_rejections** | Cumulative number of indexing requests rejected in the primary stage. | 1 | Sum(Int) | <ul> </ul> |
 | **elasticsearch.indexing_pressure.memory.total.replica_rejections** | Number of indexing requests rejected in the replica stage. | 1 | Sum(Int) | <ul> </ul> |
@@ -44,8 +46,8 @@ These are the metrics available for this scraper.
 | **elasticsearch.node.ingest.documents.current** | Total number of documents currently being ingested. | {documents} | Sum(Int) | <ul> </ul> |
 | **elasticsearch.node.ingest.operations.failed** | Total number of failed ingest operations during the lifetime of this node. | {operation} | Sum(Int) | <ul> </ul> |
 | **elasticsearch.node.open_files** | The number of open file descriptors held by the node. | {files} | Sum(Int) | <ul> </ul> |
-| **elasticsearch.node.operations.completed** | The number of operations completed. | {operations} | Sum(Int) | <ul> <li>operation</li> </ul> |
-| **elasticsearch.node.operations.time** | Time spent on operations. | ms | Sum(Int) | <ul> <li>operation</li> </ul> |
+| **elasticsearch.node.operations.completed** | The number of operations completed by a node. | {operations} | Sum(Int) | <ul> <li>operation</li> </ul> |
+| **elasticsearch.node.operations.time** | Time spent on operations by a node. | ms | Sum(Int) | <ul> <li>operation</li> </ul> |
 | **elasticsearch.node.pipeline.ingest.documents.current** | Total number of documents currently being ingested by a pipeline. | {documents} | Sum(Int) | <ul> <li>ingest_pipeline_name</li> </ul> |
 | **elasticsearch.node.pipeline.ingest.documents.preprocessed** | Number of documents preprocessed by the ingest pipeline. | {documents} | Sum(Int) | <ul> <li>ingest_pipeline_name</li> </ul> |
 | **elasticsearch.node.pipeline.ingest.operations.failed** | Total number of failed operations for the ingest pipeline. | {operation} | Sum(Int) | <ul> <li>ingest_pipeline_name</li> </ul> |
@@ -92,6 +94,7 @@ metrics:
 | Name | Description | Type |
 | ---- | ----------- | ---- |
 | elasticsearch.cluster.name | The name of the elasticsearch cluster. | String |
+| elasticsearch.index.name | The name of the elasticsearch index. | String |
 | elasticsearch.node.name | The name of the elasticsearch node. | String |
 
 ## Metric attributes
@@ -109,6 +112,7 @@ metrics:
 | document_state (state) | The state of the document. | active, deleted |
 | fs_direction (direction) | The direction of filesystem IO. | read, write |
 | health_status (status) | The health status of the cluster. | green, yellow, red |
+| index_aggregation_type (aggregation) | Type of shard aggregation for index statistics | primary_shards, total |
 | indexing_memory_state (state) | State of the indexing memory | current, total |
 | indexing_pressure_stage (stage) | Stage of the indexing pressure | coordinating, primary, replica |
 | ingest_pipeline_name (name) | Name of the ingest pipeline. |  |

receiver/elasticsearchreceiver/factory.go

@@ -58,6 +58,7 @@ func createDefaultConfig() config.Receiver {
 		},
 		Metrics: metadata.DefaultMetricsSettings(),
 		Nodes:   []string{"_all"},
+		Indices: []string{"_all"},
 	}
 }