Add max_iterations configuration to watcher action with foreach execu… (#4139)

addresses elastic/elasticsearch#45715

Author: Mpdreamz

src/Nest/XPack/Watcher/Action/ActionBase.cs

@@ -42,6 +42,12 @@ public interface IAction
 		[IgnoreDataMember]
 		string Foreach { get; set; }
 
+		[IgnoreDataMember]
+		/// <summary>The maximum number of iterations that each watch executes. If this limit is reached,
+		/// the execution is gracefully stopped. Defaults to <c>100</c>.
+		/// </summary>
+		int? MaxIterations  { get; set; }
+
 		/// <summary>
 		/// Transforms the payload before executing the action. The transformation is only applied
 		/// for the payload for this action.
@@ -75,6 +81,9 @@ internal ActionBase() { }
 		/// <inheritdoc />
 		public string Foreach { get; set; }
 
+		/// <inheritdoc />
+		public int? MaxIterations { get; set; }
+
 		/// <inheritdoc />
 		public TransformContainer Transform { get; set; }
 
@@ -125,7 +134,8 @@ internal class ActionsFormatter : IJsonFormatter<Actions>
 			{ "pagerduty", 6 },
 			{ "foreach", 7 },
 			{ "transform", 8 },
-			{ "condition", 9 }
+			{ "condition", 9 },
+			{ "max_iterations", 10 },
 		};
 
 		public Actions Deserialize(ref JsonReader reader, IJsonFormatterResolver formatterResolver)
@@ -140,6 +150,7 @@ public Actions Deserialize(ref JsonReader reader, IJsonFormatterResolver formatt
 				Time throttlePeriod = null;
 				IAction action = null;
 				string @foreach = null;
+				int? maxIterations = null;
 				TransformContainer transform = null;
 				ConditionContainer condition = null;
 
@@ -190,6 +201,9 @@ public Actions Deserialize(ref JsonReader reader, IJsonFormatterResolver formatt
 								condition = formatterResolver.GetFormatter<ConditionContainer>()
 									.Deserialize(ref reader, formatterResolver);
 								break;
+							case 10:
+								maxIterations = reader.ReadInt32();
+								break;
 						}
 					}
 					else
@@ -201,6 +215,7 @@ public Actions Deserialize(ref JsonReader reader, IJsonFormatterResolver formatt
 					action.Name = name;
 					action.ThrottlePeriod = throttlePeriod;
 					action.Foreach = @foreach;
+					action.MaxIterations = maxIterations;
 					action.Transform = transform;
 					action.Condition = condition;
 					dictionary.Add(name, action);
@@ -238,6 +253,12 @@ public void Serialize(ref JsonWriter writer, Actions value, IJsonFormatterResolv
 						writer.WriteString(action.Foreach);
 						writer.WriteValueSeparator();
 					}
+					if (action.MaxIterations.HasValue)
+					{
+						writer.WritePropertyName("max_iterations");
+						writer.WriteInt32(action.MaxIterations.Value);
+						writer.WriteValueSeparator();
+					}
 
 					if (action.Transform != null)
 					{

src/Nest/XPack/Watcher/Action/ActionsDescriptorBase.cs

@@ -32,6 +32,7 @@ string IAction.Name
 		TransformContainer IAction.Transform { get; set; }
 		ConditionContainer IAction.Condition { get; set; }
 		string IAction.Foreach { get; set; }
+		int? IAction.MaxIterations { get; set; }
 
 		/// <inheritdoc cref="IAction.Transform"/>
 		public TDescriptor Transform(Func<TransformDescriptor, TransformContainer> selector) =>
@@ -46,5 +47,8 @@ public TDescriptor Condition(Func<ConditionDescriptor, ConditionContainer> selec
 
 		/// <inheritdoc cref="IAction.Foreach"/>
 		public TDescriptor Foreach(string @foreach) => Assign(@foreach, (a, v) => a.Foreach = v);
+
+		/// <inheritdoc cref="IAction.MaxIterations"/>
+		public TDescriptor MaxIterations(int maxIterations) => Assign(maxIterations, (a,v) => a.MaxIterations = v);
 	}
 }

src/Tests/Tests/XPack/Watcher/PutWatch/PutWatchApiTests.cs

@@ -712,7 +712,7 @@ protected override void ExpectResponse(PutWatchResponse response)
 		}
 	}
 
-	[SkipVersion("<7.3.0", "Foreach introduced in 7.3.0")]
+	[SkipVersion("<7.4.0", "Foreach introduced in 7.3.0, max iterations in 7.4.0")]
 	public class PutWatchApiWithForeachTests : ApiIntegrationTestBase<XPackCluster, PutWatchResponse, IPutWatchRequest, PutWatchDescriptor, PutWatchRequest>
 	{
 		public PutWatchApiWithForeachTests(XPackCluster cluster, EndpointUsage usage) : base(cluster, usage) { }
@@ -757,6 +757,7 @@ public PutWatchApiWithForeachTests(XPackCluster cluster, EndpointUsage usage) :
 					log_hits = new
 					{
 						@foreach = "ctx.payload.hits.hits",
+						max_iterations = 500,
 						logging = new
 						{
 							text = "Found id {{ctx.payload._id}} with field {{ctx.payload._source.numberOfCommits}}"
@@ -804,6 +805,7 @@ public PutWatchApiWithForeachTests(XPackCluster cluster, EndpointUsage usage) :
 			.Actions(a => a
 				.Logging("log_hits", i => i
 					.Foreach("ctx.payload.hits.hits")
+					.MaxIterations(500)
 					.Text("Found id {{ctx.payload._id}} with field {{ctx.payload._source.numberOfCommits}}")
 					.Transform(t => t
 						.Script(st =>st
@@ -841,6 +843,7 @@ public PutWatchApiWithForeachTests(XPackCluster cluster, EndpointUsage usage) :
 				Actions = new LoggingAction("log_hits")
 				{
 					Foreach = "ctx.payload.hits.hits",
+					MaxIterations = 500,
 					Text = "Found id {{ctx.payload._id}} with field {{ctx.payload._source.numberOfCommits}}",
 					Transform = new InlineScriptTransform("return [ 'time' : ctx.trigger.scheduled_time ]"),
 					Condition = new AlwaysCondition()