You could add a `logLevel` on task and trigger that will limit the output of every log emitted by the task, by default, we will keep the `TRACE` level
Add `logLevel` Enum property on the core task
### Feature description

Hi,

Usually, all tasks are generating logs. These logs are collected by Kestra to be display in the UI. By default, this is the ideal behavior
But sometimes we don't want that logs are collected for example when we have

- huge log size (that use a lot of space in the Kestra DB for nothing and  that are not readable in Kestra UI)
- sensible information in the logs

So it would be interesting if we should be able to disable the collect of the log by Kestra.
core/src/main/java/io/kestra/core/models/executions/TaskRun.java core/src/main/java/io/kestra/core/models/tasks/Task.java core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java core/src/main/java/io/kestra/core/runners/RunContext.java core/src/main/java/io/kestra/core/runners/RunContextLogger.java
core/src/main/java/io/kestra/core/models/executions/TaskRun.java core/src/main/java/io/kestra/core/models/tasks/Task.java core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java core/src/main/java/io/kestra/core/runners/RunContext.java core/src/main/java/io/kestra/core/runners/RunContextLogger.java
core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java core/src/test/java/io/kestra/core/runners/RunContextTest.java core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java core/src/test/resources/flows/valids/logs.yaml jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
diff --git a/core/src/main/java/io/kestra/core/models/executions/TaskRun.java b/core/src/main/java/io/kestra/core/models/executions/TaskRun.java
index 43abe06476..5f61bdb787 100644
--- a/core/src/main/java/io/kestra/core/models/executions/TaskRun.java
+++ b/core/src/main/java/io/kestra/core/models/executions/TaskRun.java
@@ -9,6 +9,7 @@
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.utils.IdUtils;
+import org.slf4j.event.Level;
 
 import java.util.ArrayList;
 import java.util.Collections;
diff --git a/core/src/main/java/io/kestra/core/models/tasks/Task.java b/core/src/main/java/io/kestra/core/models/tasks/Task.java
index fc64c825f7..a571cea4ec 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/Task.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/Task.java
@@ -13,6 +13,7 @@
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
+import org.slf4j.event.Level;
 
 import java.time.Duration;
 import java.util.Optional;
