From 11f007a7859ba5a210f5019f773299af497a31df Mon Sep 17 00:00:00 2001 From: NekoPunch Date: Tue, 28 Jul 2026 23:31:09 -0700 Subject: [PATCH] fix(collector): correct jsonpath alias parsing for rows missing path Http jsonPath collection resolved alias paths with a global "parseScript + alias" query indexed by row number, so rows missing the path (e.g. pending pods without containerStatuses) misaligned every following row. Calculates like rc=$.status.containerStatuses[0].restartCount also compiled as JEXL array access and silently evaluated to null. Alias paths are now evaluated per row, and calculates equal to an aliasField skip JEXL entirely. Fixes #3307. --- .../collect/http/HttpCollectImpl.java | 11 +-- .../collect/http/HttpCollectImplTest.java | 43 +++++++++++ .../collector/dispatch/MetricsCollect.java | 11 ++- .../dispatch/MetricsCollectTest.java | 73 +++++++++++++++++++ .../collector/util/JsonPathParser.java | 18 +++++ .../collector/util/JsonPathParserTest.java | 73 +++++++++++++++++++ 6 files changed, 219 insertions(+), 10 deletions(-) create mode 100644 hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectTest.java create mode 100644 hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/util/JsonPathParserTest.java diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImpl.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImpl.java index ab9f2dca919..ce6702bcf59 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImpl.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImpl.java @@ -704,13 +704,10 @@ private void parseResponseByJsonPath(String resp, List aliasFields, Http valueRowBuilder.addColumn(String.valueOf(value)); } else { if (alias.startsWith("$.")) { - List 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 aliasValues = JsonPathParser.parseRowWithJsonPath(objectValue, alias); + 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); } diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImplTest.java b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImplTest.java index 05199a2feb7..f69a95a32e4 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImplTest.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImplTest.java @@ -20,6 +20,7 @@ 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 @@ public CollectRep.MetricsData.Builder addValueRow(CollectRep.ValueRow valueRow) 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 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 diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java index 94b20225996..ae3937b69e5 100644 --- a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java +++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java @@ -252,11 +252,12 @@ public void calculateFields(Metrics metrics, CollectRep.MetricsData.Builder coll if (metrics.getCalculates() == null) { metrics.setCalculates(Collections.emptyList()); } + List aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList); // eg: database_pages=Database pages unconventional mapping Map fieldAliasMap = new HashMap<>(8); Map 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 void calculateFields(Metrics metrics, CollectRep.MetricsData.Builder coll .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (Pair) arr[1], (oldValue, newValue) -> newValue)); List fields = metrics.getFields(); - List aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList); Map aliasFieldValueMap = new HashMap<>(8); Map fieldValueMap = new HashMap<>(8); Map stringTypefieldValueMap = new HashMap<>(8); @@ -420,13 +420,18 @@ public void calculateFields(Metrics metrics, CollectRep.MetricsData.Builder coll * @param fieldAliasMap field alias map * @return expr */ - private Object[] transformCal(String cal, Map fieldAliasMap) { + private Object[] transformCal(String cal, Map fieldAliasMap, List 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 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); diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectTest.java b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/MetricsCollectTest.java new file mode 100644 index 00000000000..6bc9ec89149 --- /dev/null +++ b/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 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)); + } +} diff --git a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/util/JsonPathParser.java b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/util/JsonPathParser.java index 5deb848e586..97a8345b044 100644 --- a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/util/JsonPathParser.java +++ b/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 static T parseContentWithJsonPath(String content, String jsonPath, Ty 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 parseRowWithJsonPath(Object document, String jsonPath) { + if (document == null || StringUtils.isEmpty(jsonPath)) { + return Collections.emptyList(); + } + List values = ROW_PARSER.parse(document).read(jsonPath); + return values == null ? Collections.emptyList() : values; + } + } diff --git a/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/util/JsonPathParserTest.java b/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/util/JsonPathParserTest.java new file mode 100644 index 00000000000..e606ec5946d --- /dev/null +++ b/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/util/JsonPathParserTest.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.util; + +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.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 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 values = JsonPathParser.parseRowWithJsonPath(pendingRow, "$.status.containerStatuses[0].restartCount"); + + assertTrue(values.isEmpty()); + } + + @Test + void parseRowWithJsonPathReturnsAllValuesForWildcard() { + List 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 parseRowWithJsonPathHandlesNullDocumentAndEmptyPath() { + assertTrue(JsonPathParser.parseRowWithJsonPath(null, "$.status").isEmpty()); + assertTrue(JsonPathParser.parseRowWithJsonPath(row(), "").isEmpty()); + } +}