Repository: kestra-io/kestra
Pull request #6683 (2025-01-08 09:59:36)

feat(core): add log record serialization
<!-- Thanks for submitting a Pull Request to Kestra. To help us review your contribution, please follow the guidelines below:

- Make sure that your commits follow the [conventional commits](https://www.conventionalcommits.org/en/v1.0.0/) specification e.g. `feat(ui): add a new navigation menu item` or `fix(core): fix a bug in the core model` or `docs: update the README.md`. This will help us automatically generate the changelog.
- The title should briefly summarize the proposed changes.
- Provide a short overview of the change and the value it adds.
- Share a flow example to help the reviewer understand and QA the change.
- Use "closes" to automatically close an issue. For example, `closes #1234` will close issue #1234. -->

### What changes are being made and why?

<!-- Please include a brief summary of the changes included in this PR e.g. closes #1234. -->

---

### How the changes have been QAed?

<!-- Include example code that shows how this PR has been QAed. The code should present a complete yet easily reproducible flow.

```yaml
# Your example flow code here
```

Note that this is not a replacement for unit tests but rather a way to demonstrate how the changes work in a real-life scenario, as the end-user would experience them.

Remove this section if this change applies to all flows or to the documentation only. -->

---

### Setup Instructions

<!--If there are any setup requirements like API keys or trial accounts, kindly include brief bullet-points-description outlining the setup process below.

- [External System Documentation](URL)
- Steps to set up the necessary resources

If there are no setup requirements, you can remove this section.

Thank you for your contribution. ❤️  -->


sonarqubecloud[bot]: ## [![Quality Gate Passed](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/QualityGateBadge/qg-passed-20px.png 'Quality Gate Passed')](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=6683) **Quality Gate passed**  
Issues  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [2 New issues](https://sonarcloud.io/project/issues?id=kestra-io_kestra&pullRequest=6683&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/accepted-16px.png '') [0 Accepted issues](https://sonarcloud.io/project/issues?id=kestra-io_kestra&pullRequest=6683&issueStatuses=ACCEPTED)

Measures  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=kestra-io_kestra&pullRequest=6683&issueStatuses=OPEN,CONFIRMED&sinceLeakPeriod=true)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [95.5% Coverage on New Code](https://sonarcloud.io/component_measures?id=kestra-io_kestra&pullRequest=6683&metric=new_coverage&view=list)  
![](https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/passed-16px.png '') [0.0% Duplication on New Code](https://sonarcloud.io/component_measures?id=kestra-io_kestra&pullRequest=6683&metric=new_duplicated_lines_density&view=list)  
  
[See analysis details on SonarQube Cloud](https://sonarcloud.io/dashboard?id=kestra-io_kestra&pullRequest=6683)



Patch:
diff --git a/core/src/main/java/io/kestra/core/models/executions/LogEntry.java b/core/src/main/java/io/kestra/core/models/executions/LogEntry.java
index 5e8530eb8..acaaa16b7 100644
--- a/core/src/main/java/io/kestra/core/models/executions/LogEntry.java
+++ b/core/src/main/java/io/kestra/core/models/executions/LogEntry.java
@@ -1,9 +1,6 @@
 package io.kestra.core.models.executions;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
-import com.fasterxml.jackson.annotation.JsonView;
-import com.fasterxml.jackson.databind.MapperFeature;
-import com.fasterxml.jackson.databind.json.JsonMapper;
 import io.kestra.core.models.DeletedInterface;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.flows.Flow;
@@ -12,7 +9,6 @@ import io.kestra.core.models.triggers.TriggerContext;
 import io.micronaut.core.annotation.Nullable;
 import io.swagger.v3.oas.annotations.Hidden;
 import lombok.Builder;
-import lombok.SneakyThrows;
 import lombok.Value;
 import org.slf4j.event.Level;
 
@@ -28,46 +24,36 @@ import java.util.stream.Stream;
 public class LogEntry implements DeletedInterface, TenantInterface {
     @Hidden
     @Pattern(regexp = "^[a-z0-9][a-z0-9_-]*")
-    @JsonView(MessageView.class)
     String tenantId;
 
     @NotNull
-    @JsonView(MessageView.class)
     String namespace;
 
     @NotNull
-    @JsonView(MessageView.class)
     String flowId;
 
     @Nullable
-    @JsonView(MessageView.class)
     String taskId;
 
     @Nullable
-    @JsonView(MessageView.class)
     String executionId;
 
     @Nullable
-    @JsonView(MessageView.class)
     String taskRunId;
 
     @Nullable
     @JsonInclude
-    @JsonView(MessageView.class)
     Integer attemptNumber;
 
     @Nullable
-    @JsonView(MessageView.class)
     String triggerId;
 
     Instant timestamp;
 
     Level level;
 
-    @JsonView(MessageView.class)
     String thread;
 
-    @JsonView(MessageView.class)
     String message;
 
     @NotNull
@@ -142,16 +128,12 @@ public class LogEntry implements DeletedInterface, TenantInterface {
             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
 
-    @SneakyThrows
-    public String toJson(){
-        JsonMapper mapper = JsonMapper.builder()
-            .configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false)
-            .serializationInclusion(JsonInclude.Include.ALWAYS)
-            .build();
-        return mapper.writerWithView(MessageView.class)
-            .writeValueAsString(this);
+    public Map<String, String> toLogMap() {
+        Map<String, String> map = this.toMap();
+        map.put("attemptNumber", this.attemptNumber != null ? String.valueOf(this.attemptNumber) : null);
+        map.put("thread", this.thread);
+        map.put("message", this.message);
+        return map;
     }
 
-    public static class MessageView{}
-
 }
diff --git a/core/src/main/java/io/kestra/core/models/tasks/logs/LogRecord.java b/core/src/main/java/io/kestra/core/models/tasks/logs/LogRecord.java
index e764c3217..8f63ac947 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/logs/LogRecord.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/logs/LogRecord.java
@@ -1,6 +1,8 @@
 package io.kestra.core.models.tasks.logs;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.KeyValue;
 import io.opentelemetry.api.common.Value;
 import io.opentelemetry.api.logs.Severity;
 import io.opentelemetry.api.trace.SpanContext;
@@ -8,6 +10,7 @@ import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
 import io.opentelemetry.sdk.logs.data.Body;
 import io.opentelemetry.sdk.logs.data.LogRecordData;
 import io.opentelemetry.sdk.resources.Resource;
+import java.util.List;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Getter;
@@ -26,11 +29,13 @@ public class LogRecord implements LogRecordData {
     SpanContext spanContext;
     Severity severity;
     String severityText;
-    Value<String> bodyValue;
     Attributes attributes;
     int totalAttributeCount;
+    Value<List<KeyValue>> bodyValue;
 
+    @JsonIgnore
     public Body getBody(){
         throw new UnsupportedOperationException();
     }
+
 }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/models/executions/LogEntryTest.java b/core/src/test/java/io/kestra/core/models/executions/LogEntryTest.java
index b77c49b65..4934e9b7e 100644
--- a/core/src/test/java/io/kestra/core/models/executions/LogEntryTest.java
+++ b/core/src/test/java/io/kestra/core/models/executions/LogEntryTest.java
@@ -3,12 +3,13 @@ package io.kestra.core.models.executions;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 
+import java.util.Map;
 import org.junit.jupiter.api.Test;
 
 public class LogEntryTest {
 
     @Test
-    public void should_format_to_json_full_log_entry(){
+    public void should_format_to_log_map(){
         LogEntry logEntry = LogEntry.builder()
             .tenantId("tenantId")
             .namespace("namespace")
@@ -21,15 +22,24 @@ public class LogEntryTest {
             .thread("thread")
             .message("message")
             .build();
-        assertThat(logEntry.toJson(), is("{\"tenantId\":\"tenantId\",\"namespace\":\"namespace\",\"flowId\":\"flowId\",\"taskId\":\"taskId\",\"executionId\":\"executionId\",\"taskRunId\":\"taskRunId\",\"attemptNumber\":1,\"triggerId\":\"triggerId\",\"thread\":\"thread\",\"message\":\"message\"}"));
+        Map<String, String> logMap = logEntry.toLogMap();
+        assertThat(logMap.get("tenantId"), is("tenantId"));
+        assertThat(logMap.get("namespace"), is("namespace"));
+        assertThat(logMap.get("flowId"), is("flowId"));
+        assertThat(logMap.get("taskId"), is("taskId"));
+        assertThat(logMap.get("executionId"), is("executionId"));
+        assertThat(logMap.get("taskRunId"), is("taskRunId"));
+        assertThat(logMap.get("attemptNumber"), is("1"));
+        assertThat(logMap.get("triggerId"), is("triggerId"));
+        assertThat(logMap.get("thread"), is("thread"));
+        assertThat(logMap.get("message"), is("message"));
     }
 
     @Test
-    public void should_format_to_json_empty_log_entry(){
+    public void should_format_to_log_map_empty_log_entry(){
         LogEntry logEntry = LogEntry.builder().build();
-        assertThat(logEntry.toJson(), is("{\"tenantId\":null,\"namespace\":null,\"flowId\":null,\"taskId\":null,\"executionId\":null,\"taskRunId\":null,\"attemptNumber\":null,\"triggerId\":null,\"thread\":null,\"message\":null}"));
+        Map<String, String> map = logEntry.toMap();
+        assertThat(map.size(), is(0));
     }
 
-
-
 }
diff --git a/core/src/test/java/io/kestra/core/models/tasks/logs/LogRecordTest.java b/core/src/test/java/io/kestra/core/models/tasks/logs/LogRecordTest.java
new file mode 100644
index 000000000..1f22cdcaf
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/models/tasks/logs/LogRecordTest.java
@@ -0,0 +1,49 @@
+package io.kestra.core.models.tasks.logs;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.kestra.core.models.executions.LogEntry;
+import io.opentelemetry.api.common.Value;
+import io.opentelemetry.api.logs.Severity;
+import java.time.Instant;
+import java.util.Map.Entry;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+import org.slf4j.event.Level;
+
+public class LogRecordTest {
+
+    @Test
+    void should_convert_log_record_to_string() throws JsonProcessingException {
+        LogRecord logRecord = LogRecord.builder()
+            .timestampEpochNanos(1322907330123456789L)
+            .observedTimestampEpochNanos(1322907330123456789L)
+            .severity(Severity.ERROR)
+            .severityText("ERROR")
+            .bodyValue(Value.of(
+                LogEntry.builder()
+                    .tenantId("tenantId")
+                    .namespace("namespace")
+                    .flowId("flowId")
+                    .taskId("taskId")
+                    .executionId("executionId")
+                    .taskRunId("taskRunId")
+                    .attemptNumber(1)
+                    .triggerId("triggerId")
+                    .timestamp(Instant.parse("2011-12-03T10:15:30.123456789Z"))
+                    .level(Level.INFO)
+                    .thread("thread")
+                    .message("message")
+                    .build()
+                    .toLogMap()
+                    .entrySet().stream()
+                    .collect(Collectors.toMap(Entry::getKey, entry -> Value.of(entry.getValue())))
+            ))
+            .build();
+        String log = new ObjectMapper().writeValueAsString(logRecord);
+        assertThat(log, is("{\"resource\":null,\"instrumentationScopeInfo\":null,\"timestampEpochNanos\":1322907330123456789,\"observedTimestampEpochNanos\":1322907330123456789,\"spanContext\":null,\"severity\":\"ERROR\",\"severityText\":\"ERROR\",\"attributes\":null,\"totalAttributeCount\":0,\"bodyValue\":{\"value\":[{\"key\":\"taskRunId\",\"value\":{\"value\":\"taskRunId\",\"type\":\"STRING\"}},{\"key\":\"attemptNumber\",\"value\":{\"value\":\"1\",\"type\":\"STRING\"}},{\"key\":\"executionId\",\"value\":{\"value\":\"executionId\",\"type\":\"STRING\"}},{\"key\":\"triggerId\",\"value\":{\"value\":\"triggerId\",\"type\":\"STRING\"}},{\"key\":\"tenantId\",\"value\":{\"value\":\"tenantId\",\"type\":\"STRING\"}},{\"key\":\"namespace\",\"value\":{\"value\":\"namespace\",\"type\":\"STRING\"}},{\"key\":\"thread\",\"value\":{\"value\":\"thread\",\"type\":\"STRING\"}},{\"key\":\"message\",\"value\":{\"value\":\"message\",\"type\":\"STRING\"}},{\"key\":\"flowId\",\"value\":{\"value\":\"flowId\",\"type\":\"STRING\"}},{\"key\":\"taskId\",\"value\":{\"value\":\"taskId\",\"type\":\"STRING\"}}],\"type\":\"KEY_VALUE_LIST\"}}"));
+    }
+}