@@ -53,6 +54,8 @@ abstract public class Task {
     @Valid
     private WorkerGroup workerGroup;
 
+    private Level logLevel;
+
     public Optional<Task> findById(String id) {
         if (this.getId().equals(id)) {
             return Optional.of(this);
diff --git a/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java b/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
index caf232712f..b7c917b4de 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java
@@ -11,6 +11,7 @@
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
+import org.slf4j.event.Level;
 
 import java.util.List;
 import javax.validation.Valid;
@@ -52,4 +53,6 @@ abstract public class AbstractTrigger {
 
     @Valid
     private WorkerGroup workerGroup;
+
+    private Level minLogLevel;
 }
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
index 51cf36ad28..5dbd5577de 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContext.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -83,7 +83,7 @@ public RunContext(ApplicationContext applicationContext, Flow flow, Execution ex
     public RunContext(ApplicationContext applicationContext, Flow flow, Task task, Execution execution, TaskRun taskRun) {
         this.initBean(applicationContext);
         this.initContext(flow, task, execution, taskRun);
-        this.initLogger(taskRun);
+        this.initLogger(taskRun, task);
     }
 
     /**
@@ -136,13 +136,14 @@ private void initContext(Flow flow, Task task, Execution execution, TaskRun task
     }
 
     @SuppressWarnings("unchecked")
-    private void initLogger(TaskRun taskRun) {
+    private void initLogger(TaskRun taskRun, Task task) {
         this.runContextLogger = new RunContextLogger(
             applicationContext.findBean(
                 QueueInterface.class,
                 Qualifiers.byName(QueueFactoryInterface.WORKERTASKLOG_NAMED)
             ).orElseThrow(),
-            LogEntry.of(taskRun)
+            LogEntry.of(taskRun),
+            task.getLogLevel()
         );
     }
 
@@ -153,7 +154,8 @@ private void initLogger(Execution execution) {
                 QueueInterface.class,
                 Qualifiers.byName(QueueFactoryInterface.WORKERTASKLOG_NAMED)
             ).orElseThrow(),
-            LogEntry.of(execution)
+            LogEntry.of(execution),
+            null
         );
     }
 
@@ -164,7 +166,8 @@ private void initLogger(TriggerContext triggerContext, AbstractTrigger trigger)
                 QueueInterface.class,
                 Qualifiers.byName(QueueFactoryInterface.WORKERTASKLOG_NAMED)
             ).orElseThrow(),
-            LogEntry.of(triggerContext, trigger)
+            LogEntry.of(triggerContext, trigger),
+            trigger.getMinLogLevel()
         );
     }
 
@@ -175,7 +178,8 @@ private void initLogger(Flow flow, AbstractTrigger trigger) {
                 QueueInterface.class,
                 Qualifiers.byName(QueueFactoryInterface.WORKERTASKLOG_NAMED)
             ).orElseThrow(),
-            LogEntry.of(flow, trigger)
+            LogEntry.of(flow, trigger),
+            trigger.getMinLogLevel()
         );
     }
 
@@ -406,7 +410,7 @@ public RunContext forScheduler(TriggerContext triggerContext, AbstractTrigger tr
 
     public RunContext forWorker(ApplicationContext applicationContext, WorkerTask workerTask) {
         this.initBean(applicationContext);
-        this.initLogger(workerTask.getTaskRun());
+        this.initLogger(workerTask.getTaskRun(), workerTask.getTask());
 
         Map<String, Object> clone = new HashMap<>(this.variables);
 
diff --git a/core/src/main/java/io/kestra/core/runners/RunContextLogger.java b/core/src/main/java/io/kestra/core/runners/RunContextLogger.java
index 3c29b9f678..3cd4e497ee 100644
--- a/core/src/main/java/io/kestra/core/runners/RunContextLogger.java
+++ b/core/src/main/java/io/kestra/core/runners/RunContextLogger.java
@@ -26,13 +26,14 @@ public class RunContextLogger {
     private Logger logger;
     private QueueInterface<LogEntry> logQueue;
     private LogEntry logEntry;
+    private Level loglevel;
 
     @VisibleForTesting
     public RunContextLogger() {
         this.loggerName = "unit-test";
     }
 
-    public RunContextLogger(QueueInterface<LogEntry> logQueue, LogEntry logEntry) {
+    public RunContextLogger(QueueInterface<LogEntry> logQueue, LogEntry logEntry, org.slf4j.event.Level loglevel) {
         if (logEntry.getExecutionId() != null) {
             this.loggerName = "flow." + logEntry.getFlowId() + "." + logEntry.getExecutionId() + (logEntry.getTaskRunId() != null ? "." + logEntry.getTaskRunId() : "");
         } else {
@@ -40,6 +41,7 @@ public RunContextLogger(QueueInterface<LogEntry> logQueue, LogEntry logEntry) {
         }
         this.logQueue = logQueue;
         this.logEntry = logEntry;
+        this.loglevel = loglevel == null ? Level.TRACE : Level.toLevel(loglevel.toString());
     }
 
     private static List<LogEntry> logEntry(ILoggingEvent event, String message, org.slf4j.event.Level level, LogEntry logEntry) {
@@ -138,7 +140,7 @@ public org.slf4j.Logger logger() {
             forwardAppender.start();
             this.logger.addAppender(forwardAppender);
 
-            this.logger.setLevel(Level.TRACE);
+            this.logger.setLevel(this.loglevel);
             this.logger.setAdditive(true);
         }
 

diff --git a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
index 8dae4c0cdb..7b541cfba2 100644
--- a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
+++ b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
@@ -128,7 +128,7 @@ void restartDynamic() throws Exception {
     @Test
     void replayFromBeginning() throws Exception {
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "logs");
-        assertThat(execution.getTaskRunList(), hasSize(3));
+        assertThat(execution.getTaskRunList(), hasSize(4));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
         Execution restart = executionService.replay(execution, null, null);
@@ -148,7 +148,7 @@ void replayFromBeginning() throws Exception {
     @Test
     void replaySimple() throws Exception {
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "logs");
-        assertThat(execution.getTaskRunList(), hasSize(3));
+        assertThat(execution.getTaskRunList(), hasSize(4));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
         Execution restart = executionService.replay(execution, execution.getTaskRunList().get(1).getId(), null);
diff --git a/core/src/test/java/io/kestra/core/runners/RunContextTest.java b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
index 3e8901ac26..076ed9b4a7 100644
--- a/core/src/test/java/io/kestra/core/runners/RunContextTest.java
+++ b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
@@ -58,7 +58,7 @@ void logs() throws TimeoutException {
 
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "logs");
 
-        assertThat(execution.getTaskRunList(), hasSize(3));
+        assertThat(execution.getTaskRunList(), hasSize(4));
 
         matchingLog = TestsUtils.awaitLog(logs, log -> Objects.equals(log.getTaskRunId(), execution.getTaskRunList().get(0).getId()));
         assertThat(matchingLog, notNullValue());
@@ -74,6 +74,9 @@ void logs() throws TimeoutException {
         assertThat(matchingLog, notNullValue());
         assertThat(matchingLog.getLevel(), is(Level.ERROR));
         assertThat(matchingLog.getMessage(), is("third logs"));
+
+        matchingLog = TestsUtils.awaitLog(logs, log -> Objects.equals(log.getTaskRunId(), execution.getTaskRunList().get(3).getId()));
+        assertThat(matchingLog, nullValue());
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java b/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
index 2b5d36caa9..6e7f44ca2f 100644
--- a/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
+++ b/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
@@ -212,7 +212,7 @@ void invalidProperty() {
             () -> this.parse("flows/invalids/invalid-property.yaml")
         );
 
-        assertThat(exception.getMessage(), is("Unrecognized field \"invalid\" (class io.kestra.core.tasks.debugs.Return), not marked as ignorable (8 known properties: \"timeout\", \"format\", \"retry\", \"type\", \"id\", \"description\", \"workerGroup\", \"disabled\"])"));
+        assertThat(exception.getMessage(), is("Unrecognized field \"invalid\" (class io.kestra.core.tasks.debugs.Return), not marked as ignorable (9 known properties: \"logLevel\", \"timeout\", \"format\", \"retry\", \"type\", \"id\", \"description\", \"workerGroup\", \"disabled\"])"));
         assertThat(exception.getConstraintViolations().size(), is(1));
         assertThat(exception.getConstraintViolations().iterator().next().getPropertyPath().toString(), is("io.kestra.core.models.flows.Flow[\"tasks\"]->java.util.ArrayList[0]->io.kestra.core.tasks.debugs.Return[\"invalid\"]"));
     }
diff --git a/core/src/test/resources/flows/valids/logs.yaml b/core/src/test/resources/flows/valids/logs.yaml
index 139c3d5a9f..c14322e777 100644
--- a/core/src/test/resources/flows/valids/logs.yaml
+++ b/core/src/test/resources/flows/valids/logs.yaml
@@ -23,3 +23,8 @@ tasks:
   type: io.kestra.core.tasks.log.Log
   message: third {{flow.id}}
   level: ERROR
+- id: t4
+  type: io.kestra.core.tasks.log.Log
+  message: fourth {{task.id}}
+  level: TRACE
+  logLevel: INFO
\ No newline at end of file
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
index c31275faf4..ccdbd13d6a 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -102,7 +102,7 @@ void full() throws TimeoutException, QueueException {
     void logs() throws TimeoutException {
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "logs", null, null, Duration.ofSeconds(60));
 
-        assertThat(execution.getTaskRunList(), hasSize(3));
+        assertThat(execution.getTaskRunList(), hasSize(4));
     }
 
     @Test
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
index b325ed2780..d3fac39c0a 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/PluginControllerTest.java
@@ -138,7 +138,7 @@ void taskWithBase() throws URISyntaxException {
             Map<String, Map<String, Object>> properties = (Map<String, Map<String, Object>>) doc.getSchema().getProperties().get("properties");
 
             assertThat(doc.getMarkdown(), containsString("io.kestra.plugin.templates.ExampleTask"));
-            assertThat(properties.size(), is(11));
+            assertThat(properties.size(), is(12));
             assertThat(properties.get("id").size(), is(4));
             assertThat(((Map<String, Object>) doc.getSchema().getOutputs().get("properties")).size(), is(1));
         });
