Repository: kestra-io/kestra (java)

Issue:
Task "Count" could take an input "Namespaces" 
### Feature description

**Feature Request**: Extend io.kestra.core.tasks.executions.counts to Monitor by Namespaces

**Background:**
The current implementation of the io.kestra.core.tasks.executions.counts task provides execution counts for a list of flows. This can be used to send alerts if specific conditions about execution counts are met. While monitoring based on specific flow names is useful, there are use cases where one might want to monitor execution counts across an entire namespace.

**Feature:**
Extend the io.kestra.core.tasks.executions.counts task to accept a list of Namespaces as input, in addition to the existing Flows. If a namespace is provided, the task should monitor the counts for all the flows within that namespace.

**Input Parameters:**
Flows (Existing): A list of flow IDs to monitor.
Namespaces (New): A list of namespaces to monitor. If a namespace is provided, all the flows within that namespace should be monitored.

**Output:**
The output should remain consistent with the current format, detailing the flows that met the specified condition. If the flows from a namespace meet the condition, those specific flow details should be included in the output.

**Use Case:**
By allowing monitoring based on namespaces, users can keep a more holistic check on execution counts, especially useful when managing a large number of flows grouped under a specific namespace. This ensures better observability and quicker response times in case of anomalies.

Example:
Given a scenario where we want to monitor the execution counts for all flows under the payment and shipping namespaces:
```
io.kestra.core.tasks.executions.counts:
  Namespaces:
    - x.y.z
    - a.b.c
```


Note : Disabled flow should not be taking in consideration


Pull request description:
feat(core): Added `namespaces` property to the Count task
close #2163 

Patch:
diff --git a/core/src/main/java/io/kestra/core/models/executions/statistics/ExecutionCount.java b/core/src/main/java/io/kestra/core/models/executions/statistics/ExecutionCount.java
index 9521855a4..6209712d0 100644
--- a/core/src/main/java/io/kestra/core/models/executions/statistics/ExecutionCount.java
+++ b/core/src/main/java/io/kestra/core/models/executions/statistics/ExecutionCount.java
@@ -9,7 +9,6 @@ public class ExecutionCount {
     @NotNull
     String namespace;
 
-    @NotNull
     String flowId;
 
     @NotNull
diff --git a/core/src/main/java/io/kestra/core/repositories/ExecutionRepositoryInterface.java b/core/src/main/java/io/kestra/core/repositories/ExecutionRepositoryInterface.java
index 7c6d29821..e72b34790 100644
--- a/core/src/main/java/io/kestra/core/repositories/ExecutionRepositoryInterface.java
+++ b/core/src/main/java/io/kestra/core/repositories/ExecutionRepositoryInterface.java
@@ -167,11 +167,11 @@ public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Ex
 
     List<ExecutionCount> executionCounts(
         @Nullable String tenantId,
-        List<Flow> flows,
+        @Nullable List<Flow> flows,
         @Nullable List<State.Type> states,
         @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate
-    );
+        @Nullable ZonedDateTime endDate,
+        @Nullable List<String> namespaces);
 
     Execution save(Execution execution);
 
diff --git a/core/src/main/java/io/kestra/plugin/core/execution/Count.java b/core/src/main/java/io/kestra/plugin/core/execution/Count.java
index b34179d11..2d9fa4686 100644
--- a/core/src/main/java/io/kestra/plugin/core/execution/Count.java
+++ b/core/src/main/java/io/kestra/plugin/core/execution/Count.java
@@ -14,15 +14,13 @@ import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.services.FlowService;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 import org.slf4j.Logger;
 
 import java.time.ZonedDateTime;
 import java.util.List;
-import java.util.stream.Collectors;
-import jakarta.validation.constraints.NotEmpty;
-import jakarta.validation.constraints.NotNull;
 
 import static io.kestra.core.utils.Rethrow.throwPredicate;
 
