diff --git a/pkg/tests/api/elasticsearch/elasticsearch_test.go b/pkg/tests/api/elasticsearch/elasticsearch_test.go index 09277c944f0..651dd74e9e2 100644 --- a/pkg/tests/api/elasticsearch/elasticsearch_test.go +++ b/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) diff --git a/pkg/tsdb/elasticsearch/client/client.go b/pkg/tsdb/elasticsearch/client/client.go index fbb3e09f092..49b24651609 100644 --- a/pkg/tsdb/elasticsearch/client/client.go +++ b/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") diff --git a/pkg/tsdb/elasticsearch/client/cluster_info.go b/pkg/tsdb/elasticsearch/client/cluster_info.go new file mode 100644 index 00000000000..eb89189804f --- /dev/null +++ b/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 +} diff --git a/pkg/tsdb/elasticsearch/client/cluster_info_test.go b/pkg/tsdb/elasticsearch/client/cluster_info_test.go new file mode 100644 index 00000000000..0fdcc46e813 --- /dev/null +++ b/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()) + }) +} diff --git a/pkg/tsdb/elasticsearch/elasticsearch.go b/pkg/tsdb/elasticsearch/elasticsearch.go index 40073bf1740..0432bbcee20 100644 --- a/pkg/tsdb/elasticsearch/elasticsearch.go +++ b/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 } diff --git a/pkg/tsdb/elasticsearch/elasticsearch_test.go b/pkg/tsdb/elasticsearch/elasticsearch_test.go index 8ab3cabc7e5..35ec1f814ce 100644 --- a/pkg/tsdb/elasticsearch/elasticsearch_test.go +++ b/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), } diff --git a/pkg/tsdb/elasticsearch/healthcheck.go b/pkg/tsdb/elasticsearch/healthcheck.go index 928945691de..cb5d0a866db 100644 --- a/pkg/tsdb/elasticsearch/healthcheck.go +++ b/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" diff --git a/public/app/plugins/datasource/elasticsearch/configuration/ApiKeyConfig.tsx b/public/app/plugins/datasource/elasticsearch/configuration/ApiKeyConfig.tsx new file mode 100644 index 00000000000..8433160d4f2 --- /dev/null +++ b/public/app/plugins/datasource/elasticsearch/configuration/ApiKeyConfig.tsx @@ -0,0 +1,22 @@ +import { onUpdateDatasourceSecureJsonDataOption, updateDatasourcePluginResetOption } from '@grafana/data'; +import { InlineField, SecretInput } from '@grafana/ui'; + +import { Props } from './ConfigEditor'; + +export const ApiKeyConfig = (props: Props) => { + const { options } = props; + + return ( + + updateDatasourcePluginResetOption(props, 'apiKey')} + onChange={onUpdateDatasourceSecureJsonDataOption(props, 'apiKey')} + /> + + ); +}; diff --git a/public/app/plugins/datasource/elasticsearch/configuration/ConfigEditor.tsx b/public/app/plugins/datasource/elasticsearch/configuration/ConfigEditor.tsx index 5961ebef510..57e3f8dc92a 100644 --- a/public/app/plugins/datasource/elasticsearch/configuration/ConfigEditor.tsx +++ b/public/app/plugins/datasource/elasticsearch/configuration/ConfigEditor.tsx @@ -14,14 +14,15 @@ import { import { config } from '@grafana/runtime'; import { Alert, SecureSocksProxySettings, Divider, Stack } from '@grafana/ui'; -import { ElasticsearchOptions } from '../types'; +import { ElasticsearchOptions, ElasticsearchSecureJsonData } from '../types'; +import { ApiKeyConfig } from './ApiKeyConfig'; import { DataLinks } from './DataLinks'; import { ElasticDetails } from './ElasticDetails'; import { LogsConfig } from './LogsConfig'; import { coerceOptions, isValidOptions } from './utils'; -export type Props = DataSourcePluginOptionsEditorProps; +export type Props = DataSourcePluginOptionsEditorProps; export const ConfigEditor = (props: Props) => { const { options, onOptionsChange } = props; @@ -48,6 +49,16 @@ export const ConfigEditor = (props: Props) => { authProps.selectedMethod = options.jsonData.sigV4Auth ? 'custom-sigv4' : authProps.selectedMethod; } + authProps.customMethods = [ + { + id: 'custom-api-key', + label: 'API Key', + description: 'API Key authentication', + component: , + }, + ]; + authProps.selectedMethod = options.jsonData.apiKeyAuth ? 'custom-api-key' : authProps.selectedMethod; + return ( <> {options.access === 'direct' && ( @@ -73,6 +84,7 @@ export const ConfigEditor = (props: Props) => { jsonData: { ...options.jsonData, sigV4Auth: method === 'custom-sigv4', + apiKeyAuth: method === 'custom-api-key', oauthPassThru: method === AuthMethod.OAuthForward, }, }); diff --git a/public/app/plugins/datasource/elasticsearch/types.ts b/public/app/plugins/datasource/elasticsearch/types.ts index 4645a2a824f..d435f1b4594 100644 --- a/public/app/plugins/datasource/elasticsearch/types.ts +++ b/public/app/plugins/datasource/elasticsearch/types.ts @@ -64,6 +64,11 @@ export interface ElasticsearchOptions extends DataSourceJsonData { sigV4Auth?: boolean; oauthPassThru?: boolean; defaultQueryMode?: QueryType; + apiKeyAuth?: boolean; +} + +export interface ElasticsearchSecureJsonData { + apiKey?: string; } export type QueryType = 'metrics' | 'logs' | 'raw_data' | 'raw_document';