Repository: kestra-io/kestra (java)

Issue:
Add a pebble function `fileSize()` that returns the file size in bytes
### Feature description

This is needed for use cases when you need to check the file size within a condition and perhaps skip some downstream tasks if the file is empty.

This can be a function that takes an internal storage file URI as a single input parameter: `fileSize(URI)`

Pull request description:
feat(core): Implement a pebble function fileSize() that returns the file size in bytes
<!-- 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. -->

### Closes #5902 

### What changes are being made and why?

<!-- Please include a brief summary of the changes included in this PR e.g. closes #1234. -->
---

The new pebble function is created to return the file size that takes the internal storage URI. The implementation followed some logic from the `read()` pebble function regarding the way it gets the files and handles the cases around current and parent execution. The only difference is that the **fileSize** function is implemented to handle **only** the case when the path is Internal Storage URI, meaning it starts with "kestra:///".  The namespace file is not covered.


### 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. -->

---
I tested this feature through unit tests. I am open to suggestions on how to test it further.


Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/Extension.java b/core/src/main/java/io/kestra/core/runners/pebble/Extension.java
index 24a94d156..3be71f939 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/Extension.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/Extension.java
@@ -41,6 +41,9 @@ public class Extension extends AbstractExtension {
     @Nullable
     private RenderOnceFunction renderOnceFunction;
 
+    @Inject
+    private FileSizeFunction fileSizeFunction;
+
     @Override
     public List<TokenParser> getTokenParsers() {
         return null;
@@ -131,7 +134,7 @@ public class Extension extends AbstractExtension {
         functions.put("yaml", new YamlFunction());
         functions.put("printContext", new PrintContextFunction());
         functions.put("fromIon", new FromIonFunction());
-
+        functions.put("fileSize", fileSizeFunction);
         return functions;
     }
 
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/FileSizeFunction.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/FileSizeFunction.java
new file mode 100644
index 000000000..d55312ac3
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/FileSizeFunction.java
@@ -0,0 +1,101 @@
+package io.kestra.core.runners.pebble.functions;
+
+import io.kestra.core.storages.FileAttributes;
+import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.utils.Slugify;
+import io.pebbletemplates.pebble.error.PebbleException;
+import io.pebbletemplates.pebble.template.EvaluationContext;
+import io.pebbletemplates.pebble.template.PebbleTemplate;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import io.pebbletemplates.pebble.extension.Function;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+
+@Singleton
+public class FileSizeFunction implements Function {
+    private static final String ERROR_MESSAGE = "The 'fileSize' function expects an argument 'path' that is a path to the internal storage URI.";
+    private static final String KESTRA_SCHEME = "kestra:///";
+    private static final String TRIGGER = "trigger";
+    private static final String NAMESPACE = "namespace";
+    private static final String ID  = "id";
+
+    @Inject
+    private StorageInterface storageInterface;
+
+    @Override
+    public List<String> getArgumentNames() {
+        return List.of("path");
+    }
+
+    @Override
+    public Object execute(Map<String, Object> args, PebbleTemplate self, EvaluationContext context, int lineNumber) {
+
+        if (!args.containsKey("path")) {
+            throw new PebbleException(null, ERROR_MESSAGE, lineNumber, self.getName());
+        }
+
+        Object path = args.get("path");
+        URI uri = getUriFromThePath(path, lineNumber, self);
+
+        try {
+            return getFileSizeFromInternalStorageUri(context, uri);
+        } catch (IOException e) {
+            throw new PebbleException(e, e.getMessage(), lineNumber, self.getName());
+        }
+
+    }
+
+    private URI getUriFromThePath(Object path, int lineNumber, PebbleTemplate self) {
+        if (path instanceof URI u) {
+            return u;
+        } else if (path instanceof String str && str.startsWith(KESTRA_SCHEME)) {
+            return URI.create(str);
+        } else {
+            throw new PebbleException(null, "Unable to create the URI from the path " + path, lineNumber, self.getName());
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private long getFileSizeFromInternalStorageUri(EvaluationContext context, URI path) throws IOException {
+        Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
+        Map<String, String> execution = (Map<String, String>) context.getVariable("execution");
+
+        boolean isFileFromCurrentExecution = isFileUriValid(flow.get(NAMESPACE), flow.get(ID), execution.get(ID), path);
+
+        if (!isFileFromCurrentExecution) {
+            checkIfFileFromParentExecution(context, path);
+        }
+
+        FileAttributes fileAttributes = storageInterface.getAttributes(flow.get("tenantId"), path);
+        return fileAttributes.getSize();
+    }
+
+    private void checkIfFileFromParentExecution(EvaluationContext context, URI path) {
+        if (context.getVariable(TRIGGER) != null) {
+            // if there is a trigger of type execution, we also allow accessing a file from the parent execution
+            Map<String, String> trigger = (Map<String, String>) context.getVariable(TRIGGER);
+
+            if (!isFileUriValid(trigger.get(NAMESPACE), trigger.get("flowId"), trigger.get("executionId"), path)) {
+                throw new IllegalArgumentException("Unable to read the file '" + path + "' as it didn't belong to the current execution");
+            }
+        }
+        else {
+            throw new IllegalArgumentException("Unable to read the file '" + path + "' as it didn't belong to the current execution");
+        }
+    }
+
+    private boolean isFileUriValid(String namespace, String flowId, String executionId, URI path) {
+        // Internal storage URI should be: kestra:///$namespace/$flowId/executions/$executionId/tasks/$taskName/$taskRunId/$random.ion or kestra:///$namespace/$flowId/executions/$executionId/trigger/$triggerName/$random.ion
+        // We check that the file is for the given flow execution
+        if (namespace == null || flowId == null || executionId == null) {
+            return false;
+        }
+
+        String authorizedBasePath = KESTRA_SCHEME + namespace.replace(".", "/") + "/" + Slugify.of(flowId) + "/executions/" + executionId + "/";
+        return path.toString().startsWith(authorizedBasePath);
+    }
+}
diff --git a/ui/src/assets/documentations/basic.md b/ui/src/assets/documentations/basic.md
index 41c97874d..c38759700 100644
--- a/ui/src/assets/documentations/basic.md
+++ b/ui/src/assets/documentations/basic.md
@@ -155,11 +155,12 @@ The table below lists common Pebble expressions and functions.
 | `{{ outputs.taskId.outputAttribute }}`                                                             | Accesses task output attribute.                                                                                                 |
 | `{{ error.taskId }}`                                                                               | In case of failure, accesses the task identifier of the last task that fail.                                                    |
 | `{{ error.message }}`                                                                              | In case of failure, accesses the last error message.                                                                            |
-| `{{ error.stackTrace }}`                                                                               | In case of failure, accesses the last error stack trace.                                                                        |
+| `{{ error.stackTrace }}`                                                                           | In case of failure, accesses the last error stack trace.                                                                        |
 | `{{ range(0, 3) }}`                                                                                | Generates a list from 0 to 3.                                                                                                   |
 | `{{ block("post") }}`                                                                              | Renders the contents of the ["post" block](https://kestra.io/docs/concepts/expression/function#block).                          |
 | `{{ currentEachOutput(outputs.first) }}`                                                           | Retrieves the current output of a sibling task.                                                                                 |
 | `{{ fromJson('{"foo": [666, 1, 2]}').foo[0] }}`                                                    | Converts a JSON string to an object and accesses its properties.                                                                |
+| `{{ fromIon(read(someItem)).someField }}`                                                          | Converts a ION string to an object and accesses its properties.                                                                 |
 | `{{ yaml('foo: [666, 1, 2]').foo[0] }}`                                                            | Converts a YAML string to an object and accesses its properties.                                                                |
 | `{{ max(user.age, 80) }}`                                                                          | Returns the largest of its numerical arguments.                                                                                 |
 | `{{ min(user.age, 80) }}`                                                                          | Returns the smallest of its numerical arguments.                                                                                |
@@ -220,7 +221,8 @@ The table below lists Pebble filter expressions:
 | `yaml`           | `{{ myObject \| yaml }}` — Converts `myObject` into a YAML string.                                                               |
 | `indent`         | `{{ "Hello\nworld" \| indent(4) }}` — Adds 4 spaces before each line except the first, resulting in "Hello\n    world".          |
 | `nindent`        | `{{ "Hello\nworld" \| nindent(4) }}` — Adds a newline and then 4 spaces before each line, resulting in "\n    Hello\n    world". |
-| `toJson`           | `{{ myObject \| toJson }}` — Converts `myObject` into a JSON string.                                                               |
+| `toJson`         | `{{ myObject \| toJson }}` — Converts `myObject` into a JSON string.                                                             |
+| `toIon`          | `{{ myObject \| toIon }}` — Converts `myObject` into a ION string.                                                               |
 | `jq`             | `{{ myObject \| jq(".foo") }}` — Applies JQ expression to extract the "foo" property from `myObject`.                            |
 | `length`         | `{{ "Hello" \| length }}` — Returns the length of "Hello", which is 5.                                                           |
 | `merge`          | `{{ [1, 2] \| merge([3, 4]) }}` — Merges two lists, resulting in [1, 2, 3, 4].                                                   |


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/pebble/functions/FileSizeFunctionTest.java b/core/src/test/java/io/kestra/core/runners/pebble/functions/FileSizeFunctionTest.java
new file mode 100644
index 000000000..2a689df10
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/runners/pebble/functions/FileSizeFunctionTest.java
@@ -0,0 +1,170 @@
+package io.kestra.core.runners.pebble.functions;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.runners.VariableRenderer;
+import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.utils.IdUtils;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hibernate.validator.internal.util.Contracts.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+@KestraTest
+public class FileSizeFunctionTest {
+
+    private static final String NAMESPACE = "my.namespace";
+    private static final String FLOW = "flow";
+    private static final String FILE_TEXT = "Hello from a task output";
+    private static final String FILE_SIZE = "24";
+
+    @Inject
+    StorageInterface storageInterface;
+
+    @Inject
+    VariableRenderer variableRenderer;
+
+    @Test
+    void returnsCorrectSize_givenStringUri_andCurrentExecution() throws IOException, IllegalVariableEvaluationException {
+        String executionId = IdUtils.create();
+        URI internalStorageURI = getInternalStorageURI(executionId);
+        URI internalStorageFile = getInternalStorageFile(internalStorageURI);
+
+        // test for an authorized execution
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", FLOW,
+                "namespace", NAMESPACE),
+            "execution", Map.of("id", executionId)
+        );
+
+        String size = variableRenderer.render("{{ fileSize('" + internalStorageFile + "') }}", variables);
+        assertThat(size, is(FILE_SIZE));
+    }
+
+    @Test
+    void returnsCorrectSize_givenStringUri_andParentExecution() throws IOException, IllegalVariableEvaluationException {
+        String executionId = IdUtils.create();
+        URI internalStorageURI = getInternalStorageURI(executionId);
+        URI internalStorageFile = getInternalStorageFile(internalStorageURI);
+
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", "subflow",
+                "namespace", NAMESPACE),
+            "execution", Map.of("id", IdUtils.create()),
+            "trigger", Map.of(
+                "flowId", FLOW,
+                "namespace", NAMESPACE,
+                "executionId", executionId
+            )
+        );
+
+        String size = variableRenderer.render("{{ fileSize('" + internalStorageFile + "') }}", variables);
+        assertThat(size, is(FILE_SIZE));
+    }
+
+    @Test
+    void shouldThrowIllegalArgumentException_givenMissingTrigger_andParentExecution() throws IOException {
+        String executionId = IdUtils.create();
+        URI internalStorageURI = getInternalStorageURI(executionId);
+        URI internalStorageFile = getInternalStorageFile(internalStorageURI);
+
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", "subflow",
+                "namespace", NAMESPACE),
+            "execution", Map.of("id", IdUtils.create())
+        );
+
+        Exception ex = assertThrows(
+            IllegalArgumentException.class,
+            () -> variableRenderer.render("{{ fileSize('" + internalStorageFile + "') }}", variables)
+        );
+
+        assertTrue(ex.getMessage().startsWith("Unable to read the file"), "Exception message doesn't match expected one");
+    }
+
+    @Test
+    void shouldThrowIllegalArgumentException_givenTrigger_andParentExecution_andMissingNamespace() throws IOException {
+        String executionId = IdUtils.create();
+        URI internalStorageURI = getInternalStorageURI(executionId);
+        URI internalStorageFile = getInternalStorageFile(internalStorageURI);
+
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", "subflow",
+                "namespace", NAMESPACE),
+            "execution", Map.of("id", IdUtils.create()),
+            "trigger", Map.of(
+                "flowId", FLOW,
+                "executionId", executionId
+            )
+        );
+
+        Exception ex = assertThrows(
+            IllegalArgumentException.class,
+            () -> variableRenderer.render("{{ fileSize('" + internalStorageFile + "') }}", variables)
+        );
+
+        assertTrue(ex.getMessage().startsWith("Unable to read the file"), "Exception message doesn't match expected one");
+    }
+
+    @Test
+    void returnsCorrectSize_givenUri_andCurrentExecution() throws IOException, IllegalVariableEvaluationException {
+        String executionId = IdUtils.create();
+        URI internalStorageURI = getInternalStorageURI(executionId);
+        URI internalStorageFile = getInternalStorageFile(internalStorageURI);
+
+        // test for an authorized execution
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", FLOW,
+                "namespace", NAMESPACE),
+            "execution", Map.of("id", executionId),
+            "file", internalStorageFile
+        );
+
+        String size = variableRenderer.render("{{ fileSize(file) }}", variables);
+        assertThat(size, is(FILE_SIZE));
+    }
+
+    @Test
+    void returnsCorrectSize_givenUri_andParentExecution() throws IOException, IllegalVariableEvaluationException {
+        String executionId = IdUtils.create();
+        URI internalStorageURI = getInternalStorageURI(executionId);
+        URI internalStorageFile = getInternalStorageFile(internalStorageURI);
+
+        Map<String, Object> variables = Map.of(
+            "flow", Map.of(
+                "id", "subflow",
+                "namespace", NAMESPACE),
+            "execution", Map.of("id", IdUtils.create()),
+            "trigger", Map.of(
+                "flowId", FLOW,
+                "namespace", NAMESPACE,
+                "executionId", executionId
+            ),
+            "file", internalStorageFile
+        );
+
+        String size = variableRenderer.render("{{ fileSize(file) }}", variables);
+        assertThat(size, is(FILE_SIZE));
+    }
+
+    private URI getInternalStorageURI(String executionId) {
+        return URI.create("/" + NAMESPACE.replace(".", "/") + "/" + FLOW + "/executions/" + executionId + "/tasks/task/" + IdUtils.create() + "/123456.ion");
+    }
+
+    private URI getInternalStorageFile(URI internalStorageURI) throws IOException {
+        return storageInterface.put(null, internalStorageURI, new ByteArrayInputStream(FILE_TEXT.getBytes()));
+    }
+}