Client: Add compatibility mode, tests, config option and documentation

Author: Anaethelion

.doc/connecting.asciidoc

@@ -60,6 +60,19 @@ cfg := elasticsearch.Config{
 es, err := elasticsearch.NewClient(cfg)
 ------------------------------------
 
+[discrete]
+[[compatibility-mode]]
+==== Compatibility Mode
+
+Elasticsearch server 8.0 introduces a new compatibility mode allowing for a smoother transition from 7 to 8.
+Using the compatibility mode, the client sends versioned headers with each request instructing the server on the expected behaviour.
+------------------------------------
+Content-Type: application/vnd.elasticsearch+json; compatible-with=7
+Accept: application/vnd.elasticsearch+json; compatible-with=7
+------------------------------------
+This allows you to update the server first, then the client.
+
+To enable this, the latest 7.x client can use the environment variable `ELASTIC_CLIENT_APIVERSIONING` or the configuration option `config.EnableCompatibilityMode` in the client Config.
 
 [discrete]
 [[client-usage]]

elasticsearch.go

@@ -82,13 +82,14 @@ type Config struct {
 	EnableRetryOnTimeout bool  // Default: false.
 	MaxRetries           int   // Default: 3.
 
-	CompressRequestBody bool // Default: false.
+	CompressRequestBody  bool // Default: false.
+	DiscoverNodesOnStart bool // Discover nodes when initializing the client. Default: false.
 
-	DiscoverNodesOnStart  bool          // Discover nodes when initializing the client. Default: false.
 	DiscoverNodesInterval time.Duration // Discover nodes periodically. Default: disabled.
 
-	EnableMetrics     bool // Enable the metrics collection.
-	EnableDebugLogger bool // Enable the debug logging.
+	EnableMetrics           bool // Enable the metrics collection.
+	EnableDebugLogger       bool // Enable the debug logging.
+	EnableCompatibilityMode bool // Enable sends compatibility header
 
 	DisableMetaHeader    bool // Disable the additional "X-Elastic-Client-Meta" HTTP header.
 	UseResponseCheckOnly bool
@@ -204,6 +205,7 @@ func NewClient(cfg Config) (*Client, error) {
 		RetryBackoff:         cfg.RetryBackoff,
 
 		CompressRequestBody: cfg.CompressRequestBody,
+		CompatibilityHeader: cfg.EnableCompatibilityMode,
 
 		EnableMetrics:     cfg.EnableMetrics,
 		EnableDebugLogger: cfg.EnableDebugLogger,

elasticsearch_integration_test.go

@@ -158,6 +158,23 @@ func TestClientTransport(t *testing.T) {
 			t.Fatalf("Expected net.OpError, but got: %T", err)
 		}
 	})
+
+	t.Run("Compatibility Header", func(t *testing.T) {
+		client, err := elasticsearch.NewClient(elasticsearch.Config{EnableCompatibilityMode: true})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		res, err := client.Info()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		contentType := res.Header.Get("content-type")
+		if contentType != "application/vnd.elasticsearch+json;compatible-with=7" {
+			t.Fatalf("Unexpected content-type header, want \"application/vnd.elasticsearch+json;compatible-with=7\", got: %s", contentType)
+		}
+	})
 }
 
 type CustomTransport struct {

elasticsearch_internal_test.go

@@ -696,6 +696,7 @@ func TestResponseCheckOnly(t *testing.T) {
 	}
 }
 
+
 func TestProductCheckError(t *testing.T) {
 	var requestPaths []string
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -731,3 +732,45 @@ func TestProductCheckError(t *testing.T) {
 		t.Fatalf("product check should be valid, got : %v", c.productCheckSuccess)
 	}
 }
+
+
+func TestCompatibilityHeader(t *testing.T) {
+	client, err := NewClient(Config{
+		Transport: &mockTransp{RoundTripFunc: func(request *http.Request) (*http.Response, error) {
+			if request.URL.Path == "/" || request.URL.Path == "/test-index/_search" {
+				accept := request.Header.Get("Accept")
+				if accept != "application/vnd.elasticsearch+json;compatible-with=7" {
+					t.Fatalf("Invalid header, expected: %s, got: %s",
+						"application/vnd.elasticsearch+json;compatible-with=7",
+						accept,
+					)
+				}
+
+				if request.URL.Path == "/test-index/_search" {
+					contentType := request.Header.Get("Content-Type")
+					if contentType != "application/vnd.elasticsearch+json;compatible-with=7" {
+						t.Fatalf("Invalid header, expected: %s, got: %s",
+							"application/vnd.elasticsearch+json;compatible-with=7",
+							contentType,
+						)
+					}
+				}
+			} else {
+				t.Fatal("Unexpected route called")
+			}
+
+			return defaultRoundTripFunc(request)
+		}},
+		Addresses:               []string{"http://127.0.0.1:9200"},
+		EnableCompatibilityMode: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	client.Info()
+	client.Search(
+		client.Search.WithIndex("test-index"),
+		client.Search.WithBody(strings.NewReader("{}")),
+	)
+}

estransport/estransport.go

@@ -92,6 +92,7 @@ type Config struct {
 	RetryBackoff         func(attempt int) time.Duration
 
 	CompressRequestBody bool
+	CompatibilityHeader bool
 
 	EnableMetrics     bool
 	EnableDebugLogger bool
@@ -129,6 +130,7 @@ type Client struct {
 	discoverNodesTimer    *time.Timer
 
 	compressRequestBody bool
+	compatibilityHeader bool
 
 	metrics *metrics
 
@@ -194,6 +196,7 @@ func New(cfg Config) (*Client, error) {
 		discoverNodesInterval: cfg.DiscoverNodesInterval,
 
 		compressRequestBody: cfg.CompressRequestBody,
+		compatibilityHeader: cfg.CompatibilityHeader,
 
 		transport: cfg.Transport,
 		logger:    cfg.Logger,
@@ -240,7 +243,7 @@ func (c *Client) Perform(req *http.Request) (*http.Response, error) {
 	)
 
 	// Compatibility Header
-	if compatibilityHeader {
+	if compatibilityHeader || c.compatibilityHeader {
 		if req.Body != nil {
 			req.Header.Set("Content-Type", "application/vnd.elasticsearch+json;compatible-with=7")
 		}