Fix #3404 bulk all improvements (#3405)

* Add more explicit support for what to do when dropped documents are encountered. Now defaults to halting the BulkAll which is a saner default

* fix audit reporting negative durations

* Add FailedOverOnAllNodes audit event

* BulkAll now handles bad responses from the _bulk itself as well and optionally bails out early

* Fixed tests now excepting FailedOverAllNodes

* ContinueAfterDroppedDocuments should not be tested for nullable arg in the code standards test

Author: Mpdreamz

src/Elasticsearch.Net/Auditing/Audit.cs

@@ -41,6 +41,7 @@ public Audit(AuditEvent type, DateTime started)
 		public override string ToString()
 		{
 			var took = Ended - Started;
+			if (took < TimeSpan.Zero) took = TimeSpan.Zero;
 			return $"Node: {Node?.Uri}, Event: {Event.GetStringValue()} NodeAlive: {Node?.IsAlive}, Took: {took}";
 		}
 	}

src/Elasticsearch.Net/Auditing/AuditEvent.cs

@@ -21,5 +21,6 @@ public enum AuditEvent
 		BadRequest,
 		NoNodesAttempted,
 		CancellationRequested,
+		FailedOverAllNodes,
 	}
 }

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

@@ -526,13 +526,21 @@ public ElasticsearchClientException CreateClientException<TResponse>(
 				pipelineFailure = PipelineFailure.MaxRetriesReached;
 				this.Audit(MaxRetriesReached);
 				exceptionMessage = "Maximum number of retries reached";
+
+				var now = this._dateTimeProvider.Now();
+				// TODO make AliveNodes on IConnectionPool public in 7.0 (default interface C# 8 FTW)
+				var activeNodes = this._connectionPool.Nodes.Count(n => n.IsAlive || n.DeadUntil <= now);
+				if (this.Retried >= activeNodes)
+				{
+                    this.Audit(FailedOverAllNodes);
+                    exceptionMessage += ", failed over to all the known alive nodes before failing";
+				}
 			}
 
 			exceptionMessage += $". Call: {resource}";
 			if (response != null && response.TryGetServerErrorReason(out var reason))
 				exceptionMessage += $". ServerError: {reason}";
 
