[Fix apache#3486] Arrays

kogito-serverless-workflow/kogito-serverless-workflow-monitoring/pom.xml

@@ -20,5 +20,21 @@
       <groupId>org.kie.kogito</groupId>
       <artifactId>kogito-serverless-workflow-runtime</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-inline</artifactId>
+      <scope>test</scope>
+    </dependency>
+
   </dependencies>
 </project>

kogito-serverless-workflow/kogito-serverless-workflow-monitoring/src/main/java/org/kie/kogito/serverless/workflow/monitoring/SonataFlowMetricProcessEventListener.java

@@ -1,50 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
 package org.kie.kogito.serverless.workflow.monitoring;
 
+import java.util.Iterator;
+
 import org.kie.api.event.process.ProcessStartedEvent;
 import org.kie.kogito.KogitoGAV;
 import org.kie.kogito.internal.process.runtime.KogitoProcessInstance;
+import org.kie.kogito.jackson.utils.JsonObjectUtils;
 import org.kie.kogito.monitoring.core.common.process.MetricsProcessEventListener;
 import org.kie.kogito.serverless.workflow.SWFConstants;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Tag;
 
 public class SonataFlowMetricProcessEventListener extends MetricsProcessEventListener {
 
-    public SonataFlowMetricProcessEventListener(KogitoGAV gav, MeterRegistry meterRegistry) {
+    public enum ArrayStoreMode {
+        STRING,
+        JSON_STRING,
+        MULTI_PARAM
+    }
+
+    static final String INPUT_PARAMS_COUNTER_NAME = "sonataflow_input_parameters_counter";
+
+    private ArrayStoreMode arrayStoreMode;
+
+    public SonataFlowMetricProcessEventListener(KogitoGAV gav, MeterRegistry meterRegistry, ArrayStoreMode arrayStoreMode) {
         super("sonataflow-process-monitoring-listener", gav, meterRegistry);
+        this.arrayStoreMode = arrayStoreMode;
     }
 
     @Override
     public void beforeProcessStarted(ProcessStartedEvent event) {
         final KogitoProcessInstance processInstance = (KogitoProcessInstance) event.getProcessInstance();
         Object node = processInstance.getVariables().get(SWFConstants.DEFAULT_WORKFLOW_VAR);
         if (node instanceof ObjectNode) {
-            registerObject(processInstance.getProcessId(), (ObjectNode) node, null);
+            registerObject(processInstance.getProcessId(), null, (ObjectNode) node);
         }
-
     }
 
-    private void registerObject(String processId, ObjectNode node, String prefix) {
-        node.fields().forEachRemaining(e -> registerInputParam(processId, e.getKey(), e.getValue(), prefix));
+    final void registerObject(String processId, String key, ObjectNode node) {
+        node.fields().forEachRemaining(e -> registerInputParam(processId, concat(key, e.getKey(), '.'), e.getValue()));
     }
 
-    private void registerInputParam(String processId, String key, JsonNode value, String prefix) {
-        if (value.isObject()) {
-            registerObject(processId, (ObjectNode) value, concat(prefix, key));
+    private void registerInputParam(String processId, String key, JsonNode value) {
+        if (value instanceof ObjectNode) {
+            registerObject(processId, key, (ObjectNode) value);
+        } else if (value instanceof ArrayNode) {
+            registerArray(processId, key, (ArrayNode) value);
         } else {
-            registerInputParam(processId, concat(prefix, key), value.toString());
+            registerValue(processId, key, value.asText());
         }
     }
 
-    private String concat(String prefix, String key) {
-        return prefix == null ? key : prefix + "." + key;
+    private void registerArray(String processId, String key, ArrayNode node) {
+        if (arrayStoreMode == ArrayStoreMode.MULTI_PARAM) {
+            Iterator<JsonNode> iter = node.elements();
+            for (int i = 0; iter.hasNext(); i++) {
+                registerInputParam(processId, concat(key, "[" + i + "]"), iter.next());
+            }
+        } else if (arrayStoreMode == ArrayStoreMode.JSON_STRING) {
+            registerValue(processId, key, node.toString());
+        } else if (arrayStoreMode == ArrayStoreMode.STRING) {
+            registerValue(processId, key, JsonObjectUtils.toJavaValue(node).toString());
+        }
+    }
+
+    private void registerValue(String processId, String key, String value) {
+        buildCounter(INPUT_PARAMS_COUNTER_NAME, "Input parameters", processId, Tag.of("param_name", key), Tag.of("param_value", value)).increment();
     }
 
-    private void registerInputParam(String processId, String key, String value) {
-        buildCounter("sonataflow_input_parameters_counter", "Input parameters", processId, Tag.of("param_name", key), Tag.of("param_value", value)).increment();
+    private String concat(String prefix, String key, char prefixChar) {
+        return prefix == null ? key : prefix + prefixChar + key;
+    }
+
+    private String concat(String prefix, String key) {
+        return prefix == null ? key : prefix + key;
     }
 }

kogito-serverless-workflow/kogito-serverless-workflow-monitoring/src/test/java/org/kie/kogito/serverless/workflow/monitoring/SonataFlowMetricProcessEventListenerTest.java

@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.kie.kogito.serverless.workflow.monitoring;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.kie.kogito.KogitoGAV;
+import org.kie.kogito.jackson.utils.ObjectMapperFactory;
+import org.kie.kogito.serverless.workflow.monitoring.SonataFlowMetricProcessEventListener.ArrayStoreMode;
+import org.mockito.MockedStatic;
+
+import com.fasterxml.jackson.databind.node.ArrayNode;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class SonataFlowMetricProcessEventListenerTest {
+
+    private static final String PROCESS_ID = "testMetric";
+
+    private Counter counter;
+    private MeterRegistry meterRegistry;
+    private KogitoGAV kogitoGAV;
+    private Counter.Builder builder;
+    private MockedStatic<Counter> factory;
+
+    @BeforeEach
+    void setup() {
+        counter = mock(Counter.class);
+        meterRegistry = mock(MeterRegistry.class);
+        builder = mock(Counter.Builder.class);
+        factory = mockStatic(Counter.class);
+        factory.when(() -> Counter.builder(SonataFlowMetricProcessEventListener.INPUT_PARAMS_COUNTER_NAME)).thenReturn(builder);
+        when(builder.register(meterRegistry)).thenReturn(counter);
+        when(builder.description(anyString())).thenReturn(builder);
+        when(builder.tag(anyString(), anyString())).thenReturn(builder);
+        kogitoGAV = new KogitoGAV("org.kogito", "test-artifact", "999-SNAPTHOT");
+    }
+
+    @AfterEach
+    void clean() {
+        factory.close();
+    }
+
+    @Test
+    void testSimpleCall() {
+        SonataFlowMetricProcessEventListener listener = new SonataFlowMetricProcessEventListener(kogitoGAV, meterRegistry, ArrayStoreMode.JSON_STRING);
+        listener.registerObject(PROCESS_ID, null, ObjectMapperFactory.get().createObjectNode().put("number", 1));
+        listener.registerObject(PROCESS_ID, null, ObjectMapperFactory.get().createObjectNode().put("number", 2));
+        verify(builder, times(2)).tag("process_id", PROCESS_ID);
+        verify(builder, times(2)).tag("param_name", "number");
+        verify(builder).tag("param_value", "1");
+        verify(builder).tag("param_value", "2");
+        verify(counter, times(2)).increment();
+    }
+
+    @Test
+    void testComplexCall() {
+        SonataFlowMetricProcessEventListener listener = new SonataFlowMetricProcessEventListener(kogitoGAV, meterRegistry, ArrayStoreMode.JSON_STRING);
+        listener.registerObject(PROCESS_ID, null,
+                ObjectMapperFactory.get().createObjectNode().set("team", ObjectMapperFactory.get().createObjectNode().put("name", "Real Betis Balompie").put("age", 117)));
+        verify(builder, times(2)).tag("process_id", PROCESS_ID);
+        verify(builder).tag("param_name", "team.name");
+        verify(builder).tag("param_value", "Real Betis Balompie");
+        verify(builder).tag("param_name", "team.age");
+        verify(builder).tag("param_value", "117");
+        verify(counter, times(2)).increment();
+    }
+
+    @Test
+    void testArrayMultiParam() {
+        SonataFlowMetricProcessEventListener listener = new SonataFlowMetricProcessEventListener(kogitoGAV, meterRegistry, ArrayStoreMode.MULTI_PARAM);
+        listener.registerObject(PROCESS_ID, null,
+                ObjectMapperFactory.get().createObjectNode().set("teams",
+                        ObjectMapperFactory.get().createArrayNode().add(ObjectMapperFactory.get().createObjectNode().put("name", "Real Betis Balompie"))
+                                .add(ObjectMapperFactory.get().createObjectNode().put("name", "Real Sociedad"))));
+        verify(builder, times(2)).tag("process_id", PROCESS_ID);
+        verify(builder).tag("param_name", "teams[0].name");
+        verify(builder).tag("param_value", "Real Betis Balompie");
+        verify(builder).tag("param_name", "teams[1].name");
+        verify(builder).tag("param_value", "Real Sociedad");
+        verify(counter, times(2)).increment();
+    }
+
+    @Test
+    void testArrayJsonString() {
+        SonataFlowMetricProcessEventListener listener = new SonataFlowMetricProcessEventListener(kogitoGAV, meterRegistry, ArrayStoreMode.JSON_STRING);
+        ArrayNode arrayNode = ObjectMapperFactory.get().createArrayNode().add(ObjectMapperFactory.get().createObjectNode().put("name", "Real Betis Balompie"))
+                .add(ObjectMapperFactory.get().createObjectNode().put("name", "Real Sociedad"));
+        listener.registerObject(PROCESS_ID, null,
+                ObjectMapperFactory.get().createObjectNode().set("teams", arrayNode));
+        verify(builder).tag("process_id", PROCESS_ID);
+        verify(builder).tag("param_name", "teams");
+        verify(builder).tag("param_value", arrayNode.toString());
+        verify(counter).increment();
+    }
+
+    @Test
+    void testArrayString() {
+        SonataFlowMetricProcessEventListener listener = new SonataFlowMetricProcessEventListener(kogitoGAV, meterRegistry, ArrayStoreMode.STRING);
+        ArrayNode arrayNode = ObjectMapperFactory.get().createArrayNode().add(ObjectMapperFactory.get().createObjectNode().put("name", "Real Betis Balompie"))
+                .add(ObjectMapperFactory.get().createObjectNode().put("name", "Real Sociedad"));
+        listener.registerObject(PROCESS_ID, null,
+                ObjectMapperFactory.get().createObjectNode().set("teams", arrayNode));
+        verify(builder).tag("process_id", PROCESS_ID);
+        verify(builder).tag("param_name", "teams");
+        verify(builder).tag("param_value", "[{name=Real Betis Balompie}, {name=Real Sociedad}]");
+        verify(counter).increment();
+    }
+
+}

quarkus/addons/monitoring/sonataflow/src/main/java/org/kie/sonataflow/monitoring/SonataFlowMetricEventListenerFactory.java

@@ -1,8 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
 package org.kie.sonataflow.monitoring;
 
+import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.kie.kogito.KogitoGAV;
 import org.kie.kogito.config.ConfigBean;
 import org.kie.kogito.internal.process.event.KogitoProcessEventListener;
+import org.kie.kogito.serverless.workflow.monitoring.SonataFlowMetricProcessEventListener.ArrayStoreMode;
 import org.kie.kogito.serverless.workflow.monitoring.SonataFlowMetricProcessEventListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -21,10 +41,13 @@ public class SonataFlowMetricEventListenerFactory {
     @Inject
     ConfigBean configBean;
 
+    @ConfigProperty(name = "kie.monitoring.sonataflow.arrays.store", defaultValue = "JSON_STRING")
+    ArrayStoreMode arrayStoreMode;
+
     @Produces
     public KogitoProcessEventListener produceProcessListener() {
         LOGGER.info("Producing sonataflow listener for process monitoring.");
         return new SonataFlowMetricProcessEventListener(
-                configBean.getGav().orElse(KogitoGAV.EMPTY_GAV), Metrics.globalRegistry);
+                configBean.getGav().orElse(KogitoGAV.EMPTY_GAV), Metrics.globalRegistry, arrayStoreMode);
     }
 }