Introduce compatibility header (#5438)

* stage

* update to latest libs from elastic/elasticsearch-net-abstractions and remove nightlies from nuget.config

(cherry picked from commit 1e8b465)

* Adds support for vendored mimetype headers

Changes our default `Accept` and `Content-Type` request headers over to
the new vendor specific headers that includes compatible with
informations to start powering structurally versioned API's

```
Accept:       application/vnd.elasticsearch+json;compatible-with=7
Content-Type: application/vnd.elasticsearch+json; compatible-with=7
```

On 7.x Elasticsearch does not return the vendor type we send for
`Accept` but is still hardwired to return:

```
content-type: application/json; charset=UTF-8
```

We are therefor lenient in our Accept to Response ContentType
validations.

We do these assertions to prevent the client from deserializing HTML
from possible proxies in the middle of the client and server.

* Make sure compatible with is calculated from assembly version

* Make sure InMemoryConnection still defaults to the default mimetype

* update nuget.config

* Make sure RequestDataContent does not use constructor to set mimetype since its too strict

* try to isolate constant header usage and add tests

* Randomly enable ApiVersioning in tests if we are testing against 7.12.0 and higher

* make sure we can talk to 8.x cluster

* Fix isvalid json mimetype check

* Add space after ; because MediaType .ToString() does this too

* Fix FixedResponseClient default content type

* update StartsWith to ignore culture and case

* Add support for configuring via env var

Enabled when ELASTIC_CLIENT_APIVERSIONING is set to "true" or "1".

Co-authored-by: Steve Gordon <sgordon@hotmail.co.uk>

Author: Mpdreamz

src/Elasticsearch.Net.VirtualizedCluster/Rules/RuleBase.cs

@@ -69,14 +69,14 @@ public TRule ReturnResponse<T>(T response)
 				r = ms.ToArray();
 			}
 			Self.ReturnResponse = r;
-			Self.ReturnContentType = RequestData.MimeType;
+			Self.ReturnContentType = RequestData.DefaultJsonMimeType;
 			return (TRule)this;
 		}
 
-		public TRule ReturnByteResponse(byte[] response, string responseContentType = RequestData.MimeType)
+		public TRule ReturnByteResponse(byte[] response, string responseContentType = null)
 		{
 			Self.ReturnResponse = response;
-			Self.ReturnContentType = responseContentType;
+			Self.ReturnContentType = responseContentType ?? RequestData.DefaultJsonMimeType;
 			return (TRule)this;
 		}
 	}

src/Elasticsearch.Net/Configuration/ConnectionConfiguration.cs