@@ -76,10 +74,9 @@ import static io.kestra.core.utils.Rethrow.throwPredicate;
     aliases = "io.kestra.core.tasks.executions.Counts"
 )
 public class Count extends Task implements RunnableTask<Count.Output> {
-    @NotNull
-    @NotEmpty
     @Schema(
-        title = "A list of flows to be filtered."
+        title = "A list of flows to be filtered.",
+        description = "If not provided, namespaces must be set."
     )
     @PluginProperty
     protected List<Flow> flows;
@@ -114,6 +111,9 @@ public class Count extends Task implements RunnableTask<Count.Output> {
     @PluginProperty(dynamic = true)
     protected String expression;
 
+    @PluginProperty
+    protected List<String> namespaces;
+
     @Override
     public Output run(RunContext runContext) throws Exception {
         Logger logger = runContext.logger();
@@ -121,18 +121,28 @@ public class Count extends Task implements RunnableTask<Count.Output> {
             .getApplicationContext()
             .getBean(ExecutionRepositoryInterface.class);
 
+        if (flows == null && namespaces == null) {
+            throw new IllegalArgumentException("You must provide a list of flows or namespaces");
+        }
+
         var flowInfo = runContext.flowInfo();
 
         // check that all flows are allowed
         FlowService flowService = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowService.class);
-        flows.forEach(flow -> flowService.checkAllowedNamespace(flowInfo.tenantId(), flow.getNamespace(), flowInfo.tenantId(), flowInfo.namespace()));
+        if (flows != null) {
+            flows.forEach(flow -> flowService.checkAllowedNamespace(flowInfo.tenantId(), flow.getNamespace(), flowInfo.tenantId(), flowInfo.namespace()));
+        }
+        if (namespaces != null) {
+            namespaces.forEach(namespace -> flowService.checkAllowedNamespace(flowInfo.tenantId(), namespace, flowInfo.tenantId(), flowInfo.namespace()));
+        }
 
         List<ExecutionCount> executionCounts = executionRepository.executionCounts(
             flowInfo.tenantId(),
             flows,
             this.states,
             startDate != null ? ZonedDateTime.parse(runContext.render(startDate)) : null,
-            endDate != null ? ZonedDateTime.parse(runContext.render(endDate)) : null
+            endDate != null ? ZonedDateTime.parse(runContext.render(endDate)) : null,
+            namespaces
         );
 
         logger.trace("{} flows matching filters", executionCounts.size());
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
index c4790dbc4..51bb24a7a 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcExecutionRepository.java
@@ -18,6 +18,7 @@ import io.kestra.core.repositories.ExecutionRepositoryInterface;
 import io.kestra.core.runners.Executor;
 import io.kestra.core.runners.ExecutorState;
 import io.kestra.core.utils.DateUtils;
+import io.kestra.core.utils.ListUtils;
 import io.kestra.core.utils.NamespaceUtils;
 import io.kestra.jdbc.runner.AbstractJdbcExecutorStateStorage;
 import io.kestra.jdbc.runner.JdbcIndexerInterface;
@@ -99,7 +100,9 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
         return false;
     }
 
-    /** {@inheritDoc} **/
+    /**
+     * {@inheritDoc}
+     **/
     @Override
     public Flux<Execution> findAllByTriggerExecutionId(String tenantId,
                                                        String triggerExecutionId) {
@@ -804,8 +807,8 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
         List<Flow> flows,
         @Nullable List<State.Type> states,
         @Nullable ZonedDateTime startDate,
-        @Nullable ZonedDateTime endDate
-    ) {
+        @Nullable ZonedDateTime endDate,
+        @Nullable List<String> namespaces) {
         ZonedDateTime finalStartDate = startDate == null ? ZonedDateTime.now().minusDays(30) : startDate;
         ZonedDateTime finalEndDate = endDate == null ? ZonedDateTime.now() : endDate;
 
@@ -835,16 +838,24 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
                     select = select.and(this.statesFilter(states));
                 }
 
-                // add flow & namespace filters
-                select = select.and(DSL.or(
-                    flows
+                List<Condition> orConditions = new ArrayList<>();
+                orConditions.addAll(ListUtils.emptyOnNull(flows)
+                    .stream()
+                    .map(flow -> DSL.and(
+                        field("namespace").eq(flow.getNamespace()),
+                        field("flow_id").eq(flow.getFlowId())
+                    ))
+                    .toList());
+
+                orConditions.addAll(
+                    ListUtils.emptyOnNull(namespaces)
                         .stream()
-                        .map(flow -> DSL.and(
-                            field("namespace").eq(flow.getNamespace()),
-                            field("flow_id").eq(flow.getFlowId())
-                        ))
+                        .map(np -> field("namespace").eq(np))
                         .toList()
-                ));
+                );
+
+                // add flows filters
+                select = select.and(DSL.or(orConditions));
 
                 // map result to flow
                 return select
@@ -865,22 +876,40 @@ public abstract class AbstractJdbcExecutionRepository extends AbstractJdbcReposi
                     .toList();
             });
 
