Elasticsearch: Add support for serverless connections (#114855)

* serverless connecction

* Adding api key

* fix

* addressing pr comments

* fixing tests

* refactoring

* changing to value semantic

* addressing pr comments

* minor changes

---------

Co-authored-by: Lucas Francisco Lopez <lucas.lopez@elastic.co>

pkg/tests/api/elasticsearch/elasticsearch_test.go

@@ -24,6 +24,24 @@ func TestMain(m *testing.M) {
 	testsuite.Run(m)
 }
 
+// mockElasticsearchHandler returns a handler that mocks Elasticsearch endpoints.
+// It responds to GET / with cluster info (required for datasource initialization)
+// and returns 401 Unauthorized for all other requests.
+func mockElasticsearchHandler(onRequest func(r *http.Request)) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.Method == http.MethodGet && r.URL.Path == "/":
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = w.Write([]byte(`{"version":{"build_flavor":"default","number":"8.0.0"}}`))
+		default:
+			if onRequest != nil {
+				onRequest(r)
+			}
+			w.WriteHeader(http.StatusUnauthorized)
+		}
+	}
+}
+
 func TestIntegrationElasticsearch(t *testing.T) {
 	testutil.SkipIntegrationTestInShortMode(t)
 
@@ -35,9 +53,8 @@ func TestIntegrationElasticsearch(t *testing.T) {
 	ctx := context.Background()
 
 	var outgoingRequest *http.Request
-	outgoingServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	outgoingServer := httptest.NewServer(mockElasticsearchHandler(func(r *http.Request) {
 		outgoingRequest = r
-		w.WriteHeader(http.StatusUnauthorized)
 	}))
 	t.Cleanup(outgoingServer.Close)
 

pkg/tsdb/elasticsearch/client/client.go

@@ -35,6 +35,7 @@ type DatasourceInfo struct {
 	Interval                   string
 	MaxConcurrentShardRequests int64
 	IncludeFrozen              bool
+	ClusterInfo                ClusterInfo
 }
 
 type ConfiguredFields struct {
@@ -197,7 +198,11 @@ func (c *baseClientImpl) createMultiSearchRequests(searchRequests []*SearchReque
 
 func (c *baseClientImpl) getMultiSearchQueryParameters() string {
 	var qs []string
-	qs = append(qs, fmt.Sprintf("max_concurrent_shard_requests=%d", c.ds.MaxConcurrentShardRequests))
+	// if the build flavor is not serverless, we can use the max concurrent shard requests
+	// this is because serverless clusters do not support max concurrent shard requests
+	if !c.ds.ClusterInfo.IsServerless() && c.ds.MaxConcurrentShardRequests > 0 {
+		qs = append(qs, fmt.Sprintf("max_concurrent_shard_requests=%d", c.ds.MaxConcurrentShardRequests))
+	}
 
 	if c.ds.IncludeFrozen {
 		qs = append(qs, "ignore_throttled=false")

pkg/tsdb/elasticsearch/client/cluster_info.go

@@ -0,0 +1,51 @@
+package es
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+)
+
+type VersionInfo struct {
+	BuildFlavor string `json:"build_flavor"`
+}
+
+// ClusterInfo represents Elasticsearch cluster information returned from the root endpoint.
+// It is used to determine cluster capabilities and configuration like whether the cluster is serverless.
+type ClusterInfo struct {
+	Version VersionInfo `json:"version"`
+}
+
+const (
+	BuildFlavorServerless = "serverless"
+)
+
+// GetClusterInfo fetches cluster information from the Elasticsearch root endpoint.
+// It returns the cluster build flavor which is used to determine if the cluster is serverless.
+func GetClusterInfo(httpCli *http.Client, url string) (clusterInfo ClusterInfo, err error) {
+	resp, err := httpCli.Get(url)
+	if err != nil {
+		return ClusterInfo{}, fmt.Errorf("error getting ES cluster info: %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return ClusterInfo{}, fmt.Errorf("unexpected status code %d getting ES cluster info", resp.StatusCode)
+	}
+
+	defer func() {
+		if closeErr := resp.Body.Close(); closeErr != nil && err == nil {
+			err = fmt.Errorf("error closing response body: %w", closeErr)
+		}
+	}()
+
+	err = json.NewDecoder(resp.Body).Decode(&clusterInfo)
+	if err != nil {
+		return ClusterInfo{}, fmt.Errorf("error decoding ES cluster info: %w", err)
+	}
+
+	return clusterInfo, nil
+}
+
+func (ci ClusterInfo) IsServerless() bool {
+	return ci.Version.BuildFlavor == BuildFlavorServerless
+}

pkg/tsdb/elasticsearch/client/cluster_info_test.go

@@ -0,0 +1,188 @@
+package es
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetClusterInfo(t *testing.T) {
+	t.Run("Should successfully get cluster info", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+			rw.Header().Set("Content-Type", "application/json")
+			_, err := rw.Write([]byte(`{
+				"name": "test-cluster",
+				"cluster_name": "elasticsearch",
+				"cluster_uuid": "abc123",
+				"version": {
+					"number": "8.0.0",
+					"build_flavor": "default",
+					"build_type": "tar",
+					"build_hash": "abc123",
+					"build_date": "2023-01-01T00:00:00.000Z",
+					"build_snapshot": false,
+					"lucene_version": "9.0.0"
+				}
+			}`))
+			require.NoError(t, err)
+		}))
+
+		t.Cleanup(func() {
+			ts.Close()
+		})
+
+		clusterInfo, err := GetClusterInfo(ts.Client(), ts.URL)
+
+		require.NoError(t, err)
+		require.NotNil(t, clusterInfo)
+		assert.Equal(t, "default", clusterInfo.Version.BuildFlavor)
+	})
+
+	t.Run("Should successfully get serverless cluster info", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+			rw.Header().Set("Content-Type", "application/json")
+			_, err := rw.Write([]byte(`{
+				"name": "serverless-cluster",
+				"cluster_name": "elasticsearch",
+				"cluster_uuid": "def456",
+				"version": {
+					"number": "8.11.0",
+					"build_flavor": "serverless",
+					"build_type": "docker",
+					"build_hash": "def456",
+					"build_date": "2023-11-01T00:00:00.000Z",
+					"build_snapshot": false,
+					"lucene_version": "9.8.0"
+				}
+			}`))
+			require.NoError(t, err)
+		}))
+
+		t.Cleanup(func() {
+			ts.Close()
+		})
+
+		clusterInfo, err := GetClusterInfo(ts.Client(), ts.URL)
+
+		require.NoError(t, err)
+		require.NotNil(t, clusterInfo)
+		assert.Equal(t, "serverless", clusterInfo.Version.BuildFlavor)
+		assert.True(t, clusterInfo.IsServerless())
+	})
+
+	t.Run("Should return error when HTTP request fails", func(t *testing.T) {
+		clusterInfo, err := GetClusterInfo(http.DefaultClient, "http://invalid-url-that-does-not-exist.local:9999")
+
+		require.Error(t, err)
+		require.Equal(t, ClusterInfo{}, clusterInfo)
+		assert.Contains(t, err.Error(), "error getting ES cluster info")
+	})
+
+	t.Run("Should return error when response body is invalid JSON", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+			rw.Header().Set("Content-Type", "application/json")
+			_, err := rw.Write([]byte(`{"invalid json`))
+			require.NoError(t, err)
+		}))
+
+		t.Cleanup(func() {
+			ts.Close()
+		})
+
+		clusterInfo, err := GetClusterInfo(ts.Client(), ts.URL)
+
+		require.Error(t, err)
+		require.Equal(t, ClusterInfo{}, clusterInfo)
+		assert.Contains(t, err.Error(), "error decoding ES cluster info")
+	})
+
+	t.Run("Should handle empty version object", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+			rw.Header().Set("Content-Type", "application/json")
+			_, err := rw.Write([]byte(`{
+				"name": "test-cluster",
+				"version": {}
+			}`))
+			require.NoError(t, err)
+		}))
+
+		t.Cleanup(func() {
+			ts.Close()
+		})
+
+		clusterInfo, err := GetClusterInfo(ts.Client(), ts.URL)
+
+		require.NoError(t, err)
+		require.Equal(t, ClusterInfo{}, clusterInfo)
+		assert.Equal(t, "", clusterInfo.Version.BuildFlavor)
+		assert.False(t, clusterInfo.IsServerless())
+	})
+
+	t.Run("Should handle HTTP error status codes", func(t *testing.T) {
+		ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+			rw.WriteHeader(http.StatusUnauthorized)
+			_, err := rw.Write([]byte(`{"error": "Unauthorized"}`))
+			require.NoError(t, err)
+		}))
+
+		t.Cleanup(func() {
+			ts.Close()
+		})
+
+		clusterInfo, err := GetClusterInfo(ts.Client(), ts.URL)
+
+		require.Error(t, err)
+		require.Equal(t, ClusterInfo{}, clusterInfo)
+		assert.Contains(t, err.Error(), "unexpected status code 401 getting ES cluster info")
+	})
+}
+
+func TestClusterInfo_IsServerless(t *testing.T) {
+	t.Run("Should return true when build_flavor is serverless", func(t *testing.T) {
+		clusterInfo := ClusterInfo{
+			Version: VersionInfo{
+				BuildFlavor: BuildFlavorServerless,
+			},
+		}
+
+		assert.True(t, clusterInfo.IsServerless())
+	})
+
+	t.Run("Should return false when build_flavor is default", func(t *testing.T) {
+		clusterInfo := ClusterInfo{
+			Version: VersionInfo{
+				BuildFlavor: "default",
+			},
+		}
+
+		assert.False(t, clusterInfo.IsServerless())
+	})
+
+	t.Run("Should return false when build_flavor is empty", func(t *testing.T) {
+		clusterInfo := ClusterInfo{
+			Version: VersionInfo{
+				BuildFlavor: "",
+			},
+		}
+
+		assert.False(t, clusterInfo.IsServerless())
+	})
+
+	t.Run("Should return false when build_flavor is unknown value", func(t *testing.T) {
+		clusterInfo := ClusterInfo{
+			Version: VersionInfo{
+				BuildFlavor: "unknown",
+			},
+		}
+
+		assert.False(t, clusterInfo.IsServerless())
+	})
+
+	t.Run("should return false when cluster info is empty", func(t *testing.T) {
+		clusterInfo := ClusterInfo{}
+		assert.False(t, clusterInfo.IsServerless())
+	})
+}