@@ -131,6 +131,8 @@ public ConnectionConfiguration(IConnectionPool connectionPool, IConnection conne
 	public abstract class ConnectionConfiguration<T> : IConnectionConfigurationValues
 		where T : ConnectionConfiguration<T>
 	{
+		internal const string ApiVersioningEnvironmentVariableName = "ELASTIC_CLIENT_APIVERSIONING";
+
 		private readonly IConnection _connection;
 		private readonly IConnectionPool _connectionPool;
 		private readonly NameValueCollection _headers = new NameValueCollection();
@@ -176,6 +178,7 @@ public abstract class ConnectionConfiguration<T> : IConnectionConfigurationValue
 		//public static IMemoryStreamFactory Default { get; } = RecyclableMemoryStreamFactory.Default;
 		public static IMemoryStreamFactory DefaultMemoryStreamFactory { get; } = Elasticsearch.Net.MemoryStreamFactory.Default;
 		private bool _enableThreadPoolStats;
+		private bool _enableApiVersioningHeader;
 
 		private string _userAgent = ConnectionConfiguration.DefaultUserAgent;
 		private readonly Func<HttpMethod, int, bool> _statusCodeToResponseSuccess;
@@ -205,6 +208,14 @@ protected ConnectionConfiguration(IConnectionPool connectionPool, IConnection co
 				_apiKeyAuthCredentials = cloudPool.ApiKeyCredentials;
 				_enableHttpCompression = true;
 			}
+
+			var apiVersioningEnabled = Environment.GetEnvironmentVariable(ApiVersioningEnvironmentVariableName);
+			_enableApiVersioningHeader = string.IsNullOrEmpty(apiVersioningEnabled) switch
+			{
+				false when bool.TryParse(apiVersioningEnabled, out var isEnabled) => isEnabled,
+				false when int.TryParse(apiVersioningEnabled, out var isEnabledValue) => isEnabledValue == 1,
+				_ => _enableApiVersioningHeader
+			};
 		}
 
 		protected IElasticsearchSerializer UseThisRequestResponseSerializer { get; set; }
@@ -257,8 +268,9 @@ protected ConnectionConfiguration(IConnectionPool connectionPool, IConnection co
 		bool IConnectionConfigurationValues.TransferEncodingChunked => _transferEncodingChunked;
 		bool IConnectionConfigurationValues.EnableTcpStats => _enableTcpStats;
 		bool IConnectionConfigurationValues.EnableThreadPoolStats => _enableThreadPoolStats;
-		
+
 		MetaHeaderProvider IConnectionConfigurationValues.MetaHeaderProvider { get; } = new MetaHeaderProvider();
+		bool IConnectionConfigurationValues.EnableApiVersioningHeader => _enableApiVersioningHeader;
 
 		void IDisposable.Dispose() => DisposeManagedResources();
 
@@ -342,7 +354,7 @@ public T SniffOnConnectionFault(bool sniffsOnConnectionFault = true) =>
 		public T DisableAutomaticProxyDetection(bool disable = true) => Assign(disable, (a, v) => a._disableAutomaticProxyDetection = v);
 
 		/// <summary>
-		/// Disables the meta header which is included on all requests by default. This header contains lightweight information 
+		/// Disables the meta header which is included on all requests by default. This header contains lightweight information
 		/// about the client and runtime.
 		/// </summary>
 		public T DisableMetaHeader(bool disable = true) => Assign(disable, (a, v) => a._disableMetaHeader = v);
@@ -602,6 +614,9 @@ public T SkipDeserializationForStatusCodes(params int[] statusCodes) =>
 		/// </summary>
 		public T MemoryStreamFactory(IMemoryStreamFactory memoryStreamFactory) => Assign(memoryStreamFactory, (a, v) => a._memoryStreamFactory = v);
 
+		/// <inheritdoc cref="IConnectionConfigurationValues.EnableApiVersioningHeader"/>
+		public T EnableApiVersioningHeader(bool enable = true) => Assign(enable, (a, v) => a._enableApiVersioningHeader = v);
+
 		public T EnableTcpStats(bool enableTcpStats = true) => Assign(enableTcpStats, (a, v) => a._enableTcpStats = v);
 
 		public T EnableThreadPoolStats(bool enableThreadPoolStats = true) => Assign(enableThreadPoolStats, (a, v) => a._enableThreadPoolStats = v);

src/Elasticsearch.Net/Configuration/IConnectionConfigurationValues.cs

@@ -280,5 +280,11 @@ public interface IConnectionConfigurationValues : IDisposable
 		/// Produces the client meta header for a request.
 		/// </summary>
 		MetaHeaderProvider MetaHeaderProvider { get; }
+
+		/// <summary>
+		/// Enables the client to start sending the API versioning header. This allows a `7.x` client to talk to `8.x` Elasticsearch.
+		/// <para> NOTE: You need at least Elasticsearch 7.11 and higher before you can enable this setting on the client</para>
+		/// </summary>
+		bool EnableApiVersioningHeader { get; }
 	}
 }

src/Elasticsearch.Net/Connection/Content/RequestDataContent.cs

