ElasticsearchNode should use DesiredPort from NodeConfiguration (#2794)

* ElasticsearchNode should use DesiredPort from NodeConfiguration

Share ElasticsearchVersion instances

* Allow more time for flaky integration tests

Increase number of documents and wait time for BulkAll cancellation and dispose tests
Increase request timeout for Watcher inline execution test

(cherry picked from commit 9b0d36b)

Author: russcam

src/Tests/Document/Multiple/BulkAll/BulkAllApiTests.cs

@@ -80,7 +80,7 @@ public void DisposingObservableCancelsBulkAll()
 			var handle = new ManualResetEvent(false);
 
 			var size = 1000;
-			var pages = 100;
+			var pages = 1000;
 			var seenPages = 0;
 			var numberOfDocuments = size * pages;
 			var documents = this.CreateLazyStreamOfDocuments(numberOfDocuments);
@@ -104,7 +104,7 @@ public void DisposingObservableCancelsBulkAll()
 			observableBulk.Subscribe(bulkObserver);
 
 			//we wait N seconds to see some bulks
-			handle.WaitOne(TimeSpan.FromSeconds(1));
+			handle.WaitOne(TimeSpan.FromSeconds(3));
 			observableBulk.Dispose();
 			//we wait N seconds to give in flight request a chance to cancel
 			handle.WaitOne(TimeSpan.FromSeconds(3));
@@ -123,7 +123,7 @@ public void CancelBulkAll()
 			var handle = new ManualResetEvent(false);
 
 			var size = 1000;
-			var pages = 100;
+			var pages = 1000;
 			var seenPages = 0;
 			var numberOfDocuments = size * pages;
 			var documents = this.CreateLazyStreamOfDocuments(numberOfDocuments);
@@ -149,7 +149,7 @@ public void CancelBulkAll()
 			observableBulk.Subscribe(bulkObserver);
 
 			//we wait Nseconds to see some bulks
-			handle.WaitOne(TimeSpan.FromSeconds(1));
+			handle.WaitOne(TimeSpan.FromSeconds(3));
 			tokenSource.Cancel();
 			//we wait Nseconds to give in flight request a chance to cancel
 			handle.WaitOne(TimeSpan.FromSeconds(3));

src/Tests/Framework/Configuration/EnvironmentConfiguration.cs

@@ -11,7 +11,7 @@ public class EnvironmentConfiguration : TestConfigurationBase
 	{
 		public override bool TestAgainstAlreadyRunningElasticsearch { get; protected set; } = false;
 		public override bool ForceReseed { get; protected set; } = true;
-		public override ElasticsearchVersion ElasticsearchVersion { get; protected set; } = new ElasticsearchVersion("2.0.0");
+		public override ElasticsearchVersion ElasticsearchVersion { get; protected set; } = ElasticsearchVersion.GetOrAdd("2.0.0");
 		public override TestMode Mode { get; protected set; } = TestMode.Unit;
 		public override string ClusterFilter { get; protected set; }
 		public override string TestFilter { get; protected set; }
@@ -23,7 +23,7 @@ public EnvironmentConfiguration()
 			var version = Environment.GetEnvironmentVariable("NEST_INTEGRATION_VERSION");
 			if (!string.IsNullOrEmpty(version)) Mode = TestMode.Integration;
 
-			this.ElasticsearchVersion = new ElasticsearchVersion(string.IsNullOrWhiteSpace(version) ? "2.0.0" : version);
+			this.ElasticsearchVersion = ElasticsearchVersion.GetOrAdd(string.IsNullOrWhiteSpace(version) ? "2.0.0" : version);
 			this.ClusterFilter = Environment.GetEnvironmentVariable("NEST_INTEGRATION_CLUSTER");
 			this.TestFilter = Environment.GetEnvironmentVariable("NEST_TEST_FILTER");
 		}

src/Tests/Framework/Configuration/Versions/ElasticsearchVersion.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Linq;
+using System.Net;
 using System.Threading;
 using System.Xml.Linq;
 using Version = SemVer.Version;
@@ -14,6 +15,7 @@ public class ElasticsearchVersion : Version
 		private static readonly object _lock = new { };
 		private static readonly ConcurrentDictionary<string, string> SnapshotVersions = new ConcurrentDictionary<string, string>();
 		private static readonly string SonaTypeUrl = "https://oss.sonatype.org/content/repositories/snapshots/org/elasticsearch/distribution/zip/elasticsearch";
+		private static readonly ConcurrentDictionary<string, ElasticsearchVersion> Versions = new ConcurrentDictionary<string, ElasticsearchVersion>();
 		private static readonly string OldDownloadUrl = "https://download.elasticsearch.org/elasticsearch/release/org/elasticsearch/distribution/zip/elasticsearch";
 
 		private string RootUrl => this.IsSnapshot
@@ -119,5 +121,8 @@ public ElasticsearchVersion(string version) : base(TranslateConfigVersion(versio
 		public bool IsSnapshot => this.Version?.ToLower().Contains("snapshot") ?? false;
 
 		public override string ToString() => this.Version;
+
+		public static ElasticsearchVersion GetOrAdd(string version) =>
+			Versions.GetOrAdd(version, v => new ElasticsearchVersion(v));
 	}
 }

src/Tests/Framework/Configuration/YamlConfiguration.cs

@@ -23,7 +23,7 @@ public YamlConfiguration(string configurationFile)
 				.ToDictionary(ConfigName, ConfigValue);
 
 			this.Mode = GetTestMode(config["mode"]);
-			this.ElasticsearchVersion = new ElasticsearchVersion(config["elasticsearch_version"]);
+			this.ElasticsearchVersion = ElasticsearchVersion.GetOrAdd(config["elasticsearch_version"]);
 			this.ForceReseed = bool.Parse(config["force_reseed"]);
 			this.TestAgainstAlreadyRunningElasticsearch = bool.Parse(config["test_against_already_running_elasticsearch"]);
 			this.ClusterFilter = config.ContainsKey("cluster_filter") ? config["cluster_filter"] : null;

src/Tests/Framework/ManagedElasticsearch/Nodes/ElasticsearchNode.cs

@@ -26,17 +26,21 @@ public class ElasticsearchNode : IDisposable
 		private readonly NodeConfiguration _config;
 
 		public ElasticsearchVersion Version => _config.ElasticsearchVersion;
+
 		public NodeFileSystem FileSystem { get; }
 
 		public bool Started { get; private set; }
 
-		public int Port { get; private set; } = 9200;
+		public int Port { get; private set; }
+
+		private bool RunningOnCI { get; }
 
 		public ElasticsearchNode(NodeConfiguration config)
 		{
 			this._config = config;
 			this.FileSystem = config.FileSystem;
-
+			this.RunningOnCI = !string.IsNullOrEmpty(Environment.GetEnvironmentVariable("TEAMCITY_VERSION"));
+			this.Port = config.DesiredPort;
 			if (this._config.RunIntegrationTests && !this._config.TestAgainstAlreadyRunningElasticsearch) return;
 		}
 
@@ -88,7 +92,7 @@ public void Start(string[] settings)
 				{
 					var subscription = Observable.Using(() => process, p => p.Start())
 						.Select(c => new ElasticsearchConsoleOut(this._config.ElasticsearchVersion, c.Error, c.Data))
-						.Subscribe(s => this.HandleConsoleMessage(s, handle), e => { throw e; }, () => handle.Set());
+						.Subscribe(s => this.HandleConsoleMessage(s, handle), e => throw e, () => handle.Set());
 					this._composite.Add(subscription);
 
 					if (!handle.WaitOne(timeout, true))
@@ -111,20 +115,24 @@ private bool UseAlreadyRunningInstance()
 
 		private void HandleConsoleMessage(ElasticsearchConsoleOut consoleOut, XplatManualResetEvent handle)
 		{
-			Console.WriteLine(consoleOut.Data);
 			//no need to snoop for metadata if we already started
 			if (!this._config.RunIntegrationTests || this.Started) return;
 
-			if (consoleOut.Error && !this.Started && !string.IsNullOrWhiteSpace(consoleOut.Data)) 
-				throw new Exception("Error out:" + consoleOut.Data);
+			//if we are running on CI and not started dump elasticsearch stdout/err
+			{
+				if (consoleOut.Error) Console.Error.WriteLine(consoleOut.Data);
+				else Console.WriteLine(consoleOut.Data);
+			}
+
+			if (consoleOut.Error && !this.Started && !string.IsNullOrWhiteSpace(consoleOut.Data)) throw new Exception(consoleOut.Data);
 
 			string version;
 			int? pid;
 			int port;
 
 			if (this.ProcessId == null && consoleOut.TryParseNodeInfo(out version, out pid))
 			{
-				var startedVersion = new ElasticsearchVersion(version);
+				var startedVersion = ElasticsearchVersion.GetOrAdd(version);
 				this.ProcessId = pid;
 				if (this.Version != startedVersion)
 					throw new Exception($"Booted elasticsearch is version {startedVersion} but the test config dictates {this.Version}");

src/Tests/Framework/ManagedElasticsearch/Nodes/NodeConfiguration.cs

@@ -56,9 +56,9 @@ public NodeConfiguration(ITestConfiguration configuration, ClusterBase cluster)
 			this.DesiredPort = cluster.DesiredPort;
 
 			var attr = v.Major >= 5 ? "attr." : "";
-			var indexedOrStored = v > new ElasticsearchVersion("5.0.0-alpha1") ? "stored" : "indexed";
-			var shieldOrSecurity = v > new ElasticsearchVersion("5.0.0-alpha1") ? "xpack.security" : "shield";
-			var es = v > new ElasticsearchVersion("5.0.0-alpha2") ? "" : "es.";
+			var indexedOrStored = v > ElasticsearchVersion.GetOrAdd("5.0.0-alpha1") ? "stored" : "indexed";
+			var shieldOrSecurity = v > ElasticsearchVersion.GetOrAdd("5.0.0-alpha1") ? "xpack.security" : "shield";
+			var es = v > ElasticsearchVersion.GetOrAdd("5.0.0-alpha2") ? "" : "es.";
 			var b = this.XPackEnabled.ToString().ToLowerInvariant();
 			var sslEnabled = this.EnableSsl.ToString().ToLowerInvariant();
 			this.DefaultNodeSettings = new List<string>
@@ -74,6 +74,8 @@ public NodeConfiguration(ITestConfiguration configuration, ClusterBase cluster)
 				$"{es}node.{attr}testingcluster=true",
 				$"{es}{shieldOrSecurity}.enabled={b}",
 			};
+			if (v >= new ElasticsearchVersion("5.4.0"))
+				this.DefaultNodeSettings.Add($"{es}search.remote.connect=true");
 		}
 
 		public string[] CreateSettings(string[] additionalSettings)

src/Tests/Framework/ManagedElasticsearch/Plugins/ElasticsearchPluginCollection.cs

@@ -8,21 +8,21 @@ public class ElasticsearchPluginCollection : KeyedCollection<ElasticsearchPlugin
 		public static ElasticsearchPluginCollection Supported { get; } =
 			new ElasticsearchPluginCollection
 			{
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.DeleteByQuery, version => version < new ElasticsearchVersion("5.0.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.CloudAzure, version => version < new ElasticsearchVersion("5.0.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.License, version => version < new ElasticsearchVersion("5.0.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.Shield, version => version < new ElasticsearchVersion("5.0.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.Watcher, version => version < new ElasticsearchVersion("5.0.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.Graph, v => v >= new ElasticsearchVersion("2.3.0") && v < new ElasticsearchVersion("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.DeleteByQuery, version => version < ElasticsearchVersion.GetOrAdd("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.CloudAzure, version => version < ElasticsearchVersion.GetOrAdd("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.License, version => version < ElasticsearchVersion.GetOrAdd("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.Shield, version => version < ElasticsearchVersion.GetOrAdd("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.Watcher, version => version < ElasticsearchVersion.GetOrAdd("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.Graph, v => v >= ElasticsearchVersion.GetOrAdd("2.3.0") && v < ElasticsearchVersion.GetOrAdd("5.0.0")),
 
 				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.MapperMurmer3),
 				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.AnalysisKuromoji),
 				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.AnalysisIcu),
 
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.MapperAttachments, version => version >= new ElasticsearchVersion("2.4.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.XPack, version => version >= new ElasticsearchVersion("5.0.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.IngestGeoIp, version => version >= new ElasticsearchVersion("5.0.0")),
-				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.IngestAttachment, version => version >= new ElasticsearchVersion("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.MapperAttachments, version => version >= ElasticsearchVersion.GetOrAdd("2.4.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.XPack, version => version >= ElasticsearchVersion.GetOrAdd("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.IngestGeoIp, version => version >= ElasticsearchVersion.GetOrAdd("5.0.0")),
+				new ElasticsearchPluginConfiguration(ElasticsearchPlugin.IngestAttachment, version => version >= ElasticsearchVersion.GetOrAdd("5.0.0")),
 			};
 
 		protected override ElasticsearchPlugin GetKeyForItem(ElasticsearchPluginConfiguration item) => item.Plugin;

src/Tests/Framework/ManagedElasticsearch/Tasks/ValidationTasks/ValidateRunningVersion.cs

@@ -17,8 +17,8 @@ public override void Validate(IElasticClient client, NodeConfiguration configura
 			if (!alreadyUp.IsValid) return;
 			var v = configuration.ElasticsearchVersion;
 
-			var alreadyUpVersion = new ElasticsearchVersion(alreadyUp.Version.Number);
-			var alreadyUpSnapshotVersion = new ElasticsearchVersion(alreadyUp.Version.Number + "-SNAPSHOT");
+			var alreadyUpVersion = ElasticsearchVersion.GetOrAdd(alreadyUp.Version.Number);
+			var alreadyUpSnapshotVersion = ElasticsearchVersion.GetOrAdd(alreadyUp.Version.Number + "-SNAPSHOT");
 			if (v != alreadyUpVersion && v != alreadyUpSnapshotVersion)
 				throw new Exception($"running elasticsearch is version {alreadyUpVersion} but the test config dictates {v}");
 		}

src/Tests/XPack/Watcher/ExecuteWatch/ExecuteWatchApiTests.cs

@@ -549,6 +549,9 @@ protected override LazyResponses ClientUsage() => Calls(
 						.Body("{}")
 					)
 				)
+			)
+			.RequestConfiguration(r => r
+				.RequestTimeout(TimeSpan.FromMinutes(2))
 			);
 
 		protected override ExecuteWatchRequest Initializer =>
@@ -616,6 +619,10 @@ protected override LazyResponses ClientUsage() => Calls(
 						Method = HttpInputMethod.Post,
 						Body = "{}"
 					}
+				},
+				RequestConfiguration = new RequestConfiguration
+				{
+					RequestTimeout = TimeSpan.FromMinutes(2)
 				}
 			};