Reintroduce Elasticsearch compatibility from OpenDistro datasource (#5)

* Allow flavor selection in datasource config

* Add support for Elasticsearch to QueryBuilder

* Show only supported metric aggregations bases on configured flavor

* Restore maxConcurrentShardRequests settings

* ES compatibility for getFields

* ES compatibility for getTerms

* fix watch issue

* default to OpenSearch when flavor is not specified

* reintroduce BE compatibility with ES

* Restore compatibility for Elasticsearch in BE tests

* Merge version & flavor in a single select

* Fix configeditor tests

* Fix provisioning example

* Fix config editor test

* Apply suggestions from code review

Co-authored-by: Piotr Jamróz <[email protected]>

Co-authored-by: Piotr Jamróz <[email protected]>

Author: Elfo404

README.md

@@ -208,14 +208,16 @@ Here are some provisioning examples for this data source.
 apiVersion: 1
 
 datasources:
-  - name: Elastic
-    type: grafana-es-open-distro-datasource
+  - name: OpenSearch
+    type: grafana-opensearch-datasource
     access: proxy
     database: '[metrics-]YYYY.MM.DD'
     url: http://localhost:9200
     jsonData:
       interval: Daily
       timeField: '@timestamp'
+      version: '1.0.0'
+      flavor: 'opensearch'
 ```
 
 or, for logs:
@@ -225,14 +227,15 @@ apiVersion: 1
 
 datasources:
   - name: elasticsearch-v7-filebeat
-    type: grafana-es-open-distro-datasource
+    type: grafana-opensearch-datasource
     access: proxy
     database: '[filebeat-]YYYY.MM.DD'
     url: http://localhost:9200
     jsonData:
       interval: Daily
       timeField: '@timestamp'
-      version: '1.0.0'
+      version: '7.0.0'
+      flavor: 'elasticsearch'
       logMessageField: message
       logLevelField: fields.level
 ```

package.json

@@ -14,9 +14,6 @@
   "author": "Grafana Labs",
   "license": "Apache-2.0",
   "dependencies": {
-    "@emotion/core": "10.0.27",
-    "@emotion/css": "11.1.3",
-    "@emotion/react": "11.1.5",
     "@grafana/aws-sdk": "0.0.3",
     "@grafana/data": "7.5.7",
     "@grafana/runtime": "7.5.7",

pkg/opensearch/client/client.go

@@ -96,6 +96,7 @@ func GetSigV4Config(ds *backend.DataSourceInstanceSettings) (*sigv4.Config, erro
 // Client represents a client which can interact with OpenSearch api
 type Client interface {
 	GetVersion() *semver.Version
+	GetFlavor() Flavor
 	GetTimeField() string
 	GetMinInterval(queryInterval string) (time.Duration, error)
 	GetIndex() string
@@ -126,12 +127,14 @@ var NewClient = func(ctx context.Context, ds *backend.DataSourceInstanceSettings
 
 	version, err := extractVersion(jsonData.Get("version"))
 	if err != nil {
-		return nil, fmt.Errorf("opensearch version is required, err=%v", err)
+		return nil, fmt.Errorf("version is required, err=%v", err)
 	}
 
+	flavor := jsonData.Get("flavor").MustString(string(OpenSearch))
+
 	timeField, err := jsonData.Get("timeField").String()
 	if err != nil {
-		return nil, fmt.Errorf("opensearch time field name is required, err=%v", err)
+		return nil, fmt.Errorf("time field name is required, err=%v", err)
 	}
 
 	indexInterval := jsonData.Get("interval").MustString()
@@ -156,6 +159,7 @@ var NewClient = func(ctx context.Context, ds *backend.DataSourceInstanceSettings
 		ctx:       ctx,
 		ds:        ds,
 		version:   version,
+		flavor:    Flavor(flavor),
 		timeField: timeField,
 		indices:   indices,
 		index:     index,
@@ -166,6 +170,7 @@ var NewClient = func(ctx context.Context, ds *backend.DataSourceInstanceSettings
 type baseClientImpl struct {
 	ctx          context.Context
 	ds           *backend.DataSourceInstanceSettings
+	flavor       Flavor
 	version      *semver.Version
 	timeField    string
 	indices      []string
@@ -174,6 +179,10 @@ type baseClientImpl struct {
 	debugEnabled bool
 }
 
+func (c *baseClientImpl) GetFlavor() Flavor {
+	return c.flavor
+}
+
 func (c *baseClientImpl) GetVersion() *semver.Version {
 	return c.version
 }
@@ -384,19 +393,42 @@ func (c *baseClientImpl) createMultiSearchRequests(searchRequests []*SearchReque
 			interval: searchReq.Interval,
 		}
 
+		if c.flavor == Elasticsearch {
+			if c.version.Major() < 5 {
+				mr.header["search_type"] = "count"
+			} else {
+				allowedVersionRange, _ := semver.NewConstraint(">=5.6.0, <7.0.0")
+
+				if allowedVersionRange.Check(c.version) {
+					maxConcurrentShardRequests := c.getSettings().Get("maxConcurrentShardRequests").MustInt(256)
+					if maxConcurrentShardRequests == 0 {
+						maxConcurrentShardRequests = 256
+					}
+					mr.header["max_concurrent_shard_requests"] = maxConcurrentShardRequests
+				}
+			}
+		}
+
 		multiRequests = append(multiRequests, &mr)
 	}
 
 	return multiRequests
 }
 
 func (c *baseClientImpl) getMultiSearchQueryParameters() string {
-	maxConcurrentShardRequests := c.getSettings().Get("maxConcurrentShardRequests").MustInt(5)
-	return fmt.Sprintf("max_concurrent_shard_requests=%d", maxConcurrentShardRequests)
+	if c.version.Major() >= 7 || c.flavor == OpenSearch {
+		maxConcurrentShardRequests := c.getSettings().Get("maxConcurrentShardRequests").MustInt(5)
+		if maxConcurrentShardRequests == 0 {
+			maxConcurrentShardRequests = 5
+		}
+		return fmt.Sprintf("max_concurrent_shard_requests=%d", maxConcurrentShardRequests)
+	}
+
+	return ""
 }
 
 func (c *baseClientImpl) MultiSearch() *MultiSearchRequestBuilder {
-	return NewMultiSearchRequestBuilder(c.GetVersion())
+	return NewMultiSearchRequestBuilder(c.GetFlavor(), c.GetVersion())
 }
 
 func (c *baseClientImpl) EnableDebug() {

pkg/opensearch/client/models.go

@@ -9,6 +9,13 @@ import (
 	"github.com/grafana/opensearch-datasource/pkg/tsdb"
 )
 
+type Flavor string
+
+const (
+	Elasticsearch Flavor = "elasticsearch"
+	OpenSearch    Flavor = "opensearch"
+)
+
 type response struct {
 	httpResponse *http.Response
 	reqInfo      *SearchRequestInfo

pkg/opensearch/client/search_request.go

@@ -9,6 +9,7 @@ import (
 
 // SearchRequestBuilder represents a builder which can build a search request
 type SearchRequestBuilder struct {
+	flavor       Flavor
 	version      *semver.Version
 	interval     tsdb.Interval
 	index        string
@@ -20,8 +21,9 @@ type SearchRequestBuilder struct {
 }
 
 // NewSearchRequestBuilder create a new search request builder
-func NewSearchRequestBuilder(version *semver.Version, interval tsdb.Interval) *SearchRequestBuilder {
+func NewSearchRequestBuilder(flavor Flavor, version *semver.Version, interval tsdb.Interval) *SearchRequestBuilder {
 	builder := &SearchRequestBuilder{
+		flavor:      flavor,
 		version:     version,
 		interval:    interval,
 		sort:        make(map[string]interface{}),
@@ -88,7 +90,13 @@ func (b *SearchRequestBuilder) SortDesc(field, unmappedType string) *SearchReque
 // AddDocValueField adds a doc value field to the search request
 func (b *SearchRequestBuilder) AddDocValueField(field string) *SearchRequestBuilder {
 	b.customProps["script_fields"] = make(map[string]interface{})
-	b.customProps["docvalue_fields"] = []string{field}
+
+	if b.version.Major() < 5 && b.flavor == Elasticsearch {
+		b.customProps["fielddata_fields"] = []string{field}
+		b.customProps["fields"] = []string{"*", "_source"}
+	} else {
+		b.customProps["docvalue_fields"] = []string{field}
+	}
 
 	return b
 }
@@ -110,20 +118,22 @@ func (b *SearchRequestBuilder) Agg() AggBuilder {
 
 // MultiSearchRequestBuilder represents a builder which can build a multi search request
 type MultiSearchRequestBuilder struct {
+	flavor          Flavor
 	version         *semver.Version
 	requestBuilders []*SearchRequestBuilder
 }
 
 // NewMultiSearchRequestBuilder creates a new multi search request builder
-func NewMultiSearchRequestBuilder(version *semver.Version) *MultiSearchRequestBuilder {
+func NewMultiSearchRequestBuilder(flavor Flavor, version *semver.Version) *MultiSearchRequestBuilder {
 	return &MultiSearchRequestBuilder{
+		flavor:  flavor,
 		version: version,
 	}
 }
 
 // Search initiates and returns a new search request builder
 func (m *MultiSearchRequestBuilder) Search(interval tsdb.Interval) *SearchRequestBuilder {
-	b := NewSearchRequestBuilder(m.version, interval)
+	b := NewSearchRequestBuilder(m.flavor, m.version, interval)
 	m.requestBuilders = append(m.requestBuilders, b)
 	return b
 }
@@ -266,6 +276,7 @@ type AggBuilder interface {
 type aggBuilderImpl struct {
 	AggBuilder
 	aggDefs []*aggDef
+	flavor  Flavor
 	version *semver.Version
 }
 
@@ -358,7 +369,7 @@ func (b *aggBuilderImpl) Terms(key, field string, fn func(a *TermsAggregation, b
 		fn(innerAgg, builder)
 	}
 
-	if len(innerAgg.Order) > 0 {
+	if (b.version.Major() >= 6 || b.flavor == OpenSearch) && len(innerAgg.Order) > 0 {
 		if orderBy, exists := innerAgg.Order[termsOrderTerm]; exists {
 			innerAgg.Order["_key"] = orderBy
 			delete(innerAgg.Order, termsOrderTerm)

pkg/opensearch/client/search_request_test.go

@@ -16,9 +16,9 @@ func TestSearchRequest(t *testing.T) {
 	Convey("Test OpenSearch search request", t, func() {
 		timeField := "@timestamp"
 		// TODO: Check this
-		Convey("Given new search request builder for es version 5", func() {
+		Convey("Given new search request builder for es OpenSearch 1.0.0", func() {
 			version, _ := semver.NewVersion("1.0.0")
-			b := NewSearchRequestBuilder(version, tsdb.Interval{Value: 15 * time.Second, Text: "15s"})
+			b := NewSearchRequestBuilder(OpenSearch, version, tsdb.Interval{Value: 15 * time.Second, Text: "15s"})
 
 			Convey("When building search request", func() {
 				sr, err := b.Build()
@@ -391,14 +391,69 @@ func TestSearchRequest(t *testing.T) {
 				})
 			})
 		})
+
+		Convey("Given new search request builder for Elasticsearch 2.0.0", func() {
+			version, _ := semver.NewVersion("2.0.0")
+
+			b := NewSearchRequestBuilder(Elasticsearch, version, tsdb.Interval{Value: 15 * time.Second, Text: "15s"})
+
+			Convey("When adding doc value field", func() {
+				b.AddDocValueField(timeField)
+
+				Convey("should set correct props", func() {
+					fields, ok := b.customProps["fields"].([]string)
+					So(ok, ShouldBeTrue)
+					So(fields, ShouldHaveLength, 2)
+					So(fields[0], ShouldEqual, "*")
+					So(fields[1], ShouldEqual, "_source")
+
+					scriptFields, ok := b.customProps["script_fields"].(map[string]interface{})
+					So(ok, ShouldBeTrue)
+					So(scriptFields, ShouldHaveLength, 0)
+
+					fieldDataFields, ok := b.customProps["fielddata_fields"].([]string)
+					So(ok, ShouldBeTrue)
+					So(fieldDataFields, ShouldHaveLength, 1)
+					So(fieldDataFields[0], ShouldEqual, timeField)
+				})
+
+				Convey("When building search request", func() {
+					sr, err := b.Build()
+					So(err, ShouldBeNil)
+
+					Convey("When marshal to JSON should generate correct json", func() {
+						body, err := json.Marshal(sr)
+						So(err, ShouldBeNil)
+						json, err := simplejson.NewJson(body)
+						So(err, ShouldBeNil)
+
+						scriptFields, err := json.Get("script_fields").Map()
+						So(err, ShouldBeNil)
+						So(scriptFields, ShouldHaveLength, 0)
+
+						fields, err := json.Get("fields").StringArray()
+						So(err, ShouldBeNil)
+						So(fields, ShouldHaveLength, 2)
+						So(fields[0], ShouldEqual, "*")
+						So(fields[1], ShouldEqual, "_source")
+
+						fieldDataFields, err := json.Get("fielddata_fields").StringArray()
+						So(err, ShouldBeNil)
+						So(fieldDataFields, ShouldHaveLength, 1)
+						So(fieldDataFields[0], ShouldEqual, timeField)
+					})
+				})
+			})
+		})
+
 	})
 }
 
 func TestMultiSearchRequest(t *testing.T) {
 	Convey("Test OpenSearch multi search request", t, func() {
 		Convey("Given new multi search request builder", func() {
 			version, _ := semver.NewVersion("1.0.0")
-			b := NewMultiSearchRequestBuilder(version)
+			b := NewMultiSearchRequestBuilder(OpenSearch, version)
 
 			Convey("When adding one search request", func() {
 				b.Search(tsdb.Interval{Value: 15 * time.Second, Text: "15s"})