pkg/tsdb/elasticsearch/elasticsearch.go

@@ -88,6 +88,14 @@ func newInstanceSettings(httpClientProvider *httpclient.Provider) datasource.Ins
 			httpCliOpts.SigV4.Service = "es"
 		}
 
+		apiKeyAuth, ok := jsonData["apiKeyAuth"].(bool)
+		if ok && apiKeyAuth {
+			apiKey := settings.DecryptedSecureJSONData["apiKey"]
+			if apiKey != "" {
+				httpCliOpts.Header.Add("Authorization", "ApiKey "+apiKey)
+			}
+		}
+
 		httpCli, err := httpClientProvider.New(httpCliOpts)
 		if err != nil {
 			return nil, err
@@ -151,6 +159,11 @@ func newInstanceSettings(httpClientProvider *httpclient.Provider) datasource.Ins
 			includeFrozen = false
 		}
 
+		clusterInfo, err := es.GetClusterInfo(httpCli, settings.URL)
+		if err != nil {
+			return nil, err
+		}
+
 		configuredFields := es.ConfiguredFields{
 			TimeField:       timeField,
 			LogLevelField:   logLevelField,
@@ -166,6 +179,7 @@ func newInstanceSettings(httpClientProvider *httpclient.Provider) datasource.Ins
 			ConfiguredFields:           configuredFields,
 			Interval:                   interval,
 			IncludeFrozen:              includeFrozen,
+			ClusterInfo:                clusterInfo,
 		}
 		return model, nil
 	}