-
 			var clientException = new ElasticsearchClientException(pipelineFailure, exceptionMessage, innerException)
 			{
 				Request = data,

src/Nest/Document/Multiple/BulkAll/BulkAllObservable.cs

@@ -21,6 +21,7 @@ public class BulkAllObservable<T> : IDisposable, IObservable<IBulkAllResponse> w
 		private readonly CancellationToken _compositeCancelToken;
 		private readonly CancellationTokenSource _compositeCancelTokenSource;
 		private readonly Func<IBulkResponseItem, T, bool> _retryPredicate;
+		private Action<IBulkResponseItem, T> _droppedDocumentCallBack;
 
 		public BulkAllObservable(
 			IElasticClient client,
@@ -34,6 +35,7 @@ public BulkAllObservable(
 			this._backOffTime = (this._partionedBulkRequest?.BackOffTime?.ToTimeSpan() ?? CoordinatedRequestDefaults.BulkAllBackOffTimeDefault);
 			this._bulkSize = this._partionedBulkRequest.Size ?? CoordinatedRequestDefaults.BulkAllSizeDefault;
 			this._retryPredicate = this._partionedBulkRequest.RetryDocumentPredicate ?? RetryBulkActionPredicate;
+			this._droppedDocumentCallBack = this._partionedBulkRequest.DroppedDocumentCallback ?? DroppedDocumentCallbackDefault;
 			this._maxDegreeOfParallelism = _partionedBulkRequest.MaxDegreeOfParallelism ?? CoordinatedRequestDefaults.BulkAllMaxDegreeOfParallelismDefault;
 			this._compositeCancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 			this._compositeCancelToken = this._compositeCancelTokenSource.Token;
@@ -53,9 +55,6 @@ public IDisposable Subscribe(IObserver<IBulkAllResponse> observer)
 			return this;
 		}
 
-		private static ElasticsearchClientException Throw(string message, IApiCallDetails details) =>
-			new ElasticsearchClientException(PipelineFailure.BadResponse, message, details);
-
 		private void BulkAll(IObserver<IBulkAllResponse> observer)
 		{
 			var documents = this._partionedBulkRequest.Documents;
@@ -116,29 +115,85 @@ private async Task<IBulkAllResponse> BulkAsync(IList<T> buffer, long page, int b
 
 			this._compositeCancelToken.ThrowIfCancellationRequested();
 
-			var retryDocuments = response.Items.Zip(buffer, (i, d) => new {i, d})
-				.Where(x=> !response.IsValid && this._retryPredicate(x.i, x.d))
-				.Select(x => x.d)
+			if (!response.ApiCall.Success)
+				return await this.HandleBulkRequest(buffer, page, backOffRetries, response);
+
+			var documentsWithResponse = response.Items.Zip(buffer, Tuple.Create).ToList();
+
+			this.HandleDroppedDocuments(documentsWithResponse, response);
+
+			var retryDocuments = documentsWithResponse
+				.Where(x=> !response.IsValid && this._retryPredicate(x.Item1, x.Item2))
+				.Select(x => x.Item2)
 				.ToList();
 
 			if (retryDocuments.Count > 0 && backOffRetries < this._backOffRetries)
+				return await this.RetryDocuments(page, ++backOffRetries, retryDocuments);
+			else if (retryDocuments.Count > 0)
+				throw this.ThrowOnBadBulk(response, $"Bulk indexing failed and after retrying {backOffRetries} times");
+
+			this._partionedBulkRequest.BackPressure?.Release();
+			return new BulkAllResponse { Retries = backOffRetries, Page = page };
+		}
+
+		private void HandleDroppedDocuments(List<Tuple<IBulkResponseItem, T>> documentsWithResponse, IBulkResponse response)
+		{
+			var droppedDocuments = documentsWithResponse
+				.Where(x => !response.IsValid && !this._retryPredicate(x.Item1, x.Item2))
+				.ToList();
+			if (droppedDocuments.Count > 0 && !this._partionedBulkRequest.ContinueAfterDroppedDocuments)
+				throw this.ThrowOnBadBulk(response, $"BulkAll halted after receiving failures that can not be retried from _bulk");
+			else if (droppedDocuments.Count > 0 && this._partionedBulkRequest.ContinueAfterDroppedDocuments)
 			{
-				this._incrementRetries();
-				await Task.Delay(this._backOffTime, this._compositeCancelToken).ConfigureAwait(false);
-				return await this.BulkAsync(retryDocuments, page, ++backOffRetries).ConfigureAwait(false);
+				foreach (var dropped in droppedDocuments) this._droppedDocumentCallBack(dropped.Item1, dropped.Item2);
 			}
-			if (retryDocuments.Count > 0)
+		}
+
+		private async Task<IBulkAllResponse> HandleBulkRequest(IList<T> buffer, long page, int backOffRetries, IBulkResponse response)
+		{
+			var clientException = response.ApiCall.OriginalException as ElasticsearchClientException;
+			//TODO expose this on IAPiCallDetails as RetryLater in 7.0?
+			var failureReason = clientException?.FailureReason.GetValueOrDefault(PipelineFailure.Unexpected);
+			switch (failureReason)
 			{
-				this._incrementFailed();
-				this._partionedBulkRequest.BackPressure?.Release();
-				throw Throw($"Bulk indexing failed and after retrying {backOffRetries} times", response.ApiCall);
+				case PipelineFailure.MaxRetriesReached:
+					//TODO move this to its own PipelineFailure classification in 7.0
+					if (response.ApiCall.AuditTrail.Last().Event == AuditEvent.FailedOverAllNodes)
+						throw this.ThrowOnBadBulk(response, $"BulkAll halted after attempted bulk failed over all the active nodes");
+					return await this.RetryDocuments(page, ++backOffRetries, buffer);
+				case PipelineFailure.CouldNotStartSniffOnStartup:
+				case PipelineFailure.BadAuthentication:
+				case PipelineFailure.NoNodesAttempted:
+				case PipelineFailure.SniffFailure:
+				case PipelineFailure.Unexpected:
+					throw this.ThrowOnBadBulk(response,
+						$"BulkAll halted after {nameof(PipelineFailure)}{failureReason.GetStringValue()} from _bulk");
+				default:
+					return await this.RetryDocuments(page, ++backOffRetries, buffer);
 			}
+		}
+
+		private async Task<IBulkAllResponse> RetryDocuments(long page, int backOffRetries, IList<T> retryDocuments)
+		{
+			this._incrementRetries();
+			await Task.Delay(this._backOffTime, this._compositeCancelToken).ConfigureAwait(false);
+			return await this.BulkAsync(retryDocuments, page, backOffRetries).ConfigureAwait(false);
+		}
+
+		private Exception ThrowOnBadBulk(IElasticsearchResponse response, string message)
+		{
+			this._incrementFailed();
 			this._partionedBulkRequest.BackPressure?.Release();
-			return new BulkAllResponse {Retries = backOffRetries, Page = page};
+			return Throw(message, response.ApiCall);
 		}
+		private static ElasticsearchClientException Throw(string message, IApiCallDetails details) =>
+			new ElasticsearchClientException(PipelineFailure.BadResponse, message, details);
+
 
 		private static bool RetryBulkActionPredicate(IBulkResponseItem bulkResponseItem, T d) => bulkResponseItem.Status == 429;
 
+		private static void DroppedDocumentCallbackDefault(IBulkResponseItem bulkResponseItem, T d) { }
+
 		public bool IsDisposed { get; private set; }
 
 		public void Dispose()

src/Nest/Document/Multiple/BulkAll/BulkAllRequest.cs

@@ -73,6 +73,20 @@ public interface IBulkAllRequest<T> where T : class
 		/// A predicate which controls which documents should be retried, defaults to failed bulk items with status code 429
 		/// </summary>
 		Func<IBulkResponseItem, T, bool> RetryDocumentPredicate { get; set; }
+
+		/// <summary>
+		/// Halt the bulk all request if any of the documents returned is a failure that can not be retried.
+		/// When true, will feed dropped documents to <see cref="DroppedDocumentCallback"/>
+		/// </summary>
+		bool ContinueAfterDroppedDocuments { get; set; }
+
+		/// <summary>
+		/// If <see cref="ContinueAfterDroppedDocuments"/> is set to true dropped messages will be fed through
+		/// this callback. Use this if you don't expect many failures and want to feed these dropped messages in a dead letter queue
+		/// for instance.
+		/// </summary>
+		Action<IBulkResponseItem, T> DroppedDocumentCallback { get; set; }
+
 	}
 
 	public class BulkAllRequest<T>  : IBulkAllRequest<T>
@@ -108,9 +122,12 @@ public class BulkAllRequest<T>  : IBulkAllRequest<T>
 		public Action<BulkDescriptor, IList<T>> BufferToBulk { get; set; }
 		/// <inheritdoc />
 		public ProducerConsumerBackPressure BackPressure { get; set; }
-
 		/// <inheritdoc />
 		public Func<IBulkResponseItem, T, bool> RetryDocumentPredicate { get; set; }
+		/// <inheritdoc />
+		public bool ContinueAfterDroppedDocuments { get; set; }
+		/// <inheritdoc />
+		public Action<IBulkResponseItem, T> DroppedDocumentCallback { get; set; }
 
 		public BulkAllRequest(IEnumerable<T> documents)
 		{
@@ -141,6 +158,8 @@ public class BulkAllDescriptor<T> : DescriptorBase<BulkAllDescriptor<T>, IBulkAl
 		Action<BulkDescriptor, IList<T>>  IBulkAllRequest<T>.BufferToBulk { get; set; }
 		ProducerConsumerBackPressure IBulkAllRequest<T>.BackPressure { get; set; }
 		Func<IBulkResponseItem, T, bool> IBulkAllRequest<T>.RetryDocumentPredicate { get; set; }
+		bool IBulkAllRequest<T>.ContinueAfterDroppedDocuments { get; set; }
+		Action<IBulkResponseItem, T> IBulkAllRequest<T>.DroppedDocumentCallback { get; set; }
 
 		public BulkAllDescriptor(IEnumerable<T> documents)
 		{
@@ -209,5 +228,16 @@ public BulkAllDescriptor<T> RetryDocumentPredicate(Func<IBulkResponseItem, T, bo
 		public BulkAllDescriptor<T> BackPressure(int maxConcurrency, int? backPressureFactor = null) =>
 			Assign(a => a.BackPressure = new ProducerConsumerBackPressure(backPressureFactor, maxConcurrency));
 
+		/// <inheritdoc cref="IBulkAllRequest{T}.ContinueAfterDroppedDocuments" />
+		public BulkAllDescriptor<T> ContinueAfterDroppedDocuments(bool proceed = true) => Assign(p => p.ContinueAfterDroppedDocuments = proceed);
+
+		/// <summary>
+		/// If <see cref="ContinueAfterDroppedDocuments"/> is set to false (not the default) dropped messages will be fed through
+		/// this callback. Use this if you don't expect many failures and want to feed these dropped messages in a dead letter queue
+		/// for instance.
+		/// </summary>
+		public BulkAllDescriptor<T> DroppedDocumentCallback(Action<IBulkResponseItem, T> callback) =>
+			Assign(p => p.DroppedDocumentCallback = callback);
+
 	}
 }

src/Nest/Document/Multiple/BulkAll/BulkAllResponse.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using Newtonsoft.Json;
 
 namespace Nest
@@ -13,6 +14,7 @@ public interface IBulkAllResponse
 
 		/// <summary>The number of back off retries were needed to store this document.</summary>
 		int Retries { get; }
+
 	}
 
 	/// <inheritdoc />

src/Tests/Tests/ClientConcepts/ConnectionPooling/MaxRetries/RespectsMaxRetry.doc.cs

@@ -157,7 +157,8 @@ public async Task RetriesAreLimitedByNodesInPool()
 				new ClientCall {
 					{ BadResponse, 9200 },
 					{ BadResponse, 9201 },
-					{ MaxRetriesReached }
+					{ MaxRetriesReached },
+					{ FailedOverAllNodes }
 				}
 			);
 		}

