Client: Reeintegrate compatibility header with external transport (#379)

Author: Anaethelion

elasticsearch.go

@@ -26,6 +26,7 @@ import (
 	"os"
 	"regexp"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -45,8 +46,9 @@ const (
 )
 
 var (
-	userAgent           string
-	reGoVersion         = regexp.MustCompile(`go(\d+\.\d+\..+)`)
+	esCompatHeader = "ELASTIC_CLIENT_APIVERSIONING"
+	userAgent      string
+	reGoVersion    = regexp.MustCompile(`go(\d+\.\d+\..+)`)
 )
 
 func init() {
@@ -82,14 +84,15 @@ type Config struct {
 	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.
 
 	RetryBackoff func(attempt int) time.Duration // Optional backoff duration. Default: nil.
 
-	Transport http.RoundTripper    // The HTTP transport object.
+	Transport http.RoundTripper         // The HTTP transport object.
 	Logger    elastictransport.Logger   // The logger object.
 	Selector  elastictransport.Selector // The selector object.
 
@@ -100,8 +103,9 @@ type Config struct {
 // Client represents the Elasticsearch client.
 //
 type Client struct {
-	*esapi.API // Embeds the API methods
-	Transport  elastictransport.Interface
+	*esapi.API          // Embeds the API methods
+	Transport           elastictransport.Interface
+	compatibilityHeader bool
 
 	disableMetaHeader   bool
 	productCheckMu      sync.RWMutex
@@ -206,9 +210,13 @@ func NewClient(cfg Config) (*Client, error) {
 		return nil, fmt.Errorf("error creating transport: %s", err)
 	}
 
+	compatHeaderEnv := os.Getenv(esCompatHeader)
+	compatibilityHeader, _ := strconv.ParseBool(compatHeaderEnv)
+
 	client := &Client{
-		Transport: tp,
-		disableMetaHeader: cfg.DisableMetaHeader,
+		Transport:           tp,
+		disableMetaHeader:   cfg.DisableMetaHeader,
+		compatibilityHeader: cfg.EnableCompatibilityMode || compatibilityHeader,
 	}
 	client.API = esapi.New(client)
 
@@ -222,6 +230,12 @@ 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) {
+	// Compatibility Header
+	if c.compatibilityHeader {
+		req.Header.Set("Content-Type", "application/vnd.elasticsearch+json;compatible-with=8")
+		req.Header.Set("Accept", "application/vnd.elasticsearch+json;compatible-with=8")
+	}
+
 	if !c.disableMetaHeader {
 		existingMetaHeader := req.Header.Get(HeaderClientMeta)
 		if existingMetaHeader != "" {

elasticsearch_internal_test.go

@@ -32,6 +32,7 @@ import (
 	"os"
 	"reflect"
 	"regexp"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -407,7 +408,7 @@ func TestResponseCheckOnly(t *testing.T) {
 			name: "Valid answer without header",
 			response: &http.Response{
 				StatusCode: http.StatusOK,
-				Body: ioutil.NopCloser(strings.NewReader("{}")),
+				Body:       ioutil.NopCloser(strings.NewReader("{}")),
 			},
 			wantErr: true,
 		},
@@ -423,7 +424,7 @@ func TestResponseCheckOnly(t *testing.T) {
 			name: "Valid answer without header and response check",
 			response: &http.Response{
 				StatusCode: http.StatusOK,
-				Body: ioutil.NopCloser(strings.NewReader("{}")),
+				Body:       ioutil.NopCloser(strings.NewReader("{}")),
 			},
 			wantErr: true,
 		},
@@ -558,3 +559,94 @@ func TestFingerprint(t *testing.T) {
 		t.Fatalf("unexpected payload returned: expected: %s, got: %s", body, data)
 	}
 }
+
+func TestCompatibilityHeader(t *testing.T) {
+	tests := []struct {
+		name                string
+		envVar              bool
+		configVar           bool
+		compatibilityHeader bool
+		bodyPresent         bool
+		expectsHeader       []string
+	}{
+		{
+			name:                "Compatibility header disabled",
+			compatibilityHeader: false,
+			bodyPresent:         false,
+			envVar:              false,
+			configVar:           false,
+			expectsHeader:       []string{"application/json"},
+		},
+		{
+			name:                "Compatibility header enabled with env var",
+			compatibilityHeader: true,
+			bodyPresent:         false,
+			envVar:              true,
+			configVar:           false,
+			expectsHeader:       []string{"application/vnd.elasticsearch+json;compatible-with=8"},
+		},
+		{
+			name:                "Compatibility header enabled with body with env var",
+			compatibilityHeader: true,
+			bodyPresent:         true,
+			envVar:              true,
+			configVar:           false,
+			expectsHeader:       []string{"application/vnd.elasticsearch+json;compatible-with=8"},
+		},
+		{
+			name:                "Compatibility header enabled in conf",
+			compatibilityHeader: true,
+			bodyPresent:         false,
+			envVar:              false,
+			configVar:           true,
+			expectsHeader:       []string{"application/vnd.elasticsearch+json;compatible-with=8"},
+		},
+		{
+			name:                "Compatibility header enabled with body in conf",
+			compatibilityHeader: true,
+			bodyPresent:         true,
+			envVar:              false,
+			configVar:           true,
+			expectsHeader:       []string{"application/vnd.elasticsearch+json;compatible-with=8"},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Setenv(esCompatHeader, strconv.FormatBool(test.compatibilityHeader))
+
+			c, _ := NewClient(Config{
+				EnableCompatibilityMode: test.configVar,
+				Addresses:               []string{},
+				Transport: &mockTransp{
+					RoundTripFunc: func(req *http.Request) (*http.Response, error) {
+						if test.compatibilityHeader {
+							if !reflect.DeepEqual(req.Header["Accept"], test.expectsHeader) {
+								t.Errorf("Compatibility header enabled but header is, not in request headers, got: %s, want: %s", req.Header["Accept"], test.expectsHeader)
+							}
+
+							if test.bodyPresent {
+								if !reflect.DeepEqual(req.Header["Content-Type"], test.expectsHeader) {
+									t.Errorf("Compatibility header with Body enabled, not in request headers, got: %s, want: %s", req.Header["Content-Type"], test.expectsHeader)
+								}
+							}
+						}
+
+						return &http.Response{
+							StatusCode: http.StatusOK,
+							Status:     "MOCK",
+							Body:       ioutil.NopCloser(strings.NewReader("{}")),
+						}, nil
+					},
+				},
+			})
+
+			req := &http.Request{URL: &url.URL{}, Header: make(http.Header)}
+			if test.bodyPresent {
+				req.Body = ioutil.NopCloser(strings.NewReader("{}"))
+			}
+
+			_, _ = c.Perform(req)
+		})
+	}
+}