RockteMQ-AI commented on code in PR #495: URL: https://github.com/apache/rocketmq-connect/pull/495#discussion_r3839496441
########## metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/PrometheusSampleBuilder.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.rocketmq.connect.metrics; + +import io.prometheus.client.Collector; +import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder; +import java.util.Arrays; +import java.util.List; +import org.apache.commons.lang3.StringUtils; + +public class PrometheusSampleBuilder implements SampleBuilder { + private static final List<String> SOURCE_TASK_LABEL_NAMES = Arrays.asList("metric_group", "data_type", "connector", "task"); + + @Override + public Collector.MetricFamilySamples.Sample createSample(String dropwizardName, String nameSuffix, + List<String> additionalLabelNames, List<String> additionalLabelValues, double value) { + String suffix = nameSuffix == null ? "" : nameSuffix; + List<String> labelValues = sanitizeLabelValues(dropwizardName); + return new Collector.MetricFamilySamples.Sample(sanitizeMetricName(dropwizardName + suffix), SOURCE_TASK_LABEL_NAMES, labelValues, value); Review Comment: The createSample method completely ignores the additionalLabelNames and additionalLabelValues parameters. These carry the 'quantile' label (e.g., name="quantile", value="0.75") that differentiates histogram percentiles in Prometheus SUMMARY metrics. By always using SOURCE_TASK_LABEL_NAMES and discarding the additional labels, all percentile samples for a histogram end up with an identical metric name and label set, making them indistinguishable and causing Prometheus to reject or arbitrarily deduplicate them. The method should merge additionalLabelNames/additionalLabelValues into the output sample's label names and values. ########## metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/PrometheusSampleBuilder.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.rocketmq.connect.metrics; + +import io.prometheus.client.Collector; +import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder; +import java.util.Arrays; +import java.util.List; +import org.apache.commons.lang3.StringUtils; + +public class PrometheusSampleBuilder implements SampleBuilder { + private static final List<String> SOURCE_TASK_LABEL_NAMES = Arrays.asList("metric_group", "data_type", "connector", "task"); + + @Override + public Collector.MetricFamilySamples.Sample createSample(String dropwizardName, String nameSuffix, + List<String> additionalLabelNames, List<String> additionalLabelValues, double value) { + String suffix = nameSuffix == null ? "" : nameSuffix; + List<String> labelValues = sanitizeLabelValues(dropwizardName); + return new Collector.MetricFamilySamples.Sample(sanitizeMetricName(dropwizardName + suffix), SOURCE_TASK_LABEL_NAMES, labelValues, value); + } + + public String sanitizeMetricName(String dropwizardName) { + return dropwizardName.split(":")[1].split(",")[1].replaceAll("-", "_"); + } + + public List<String> sanitizeLabelValues(String dropwizardName) { + String[] var = dropwizardName.split(":"); + String[] split = var[1].split(","); + + String metricGroup = split[0]; + String metricName = split[1].replaceAll("-", "_"); + String dateType = split[2]; + String var3 = split[3]; + String connectorName = StringUtils.EMPTY; + if (!StringUtils.equals(var3, "")) { + connectorName = var3.substring(var3.indexOf("=") + 1); + } + String var4 = split[4]; Review Comment: sanitizeLabelValues performs unchecked array indexing on split(":") and split(",") results, accessing indices 0 through 4 (split[4]) without any bounds validation. If any metric in the registry has a name that does not match the expected 'prefix:group,name,type,connector=X,task=Y' format (e.g., metrics registered by the framework itself or third-party libraries), this will throw ArrayIndexOutOfBoundsException, crashing the entire /metrics endpoint. sanitizeMetricName (line 37) has the same issue. Consider validating the split array length or wrapping in a try-catch that skips malformed metric names. ########## metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/DropwizardExports.java: ########## @@ -0,0 +1,234 @@ +/* + * 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.rocketmq.connect.metrics; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.Metric; +import com.codahale.metrics.MetricFilter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.Timer; +import io.prometheus.client.dropwizard.samplebuilder.DefaultSampleBuilder; +import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedMap; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.apache.rocketmq.connect.metrics.stats.Stat; + +/** + * Collect Dropwizard metrics from a MetricRegistry. + */ +public class DropwizardExports extends io.prometheus.client.Collector implements io.prometheus.client.Collector.Describable { + private static final Logger LOGGER = Logger.getLogger(DropwizardExports.class.getName()); + private MetricRegistry registry; + private MetricFilter metricFilter; + private SampleBuilder sampleBuilder; + + /** + * Creates a new DropwizardExports with a {@link DefaultSampleBuilder} and {@link MetricFilter#ALL}. + * + * @param registry a metric registry to export in prometheus. + */ + public DropwizardExports(MetricRegistry registry) { + this.registry = registry; + this.metricFilter = MetricFilter.ALL; + this.sampleBuilder = new DefaultSampleBuilder(); + } + + /** + * Creates a new DropwizardExports with a {@link DefaultSampleBuilder} and custom {@link MetricFilter}. + * + * @param registry a metric registry to export in prometheus. + * @param metricFilter a custom metric filter. + */ + public DropwizardExports(MetricRegistry registry, MetricFilter metricFilter) { + this.registry = registry; + this.metricFilter = metricFilter; + this.sampleBuilder = new DefaultSampleBuilder(); + } + + /** + * @param registry a metric registry to export in prometheus. + * @param sampleBuilder sampleBuilder to use to create prometheus samples. + */ + public DropwizardExports(MetricRegistry registry, SampleBuilder sampleBuilder) { + this.registry = registry; + this.metricFilter = MetricFilter.ALL; + this.sampleBuilder = sampleBuilder; + } + + /** + * @param registry a metric registry to export in prometheus. + * @param metricFilter a custom metric filter. + * @param sampleBuilder sampleBuilder to use to create prometheus samples. + */ + public DropwizardExports(MetricRegistry registry, MetricFilter metricFilter, SampleBuilder sampleBuilder) { + this.registry = registry; + this.metricFilter = metricFilter; + this.sampleBuilder = sampleBuilder; + } + + private static String getHelpMessage(String metricName, Metric metric) { + return String.format("Generated from Dropwizard metric import (metric=%s, type=%s)", metricName, metric.getClass().getName()); + } + + /** + * Export counter as Prometheus <a href="https://prometheus.io/docs/concepts/metric_types/#gauge";>Gauge</a>. + */ + MetricFamilySamples fromCounter(String dropwizardName, Counter counter) { + MetricFamilySamples.Sample sample = sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), new Long(counter.getCount()).doubleValue()); + return new MetricFamilySamples(sample.name, Type.GAUGE, getHelpMessage(dropwizardName, counter), Arrays.asList(sample)); + } + + /** + * Export gauge as a prometheus gauge. + */ + MetricFamilySamples fromGauge(String dropwizardName, Gauge gauge) { + Object obj = gauge.getValue(); + double value; + if (obj instanceof Number) { + value = ((Number) obj).doubleValue(); + } else if (obj instanceof Boolean) { + value = ((Boolean) obj) ? 1 : 0; + } else { + LOGGER.log(Level.FINE, String.format("Invalid type for Gauge %s: %s", sanitizeMetricName(dropwizardName), obj == null ? "null" : obj.getClass().getName())); + return null; + } + MetricFamilySamples.Sample sample = sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), value); + return new MetricFamilySamples(sample.name, Type.GAUGE, getHelpMessage(dropwizardName, gauge), Arrays.asList(sample)); + } + + /** + * Export a histogram snapshot as a prometheus SUMMARY. + * + * @param dropwizardName metric name. + * @param snapshot the histogram snapshot. + * @param count the total sample count for this snapshot. + * @param factor a factor to apply to histogram values. + */ + MetricFamilySamples fromSnapshotAndCount(String dropwizardName, Snapshot snapshot, long count, double factor, + String helpMessage) { + MetricName metricName = MetricUtils.stringToMetricName(dropwizardName); + Stat.HistogramType histogramType = Stat.HistogramType.valueOf(metricName.getType()); Review Comment: Stat.HistogramType.valueOf(metricName.getType()) throws IllegalArgumentException if the metric name's type field does not match a known enum constant. Since this is called inside collect() which iterates over ALL metrics in the registry, a single metric with an unrecognized type will crash the entire metrics collection, making the /metrics endpoint return an error for all metrics. Consider wrapping this in a try-catch that skips the individual metric and logs a warning, similar to how fromGauge handles invalid types. ########## metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/PrometheusSampleBuilder.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.rocketmq.connect.metrics; + +import io.prometheus.client.Collector; +import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder; +import java.util.Arrays; +import java.util.List; +import org.apache.commons.lang3.StringUtils; + +public class PrometheusSampleBuilder implements SampleBuilder { Review Comment: No test coverage is added for any of the new classes (DropwizardExports, PrometheusSampleBuilder, PrometheusMetricsServlet). Given the complex string-parsing logic in PrometheusSampleBuilder and the metric-type dispatch in DropwizardExports, unit tests are especially important to verify correct behavior with well-formed and malformed metric names, and to prevent regressions in the quantile label and suffix handling. ########## metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/DropwizardExports.java: ########## @@ -0,0 +1,234 @@ +/* + * 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.rocketmq.connect.metrics; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.Metric; +import com.codahale.metrics.MetricFilter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.Snapshot; +import com.codahale.metrics.Timer; +import io.prometheus.client.dropwizard.samplebuilder.DefaultSampleBuilder; +import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedMap; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.apache.rocketmq.connect.metrics.stats.Stat; + +/** + * Collect Dropwizard metrics from a MetricRegistry. + */ +public class DropwizardExports extends io.prometheus.client.Collector implements io.prometheus.client.Collector.Describable { + private static final Logger LOGGER = Logger.getLogger(DropwizardExports.class.getName()); + private MetricRegistry registry; + private MetricFilter metricFilter; + private SampleBuilder sampleBuilder; + + /** + * Creates a new DropwizardExports with a {@link DefaultSampleBuilder} and {@link MetricFilter#ALL}. + * + * @param registry a metric registry to export in prometheus. + */ + public DropwizardExports(MetricRegistry registry) { + this.registry = registry; + this.metricFilter = MetricFilter.ALL; + this.sampleBuilder = new DefaultSampleBuilder(); + } + + /** + * Creates a new DropwizardExports with a {@link DefaultSampleBuilder} and custom {@link MetricFilter}. + * + * @param registry a metric registry to export in prometheus. + * @param metricFilter a custom metric filter. + */ + public DropwizardExports(MetricRegistry registry, MetricFilter metricFilter) { + this.registry = registry; + this.metricFilter = metricFilter; + this.sampleBuilder = new DefaultSampleBuilder(); + } + + /** + * @param registry a metric registry to export in prometheus. + * @param sampleBuilder sampleBuilder to use to create prometheus samples. + */ + public DropwizardExports(MetricRegistry registry, SampleBuilder sampleBuilder) { + this.registry = registry; + this.metricFilter = MetricFilter.ALL; + this.sampleBuilder = sampleBuilder; + } + + /** + * @param registry a metric registry to export in prometheus. + * @param metricFilter a custom metric filter. + * @param sampleBuilder sampleBuilder to use to create prometheus samples. + */ + public DropwizardExports(MetricRegistry registry, MetricFilter metricFilter, SampleBuilder sampleBuilder) { + this.registry = registry; + this.metricFilter = metricFilter; + this.sampleBuilder = sampleBuilder; + } + + private static String getHelpMessage(String metricName, Metric metric) { + return String.format("Generated from Dropwizard metric import (metric=%s, type=%s)", metricName, metric.getClass().getName()); + } + + /** + * Export counter as Prometheus <a href="https://prometheus.io/docs/concepts/metric_types/#gauge";>Gauge</a>. + */ + MetricFamilySamples fromCounter(String dropwizardName, Counter counter) { + MetricFamilySamples.Sample sample = sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), new Long(counter.getCount()).doubleValue()); + return new MetricFamilySamples(sample.name, Type.GAUGE, getHelpMessage(dropwizardName, counter), Arrays.asList(sample)); + } + + /** + * Export gauge as a prometheus gauge. + */ + MetricFamilySamples fromGauge(String dropwizardName, Gauge gauge) { + Object obj = gauge.getValue(); + double value; + if (obj instanceof Number) { + value = ((Number) obj).doubleValue(); + } else if (obj instanceof Boolean) { + value = ((Boolean) obj) ? 1 : 0; + } else { + LOGGER.log(Level.FINE, String.format("Invalid type for Gauge %s: %s", sanitizeMetricName(dropwizardName), obj == null ? "null" : obj.getClass().getName())); + return null; + } + MetricFamilySamples.Sample sample = sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), value); + return new MetricFamilySamples(sample.name, Type.GAUGE, getHelpMessage(dropwizardName, gauge), Arrays.asList(sample)); + } + + /** + * Export a histogram snapshot as a prometheus SUMMARY. + * + * @param dropwizardName metric name. + * @param snapshot the histogram snapshot. + * @param count the total sample count for this snapshot. + * @param factor a factor to apply to histogram values. + */ + MetricFamilySamples fromSnapshotAndCount(String dropwizardName, Snapshot snapshot, long count, double factor, + String helpMessage) { + MetricName metricName = MetricUtils.stringToMetricName(dropwizardName); + Stat.HistogramType histogramType = Stat.HistogramType.valueOf(metricName.getType()); + List<MetricFamilySamples.Sample> samples = new ArrayList<>(); + switch (histogramType) { + case Avg: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), snapshot.getMean() * factor)); + break; + case Min: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), snapshot.getMin() * factor)); + break; + case Max: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", new ArrayList<String>(), new ArrayList<String>(), snapshot.getMax() * factor)); + break; + case Percentile_75th: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.75"), snapshot.get75thPercentile() * factor)); + break; + case Percentile_95th: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.95"), snapshot.get95thPercentile() * factor)); + break; + case Percentile_98th: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.98"), snapshot.get98thPercentile() * factor)); + break; + case Percentile_99th: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.99"), snapshot.get99thPercentile() * factor)); + break; + case Percentile_999th: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.999"), snapshot.get999thPercentile() * factor)); + break; + default: + samples = Arrays.asList(sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.5"), snapshot.getMedian() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.5"), snapshot.getMedian() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.75"), snapshot.get75thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.95"), snapshot.get95thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.98"), snapshot.get98thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.99"), snapshot.get99thPercentile() * factor), sampleBuilder.createSample(dropwizardName, "", Arrays.asList("quantile"), Arrays.asList("0.999"), snapshot.get999thPercentile() * factor), sampleBuilder. createSample(dropwizardName, "_count", new ArrayList<String>(), new ArrayList<String>(), count)); Review Comment: The default case of the switch in fromSnapshotAndCount creates a duplicate sample: two samples both with quantile="0.5" and snapshot.getMedian(). The original Prometheus DropwizardExports only includes the median sample once. This duplicate should be removed — the second createSample call with Arrays.asList("0.5") and snapshot.getMedian() is redundant. ########## metric-exporter/src/main/java/org/apache/rocketmq/connect/metrics/PrometheusSampleBuilder.java: ########## @@ -0,0 +1,59 @@ +/* + * 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.rocketmq.connect.metrics; + +import io.prometheus.client.Collector; +import io.prometheus.client.dropwizard.samplebuilder.SampleBuilder; +import java.util.Arrays; +import java.util.List; +import org.apache.commons.lang3.StringUtils; + +public class PrometheusSampleBuilder implements SampleBuilder { + private static final List<String> SOURCE_TASK_LABEL_NAMES = Arrays.asList("metric_group", "data_type", "connector", "task"); + + @Override + public Collector.MetricFamilySamples.Sample createSample(String dropwizardName, String nameSuffix, + List<String> additionalLabelNames, List<String> additionalLabelValues, double value) { + String suffix = nameSuffix == null ? "" : nameSuffix; + List<String> labelValues = sanitizeLabelValues(dropwizardName); + return new Collector.MetricFamilySamples.Sample(sanitizeMetricName(dropwizardName + suffix), SOURCE_TASK_LABEL_NAMES, labelValues, value); Review Comment: sanitizeMetricName(dropwizardName + suffix) appends the suffix to the full dropwizardName string before parsing, but sanitizeMetricName extracts only the 2nd comma-separated field (split(":")[1].split(",")[1]). The suffix (e.g., "_count") lands on the last field and is silently lost. This means the histogram count sample (which uses nameSuffix="_count") gets the same metric name as the percentile samples, producing conflicting samples with the same name and label set. The suffix should be appended to the extracted metric name, not to the raw input string. ########## rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/rest/RestHandler.java: ########## @@ -60,6 +65,21 @@ public RestHandler(AbstractConnectController connectController) { this.connectController = connectController; pluginsResource = new ConnectorPluginsResource(connectController); + Javalin embeddedApp = Javalin.create(config -> { Review Comment: The Javalin 'embeddedApp' instance for the metrics server is a local variable in the constructor and is never stored as a field. This means there is no way to stop or shut down the metrics Jetty server when the RestHandler or connect runtime is stopped, causing a port and thread resource leak. The embeddedApp reference should be stored as a field and stopped in the appropriate shutdown method. ########## rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java: ########## @@ -148,6 +151,7 @@ public Worker(WorkerConfig workerConfig, this.executor = Executors.newCachedThreadPool(); this.connectMetrics = new ConnectMetrics(workerConfig); this.stateManagementService = stateManagementService; + CollectorRegistry.defaultRegistry.register(new DropwizardExports(connectMetrics.registry(), new PrometheusSampleBuilder())); Review Comment: DropwizardExports is registered with the global CollectorRegistry.defaultRegistry but is never unregistered when the Worker is stopped. On Worker restart (e.g., connector reconfiguration), registering the same collector type again will throw IllegalArgumentException('Collector already registered'), breaking the connector lifecycle. The registration should either use a dedicated CollectorRegistry (not the global singleton), or unregister the collector in the Worker's stop/shutdown method. ########## rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/rest/PrometheusMetricsServlet.java: ########## @@ -0,0 +1,127 @@ +/* + * 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.rocketmq.connect.runtime.rest; + +import io.prometheus.client.Collector; +import io.prometheus.client.CollectorRegistry; +import java.io.IOException; +import java.io.StringWriter; +import java.io.Writer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Enumeration; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import javax.servlet.ServletException; +import javax.servlet.http.HttpServlet; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; + +public class PrometheusMetricsServlet extends HttpServlet { + private CollectorRegistry registry; + + public PrometheusMetricsServlet() { + this(CollectorRegistry.defaultRegistry); + } + + public PrometheusMetricsServlet(CollectorRegistry registry) { + this.registry = registry; + } + + protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { + resp.setStatus(200); + resp.setContentType("text/plain; version=0.0.4; charset=utf-8"); + StringWriter writer = new StringWriter(); + + this.writeEscapedHelp(writer, registry); + resp.getOutputStream().print(writer.toString()); + } + + public void writeEscapedHelp(StringWriter writer, CollectorRegistry registry) throws IOException { + Enumeration<Collector.MetricFamilySamples> metricFamilySamplesEnumeration = registry.metricFamilySamples(); + List<Collector.MetricFamilySamples> list = new ArrayList<>(); + while (metricFamilySamplesEnumeration.hasMoreElements()) { + Collector.MetricFamilySamples metricFamilySamples = metricFamilySamplesEnumeration.nextElement(); + list.add(metricFamilySamples); + } + writeEscapedHelp(writer, list); + } + + public void writeEscapedHelp(StringWriter writer, List<Collector.MetricFamilySamples> mfs) throws IOException { + if (Objects.nonNull(mfs) && mfs.size() != 0) { + for (Collector.MetricFamilySamples metricFamilySamples : mfs) { + for (Iterator var3 = metricFamilySamples.samples.iterator(); var3.hasNext(); writer.write(10)) { + Collector.MetricFamilySamples.Sample sample = (Collector.MetricFamilySamples.Sample) var3.next(); + writer.write(sample.name); + if (sample.labelNames.size() > 0) { + writer.write(123); + + for (int i = 0; i < sample.labelNames.size(); ++i) { + writer.write((String) sample.labelNames.get(i)); + writer.write("=\""); + writeEscapedLabelValue(writer, (String) sample.labelValues.get(i)); + writer.write("\","); + } + + writer.write(125); + } + + writer.write(32); + writer.write(Collector.doubleToGoString(sample.value)); + if (sample.timestampMs != null) { + writer.write(32); + writer.write(sample.timestampMs.toString()); + } + } + } + } + + } + + private static void writeEscapedLabelValue(Writer writer, String s) throws IOException { + for (int i = 0; i < s.length(); ++i) { + char c = s.charAt(i); + switch (c) { + case '\n': + writer.append("\\n"); + break; + case '"': + writer.append("\\\""); + break; + case '\\': + writer.append("\\\\"); + break; + default: + writer.append(c); + } + } + + } + + private Set<String> parse(HttpServletRequest req) { Review Comment: The parse(HttpServletRequest) method is dead code — it is never called anywhere in the class. It appears to be copied from an upstream Prometheus servlet implementation but was not wired into doGet. Either remove it or use it to support the name[] query parameter for filtering which metrics are returned. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