@@ -30,7 +30,7 @@ internal class RequestDataContent : HttpContent
 		public RequestDataContent(RequestData requestData)
 		{
 			_requestData = requestData;
-			Headers.ContentType = new MediaTypeHeaderValue(requestData.RequestMimeType);
+			Headers.ContentType = MediaTypeHeaderValue.Parse(requestData.RequestMimeType);
 			if (requestData.HttpCompression)
 				Headers.ContentEncoding.Add("gzip");
 
@@ -50,7 +50,7 @@ Task OnStreamAvailable(RequestData data, Stream stream, HttpContent content, Tra
 		public RequestDataContent(RequestData requestData, CancellationToken token)
 		{
 			_requestData = requestData;
-			Headers.ContentType = new MediaTypeHeaderValue(requestData.RequestMimeType);
+			Headers.ContentType = MediaTypeHeaderValue.Parse(requestData.RequestMimeType);
 			if (requestData.HttpCompression)
 				Headers.ContentEncoding.Add("gzip");
 

src/Elasticsearch.Net/Connection/HttpConnection.cs

@@ -95,7 +95,7 @@ public virtual TResponse Request<TResponse>(RequestData requestData)
 
 				requestData.MadeItToResponse = true;
 				responseMessage.Headers.TryGetValues("Warning", out warnings);
-				mimeType = responseMessage.Content.Headers.ContentType?.MediaType;
+				mimeType = responseMessage.Content.Headers.ContentType?.ToString();
 
 				if (responseMessage.Content != null)
 				{
@@ -162,7 +162,7 @@ public virtual async Task<TResponse> RequestAsync<TResponse>(RequestData request
 				}
 
 				requestData.MadeItToResponse = true;
-				mimeType = responseMessage.Content.Headers.ContentType?.MediaType;
+				mimeType = responseMessage.Content.Headers.ContentType?.ToString();
 				responseMessage.Headers.TryGetValues("Warning", out warnings);
 
 				if (responseMessage.Content != null)
@@ -322,7 +322,7 @@ protected virtual HttpRequestMessage CreateRequestMessage(RequestData requestDat
 
 			requestMessage.Headers.Connection.Clear();
 			requestMessage.Headers.ConnectionClose = false;
-			requestMessage.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(requestData.Accept));
+			requestMessage.Headers.TryAddWithoutValidation("Accept", requestData.Accept);
 
 			if (!string.IsNullOrWhiteSpace(requestData.UserAgent))
 			{
@@ -372,7 +372,7 @@ private static void SetContent(HttpRequestMessage message, RequestData requestDa
 				if (requestData.HttpCompression)
 					message.Content.Headers.ContentEncoding.Add("gzip");
 
-				message.Content.Headers.ContentType = new MediaTypeHeaderValue(requestData.RequestMimeType);
+				message.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(requestData.RequestMimeType);
 			}
 		}
 
@@ -409,7 +409,7 @@ private static async Task SetContentAsync(HttpRequestMessage message, RequestDat
 				if (requestData.HttpCompression)
 					message.Content.Headers.ContentEncoding.Add("gzip");
 
-				message.Content.Headers.ContentType = new MediaTypeHeaderValue(requestData.RequestMimeType);
+				message.Content.Headers.ContentType = MediaTypeHeaderValue.Parse(requestData.RequestMimeType);
 			}
 		}
 

src/Elasticsearch.Net/Connection/InMemoryConnection.cs

@@ -25,12 +25,13 @@ public class InMemoryConnection : IConnection
 		/// </summary>
 		public InMemoryConnection() => _statusCode = 200;
 
-		public InMemoryConnection(byte[] responseBody, int statusCode = 200, Exception exception = null, string contentType = RequestData.MimeType)
+		public InMemoryConnection(byte[] responseBody, int statusCode = 200, Exception exception = null, string contentType = null)
 		{
+
 			_responseBody = responseBody;
 			_statusCode = statusCode;
 			_exception = exception;
-			_contentType = contentType;
+			_contentType = contentType ?? RequestData.DefaultJsonMimeType;
 		}
 
 		public virtual TResponse Request<TResponse>(RequestData requestData)
@@ -65,7 +66,7 @@ protected TResponse ReturnConnectionStatus<TResponse>(RequestData requestData, b
 
 			var sc = statusCode ?? _statusCode;
 			Stream s = body != null ? requestData.MemoryStreamFactory.Create(body) : requestData.MemoryStreamFactory.Create(EmptyBody);
-			return ResponseBuilder.ToResponse<TResponse>(requestData, _exception, sc, null, s, contentType ?? _contentType ?? RequestData.MimeType);
+			return ResponseBuilder.ToResponse<TResponse>(requestData, _exception, sc, null, s, contentType ?? _contentType ?? RequestData.DefaultJsonMimeType);
 		}
 
 		protected async Task<TResponse> ReturnConnectionStatusAsync<TResponse>(RequestData requestData, CancellationToken cancellationToken,

src/Elasticsearch.Net/Connection/MetaData/ClientVersionInfo.cs

@@ -15,6 +15,8 @@ internal sealed class ClientVersionInfo : VersionInfo
 
 		public static readonly ClientVersionInfo Empty = new ClientVersionInfo { Version = new Version(0, 0, 0), IsPrerelease = false };
 
+		public static readonly ClientVersionInfo LowLevelClientVersionInfo = Create<IElasticLowLevelClient>();
+
 		private ClientVersionInfo() { }
 
 		public static ClientVersionInfo Create<T>()

src/Elasticsearch.Net/Responses/ElasticsearchResponse.cs

@@ -74,7 +74,7 @@ public virtual bool TryGetServerError(out ServerError serverError)
 		{
 			serverError = null;
 			var bytes = ApiCall.ResponseBodyInBytes;
-			if (bytes == null || ResponseMimeType != RequestData.MimeType)
+			if (bytes == null || RequestData.IsJsonMimeType(ResponseMimeType))
 				return false;
 
 			using(var stream = ConnectionConfiguration.MemoryStreamFactory.Create(bytes))

src/Elasticsearch.Net/Responses/Special/BytesResponse.cs

@@ -13,7 +13,7 @@ public BytesResponse() { }
 		public override bool TryGetServerError(out ServerError serverError)
 		{
 			serverError = null;
-			if (Body == null || Body.Length == 0 || ResponseMimeType != RequestData.MimeType)
+			if (Body == null || Body.Length == 0 || !RequestData.IsJsonMimeType(ResponseMimeType))
 				return false;
 
 			using(var stream = ConnectionConfiguration.MemoryStreamFactory.Create(Body))

src/Elasticsearch.Net/Responses/Special/StringResponse.cs

@@ -15,7 +15,7 @@ public StringResponse() { }
 		public override bool TryGetServerError(out ServerError serverError)
 		{
 			serverError = null;
-			if (string.IsNullOrEmpty(Body) || ResponseMimeType != RequestData.MimeType)
+			if (string.IsNullOrEmpty(Body) || !RequestData.IsJsonMimeType(ResponseMimeType))
 				return false;
 
 			using(var stream = ConnectionConfiguration.MemoryStreamFactory.Create(Encoding.UTF8.GetBytes(Body)))

src/Elasticsearch.Net/Transport/Pipeline/RequestData.cs

@@ -13,11 +13,18 @@ namespace Elasticsearch.Net
 {
 	public class RequestData
 	{
-		public const string MimeType = "application/json";
-		public const string MimeTypeTextPlain = "text/plain";
 		public const string OpaqueIdHeader = "X-Opaque-Id";
 		public const string RunAsSecurityHeader = "es-security-runas-user";
 
+		public const string MimeTypeTextPlain = "text/plain";
+		private const string MimeTypeOld = "application/json";
+		private static readonly string MimeType = "application/vnd.elasticsearch+json; compatible-with=" + ClientVersionInfo.LowLevelClientVersionInfo.Version.Major;
+
+		public static readonly string DefaultJsonMimeType =
+			ClientVersionInfo.LowLevelClientVersionInfo.Version.Major >= 8 ? MimeType : MimeTypeOld;
+
+		public string JsonContentMimeType { get; }
+
 		public RequestData(HttpMethod method, string path, PostData data, IConnectionConfigurationValues global, IRequestParameters local,
 			IMemoryStreamFactory memoryStreamFactory
 		)
@@ -41,13 +48,15 @@ IMemoryStreamFactory memoryStreamFactory
 			Method = method;
 			PostData = data;
 
+			JsonContentMimeType = DefaultJsonBasedOnConfigurationSettings(ConnectionSettings);
+
 			if (data != null)
 				data.DisableDirectStreaming = local?.DisableDirectStreaming ?? global.DisableDirectStreaming;
 
 			Pipelined = local?.EnableHttpPipelining ?? global.HttpPipeliningEnabled;
 			HttpCompression = global.EnableHttpCompression;
-			RequestMimeType = local?.ContentType ?? MimeType;
-			Accept = local?.Accept ?? MimeType;
+			RequestMimeType = local?.ContentType ?? JsonContentMimeType;
+			Accept = local?.Accept ?? JsonContentMimeType;
 
 			if (global.Headers != null)
 				Headers = new NameValueCollection(global.Headers);
@@ -151,6 +160,37 @@ IMemoryStreamFactory memoryStreamFactory
 
 		public override string ToString() => $"{Method.GetStringValue()} {_path}";
 
+		public static string DefaultJsonBasedOnConfigurationSettings(IConnectionConfigurationValues settings) =>
+			settings.EnableApiVersioningHeader ? MimeType : MimeTypeOld;
+
+
+		public static bool IsJsonMimeType(string mimeType) =>
+			ValidResponseContentType(MimeType, mimeType) || ValidResponseContentType(MimeTypeOld, mimeType);
+
+		public static bool ValidResponseContentType(string acceptMimeType, string responseMimeType)
+		{
+			if (string.IsNullOrEmpty(acceptMimeType)) return false;
+			if (string.IsNullOrEmpty(responseMimeType)) return false;
+
+			// we a startswith check because the response can return charset information
+			// e.g: application/json; charset=UTF-8
+			if (acceptMimeType == RequestData.MimeTypeOld)
+				return responseMimeType.StartsWith(RequestData.MimeTypeOld, StringComparison.OrdinalIgnoreCase);
+
+			//vendored check
+			if (acceptMimeType == RequestData.MimeType)
+				// we check both vendored and nonvendored since on 7.x the response does not return a
+				// vendored Content-Type header on the response
+				return
+					responseMimeType == RequestData.MimeType
+					|| responseMimeType == RequestData.MimeTypeOld
+					|| responseMimeType.StartsWith(RequestData.MimeTypeOld, StringComparison.OrdinalIgnoreCase)
+					|| responseMimeType.StartsWith(RequestData.MimeType, StringComparison.OrdinalIgnoreCase);
+
+			return responseMimeType.StartsWith(acceptMimeType, StringComparison.OrdinalIgnoreCase);
+		}
+
+
 		// TODO This feels like its in the wrong place
 		private string CreatePathWithQueryStrings(string path, IConnectionConfigurationValues global, IRequestParameters request)
 		{

src/Elasticsearch.Net/Transport/Pipeline/RequestPipeline.cs

@@ -1,7 +1,7 @@
-// Licensed to Elasticsearch B.V under one or more agreements.
-// Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
-// See the LICENSE file in the project root for more information
-
+// Licensed to Elasticsearch B.V under one or more agreements.
+// Elasticsearch B.V licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information
+
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -142,7 +142,7 @@ ElasticsearchClientException exception
 			{
 				//make sure we copy over the error body in case we disabled direct streaming.
 				var s = callDetails?.ResponseBodyInBytes == null ? Stream.Null : _memoryStreamFactory.Create(callDetails.ResponseBodyInBytes);
-				var m = callDetails?.ResponseMimeType ?? RequestData.MimeType;
+				var m = callDetails?.ResponseMimeType ?? RequestData.DefaultJsonMimeType;
 				response = ResponseBuilder.ToResponse<TResponse>(data, exception, callDetails?.HttpStatusCode, null, s, m);
 			}
 

src/Elasticsearch.Net/Transport/Pipeline/ResponseBuilder.cs

@@ -25,7 +25,7 @@ public static TResponse ToResponse<TResponse>(
 			int? statusCode,
 			IEnumerable<string> warnings,
 			Stream responseStream,
-			string mimeType = RequestData.MimeType
+			string mimeType = null
 		)
 			where TResponse : class, IElasticsearchResponse, new()
 		{
@@ -43,7 +43,7 @@ public static async Task<TResponse> ToResponseAsync<TResponse>(
 			int? statusCode,
 			IEnumerable<string> warnings,
 			Stream responseStream,
-			string mimeType = RequestData.MimeType,
+			string mimeType = null,
 			CancellationToken cancellationToken = default
 		)
 			where TResponse : class, IElasticsearchResponse, new()
@@ -70,8 +70,7 @@ private static ApiCallDetails Initialize(
 					success = requestData.ConnectionSettings
 						.StatusCodeToResponseSuccess(requestData.Method, statusCode.Value);
 			}
-			//mimeType can include charset information on .NET full framework
-			if (!string.IsNullOrEmpty(mimeType) && !mimeType.StartsWith(requestData.Accept))
+			if (!RequestData.ValidResponseContentType(requestData.Accept, mimeType))
 				success = false;
 
 			var details = new ApiCallDetails
@@ -114,7 +113,7 @@ private static TResponse SetBody<TResponse>(ApiCallDetails details, RequestData
 				if (requestData.CustomResponseBuilder != null)
 					return requestData.CustomResponseBuilder.DeserializeResponse(serializer, details, responseStream) as TResponse;
 
-				return mimeType == null || !mimeType.StartsWith(requestData.Accept, StringComparison.Ordinal)
+				return !RequestData.ValidResponseContentType(requestData.Accept, mimeType)
 					? null
 					: serializer.Deserialize<TResponse>(responseStream);
 			}
@@ -146,7 +145,7 @@ private static async Task<TResponse> SetBodyAsync<TResponse>(
 				if (requestData.CustomResponseBuilder != null)
 					return await requestData.CustomResponseBuilder.DeserializeResponseAsync(serializer, details, responseStream, cancellationToken).ConfigureAwait(false) as TResponse;
 
-				return mimeType == null || !mimeType.StartsWith(requestData.Accept, StringComparison.Ordinal)
+				return !RequestData.ValidResponseContentType(requestData.Accept, mimeType)
 					? null
 					: await serializer
 						.DeserializeAsync<TResponse>(responseStream, cancellationToken)
@@ -170,7 +169,7 @@ private static bool SetSpecialTypes<TResponse>(string mimeType, byte[] bytes, IM
 			else if (responseType == typeof(DynamicResponse))
 			{
 				//if not json store the result under "body"
-				if (mimeType == null || !mimeType.StartsWith(RequestData.MimeType))
+				if (!RequestData.IsJsonMimeType(mimeType))
 				{
 					var dictionary = new DynamicDictionary();
 					dictionary["body"] = new DynamicValue(bytes.Utf8String());