[Fix #3486] Add input param counter (#3489)

* [Fix #3486] Add input param counter * [Fix #3486] Arrays
apache · May 3, 2024 · d6a74a5 · d6a74a5
1 parent 4a7e77d
commit d6a74a5
Show file tree

Hide file tree

Showing 10 changed files with 397 additions and 0 deletions.
diff --git a/kogito-bom/pom.xml b/kogito-bom/pom.xml
@@ -695,6 +695,17 @@
         <version>${project.version}</version>
         <classifier>sources</classifier>
       </dependency>
+      <dependency>
+        <groupId>org.kie</groupId>
+        <artifactId>kie-addons-quarkus-monitoring-sonataflow</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.kie</groupId>
+        <artifactId>kie-addons-quarkus-monitoring-sonataflow</artifactId>
+        <version>${project.version}</version>
+        <classifier>sources</classifier>
+      </dependency>
       <dependency>
         <groupId>org.kie</groupId>
         <artifactId>kie-addons-springboot-monitoring-core</artifactId>
@@ -1886,6 +1897,18 @@
         <version>${project.version}</version>
         <classifier>sources</classifier>
       </dependency>
+      <dependency>
+        <groupId>org.kie.kogito</groupId>
+        <artifactId>kogito-serverless-workflow-monitoring</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.kie.kogito</groupId>
+        <artifactId>kogito-serverless-workflow-monitoring</artifactId>
+        <version>${project.version}</version>
+        <classifier>sources</classifier>
+      </dependency>
+
       <dependency>
         <groupId>org.kie.kogito</groupId>
         <artifactId>kogito-serverless-workflow-openapi-parser</artifactId>

diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-monitoring/pom.xml b/kogito-serverless-workflow/kogito-serverless-workflow-monitoring/pom.xml
@@ -0,0 +1,40 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.kie.kogito</groupId>
+    <artifactId>kogito-serverless-workflow</artifactId>
+    <version>999-SNAPSHOT</version>
+  </parent>
+  <artifactId>kogito-serverless-workflow-monitoring</artifactId>
+  <name>Kogito :: Serverless Workflow :: Monitoring</name>
+   <properties>
+      <java.module.name>org.kie.kogito.serverless.workflow.monitoring</java.module.name>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.kie</groupId>
+      <artifactId>kie-addons-monitoring-core</artifactId>
+    </dependency>
+    <dependency>
+      <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>
diff --git a/...a/org/kie/kogito/serverless/workflow/monitoring/SonataFlowMetricProcessEventListener.java b/...a/org/kie/kogito/serverless/workflow/monitoring/SonataFlowMetricProcessEventListener.java
@@ -0,0 +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 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(), null, (ObjectNode) node);
+        }
+    }
+
+    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) {
+        if (value instanceof ObjectNode) {
+            registerObject(processId, key, (ObjectNode) value);
+        } else if (value instanceof ArrayNode) {
+            registerArray(processId, key, (ArrayNode) value);
+        } else {
+            registerValue(processId, key, value.asText());
+        }
+    }
+
+    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 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;
+    }
+}
diff --git a/...g/kie/kogito/serverless/workflow/monitoring/SonataFlowMetricProcessEventListenerTest.java b/...g/kie/kogito/serverless/workflow/monitoring/SonataFlowMetricProcessEventListenerTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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-SNAPSHOT");
+    }
+
+    @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();
+    }
+}
diff --git a/kogito-serverless-workflow/pom.xml b/kogito-serverless-workflow/pom.xml
@@ -57,6 +57,7 @@
     <module>kogito-serverless-workflow-executor-tests</module>
     <module>kogito-serverless-workflow-dmn-parser</module>
     <module>kogito-serverless-workflow-dmn</module>
+    <module>kogito-serverless-workflow-monitoring</module>
   </modules>
 
   <profiles>

diff --git a/quarkus/addons/monitoring/pom.xml b/quarkus/addons/monitoring/pom.xml
@@ -38,6 +38,7 @@
     <module>core</module>
     <module>prometheus</module>
     <module>elastic</module>
+    <module>sonataflow</module>
   </modules>
 
 </project>
diff --git a/quarkus/addons/monitoring/sonataflow/pom.xml b/quarkus/addons/monitoring/sonataflow/pom.xml
@@ -0,0 +1,23 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.kie</groupId>
+    <artifactId>kie-addons-quarkus-monitoring-parent</artifactId>
+    <version>999-SNAPSHOT</version>
+  </parent>
+  <artifactId>kie-addons-quarkus-monitoring-sonataflow</artifactId>
+  <name>KIE Add-On Monitoring Sonataflow</name>
+  <properties>
+    <java.module.name>org.kie.kogito.monitoring.sonataflow.quarkus</java.module.name>
+  </properties>
+  <dependencies>
+    <dependency>
+      <groupId>org.kie</groupId>
+      <artifactId>kie-addons-quarkus-monitoring-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.kie.kogito</groupId>
+      <artifactId>kogito-serverless-workflow-monitoring</artifactId>
+    </dependency>
+  </dependencies>
+</project>