fix(collector): correct jsonpath alias parsing for rows missing path (#4265)

Co-authored-by: Tomsun28 <tomsun28@outlook.com>

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImpl.java

@@ -704,13 +704,11 @@ public class HttpCollectImpl extends AbstractCollect {
                         valueRowBuilder.addColumn(String.valueOf(value));
                     } else {
                         if (alias.startsWith("$.")) {
-                            List<Object> subResults = JsonPathParser.parseContentWithJsonPath(resp, http.getParseScript() + alias.substring(1));
-                            if (subResults != null && subResults.size() > i) {
-                                Object resultValue = subResults.get(i);
-                                valueRowBuilder.addColumn(resultValue == null ? CommonConstants.NULL_VALUE : String.valueOf(resultValue));
-                            } else {
-                                valueRowBuilder.addColumn(CommonConstants.NULL_VALUE);
-                            }
+                            // per-row evaluation, a global "parseScript + alias" query would misalign rows missing the path
+                            List<Object> aliasValues = JsonPathParser.parseRowWithJsonPath(objectValue, alias);
+                            // a wildcard alias matching multiple values is kept whole and rendered as "[v1, v2]"
+                            Object resultValue = aliasValues.size() == 1 ? aliasValues.get(0) : (aliasValues.isEmpty() ? null : aliasValues);
+                            valueRowBuilder.addColumn(resultValue == null ? CommonConstants.NULL_VALUE : String.valueOf(resultValue));
                         } else {
                             addColumnForSummary(responseTime, valueRowBuilder, keywordNum, alias);
                         }

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImplTest.java

@@ -20,6 +20,7 @@ package org.apache.hertzbeat.collector.collect.http;
 import com.google.common.collect.Lists;
 import com.sun.net.httpserver.HttpServer;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
+import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.entity.job.Metrics;
 import org.apache.hertzbeat.common.entity.job.protocol.HttpProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
@@ -383,6 +384,48 @@ class HttpCollectImplTest {
         assertEquals("0.268751364291017", firstRow.getColumns(0));
     }
 
+    @Test
+    void parseResponseByJsonPathKeepsRowAlignmentWhenAliasPathMissing() throws Exception {
+        String jsonResponse = "{\"items\": ["
+                + "{\"metadata\": {\"name\": \"pod-a\"}, \"status\": {\"phase\": \"Running\","
+                + " \"containerStatuses\": [{\"name\": \"c1\", \"ready\": true, \"restartCount\": 5}]}},"
+                + "{\"metadata\": {\"name\": \"pod-b-pending\"}, \"status\": {\"phase\": \"Pending\"}},"
+                + "{\"metadata\": {\"name\": \"pod-c\"}, \"status\": {\"phase\": \"Running\","
+                + " \"containerStatuses\": [{\"name\": \"c3\", \"ready\": true, \"restartCount\": 2}]}}"
+                + "]}";
+        HttpProtocol http = HttpProtocol.builder()
+                .parseType(DispatchConstants.PARSE_JSON_PATH)
+                .parseScript("$.items.*")
+                .build();
+        List<CollectRep.ValueRow> capturedRows = new ArrayList<>();
+        CollectRep.MetricsData.Builder builder = new CollectRep.MetricsData.Builder() {
+            @Override
+            public CollectRep.MetricsData.Builder addValueRow(CollectRep.ValueRow valueRow) {
+                capturedRows.add(valueRow);
+                return super.addValueRow(valueRow);
+            }
+        };
+        Method parseMethod = HttpCollectImpl.class.getDeclaredMethod(
+                "parseResponseByJsonPath",
+                String.class,
+                List.class,
+                HttpProtocol.class,
+                CollectRep.MetricsData.Builder.class,
+                Long.class);
+        parseMethod.setAccessible(true);
+
+        parseMethod.invoke(httpCollectImpl, jsonResponse,
+                Lists.newArrayList("$.metadata.name", "$.status.containerStatuses[0].restartCount"), http, builder, 100L);
+
+        assertEquals(3, capturedRows.size());
+        assertEquals("pod-a", capturedRows.get(0).getColumns(0));
+        assertEquals("5", capturedRows.get(0).getColumns(1));
+        assertEquals("pod-b-pending", capturedRows.get(1).getColumns(0));
+        assertEquals(CommonConstants.NULL_VALUE, capturedRows.get(1).getColumns(1));
+        assertEquals("pod-c", capturedRows.get(2).getColumns(0));
+        assertEquals("2", capturedRows.get(2).getColumns(1));
+    }
+
     @Test
     void testParsePromQlLabelValue() throws Exception {
         // Create Prometheus format test data

hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java

@@ -252,11 +252,12 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
         if (metrics.getCalculates() == null) {
             metrics.setCalculates(Collections.emptyList());
         }
+        List<String> aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList);
         // eg: database_pages=Database pages unconventional mapping
         Map<String, String> fieldAliasMap = new HashMap<>(8);
         Map<String, JexlExpression> fieldExpressionMap = metrics.getCalculates()
                 .stream()
-                .map(cal -> transformCal(cal, fieldAliasMap))
+                .map(cal -> transformCal(cal, fieldAliasMap, aliasFields))
                 .filter(Objects::nonNull)
                 .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (JexlExpression) arr[1], (oldValue, newValue) -> newValue));
 
