Add processor stats to node ingest stats (#3529)

Author: Mpdreamz

src/Nest/Cluster/NodesStats/Statistics/IngestStats.cs

@@ -1,32 +1,59 @@
-using Newtonsoft.Json;
+using System.Collections.Generic;
+using Newtonsoft.Json;
 
 namespace Nest
 {
 	[JsonObject]
 	public class IngestStats
 	{
-		/// <summary>
-		/// The total number of document ingested during the lifetime of this node
-		/// </summary>
+		/// <summary> The total number of document ingested during the lifetime of this node</summary>
 		[JsonProperty("count")]
 		public long Count { get; set; }
 
-		/// <summary>
-		/// The total number of documents currently being ingested.
-		/// </summary>
+		/// <summary> The total number of documents currently being ingested. </summary>
 		[JsonProperty("current")]
 		public long Current { get; set; }
 
-		/// <summary>
-		/// The total number ingest preprocessing operations failed during the lifetime of this node
-		/// </summary>
+		/// <summary> The total number ingest preprocessing operations failed during the lifetime of this node </summary>
 		[JsonProperty("failed")]
 		public long Failed { get; set; }
 
-		/// <summary>
-		/// The total time spent on ingest preprocessing documents during the lifetime of this node
-		/// </summary>
+		/// <summary> The total time spent on ingest preprocessing documents during the lifetime of this node </summary>
 		[JsonProperty("time_in_millis")]
 		public long TimeInMilliseconds { get; set; }
+
+		[JsonProperty("processors")]
+		public IReadOnlyCollection<KeyedProcessorStats> Processors { get; internal set; } =
+			EmptyReadOnly<KeyedProcessorStats>.Collection;
+	}
+
+	[JsonConverter(typeof(KeyValueJsonConverter<KeyedProcessorStats, ProcessStats>))]
+	public class KeyedProcessorStats
+	{
+		/// <summary> The type of the processor </summary>
+		public string Type { get; set; }
+
+		/// <summary>The statistics for this processor</summary>
+		public ProcessStats Statistics { get; set; }
+	}
+
+	[JsonObject]
+	public class ProcessorStats
+	{
+		/// <summary> The total number of document ingested during the lifetime of this node </summary>
+		[JsonProperty("count")]
+		public long Count { get; internal set; }
+
+		/// <summary> The total number of documents currently being ingested. </summary>
+		[JsonProperty("current")]
+		public long Current { get; internal set; }
+
+		/// <summary> The total number ingest preprocessing operations failed during the lifetime of this node </summary>
+		[JsonProperty("failed")]
+		public long Failed { get; internal set; }
+
+		/// <summary> The total time spent on ingest preprocessing documents during the lifetime of this node </summary>
+		[JsonProperty("time_in_millis")]
+		public long TimeInMilliseconds { get; internal set; }
 	}
 }

src/Tests/Tests.Core/ManagedElasticsearch/NodeSeeders/DefaultSeeder.cs

@@ -17,6 +17,8 @@ public class DefaultSeeder
 		public const string ProjectsAliasName = "projects-alias";
 		public const string TestsIndexTemplateName = "nest_tests";
 
+		public const string PipelineName = "nest-pipeline";
+
 		private readonly IIndexSettings _defaultIndexSettings = new IndexSettings()
 		{
 			NumberOfShards = 2,
@@ -63,6 +65,7 @@ private async Task SeedNodeAsync()
 			// Ensure a clean slate by deleting everything regardless of whether they may already exist
 			await DeleteIndicesAndTemplatesAsync();
 			await ClusterSettingsAsync();
+			await PutPipeline();
 			// and now recreate everything
 			await CreateIndicesAndSeedIndexDataAsync();
 		}
@@ -89,6 +92,20 @@ public async Task ClusterSettingsAsync()
 			putSettingsResponse.ShouldBeValid();
 		}
 
+		public async Task PutPipeline()
+		{
+			if (TestConfiguration.Instance.InRange("<6.1.0")) return;
+
+			var putProcessors = await Client.PutPipelineAsync(PipelineName, pi => pi
+				.Description("A pipeline registered by the NEST test framework")
+				.Processors(pp => pp
+					.Set<Project>(s => s.Field(p => p.Metadata).Value(new { x = "y" }))
+				)
+			);
+			putProcessors.ShouldBeValid();
+		}
+
+
 		public async Task DeleteIndicesAndTemplatesAsync()
 		{
 			var tasks = new Task[]

src/Tests/Tests/Cluster/NodesStats/NodesStatsApiTests.cs

@@ -5,6 +5,7 @@
 using Nest;
 using Tests.Core.Extensions;
 using Tests.Core.ManagedElasticsearch.Clusters;
+using Tests.Core.ManagedElasticsearch.NodeSeeders;
 using Tests.Domain;
 using Tests.Framework;
 using Tests.Framework.Integration;
@@ -73,6 +74,19 @@ protected void Assert(NodeIngestStats nodeIngestStats)
 			nodeIngestStats.Should().NotBeNull();
 			nodeIngestStats.Total.Should().NotBeNull();
 			nodeIngestStats.Pipelines.Should().NotBeNull();
+			nodeIngestStats.Pipelines.Should().ContainKey(DefaultSeeder.PipelineName);
+
+			var pipelineStats = nodeIngestStats.Pipelines[DefaultSeeder.PipelineName];
+
+			pipelineStats.Should().NotBeNull();
+			pipelineStats.Processors.Should().NotBeNull().And.HaveCount(1);
+
+			var processorStats = pipelineStats.Processors.First();
+
+			processorStats.Type.Should().Be("set");
+			processorStats.Statistics.Should().NotBeNull();
+
+
 		}
 
 		protected void Assert(IReadOnlyDictionary<string, AdaptiveSelectionStats> adaptiveSelectionStats) =>