From f7716aae1328a64797419e2a36d28ac02a6878b3 Mon Sep 17 00:00:00 2001 From: Michael Koepf <47541996+michaelkoepf@users.noreply.github.com> Date: Wed, 8 Jan 2025 17:54:17 +0100 Subject: [PATCH 1/6] [hotfix] Metric reporting for non-partitioned tables Report empty string as value for partition label for non-partitioned table Issue #302 --- .../fluss/server/metrics/group/PhysicalTableMetricGroup.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java index 5d651d80a..a2a84d01b 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java @@ -78,8 +78,13 @@ public PhysicalTableMetricGroup( protected void putVariables(Map variables) { variables.put("database", physicalTablePath.getDatabaseName()); variables.put("table", physicalTablePath.getTableName()); + if (physicalTablePath.getPartitionName() != null) { variables.put("partition", physicalTablePath.getPartitionName()); + } else { + LOG.debug( + "Setting variable 'partition' for non-partitioned table to empty string to ensure consistent metric labels across metric reporters."); + variables.put("partition", ""); } } From aac74086a96fab9996bb2075216758998e3977ed Mon Sep 17 00:00:00 2001 From: Michael Koepf <47541996+michaelkoepf@users.noreply.github.com> Date: Wed, 8 Jan 2025 17:54:58 +0100 Subject: [PATCH 2/6] [hotfix] Metric reporting for non-partitioned tables Added test cases for Prometheus metric exporter Issue #302 --- ...etheusReporterDifferentLabelValueTest.java | 138 +++++++++++++----- 1 file changed, 103 insertions(+), 35 deletions(-) diff --git a/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java b/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java index db452b83a..3c563646c 100644 --- a/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java +++ b/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java @@ -24,13 +24,15 @@ import com.alibaba.fluss.metrics.util.TestHistogram; import com.alibaba.fluss.metrics.util.TestMeter; import com.alibaba.fluss.utils.NetUtils; - import com.mashape.unirest.http.exceptions.UnirestException; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import java.util.Arrays; +import java.util.stream.Stream; import static com.alibaba.fluss.metrics.prometheus.PrometheusReporterTest.pollMetrics; import static org.assertj.core.api.Assertions.assertThat; @@ -42,18 +44,9 @@ class PrometheusReporterDifferentLabelValueTest { private static final String[] LABEL_NAMES = {"label1", "label2"}; - private static final String[] LABEL_VALUES_1 = new String[] {"value1_1", "value1_2"}; - private static final String[] LABEL_VALUES_2 = new String[] {"value2_1", "value2_2"}; private static final String LOGICAL_SCOPE = "logical_scope"; private static final String METRIC_NAME = "myMetric"; - private final MetricGroup metricGroup1 = - TestUtils.createTestMetricGroup( - LOGICAL_SCOPE, TestUtils.toMap(LABEL_NAMES, LABEL_VALUES_1)); - private final MetricGroup metricGroup2 = - TestUtils.createTestMetricGroup( - LOGICAL_SCOPE, TestUtils.toMap(LABEL_NAMES, LABEL_VALUES_2)); - private PrometheusReporter reporter; @BeforeEach @@ -68,8 +61,39 @@ void tearDown() { } } - @Test - void countersCanBeAddedSeveralTimesIfTheyDifferInLabels() { + private static Stream provideParameters() { + final String[] labelValues1 = new String[] {"value1_1", "value1_2"}; + final String[] labelValues2 = new String[] {"value2_1", "value2_2"}; + final String[] labelValues3 = new String[] {"value3_1", ""}; + final String[] labelValues4 = new String[] {"values4_1", ""}; + + final MetricGroup metricGroup1 = + TestUtils.createTestMetricGroup( + LOGICAL_SCOPE, TestUtils.toMap(LABEL_NAMES, labelValues1)); + final MetricGroup metricGroup2 = + TestUtils.createTestMetricGroup( + LOGICAL_SCOPE, TestUtils.toMap(LABEL_NAMES, labelValues2)); + final MetricGroup metricGroup3 = + TestUtils.createTestMetricGroup( + LOGICAL_SCOPE, TestUtils.toMap(LABEL_NAMES, labelValues3)); + final MetricGroup metricGroup4 = + TestUtils.createTestMetricGroup( + LOGICAL_SCOPE, TestUtils.toMap(LABEL_NAMES, labelValues4)); + + return Stream.of( + Arguments.of(metricGroup1, labelValues1, metricGroup2, labelValues2), + Arguments.of(metricGroup1, labelValues1, metricGroup3, labelValues3), + Arguments.of(metricGroup3, labelValues3, metricGroup2, labelValues2), + Arguments.of(metricGroup3, labelValues3, metricGroup4, labelValues4)); + } + + @ParameterizedTest + @MethodSource("provideParameters") + void countersCanBeAddedSeveralTimesIfTheyDifferInLabelValues( + MetricGroup metricGroup1, + String[] expectedLabelValues1, + MetricGroup metricGroup2, + String[] expectedLabelValues2) { Counter counter1 = new SimpleCounter(); counter1.inc(1); Counter counter2 = new SimpleCounter(); @@ -80,16 +104,21 @@ void countersCanBeAddedSeveralTimesIfTheyDifferInLabels() { assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_1)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues1)) .isEqualTo(1.); assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_2)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues2)) .isEqualTo(2.); } - @Test - void gaugesCanBeAddedSeveralTimesIfTheyDifferInLabels() { + @ParameterizedTest + @MethodSource("provideParameters") + void gaugesCanBeAddedSeveralTimesIfTheyDifferInLabelValues( + MetricGroup metricGroup1, + String[] expectedLabelValues1, + MetricGroup metricGroup2, + String[] expectedLabelValues2) { Gauge gauge1 = () -> 3; Gauge gauge2 = () -> 4; @@ -98,16 +127,21 @@ void gaugesCanBeAddedSeveralTimesIfTheyDifferInLabels() { assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_1)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues1)) .isEqualTo(3.); assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_2)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues2)) .isEqualTo(4.); } - @Test - void metersCanBeAddedSeveralTimesIfTheyDifferInLabels() { + @ParameterizedTest + @MethodSource("provideParameters") + void metersCanBeAddedSeveralTimesIfTheyDifferInLabelValues( + MetricGroup metricGroup1, + String[] expectedLabelValues1, + MetricGroup metricGroup2, + String[] expectedLabelValues2) { Meter meter1 = new TestMeter(1, 1.0); Meter meter2 = new TestMeter(2, 2.0); @@ -116,16 +150,22 @@ void metersCanBeAddedSeveralTimesIfTheyDifferInLabels() { assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_1)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues1)) .isEqualTo(meter1.getRate()); assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_2)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues2)) .isEqualTo(meter2.getRate()); } - @Test - void histogramsCanBeAddedSeveralTimesIfTheyDifferInLabels() throws UnirestException { + @ParameterizedTest + @MethodSource("provideParameters") + void histogramsCanBeAddedSeveralTimesIfTheyDifferInLabelValues( + MetricGroup metricGroup1, + String[] expectedLabelValues1, + MetricGroup metricGroup2, + String[] expectedLabelValues2) + throws UnirestException { TestHistogram histogram1 = new TestHistogram(); histogram1.setCount(1); TestHistogram histogram2 = new TestHistogram(); @@ -135,8 +175,10 @@ void histogramsCanBeAddedSeveralTimesIfTheyDifferInLabels() throws UnirestExcept reporter.notifyOfAddedMetric(histogram2, METRIC_NAME, metricGroup2); final String exportedMetrics = pollMetrics(reporter.getPort()).getBody(); - assertThat(exportedMetrics).contains("label2=\"value1_2\",} 1.0"); - assertThat(exportedMetrics).contains("label2=\"value2_2\",} 2.0"); + assertThat(exportedMetrics) + .contains(formatAsPrometheusLabels(LABEL_NAMES, expectedLabelValues1) + " 1.0"); + assertThat(exportedMetrics) + .contains(formatAsPrometheusLabels(LABEL_NAMES, expectedLabelValues2) + " 2.0"); final String[] labelNamesWithQuantile = addToArray(LABEL_NAMES, "quantile"); for (Double quantile : PrometheusReporter.HistogramSummaryProxy.QUANTILES) { @@ -144,19 +186,24 @@ void histogramsCanBeAddedSeveralTimesIfTheyDifferInLabels() throws UnirestExcept reporter.registry.getSampleValue( getLogicalScope(METRIC_NAME), labelNamesWithQuantile, - addToArray(LABEL_VALUES_1, "" + quantile))) + addToArray(expectedLabelValues1, "" + quantile))) .isEqualTo(quantile); assertThat( reporter.registry.getSampleValue( getLogicalScope(METRIC_NAME), labelNamesWithQuantile, - addToArray(LABEL_VALUES_2, "" + quantile))) + addToArray(expectedLabelValues2, "" + quantile))) .isEqualTo(quantile); } } - @Test - void removingSingleInstanceOfMetricDoesNotBreakOtherInstances() { + @ParameterizedTest + @MethodSource("provideParameters") + void removingSingleInstanceOfMetricDoesNotBreakOtherInstances( + MetricGroup metricGroup1, + String[] expectedLabelValues1, + MetricGroup metricGroup2, + String[] expectedLabelValues2) { Counter counter1 = new SimpleCounter(); counter1.inc(1); Counter counter2 = new SimpleCounter(); @@ -167,23 +214,23 @@ void removingSingleInstanceOfMetricDoesNotBreakOtherInstances() { assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_1)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues1)) .isEqualTo(1.); assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_2)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues2)) .isEqualTo(2.); reporter.notifyOfRemovedMetric(counter2, METRIC_NAME, metricGroup2); assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_1)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues1)) .isEqualTo(1.); reporter.notifyOfRemovedMetric(counter1, METRIC_NAME, metricGroup1); assertThat( reporter.registry.getSampleValue( - getLogicalScope(METRIC_NAME), LABEL_NAMES, LABEL_VALUES_1)) + getLogicalScope(METRIC_NAME), LABEL_NAMES, expectedLabelValues2)) .isNull(); } @@ -199,4 +246,25 @@ private String[] addToArray(String[] array, String element) { labelNames[LABEL_NAMES.length] = element; return labelNames; } + + private static String formatAsPrometheusLabels(String[] labelNames, String[] labelValues) { + if (labelNames == null + || labelValues == null + || labelNames.length == 0 + || labelValues.length == 0 + || labelNames.length != labelValues.length) { + throw new IllegalStateException("Erroneous test setup!"); + } + + final StringBuilder sb = new StringBuilder(); + for (int i = 0; i < labelNames.length; i++) { + sb.append(labelNames[i]); + sb.append("=\""); + sb.append(labelValues[i]); + sb.append("\","); + } + sb.append("}"); + + return sb.toString(); + } } From 6c082e2446d9173874e8910b49760d6495a54c0b Mon Sep 17 00:00:00 2001 From: Michael Koepf <47541996+michaelkoepf@users.noreply.github.com> Date: Wed, 8 Jan 2025 18:08:44 +0100 Subject: [PATCH 3/6] [hotfix] Metric reporting for non-partitioned tables Fixed spotless errors Issue #302 --- .../prometheus/PrometheusReporterDifferentLabelValueTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java b/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java index 3c563646c..4e829cbd8 100644 --- a/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java +++ b/fluss-metrics/fluss-metrics-prometheus/src/test/java/com/alibaba/fluss/metrics/prometheus/PrometheusReporterDifferentLabelValueTest.java @@ -24,6 +24,7 @@ import com.alibaba.fluss.metrics.util.TestHistogram; import com.alibaba.fluss.metrics.util.TestMeter; import com.alibaba.fluss.utils.NetUtils; + import com.mashape.unirest.http.exceptions.UnirestException; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; From 7de022719f6e25720f377c6d7bfb88952e1be9d7 Mon Sep 17 00:00:00 2001 From: Michael Koepf <47541996+michaelkoepf@users.noreply.github.com> Date: Wed, 8 Jan 2025 18:25:02 +0100 Subject: [PATCH 4/6] [hotfix] Metric reporting for non-partitioned tables Adapted JMX reporter test cases Issue #302 --- .../java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java b/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java index 68f7fea7f..24ff5c053 100644 --- a/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java +++ b/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java @@ -23,7 +23,6 @@ import com.alibaba.fluss.metrics.util.TestHistogram; import com.alibaba.fluss.metrics.util.TestMeter; import com.alibaba.fluss.metrics.util.TestMetricGroup; - import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -35,7 +34,6 @@ import javax.management.remote.JMXConnector; import javax.management.remote.JMXConnectorFactory; import javax.management.remote.JMXServiceURL; - import java.io.IOException; import java.lang.management.ManagementFactory; import java.util.HashMap; @@ -54,6 +52,8 @@ class JMXReporterTest { static { variables = new HashMap<>(); variables.put("", "localhost"); + variables.put("key1", "value1"); + variables.put("key2", ""); metricGroup = TestMetricGroup.newBuilder() @@ -91,6 +91,7 @@ void testGenerateTable() { vars.put("key0", "value0"); vars.put("key1", "value1"); vars.put("\"key2,=;:?'", "\"value2 (test),=;:?'"); + vars.put("key3", ""); Hashtable jmxTable = JMXReporter.generateJmxTable(vars); @@ -98,6 +99,7 @@ void testGenerateTable() { assertThat(jmxTable).containsEntry("key0", "value0"); assertThat(jmxTable).containsEntry("key1", "value1"); assertThat(jmxTable).containsEntry("key2------", "value2_(test)------"); + assertThat(jmxTable).containsEntry("key3", ""); } /** From bb9806ff4d850632ea34b2cbc0821ba20fc23c00 Mon Sep 17 00:00:00 2001 From: Michael Koepf <47541996+michaelkoepf@users.noreply.github.com> Date: Wed, 8 Jan 2025 18:26:41 +0100 Subject: [PATCH 5/6] [hotfix] Metric reporting for non-partitioned tables - Fix spotless errors - Change debug message --- .../java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java | 2 ++ .../fluss/server/metrics/group/PhysicalTableMetricGroup.java | 3 +-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java b/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java index 24ff5c053..afdf11a6c 100644 --- a/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java +++ b/fluss-metrics/fluss-metrics-jmx/src/test/java/com/alibaba/fluss/metrics/jmx/JMXReporterTest.java @@ -23,6 +23,7 @@ import com.alibaba.fluss.metrics.util.TestHistogram; import com.alibaba.fluss.metrics.util.TestMeter; import com.alibaba.fluss.metrics.util.TestMetricGroup; + import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -34,6 +35,7 @@ import javax.management.remote.JMXConnector; import javax.management.remote.JMXConnectorFactory; import javax.management.remote.JMXServiceURL; + import java.io.IOException; import java.lang.management.ManagementFactory; import java.util.HashMap; diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java index a2a84d01b..5e3bb31da 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java @@ -82,8 +82,7 @@ protected void putVariables(Map variables) { if (physicalTablePath.getPartitionName() != null) { variables.put("partition", physicalTablePath.getPartitionName()); } else { - LOG.debug( - "Setting variable 'partition' for non-partitioned table to empty string to ensure consistent metric labels across metric reporters."); + LOG.debug("Setting variable 'partition' for non-partitioned table to empty string."); variables.put("partition", ""); } } From d5a3a8c2674c9de70e883a8d4c6c70dac7233049 Mon Sep 17 00:00:00 2001 From: Michael Koepf <47541996+michaelkoepf@users.noreply.github.com> Date: Thu, 9 Jan 2025 09:38:29 +0100 Subject: [PATCH 6/6] [observability] Add Grafana and Loki as log monitoring stack Adaptions according to PR review --- .../fluss/server/metrics/group/PhysicalTableMetricGroup.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java index 5e3bb31da..f85b145e4 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/PhysicalTableMetricGroup.java @@ -82,7 +82,7 @@ protected void putVariables(Map variables) { if (physicalTablePath.getPartitionName() != null) { variables.put("partition", physicalTablePath.getPartitionName()); } else { - LOG.debug("Setting variable 'partition' for non-partitioned table to empty string."); + // value of empty string indicates non-partitioned tables variables.put("partition", ""); } }