@@ -270,7 +271,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
                 .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (Pair<String, String>) arr[1], (oldValue, newValue) -> newValue));
 
         List<Metrics.Field> fields = metrics.getFields();
-        List<String> aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList);
         Map<String, String> aliasFieldValueMap = new HashMap<>(8);
         Map<String, Object> fieldValueMap = new HashMap<>(8);
         Map<String, Object> stringTypefieldValueMap = new HashMap<>(8);
@@ -420,13 +420,19 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
      * @param fieldAliasMap field alias map
      * @return expr
      */
-    private Object[] transformCal(String cal, Map<String, String> fieldAliasMap) {
+    private Object[] transformCal(String cal, Map<String, String> fieldAliasMap, List<String> aliasFields) {
         int splitIndex = cal.indexOf("=");
         if (splitIndex < 0) {
             return null;
         }
         String field = cal.substring(0, splitIndex).trim();
         String expressionStr = cal.substring(splitIndex + 1).trim().replace("\\#", "#");
+        // a direct alias reference (RHS must exactly equal an aliasField, no whitespace/case tolerance) is not a formula,
+        // JEXL parses "[0]" in such paths as array access and silently returns null
+        if (aliasFields.contains(expressionStr)) {
+            fieldAliasMap.put(field, expressionStr);
+            return null;
+        }
         JexlExpression expression;
         try {
             expression = JexlExpressionRunner.compile(expressionStr);

hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectTest.java

@@ -0,0 +1,73 @@
+/*
+ * 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.apache.hertzbeat.collector.dispatch;
+
+import java.util.List;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link MetricsCollect}
+ */
+class MetricsCollectTest {
+
+    @Test
+    void calculateFieldsMapsIndexedJsonPathAlias() {
+        Metrics metrics = Metrics.builder()
+                .name("pods")
+                .priority((byte) 0)
+                .fields(List.of(
+                        Metrics.Field.builder().field("pod").type(CommonConstants.TYPE_STRING).build(),
+                        Metrics.Field.builder().field("rc").type(CommonConstants.TYPE_STRING).build()))
+                .aliasFields(List.of("$.metadata.name", "$.status.containerStatuses[0].restartCount"))
+                .calculates(List.of(
+                        "pod=$.metadata.name",
+                        "rc=$.status.containerStatuses[0].restartCount"))
+                .build();
+
+        Timeout timeout = mock(Timeout.class);
+        WheelTimerTask timerTask = mock(WheelTimerTask.class);
+        when(timeout.task()).thenReturn(timerTask);
+        when(timerTask.getJob()).thenReturn(Job.builder().build());
+        MetricsCollect metricsCollect = new MetricsCollect(metrics, timeout, null, "test", List.of());
+
+        CollectRep.MetricsData.Builder collectData = CollectRep.MetricsData.newBuilder();
+        collectData.addValueRow(CollectRep.ValueRow.newBuilder()
+                .addColumn("pod-a").addColumn("5").build());
+        collectData.addValueRow(CollectRep.ValueRow.newBuilder()
+                .addColumn("pod-b-pending").addColumn(CommonConstants.NULL_VALUE).build());
+
+        metricsCollect.calculateFields(metrics, collectData);
+
+        List<CollectRep.ValueRow> rows = collectData.getValuesList();
+        assertEquals(2, rows.size());
+        assertEquals("pod-a", rows.get(0).getColumns(0));
+        assertEquals("5", rows.get(0).getColumns(1));
+        assertEquals("pod-b-pending", rows.get(1).getColumns(0));
+        assertEquals(CommonConstants.NULL_VALUE, rows.get(1).getColumns(1));
+    }
+}

hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/util/JsonPathParser.java

@@ -36,12 +36,16 @@ public final class JsonPathParser {
 
     private static final ParseContext PARSER;
 
+    private static final ParseContext ROW_PARSER;
+
     static {
         Configuration conf = Configuration.defaultConfiguration()
                 .addOptions(Option.DEFAULT_PATH_LEAF_TO_NULL)
                 .addOptions(Option.ALWAYS_RETURN_LIST);
         CacheProvider.setCache(new LRUCache(128));
         PARSER = JsonPath.using(conf);
+        // a single row legitimately may not contain the queried path
+        ROW_PARSER = JsonPath.using(conf.addOptions(Option.SUPPRESS_EXCEPTIONS));
     }
 
     private JsonPathParser() {
@@ -73,4 +77,18 @@ public final class JsonPathParser {
         return PARSER.parse(content).read(jsonPath, typeRef);
     }
 
+    /**
+     * use json path to parse one already-parsed row object, missing paths yield an empty list
+     * @param document parsed json object of a single row
+     * @param jsonPath jsonPath relative to the row root
+     * @return matched values, empty list when the path does not exist in this row
+     */
+    public static List<Object> parseRowWithJsonPath(Object document, String jsonPath) {
+        if (document == null || StringUtils.isEmpty(jsonPath)) {
+            return Collections.emptyList();
+        }
+        List<Object> values = ROW_PARSER.parse(document).read(jsonPath);
+        return values == null ? Collections.emptyList() : values;
+    }
+
 }

hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/util/JsonPathParserTest.java

@@ -0,0 +1,81 @@
+/*
+ * 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.apache.hertzbeat.collector.util;
+
+import com.jayway.jsonpath.PathNotFoundException;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test case for {@link JsonPathParser}
+ */
+class JsonPathParserTest {
+
+    private static final String ROW_JSON = "{\"metadata\": {\"name\": \"pod-a\"},"
+            + " \"status\": {\"phase\": \"Running\","
+            + " \"containerStatuses\": [{\"name\": \"c1\", \"ready\": true, \"restartCount\": 5}]}}";
+
+    private Object row() {
+        return JsonPathParser.parseContentWithJsonPath(ROW_JSON, "$").get(0);
+    }
+
+    @Test
+    void parseRowWithJsonPathReturnsExistingValue() {
+        List<Object> values = JsonPathParser.parseRowWithJsonPath(row(), "$.status.containerStatuses[0].restartCount");
+
+        assertEquals(1, values.size());
+        assertEquals(5, values.get(0));
+    }
+
+    @Test
+    void parseRowWithJsonPathReturnsEmptyListWhenPathMissing() {
+        Object pendingRow = JsonPathParser
+                .parseContentWithJsonPath("{\"metadata\": {\"name\": \"pod-b\"}, \"status\": {\"phase\": \"Pending\"}}", "$")
+                .get(0);
+
+        List<Object> values = JsonPathParser.parseRowWithJsonPath(pendingRow, "$.status.containerStatuses[0].restartCount");
+
+        assertTrue(values.isEmpty());
+    }
+
+    @Test
+    void parseRowWithJsonPathReturnsAllValuesForWildcard() {
+        List<Object> values = JsonPathParser.parseRowWithJsonPath(row(), "$.status.containerStatuses[0].*");
+
+        assertEquals(3, values.size());
+        assertTrue(values.contains("c1"));
+        assertTrue(values.contains(true));
+        assertTrue(values.contains(5));
+    }
+
+    @Test
+    void parseContentWithJsonPathStillThrowsWhenPathMissing() {
+        assertThrows(PathNotFoundException.class,
+                () -> JsonPathParser.parseContentWithJsonPath(ROW_JSON, "$.spec.nodeName"));
+    }
+
+    @Test
+    void parseRowWithJsonPathHandlesNullDocumentAndEmptyPath() {
+        assertTrue(JsonPathParser.parseRowWithJsonPath(null, "$.status").isEmpty());
+        assertTrue(JsonPathParser.parseRowWithJsonPath(row(), "").isEmpty());
+    }
+}