Add product check for 8.x (#324)

* Client: Implement product check

* Update compatibility matrix to reflect supported versions

Author: Anaethelion

.doc/installation.asciidoc

@@ -53,34 +53,8 @@ go run main.go
 [discrete]
 === {es} Version Compatibility
 
-The client major versions correspond to the {es} major versions: to connect to 
-{es} `7.x`, use a `7.x` version of the client, to connect to {es} `6.x`, use a 
-`6.x` version of the client, and so on.
+Language clients are forward compatible; meaning that clients support communicating
+with greater minor versions of {es}.
 
-[NOTE]
---
-While the 5.x version of the client is available, it is no longer actively 
-maintained, neither the corresponding {es} version. For more information, refer 
-to https://www.elastic.co/support/eol[Elastic product end of life dates].
---
-
-[%header,cols=2*]
-|===
-|{es} Version
-|Client Version
-
-|`master`
-|`master`
-
-|`7.x`
-|`7.x`
-
-|`6.x`
-|`6.x`
-
-|`5.x`
-|`5.x`
-|===
-
-The `master` branch of the client is compatible with the `master` branch of 
-{es}.
+Elastic language clients are also backwards compatible with lesser supported
+minor {es} versions.

README.md

@@ -12,7 +12,11 @@ The official Go client for [Elasticsearch](https://www.elastic.co/products/elast
 
 ## Compatibility
 
-The client major versions correspond to the compatible Elasticsearch major versions: to connect to Elasticsearch `7.x`, use a [`7.x`](https://github.com/elastic/go-elasticsearch/tree/7.x) version of the client, to connect to Elasticsearch `6.x`, use a [`6.x`](https://github.com/elastic/go-elasticsearch/tree/6.x) version of the client.
+Language clients are forward compatible; meaning that clients support communicating
+with greater minor versions of Elasticsearch.
+
+Elastic language clients are also backwards compatible with lesser supported
+minor Elasticsearch versions.
 
 When using Go modules, include the version in the import path, and specify either an explicit version or a branch:
 

_examples/cloudfunction/function_test.go

@@ -35,7 +35,10 @@ type MockTransport struct{}
 
 // RoundTrip returns a mock response.
 func (t *MockTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-	return &http.Response{Body: ioutil.NopCloser(strings.NewReader(`{"status":"mocked"}`))}, nil
+	return &http.Response{
+		Body: ioutil.NopCloser(strings.NewReader(`{"status":"mocked"}`)),
+		Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
+	}, nil
 }
 
 func TestHealth(t *testing.T) {

_examples/xkcdsearch/store_test.go

@@ -81,6 +81,7 @@ func TestStore(t *testing.T) {
 		Response: &http.Response{
 			StatusCode: http.StatusOK,
 			Body:       ioutil.NopCloser(strings.NewReader(`{}`)),
+			Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
 		},
 	}
 	mocktrans.RoundTripFn = func(req *http.Request) (*http.Response, error) { return mocktrans.Response, nil }

elasticsearch.go

@@ -25,6 +25,7 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/elastic/go-elasticsearch/v8/esapi"
@@ -38,7 +39,10 @@ const (
 
 // Version returns the package version as a string.
 //
-const Version = version.Client
+const (
+	Version        = version.Client
+	unknownProduct = "the client noticed that the server is not Elasticsearch and we do not support this unknown product"
+)
 
 // Config represents the client configuration.
 //
@@ -88,6 +92,9 @@ type Config struct {
 type Client struct {
 	*esapi.API // Embeds the API methods
 	Transport  estransport.Interface
+
+	productCheckMu      sync.RWMutex
+	productCheckSuccess bool
 }
 
 // NewDefaultClient creates a new client with default options.
@@ -187,7 +194,8 @@ func NewClient(cfg Config) (*Client, error) {
 		return nil, fmt.Errorf("error creating transport: %s", err)
 	}
 
-	client := &Client{Transport: tp, API: esapi.New(tp)}
+	client := &Client{Transport: tp}
+	client.API = esapi.New(client)
 
 	if cfg.DiscoverNodesOnStart {
 		go client.DiscoverNodes()
@@ -199,7 +207,54 @@ func NewClient(cfg Config) (*Client, error) {
 // Perform delegates to Transport to execute a request and return a response.
 //
 func (c *Client) Perform(req *http.Request) (*http.Response, error) {
-	return c.Transport.Perform(req)
+	// Retrieve the original request.
+	res, err := c.Transport.Perform(req)
+
+	// ResponseCheck path continues, we run the header check on the first answer from ES.
+	if err == nil {
+		checkHeader := func() error { return genuineCheckHeader(res.Header) }
+		if err := c.doProductCheck(checkHeader); err != nil {
+			res.Body.Close()
+			return nil, err
+		}
+	}
+	return res, err
+}
+
+// doProductCheck calls f if there as not been a prior successful call to doProductCheck,
+// returning nil otherwise.
+func (c *Client) doProductCheck(f func() error) error {
+	c.productCheckMu.RLock()
+	productCheckSuccess := c.productCheckSuccess
+	c.productCheckMu.RUnlock()
+
+	if productCheckSuccess {
+		return nil
+	}
+
+	c.productCheckMu.Lock()
+	defer c.productCheckMu.Unlock()
+
+	if c.productCheckSuccess {
+		return nil
+	}
+
+	if err := f(); err != nil {
+		return err
+	}
+
+	c.productCheckSuccess = true
+
+	return nil
+}
+
+// genuineCheckHeader validates the presence of the X-Elastic-Product header
+//
+func genuineCheckHeader(header http.Header) error {
+	if header.Get("X-Elastic-Product") != "Elasticsearch" {
+		return errors.New(unknownProduct)
+	}
+	return nil
 }
 
 // Metrics returns the client metrics.

elasticsearch_benchmark_test.go

@@ -35,8 +35,11 @@ var defaultResponse = http.Response{
 	Status:        "200 OK",
 	StatusCode:    200,
 	ContentLength: 2,
-	Header:        http.Header(map[string][]string{"Content-Type": {"application/json"}}),
-	Body:          ioutil.NopCloser(strings.NewReader(`{}`)),
+	Header: http.Header(map[string][]string{
+		"Content-Type":      {"application/json"},
+		"X-Elastic-Product": {"Elasticsearch"},
+	}),
+	Body: ioutil.NopCloser(strings.NewReader(`{}`)),
 }
 
 type FakeTransport struct {

elasticsearch_internal_test.go

@@ -22,15 +22,51 @@ package elasticsearch
 import (
 	"encoding/base64"
 	"errors"
+	"io/ioutil"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"os"
+	"reflect"
 	"regexp"
+	"strings"
 	"testing"
 
 	"github.com/elastic/go-elasticsearch/v8/estransport"
 )
 
+var called bool
+
+type mockTransp struct {
+	RoundTripFunc func(*http.Request) (*http.Response, error)
+}
+
+var defaultRoundTripFunc = func(req *http.Request) (*http.Response, error) {
+	response := &http.Response{Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}}}
+
+	if req.URL.Path == "/" {
+		response.Body = ioutil.NopCloser(strings.NewReader(`{
+		  "version" : {
+			"number" : "8.0.0-SNAPSHOT",
+			"build_flavor" : "default"
+		  },
+		  "tagline" : "You Know, for Search"
+		}`))
+		response.Header.Add("Content-Type", "application/json")
+	} else {
+		called = true
+	}
+
+	return response, nil
+}
+
+func (t *mockTransp) RoundTrip(req *http.Request) (*http.Response, error) {
+	if t.RoundTripFunc == nil {
+		return defaultRoundTripFunc(req)
+	}
+	return t.RoundTripFunc(req)
+}
+
 func TestClientConfiguration(t *testing.T) {
 	t.Parallel()
 
@@ -173,15 +209,6 @@ func TestClientConfiguration(t *testing.T) {
 	})
 }
 
-var called bool
-
-type mockTransp struct{}
-
-func (t *mockTransp) RoundTrip(req *http.Request) (*http.Response, error) {
-	called = true
-	return &http.Response{}, nil
-}
-
 func TestClientInterface(t *testing.T) {
 	t.Run("Transport", func(t *testing.T) {
 		c, err := NewClient(Config{Transport: &mockTransp{}})
@@ -357,3 +384,88 @@ func TestClientMetrics(t *testing.T) {
 		t.Errorf("Unexpected output: %s", m)
 	}
 }
+
+func TestResponseCheckOnly(t *testing.T) {
+	tests := []struct {
+		name                 string
+		response             *http.Response
+		requestErr           error
+		wantErr              bool
+	}{
+		{
+			name: "Valid answer with header",
+			response: &http.Response{
+				Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
+				Body:   ioutil.NopCloser(strings.NewReader("{}")),
+			},
+			wantErr: false,
+		},
+		{
+			name: "Valid answer without header",
+			response: &http.Response{
+				Body: ioutil.NopCloser(strings.NewReader("{}")),
+			},
+			wantErr: true,
+		},
+		{
+			name: "Valid answer with http error code",
+			response: &http.Response{
+				StatusCode: http.StatusUnauthorized,
+				Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
+				Body:       ioutil.NopCloser(strings.NewReader("{}")),
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c, _ := NewClient(Config{
+				Transport: &mockTransp{RoundTripFunc: func(request *http.Request) (*http.Response, error) {
+					return tt.response, tt.requestErr
+				}},
+			})
+			_, err := c.Cat.Indices()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Unexpected error, got %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+
+func TestProductCheckError(t *testing.T) {
+	var requestPaths []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestPaths = append(requestPaths, r.URL.Path)
+		if len(requestPaths) == 1 {
+			// Simulate transient error from a proxy on the first request.
+			// This must not be cached by the client.
+			w.WriteHeader(http.StatusBadGateway)
+			return
+		}
+		w.Header().Set("X-Elastic-Product", "Elasticsearch")
+		w.Write([]byte("{}"))
+	}))
+	defer server.Close()
+
+	c, _ := NewClient(Config{Addresses: []string{server.URL}, DisableRetry: true})
+	if _, err := c.Cat.Indices(); err == nil {
+		t.Fatal("expected error")
+	}
+	if c.productCheckSuccess {
+		t.Fatalf("product check should be invalid, got %v", c.productCheckSuccess)
+	}
+	if _, err := c.Cat.Indices(); err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+	if n := len(requestPaths); n != 2 {
+		t.Fatalf("expected 2 requests, got %d", n)
+	}
+	if !reflect.DeepEqual(requestPaths, []string{"/_cat/indices", "/_cat/indices"}) {
+		t.Fatalf("unexpected request paths: %s", requestPaths)
+	}
+	if !c.productCheckSuccess {
+		t.Fatalf("product check should be valid, got : %v", c.productCheckSuccess)
+	}
+}

esapi/esapi_benchmark_test.go

@@ -32,7 +32,11 @@ import (
 // TODO(karmi): Refactor into a shared mock/testing package
 
 var (
-	defaultResponse    = &http.Response{StatusCode: 200, Body: ioutil.NopCloser(strings.NewReader("MOCK"))}
+	defaultResponse    = &http.Response{
+		StatusCode: 200,
+		Body: ioutil.NopCloser(strings.NewReader("MOCK")),
+		Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
+	}
 	defaultRoundTripFn = func(*http.Request) (*http.Response, error) { return defaultResponse, nil }
 	errorRoundTripFn   = func(*http.Request) (*http.Response, error) {
 		return &http.Response{
@@ -54,6 +58,7 @@ var (
 					  },
 					  "status" : 400
 					}`)),
+			Header: http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
 		}, nil
 	}
 )