+        List<ExecutionCount> counts = new ArrayList<>();
         // fill missing with count at 0
-        return flows
-            .stream()
-            .map(flow -> result
+        if (flows != null) {
+            counts.addAll(flows
                 .stream()
-                .filter(executionCount -> executionCount.getNamespace().equals(flow.getNamespace()) &&
-                    executionCount.getFlowId().equals(flow.getFlowId())
+                .map(flow -> result
+                    .stream()
+                    .filter(executionCount -> executionCount.getNamespace().equals(flow.getNamespace()) &&
+                        executionCount.getFlowId().equals(flow.getFlowId())
+                    )
+                    .findFirst()
+                    .orElse(new ExecutionCount(
+                        flow.getNamespace(),
+                        flow.getFlowId(),
+                        0L
+                    ))
                 )
-                .findFirst()
-                .orElse(new ExecutionCount(
-                    flow.getNamespace(),
-                    flow.getFlowId(),
-                    0L
-                ))
-            )
-            .toList();
+                .toList());
+        }
+
+        if (namespaces != null) {
+            Map<String, Long> groupedByNamespace = result.stream()
+                .collect(Collectors.groupingBy(
+                    ExecutionCount::getNamespace,
+                    Collectors.summingLong(ExecutionCount::getCount)
+                ));
+
+            counts.addAll(groupedByNamespace.entrySet()
+                .stream()
+                .map(entry -> new ExecutionCount(entry.getKey(), null, entry.getValue()))
+                .toList());
+        }
+
+        return counts;
     }
 
     public List<Execution> lastExecutions(
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
index 86a7a1498..7c78e094b 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -381,6 +381,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                         List.of(new io.kestra.core.models.executions.statistics.Flow(flow.getNamespace(), flow.getId())),
                         List.of(State.Type.RUNNING, State.Type.PAUSED),
                         null,
+                        null,
                         null
                     ).getFirst();
 


Test patch:
diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java
index 87915ee72..1fbfba06c 100644
--- a/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java
+++ b/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java
@@ -591,7 +591,8 @@ public abstract class AbstractExecutionRepositoryTest {
             ),
             null,
             ZonedDateTime.now().minusDays(10),
-            ZonedDateTime.now()
+            ZonedDateTime.now(),
+            null
         );
         assertThat(result.size(), is(4));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("first")).findFirst().get().getCount(), is(2L));
@@ -608,12 +609,24 @@ public abstract class AbstractExecutionRepositoryTest {
             ),
             List.of(State.Type.SUCCESS),
             null,
+            null,
             null
         );
         assertThat(result.size(), is(3));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("first")).findFirst().get().getCount(), is(2L));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("second")).findFirst().get().getCount(), is(3L));
         assertThat(result.stream().filter(executionCount -> executionCount.getFlowId().equals("third")).findFirst().get().getCount(), is(9L));
+
+        result = executionRepository.executionCounts(
+            null,
+            null,
+            null,
+            null,
+            null,
+            List.of(NAMESPACE)
+        );
+        assertThat(result.size(), is(1));
+        assertThat(result.stream().filter(executionCount -> executionCount.getNamespace().equals(NAMESPACE)).findFirst().get().getCount(), is(14L));
     }
 
     @Test