asafm commented on code in PR #17531:
URL: https://github.com/apache/pulsar/pull/17531#discussion_r983490703
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/PrometheusMetricsGenerator.java:
##########
@@ -315,12 +314,6 @@ private static void
generateManagedLedgerBookieClientMetrics(PulsarService pulsa
return;
}
- try {
- Writer writer = new StringWriter();
- statsProvider.writeAllMetrics(writer);
- stream.write(writer.toString());
- } catch (IOException e) {
- // nop
- }
+ ((PrometheusMetricsProvider) statsProvider).generate(stream);
Review Comment:
It's weird to have a cast in software you wrote - you usually see this when
you use third-party software. Why can't you change the original variable type?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/CachingStatsLogger.java:
##########
@@ -0,0 +1,138 @@
+/**
+ * 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.pulsar.broker.stats.prometheus.metrics;
+
+import com.google.common.base.Joiner;
+import io.prometheus.client.Collector;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.Gauge;
+import org.apache.bookkeeper.stats.OpStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+/**
+ * A {@code StatsLogger} that caches the stats objects created by other {@code
StatsLogger}.
+ */
+public class CachingStatsLogger implements StatsLogger {
+
+ protected final StatsLogger underlying;
+ protected final ConcurrentMap<ScopeContext, Counter> counters;
+ protected final ConcurrentMap<ScopeContext, OpStatsLogger> opStatsLoggers;
+ protected final ConcurrentMap<ScopeContext, StatsLogger> scopeStatsLoggers;
+ protected final ConcurrentMap<ScopeContext, StatsLogger>
scopeLabelStatsLoggers;
+
+ private final String scope;
+ private final Map<String, String> labels;
+
+ public CachingStatsLogger(String scope, StatsLogger statsLogger,
Map<String, String> labels) {
+ this.scope = scope;
+ this.labels = labels;
+ this.underlying = statsLogger;
+ this.counters = new ConcurrentHashMap<>();
+ this.opStatsLoggers = new ConcurrentHashMap<>();
+ this.scopeStatsLoggers = new ConcurrentHashMap<>();
+ this.scopeLabelStatsLoggers = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public int hashCode() {
+ return underlying.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof CachingStatsLogger)) {
+ return false;
+ }
+ CachingStatsLogger another = (CachingStatsLogger) obj;
+ return underlying.equals(another.underlying);
+ }
+
+ @Override
+ public String toString() {
+ return underlying.toString();
+ }
+
+ @Override
+ public OpStatsLogger getOpStatsLogger(String name) {
+ return opStatsLoggers.computeIfAbsent(scopeContext(name), x ->
underlying.getOpStatsLogger(name));
Review Comment:
Why would you cache it under key = `scopeContext(name)` and not just `name`?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusMetricsProvider.java:
##########
@@ -24,37 +24,44 @@
import io.prometheus.client.Collector;
import java.io.IOException;
import java.io.Writer;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import org.apache.bookkeeper.stats.CachingStatsProvider;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.StatsProvider;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider;
+import org.apache.pulsar.common.util.SimpleTextOutputStream;
/**
- * A <i>Prometheus</i> based {@link StatsProvider} implementation.
+ * A <i>Prometheus</i> based {@link PrometheusRawMetricsProvider}
implementation.
*/
-public class PrometheusMetricsProvider implements StatsProvider {
+public class PrometheusMetricsProvider implements StatsProvider,
PrometheusRawMetricsProvider {
Review Comment:
Based on my reply to the comment, I think it's redundant. We should have a
single `PrometheusMetricsProvider` in `PulsarService`, and its metrics should
be written explicitly in `PrometheusMetricsGenerator`
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/PrometheusTextFormat.java:
##########
@@ -0,0 +1,179 @@
+/**
+ * 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.pulsar.broker.stats.prometheus.metrics;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pulsar.common.util.SimpleTextOutputStream;
+
+/**
+ * Logic to write metrics in Prometheus text format.
+ */
+public class PrometheusTextFormat {
+
+ Set<String> metricNameSet = new HashSet<>();
+
+ public void writeGauge(SimpleTextOutputStream w, String name,
SimpleGauge<? extends Number> gauge) {
+ // Example:
+ // # TYPE bookie_storage_entries_count gauge
+ // bookie_storage_entries_count 519
+ writeType(w, name, "gauge");
Review Comment:
Now that you have grouped the metrics together by metric name, the type
needs to be written once per metric name, so called from outside.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/CachingStatsLogger.java:
##########
@@ -0,0 +1,138 @@
+/**
+ * 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.pulsar.broker.stats.prometheus.metrics;
+
+import com.google.common.base.Joiner;
+import io.prometheus.client.Collector;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.Gauge;
+import org.apache.bookkeeper.stats.OpStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+/**
+ * A {@code StatsLogger} that caches the stats objects created by other {@code
StatsLogger}.
+ */
+public class CachingStatsLogger implements StatsLogger {
+
+ protected final StatsLogger underlying;
+ protected final ConcurrentMap<ScopeContext, Counter> counters;
+ protected final ConcurrentMap<ScopeContext, OpStatsLogger> opStatsLoggers;
+ protected final ConcurrentMap<ScopeContext, StatsLogger> scopeStatsLoggers;
+ protected final ConcurrentMap<ScopeContext, StatsLogger>
scopeLabelStatsLoggers;
+
+ private final String scope;
+ private final Map<String, String> labels;
+
+ public CachingStatsLogger(String scope, StatsLogger statsLogger,
Map<String, String> labels) {
+ this.scope = scope;
+ this.labels = labels;
+ this.underlying = statsLogger;
+ this.counters = new ConcurrentHashMap<>();
+ this.opStatsLoggers = new ConcurrentHashMap<>();
+ this.scopeStatsLoggers = new ConcurrentHashMap<>();
+ this.scopeLabelStatsLoggers = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public int hashCode() {
+ return underlying.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof CachingStatsLogger)) {
+ return false;
+ }
+ CachingStatsLogger another = (CachingStatsLogger) obj;
+ return underlying.equals(another.underlying);
+ }
+
+ @Override
+ public String toString() {
+ return underlying.toString();
+ }
+
+ @Override
+ public OpStatsLogger getOpStatsLogger(String name) {
+ return opStatsLoggers.computeIfAbsent(scopeContext(name), x ->
underlying.getOpStatsLogger(name));
+ }
+
+ @Override
+ public Counter getCounter(String name) {
+ return counters.computeIfAbsent(scopeContext(name), x ->
underlying.getCounter(name));
+ }
+
+ @Override
+ public <T extends Number> void registerGauge(String name, Gauge<T> gauge) {
+ underlying.registerGauge(name, gauge);
+ }
+
+ @Override
+ public <T extends Number> void unregisterGauge(String name, Gauge<T>
gauge) {
+ underlying.unregisterGauge(name, gauge);
+ }
+
+ @Override
+ public StatsLogger scope(String name) {
+ return scopeStatsLoggers.computeIfAbsent(scopeContext(name),
+ x -> new CachingStatsLogger(scope, underlying.scope(name),
labels));
+ }
+
+ @Override
+ public StatsLogger scopeLabel(String labelName, String labelValue) {
+ Map<String, String> newLabels = new TreeMap<>(labels);
+ newLabels.put(labelName, labelValue);
+ return scopeLabelStatsLoggers.computeIfAbsent(new
ScopeContext(completeName(""), newLabels),
+ x -> new CachingStatsLogger(scope,
underlying.scopeLabel(labelName, labelValue), newLabels));
+ }
+
+
+ @Override
+ public void removeScope(String name, StatsLogger statsLogger) {
+ scopeStatsLoggers.remove(scopeContext(name), statsLogger);
+ }
+
+ /**
+ Thread-scoped stats not currently supported.
Review Comment:
Why is it not supported?
--
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]