src/Tests/Tests/ClientConcepts/ConnectionPooling/RoundRobin/SkipDeadNodes.doc.cs

@@ -158,6 +158,7 @@ await audit.TraceCalls(
 					{ BadResponse, 9202},
 					{ BadResponse, 9203},
 					{ MaxRetriesReached },
+					{ FailedOverAllNodes },
 					{ pool => pool.Nodes.Where(n=>!n.IsAlive).Should().HaveCount(4) }
 				},
 				new ClientCall {

src/Tests/Tests/ClientConcepts/ConnectionPooling/Sticky/SkipDeadNodes.doc.cs

@@ -136,6 +136,7 @@ await audit.TraceCalls(
 					{ BadResponse, 9202},
 					{ BadResponse, 9203},
 					{ MaxRetriesReached },
+					{ FailedOverAllNodes },
 					{ pool => pool.Nodes.Where(n=>!n.IsAlive).Should().HaveCount(4) }
 				},
 				/** After all our registered nodes are marked dead we want to sample a single dead node

src/Tests/Tests/ClientConcepts/ConnectionPooling/Sticky/StickySniffingConnectionPool.doc.cs

@@ -167,6 +167,7 @@ await audit.TraceCalls(
 					{ BadResponse, 9202},
 					{ BadResponse, 9203},
 					{ MaxRetriesReached },
+					{ FailedOverAllNodes },
 					{ pool => pool.Nodes.Where(n=>!n.IsAlive).Should().HaveCount(4) }
 				},
 				/** After all our registered nodes are marked dead we want to sample a single dead node