pkg/tsdb/elasticsearch/elasticsearch_test.go

@@ -3,6 +3,8 @@ package elasticsearch
 import (
 	"context"
 	"encoding/json"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
@@ -18,8 +20,26 @@ type datasourceInfo struct {
 	Interval                   string `json:"interval"`
 }
 
+// mockElasticsearchServer creates a test HTTP server that mocks Elasticsearch cluster info endpoint
+func mockElasticsearchServer() *httptest.Server {
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		// Return a mock Elasticsearch cluster info response
+		_ = json.NewEncoder(w).Encode(map[string]interface{}{
+			"version": map[string]interface{}{
+				"build_flavor": "serverless",
+				"number":       "8.0.0",
+			},
+		})
+	}))
+}
+
 func TestNewInstanceSettings(t *testing.T) {
 	t.Run("fields exist", func(t *testing.T) {
+		server := mockElasticsearchServer()
+		defer server.Close()
+
 		dsInfo := datasourceInfo{
 			TimeField:                  "@timestamp",
 			MaxConcurrentShardRequests: 5,
@@ -28,6 +48,7 @@ func TestNewInstanceSettings(t *testing.T) {
 		require.NoError(t, err)
 
 		dsSettings := backend.DataSourceInstanceSettings{
+			URL:      server.URL,
 			JSONData: json.RawMessage(settingsJSON),
 		}
 
@@ -37,6 +58,9 @@ func TestNewInstanceSettings(t *testing.T) {
 
 	t.Run("timeField", func(t *testing.T) {
 		t.Run("is nil", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				MaxConcurrentShardRequests: 5,
 				Interval:                   "Daily",
@@ -46,6 +70,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -54,6 +79,9 @@ func TestNewInstanceSettings(t *testing.T) {
 		})
 
 		t.Run("is empty", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				MaxConcurrentShardRequests: 5,
 				Interval:                   "Daily",
@@ -64,6 +92,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -74,6 +103,9 @@ func TestNewInstanceSettings(t *testing.T) {
 
 	t.Run("maxConcurrentShardRequests", func(t *testing.T) {
 		t.Run("no maxConcurrentShardRequests", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				TimeField: "@timestamp",
 			}
@@ -81,6 +113,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -90,6 +123,9 @@ func TestNewInstanceSettings(t *testing.T) {
 		})
 
 		t.Run("string maxConcurrentShardRequests", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				TimeField:                  "@timestamp",
 				MaxConcurrentShardRequests: "10",
@@ -98,6 +134,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -107,6 +144,9 @@ func TestNewInstanceSettings(t *testing.T) {
 		})
 
 		t.Run("number maxConcurrentShardRequests", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				TimeField:                  "@timestamp",
 				MaxConcurrentShardRequests: 10,
@@ -115,6 +155,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -124,6 +165,9 @@ func TestNewInstanceSettings(t *testing.T) {
 		})
 
 		t.Run("zero maxConcurrentShardRequests", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				TimeField:                  "@timestamp",
 				MaxConcurrentShardRequests: 0,
@@ -132,6 +176,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -141,6 +186,9 @@ func TestNewInstanceSettings(t *testing.T) {
 		})
 
 		t.Run("negative maxConcurrentShardRequests", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				TimeField:                  "@timestamp",
 				MaxConcurrentShardRequests: -10,
@@ -149,6 +197,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -158,6 +207,9 @@ func TestNewInstanceSettings(t *testing.T) {
 		})
 
 		t.Run("float maxConcurrentShardRequests", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				TimeField:                  "@timestamp",
 				MaxConcurrentShardRequests: 10.5,
@@ -166,6 +218,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 
@@ -175,6 +228,9 @@ func TestNewInstanceSettings(t *testing.T) {
 		})
 
 		t.Run("invalid maxConcurrentShardRequests", func(t *testing.T) {
+			server := mockElasticsearchServer()
+			defer server.Close()
+
 			dsInfo := datasourceInfo{
 				TimeField:                  "@timestamp",
 				MaxConcurrentShardRequests: "invalid",
@@ -183,6 +239,7 @@ func TestNewInstanceSettings(t *testing.T) {
 			require.NoError(t, err)
 
 			dsSettings := backend.DataSourceInstanceSettings{
+				URL:      server.URL,
 				JSONData: json.RawMessage(settingsJSON),
 			}
 

pkg/tsdb/elasticsearch/healthcheck.go

@@ -28,7 +28,6 @@ func (s *Service) CheckHealth(ctx context.Context, req *backend.CheckHealthReque
 			Message: "Health check failed: Failed to get data source info",
 		}, nil
 	}
-
 	healthStatusUrl, err := url.Parse(ds.URL)
 	if err != nil {
 		logger.Error("Failed to parse data source URL", "error", err)
@@ -38,6 +37,14 @@ func (s *Service) CheckHealth(ctx context.Context, req *backend.CheckHealthReque
 		}, nil
 	}
 
+	// If the cluster is serverless, return a healthy result
+	if ds.ClusterInfo.IsServerless() {
+		return &backend.CheckHealthResult{
+			Status:  backend.HealthStatusOk,
+			Message: "Elasticsearch Serverless data source is healthy.",
+		}, nil
+	}
+
 	// check that ES is healthy
 	healthStatusUrl.Path = path.Join(healthStatusUrl.Path, "_cluster/health")
 	healthStatusUrl.RawQuery = "wait_for_status=yellow"