Reseed sorting and StickySniffingConnectionPool variant (#2789)

* Add SniffingSortedStickyConnectionPool

* SniffingSortedStickyConnectionPool inherits from SniffingConnectionPool, uses scoring function for node priority

* Extract common actions in StaticConnectionPool to share logic with inheritors

* Make StickyConnectionPool an override of StaticConnectionPool, so all the pools can share common code

* Continuation of #2657

- rebased and cherry-picked relevant commits
- Make the sorting a property of the base classes, this now also allows
  the regular sniffing connection to sort the responses
- Renamed SniffingSortedStickyCP to StickySniffingCP because of that
- Added documentation and tests for the StickySniffingCP
- Fixed a bug with the virtual clustering Failing with a 200 response
  after .Succeeds(times).

* fix documentation PR feedback

Author: Mpdreamz

src/Elasticsearch.Net/ConnectionPool/SniffingConnectionPool.cs

@@ -20,15 +20,20 @@ public SniffingConnectionPool(IEnumerable<Uri> uris, bool randomize = true, IDat
 		{ }
 
 		public SniffingConnectionPool(IEnumerable<Node> nodes, bool randomize = true, IDateTimeProvider dateTimeProvider = null)
-			: base(nodes, randomize, dateTimeProvider)
-		{ }
+			: base(nodes, randomize, dateTimeProvider) { }
 
+		public SniffingConnectionPool(IEnumerable<Node> nodes, Func<Node, float> nodeScorer, IDateTimeProvider dateTimeProvider = null)
+			: base(nodes, nodeScorer, dateTimeProvider) { }
+
+		/// <summary>
+		/// Obsolete overload
+		/// </summary>
+		/// <param name="predicate">UNUSED</param>
+		[Obsolete("this constructor has an unused parameter: predicate and will be removed in 6.0")]
 		public SniffingConnectionPool(IEnumerable<Node> nodes, Func<Node, bool> predicate, bool randomize = true, IDateTimeProvider dateTimeProvider = null)
 			: base(nodes, randomize, dateTimeProvider)
 		{ }
 
-		private static bool DefaultPredicate(Node node) => !node.MasterOnlyNode;
-
 		/// <inheritdoc/>
 		public override IReadOnlyCollection<Node> Nodes
 		{
@@ -56,8 +61,7 @@ public override void Reseed(IEnumerable<Node> nodes)
 			try
 			{
 				this._readerWriter.EnterWriteLock();
-				var sortedNodes = nodes
-					.OrderBy(item => this.Randomize ? this.Random.Next() : 1)
+				var sortedNodes = this.SortNodes(nodes)
 					.DistinctBy(n => n.Uri)
 					.ToList();
 

src/Elasticsearch.Net/ConnectionPool/StaticConnectionPool.cs

@@ -1,18 +1,31 @@
 using System;
 using System.Collections.Generic;
+using System.Globalization;
 using System.Linq;
 using System.Threading;
 
 namespace Elasticsearch.Net
 {
 	public class StaticConnectionPool : IConnectionPool
 	{
+		private readonly Func<Node, float> _nodeScorer;
 		protected IDateTimeProvider DateTimeProvider { get; }
 		protected Random Random { get; } = new Random();
 		protected bool Randomize { get; }
 
 		protected List<Node> InternalNodes { get; set; }
 
+		protected List<Node> AliveNodes
+		{
+			get
+			{
+				var now = DateTimeProvider.Now();
+                return this.InternalNodes
+                    .Where(n => n.IsAlive || n.DeadUntil <= now)
+                    .ToList();
+			}
+		}
+
 		/// <inheritdoc/>
 		public virtual IReadOnlyCollection<Node> Nodes => this.InternalNodes;
 
@@ -41,10 +54,16 @@ public StaticConnectionPool(IEnumerable<Uri> uris, bool randomize = true, IDateT
 		{ }
 
 		public StaticConnectionPool(IEnumerable<Node> nodes, bool randomize = true, IDateTimeProvider dateTimeProvider = null)
+			: this(nodes, null, dateTimeProvider)
 		{
-			nodes.ThrowIfEmpty(nameof(nodes));
-
 			this.Randomize = randomize;
+		}
+
+		//this constructor is protected because nodeScorer only makes sense on subclasses that support reseeding
+		//otherwise just manually sort `nodes` before instantiating.
+		protected StaticConnectionPool(IEnumerable<Node> nodes, Func<Node, float> nodeScorer, IDateTimeProvider dateTimeProvider = null)
+		{
+			nodes.ThrowIfEmpty(nameof(nodes));
 			this.DateTimeProvider = dateTimeProvider ?? Elasticsearch.Net.DateTimeProvider.Default;
 
 			var nn = nodes.ToList();
@@ -54,8 +73,8 @@ public StaticConnectionPool(IEnumerable<Node> nodes, bool randomize = true, IDat
 
 			this.UsingSsl = uris.Any(uri => uri.Scheme == "https");
 
-			this.InternalNodes = nn
-				.OrderBy(item => randomize ? this.Random.Next() : 1)
+			this._nodeScorer = nodeScorer;
+			this.InternalNodes = this.SortNodes(nn)
 				.DistinctBy(n => n.Uri)
 				.ToList();
 			this.LastUpdate = this.DateTimeProvider.Now();
@@ -69,42 +88,57 @@ public StaticConnectionPool(IEnumerable<Node> nodes, bool randomize = true, IDat
 		/// </summary>
 		public virtual IEnumerable<Node> CreateView(Action<AuditEvent, Node> audit = null)
 		{
-			//var count = this.InternalNodes.Count;
+			var nodes = this.AliveNodes;
 
-			var now = this.DateTimeProvider.Now();
-			var nodes = this.InternalNodes.Where(n => n.IsAlive || n.DeadUntil <= now)
-				.ToList();
-			var count = nodes.Count;
-			Node node;
 			var globalCursor = Interlocked.Increment(ref GlobalCursor);
 
-			if (count == 0)
+			if (nodes.Count == 0)
 			{
 				//could not find a suitable node retrying on first node off globalCursor
-				audit?.Invoke(AuditEvent.AllNodesDead, null);
-				node = this.InternalNodes[globalCursor % this.InternalNodes.Count];
-				node.IsResurrected = true;
-				audit?.Invoke(AuditEvent.Resurrection, node);
-				yield return node;
+				yield return this.RetryInternalNodes(globalCursor, audit);
 				yield break;
 			}
 
-			var localCursor = globalCursor % count;
+			var localCursor = globalCursor % nodes.Count;
+			foreach (var aliveNode in SelectAliveNodes(localCursor, nodes, audit))
+			{
+				yield return aliveNode;
+			}
+		}
+
+		protected virtual Node RetryInternalNodes(int globalCursor, Action<AuditEvent, Node> audit = null)
+		{
+			audit?.Invoke(AuditEvent.AllNodesDead, null);
+			var node = this.InternalNodes[globalCursor % this.InternalNodes.Count];
+			node.IsResurrected = true;
+			audit?.Invoke(AuditEvent.Resurrection, node);
+
+			return node;
+		}
 
-			for (var attempts = 0; attempts < count; attempts++)
+		protected virtual IEnumerable<Node> SelectAliveNodes(int cursor, List<Node> aliveNodes, Action<AuditEvent, Node> audit = null)
+		{
+			for (var attempts = 0; attempts < aliveNodes.Count; attempts++)
 			{
-				node = nodes[localCursor];
-				localCursor = (localCursor + 1) % count;
+				var node = aliveNodes[cursor];
+				cursor = (cursor + 1) % aliveNodes.Count;
 				//if this node is not alive or no longer dead mark it as resurrected
 				if (!node.IsAlive)
 				{
 					audit?.Invoke(AuditEvent.Resurrection, node);
 					node.IsResurrected = true;
 				}
+
 				yield return node;
 			}
 		}
 
+		protected IOrderedEnumerable<Node> SortNodes(IEnumerable<Node> nodes) =>
+			this._nodeScorer != null
+				? nodes.OrderByDescending(_nodeScorer)
+				: nodes.OrderBy(n => this.Randomize ? this.Random.Next() : 1);
+
+
 		void IDisposable.Dispose() => this.DisposeManagedResources();
 
 		protected virtual void DisposeManagedResources() { }

src/Elasticsearch.Net/ConnectionPool/StickyConnectionPool.cs

@@ -1,102 +1,42 @@
 using System;
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 
 namespace Elasticsearch.Net
 {
-	public class StickyConnectionPool : IConnectionPool
+	public class StickyConnectionPool : StaticConnectionPool
 	{
-		protected IDateTimeProvider DateTimeProvider { get; }
-
-		protected List<Node> InternalNodes { get; set; }
-
-		public bool UsingSsl { get; }
-		public bool SniffedOnStartup { get; set; }
-
-		public IReadOnlyCollection<Node> Nodes => this.InternalNodes;
-
-		public int MaxRetries => this.InternalNodes.Count - 1;
-
-		public bool SupportsReseeding => false;
-
-		public bool SupportsPinging => true;
-
-		public DateTime LastUpdate { get; protected set; }
-
 		public StickyConnectionPool(IEnumerable<Uri> uris, IDateTimeProvider dateTimeProvider = null)
-			: this(uris.Select(uri => new Node(uri)), dateTimeProvider)
+			: base(uris, false, dateTimeProvider)
 		{ }
 
 		public StickyConnectionPool(IEnumerable<Node> nodes, IDateTimeProvider dateTimeProvider = null)
-		{
-			nodes.ThrowIfEmpty(nameof(nodes));
-
-			this.DateTimeProvider = dateTimeProvider ?? Elasticsearch.Net.DateTimeProvider.Default;
-
-			var nn = nodes.ToList();
-			var uris = nn.Select(n => n.Uri).ToList();
-			if (uris.Select(u => u.Scheme).Distinct().Count() > 1)
-				throw new ArgumentException("Trying to instantiate a connection pool with mixed URI Schemes");
-
-			this.UsingSsl = uris.Any(uri => uri.Scheme == "https");
-
-			this.InternalNodes = nn
-				.DistinctBy(n => n.Uri)
-				.ToList();
-
-			this.LastUpdate = this.DateTimeProvider.Now();
-		}
-
-		protected int GlobalCursor = -1;
+			: base(nodes, false, dateTimeProvider)
+		{ }
 
-		public IEnumerable<Node> CreateView(Action<AuditEvent, Node> audit = null)
+		public override IEnumerable<Node> CreateView(Action<AuditEvent, Node> audit = null)
 		{
-			var now = this.DateTimeProvider.Now();
-			var nodes = this.InternalNodes.Where(n => n.IsAlive || n.DeadUntil <= now)
-				.ToList();
-			var count = nodes.Count;
-			Node node;
+			var nodes = this.AliveNodes;
 
-			if (count == 0)
+			if (nodes.Count == 0)
 			{
-				var globalCursor = Interlocked.Increment(ref GlobalCursor);
+				var globalCursor = Interlocked.Increment(ref this.GlobalCursor);
+
 				//could not find a suitable node retrying on first node off globalCursor
-				audit?.Invoke(AuditEvent.AllNodesDead, null);
-				node = this.InternalNodes[globalCursor % this.InternalNodes.Count];
-				node.IsResurrected = true;
-				audit?.Invoke(AuditEvent.Resurrection, node);
-				yield return node;
+				yield return this.RetryInternalNodes(globalCursor, audit);
 				yield break;
 			}
 
 			// If the cursor is greater than the default then it's been
 			// set already but we now have a live node so we should reset it
-			if (GlobalCursor > -1)
-			{
-				Interlocked.Exchange(ref GlobalCursor, -1);
-			}
+			if (this.GlobalCursor > -1)
+				Interlocked.Exchange(ref this.GlobalCursor, -1);
 
 			var localCursor = 0;
-
-			for (var attempts = 0; attempts < count; attempts++)
-			{
-				node = nodes[localCursor];
-				localCursor = (localCursor + 1) % count;
-				//if this node is not alive or no longer dead mark it as resurrected
-				if (!node.IsAlive)
-				{
-					audit?.Invoke(AuditEvent.Resurrection, node);
-					node.IsResurrected = true;
-				}
-				yield return node;
-			}
+			foreach (var aliveNode in this.SelectAliveNodes(localCursor, nodes, audit))
+				yield return aliveNode;
 		}
 
-		public void Reseed(IEnumerable<Node> nodes) { }
-
-		void IDisposable.Dispose() => this.DisposeManagedResources();
-
-		protected virtual void DisposeManagedResources() { }
+		public override void Reseed(IEnumerable<Node> nodes) { }
 	}
 }

src/Elasticsearch.Net/ConnectionPool/StickySniffingConnectionPool.cs

@@ -0,0 +1,44 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+
+namespace Elasticsearch.Net
+{
+	public class StickySniffingConnectionPool : SniffingConnectionPool
+	{
+		public override bool SupportsPinging => true;
+		public override bool SupportsReseeding => true;
+
+		public StickySniffingConnectionPool(IEnumerable<Uri> uris, Func<Node, float> nodeScorer, IDateTimeProvider dateTimeProvider = null)
+			: base(uris.Select(uri => new Node(uri)), nodeScorer ?? DefaultNodeScore, dateTimeProvider) { }
+
+		public StickySniffingConnectionPool(IEnumerable<Node> nodes, Func<Node, float> nodeScorer, IDateTimeProvider dateTimeProvider = null)
+			: base(nodes, nodeScorer ?? DefaultNodeScore, dateTimeProvider) { }
+
+		public override IEnumerable<Node> CreateView(Action<AuditEvent, Node> audit = null)
+		{
+			var nodes = this.AliveNodes;
+
+			if (nodes.Count == 0)
+			{
+				var globalCursor = Interlocked.Increment(ref this.GlobalCursor);
+
+				//could not find a suitable node retrying on first node off globalCursor
+				yield return this.RetryInternalNodes(globalCursor, audit);
+				yield break;
+			}
+
+			// If the cursor is greater than the default then it's been
+			// set already but we now have a live node so we should reset it
+			if (this.GlobalCursor > -1)
+				Interlocked.Exchange(ref this.GlobalCursor, -1);
+
+			var localCursor = 0;
+			foreach (var aliveNode in this.SelectAliveNodes(localCursor, nodes, audit))
+				yield return aliveNode;
+		}
+
+		private static float DefaultNodeScore(Node node) => 0f;
+	}
+}

src/Tests/ClientConcepts/ConnectionPooling/BuildingBlocks/ConnectionPooling.doc.cs

@@ -180,5 +180,38 @@ [U] public void Sticky()
 			client.ConnectionSettings.ConnectionPool
 				.Should().BeOfType<StickyConnectionPool>();
 		}
+
+		/**[[sticky-sniffing-connection-pool]]
+		* ==== Sticky Sniffing Connection Pool
+		*
+		* A type of connection pool that returns the first live node to issue a request against, such that the node is _sticky_ between requests.
+		* This implementation supports sniffing and sorting so that each instance of your application can favor a node in the same rack based
+		* on node attributes for instance.
+		*/
+		[U] public void SniffingSortedSticky()
+		{
+			var uris = Enumerable.Range(9200, 5)
+				.Select(port => new Uri($"http://localhost:{port}"));
+
+			/** a sniffing sorted sticky pool takes a second parameter `Func` takes a Node and returns a weight.
+			* Nodes will be sorted descending by weight. In the following example we score nodes that are client nodes
+			* AND in rack_id `rack_one` the highest
+			*/
+
+			var pool = new StickySniffingConnectionPool(uris, n =>
+				(n.ClientNode ? 10 : 0)
+				+ (n.Settings.TryGetValue("node.attr.rack_id", out string rackId)
+				   		&& rackId == "rack_one" ? 10 : 0));
+
+			pool.SupportsReseeding.Should().BeTrue();
+			pool.SupportsPinging.Should().BeTrue();
+
+			/** To create a client using the sticky sniffing connection pool pass
+			* the connection pool to the `ConnectionSettings` you pass to `ElasticClient`
+			*/
+			var client = new ElasticClient(new ConnectionSettings(pool));
+			client.ConnectionSettings.ConnectionPool
+				.Should().BeOfType<StickySniffingConnectionPool>();
+		}
 	}
 }