ArvinZheng commented on a change in pull request #10412: URL: https://github.com/apache/druid/pull/10412#discussion_r513125045
########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitter.java ########## @@ -0,0 +1,153 @@ +/* + * 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.druid.emitter.prometheus; + + +import com.google.common.collect.ImmutableMap; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.exporter.HTTPServer; +import io.prometheus.client.exporter.PushGateway; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.emitter.core.Emitter; +import org.apache.druid.java.util.emitter.core.Event; +import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; + +import java.io.IOException; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * + */ +public class PrometheusEmitter implements Emitter +{ + + private static final Logger log = new Logger(PrometheusEmitter.class); + private final Metrics metrics; + private final PrometheusEmitterConfig config; + private final PrometheusEmitterConfig.Strategy strategy; + private final Pattern pattern = Pattern.compile("[^a-zA-Z0-9_][^a-zA-Z0-9_]*"); Review comment: Reuse the pattern in `PrometheusEmitterConfig` ########## File path: extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json ########## @@ -0,0 +1,128 @@ +{ + "query/time" : { "dimensions" : ["dataSource", "type"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to complete a query."}, + "query/bytes" : { "dimensions" : ["dataSource", "type"], "type" : "count", "help": "Number of bytes returned in query response."}, + "query/node/time" : { "dimensions" : ["server"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to query individual historical/realtime processes."}, + "query/node/ttfb" : { "dimensions" : ["server"], "type" : "timer", "help": "Time to first byte. Seconds elapsed until Broker starts receiving the response from individual historical/realtime processes."}, + "query/node/bytes" : { "dimensions" : ["server"], "type" : "count", "help": "Number of bytes returned from querying individual historical/realtime processes."}, + "query/node/backpressure": { "dimensions" : ["server"], "type" : "timer", "help": "Seconds that the channel to this process has spent suspended due to backpressure."}, + "query/intervalChunk/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Only emitted if interval chunking is enabled. Milliseconds required to query an interval chunk. This metric is deprecated and will be removed in the future because interval chunking is deprecated."}, + + "query/segment/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to query individual segment. Includes time to page in the segment from disk."}, + "query/wait/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds spent waiting for a segment to be scanned."}, + "segment/scan/pending" : { "dimensions" : [], "type" : "gauge", "help": "Number of segments in queue waiting to be scanned."}, + "query/segmentAndCache/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to query individual segment or hit the cache (if it is enabled on the Historical process)."}, + "query/cpu/time" : { "dimensions" : ["dataSource", "type"], "type" : "timer", "conversionFactor": "1000000", "help": "Seconds of CPU time taken to complete a query"}, + + "query/count" : { "dimensions" : [], "type" : "count", "help": "Number of total queries" }, + "query/success/count" : { "dimensions" : [], "type" : "count", "help": "Number of queries successfully processed"}, + "query/failed/count" : { "dimensions" : [], "type" : "count", "help": "Number of failed queries"}, + "query/interrupted/count" : { "dimensions" : [], "type" : "count", "help": "Number of queries interrupted due to cancellation or timeout"}, + + "query/cache/delta/numEntries" : { "dimensions" : [], "type" : "count", "help": "Number of entries in cache"}, Review comment: potential bug, deltas can be negative but Prometheus counter accepts only non-negative increment. ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java ########## @@ -0,0 +1,157 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Strings; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.SimpleCollector; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.SortedSet; +import java.util.regex.Pattern; + +public class Metrics +{ + + private static final Logger log = new Logger(Metrics.class); + private final Map<String, DimensionsAndCollector> map = new HashMap<>(); + private final ObjectMapper mapper = new ObjectMapper(); + private final Pattern pattern = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*"); + + public DimensionsAndCollector getByName(String name, String service) + { + if (map.containsKey(name)) { + return map.get(name); + } else if (map.containsKey(service + "_" + name)) { + return map.get(service + "_" + name); + } else { + return null; + } + } + + public Metrics(String namespace, String path) + { + Map<String, Metric> metrics = readMap(path); + for (Map.Entry<String, Metric> entry : metrics.entrySet()) { + String name = entry.getKey(); + Metric metric = entry.getValue(); + Metric.Type type = metric.type; + String[] dimensions = metric.dimensions.toArray(new String[0]); + String formattedName = pattern.matcher(StringUtils.toLowerCase(name)).replaceAll("_"); + SimpleCollector collector = null; + if (Metric.Type.count.equals(type)) { + collector = new Counter.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.gauge.equals(type)) { + collector = new Gauge.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.timer.equals(type)) { + collector = new Histogram.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .buckets(.1, .25, .5, .75, 1, 2.5, 5, 7.5, 10, 30, 60, 120, 300) + .help(metric.help) + .register(); + } else { + log.error("Unrecognized metric type [%s]", type); + } + + if (collector != null) { + map.put(name, new DimensionsAndCollector(dimensions, collector, metric.conversionFactor)); + } + } + + } + + private Map<String, Metric> readMap(String path) + { + try { + InputStream is; + if (Strings.isNullOrEmpty(path)) { + log.info("Using default metric dimension and types"); + is = this.getClass().getClassLoader().getResourceAsStream("defaultMetrics.json"); + } else { + log.info("Using metric dimensions at types at [%s]", path); Review comment: typo, `metric dimensions at types` -> `metric dimensions and types`, maybe we could replace it by `metric configuration` ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitter.java ########## @@ -0,0 +1,153 @@ +/* + * 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.druid.emitter.prometheus; + + +import com.google.common.collect.ImmutableMap; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.exporter.HTTPServer; +import io.prometheus.client.exporter.PushGateway; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.emitter.core.Emitter; +import org.apache.druid.java.util.emitter.core.Event; +import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; + +import java.io.IOException; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * + */ +public class PrometheusEmitter implements Emitter +{ + + private static final Logger log = new Logger(PrometheusEmitter.class); + private final Metrics metrics; + private final PrometheusEmitterConfig config; + private final PrometheusEmitterConfig.Strategy strategy; + private final Pattern pattern = Pattern.compile("[^a-zA-Z0-9_][^a-zA-Z0-9_]*"); + + private HTTPServer server; + private PushGateway pushGateway; + private String identifier; + + static PrometheusEmitter of(PrometheusEmitterConfig config) + { + return new PrometheusEmitter(config); + } + + public PrometheusEmitter(PrometheusEmitterConfig config) + { + this.config = config; + this.strategy = config.getStrategy(); + metrics = new Metrics(config.getNamespace(), config.getDimensionMapPath()); + } + + + @Override + public void start() + { + if (strategy.equals(PrometheusEmitterConfig.Strategy.exporter)) { + if (server == null) { + try { + server = new HTTPServer(config.getPort()); + } + catch (IOException e) { + log.error(e, "Unable to start prometheus HTTPServer"); + } + } else { + log.error("HTTPServer is already started"); + } + } else if (strategy.equals(PrometheusEmitterConfig.Strategy.pushgateway)) { + pushGateway = new PushGateway(config.getPushGatewayAddress()); + } + + } + + @Override + public void emit(Event event) + { + if (event instanceof ServiceMetricEvent) { + emitMetric((ServiceMetricEvent) event); + } + } + + void emitMetric(ServiceMetricEvent metricEvent) Review comment: private? ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitterConfig.java ########## @@ -0,0 +1,104 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.base.Preconditions; + +import javax.annotation.Nullable; +import javax.xml.ws.BindingType; +import java.util.regex.Pattern; + +/** + * + */ +public class PrometheusEmitterConfig +{ + + Pattern pattern = Pattern.compile("[a-zA-Z_:][a-zA-Z0-9_:]*"); Review comment: final, and static maybe, so that we can reuse this object/regex in other places ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java ########## @@ -0,0 +1,157 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Strings; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.SimpleCollector; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.SortedSet; +import java.util.regex.Pattern; + +public class Metrics +{ + + private static final Logger log = new Logger(Metrics.class); + private final Map<String, DimensionsAndCollector> map = new HashMap<>(); + private final ObjectMapper mapper = new ObjectMapper(); + private final Pattern pattern = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*"); + + public DimensionsAndCollector getByName(String name, String service) + { + if (map.containsKey(name)) { + return map.get(name); + } else if (map.containsKey(service + "_" + name)) { + return map.get(service + "_" + name); + } else { + return null; + } + } + + public Metrics(String namespace, String path) + { + Map<String, Metric> metrics = readMap(path); + for (Map.Entry<String, Metric> entry : metrics.entrySet()) { + String name = entry.getKey(); + Metric metric = entry.getValue(); + Metric.Type type = metric.type; + String[] dimensions = metric.dimensions.toArray(new String[0]); + String formattedName = pattern.matcher(StringUtils.toLowerCase(name)).replaceAll("_"); + SimpleCollector collector = null; + if (Metric.Type.count.equals(type)) { + collector = new Counter.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.gauge.equals(type)) { + collector = new Gauge.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.timer.equals(type)) { + collector = new Histogram.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .buckets(.1, .25, .5, .75, 1, 2.5, 5, 7.5, 10, 30, 60, 120, 300) + .help(metric.help) + .register(); + } else { + log.error("Unrecognized metric type [%s]", type); + } + + if (collector != null) { + map.put(name, new DimensionsAndCollector(dimensions, collector, metric.conversionFactor)); + } + } + + } + + private Map<String, Metric> readMap(String path) Review comment: maybe `readConfig` or `loadConfig`? ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java ########## @@ -0,0 +1,157 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Strings; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.SimpleCollector; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.SortedSet; +import java.util.regex.Pattern; + +public class Metrics +{ + + private static final Logger log = new Logger(Metrics.class); + private final Map<String, DimensionsAndCollector> map = new HashMap<>(); + private final ObjectMapper mapper = new ObjectMapper(); + private final Pattern pattern = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*"); Review comment: Reuse the pattern in `PrometheusEmitterConfig` ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java ########## @@ -0,0 +1,157 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Strings; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.SimpleCollector; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.SortedSet; +import java.util.regex.Pattern; + +public class Metrics +{ + + private static final Logger log = new Logger(Metrics.class); + private final Map<String, DimensionsAndCollector> map = new HashMap<>(); + private final ObjectMapper mapper = new ObjectMapper(); + private final Pattern pattern = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*"); + + public DimensionsAndCollector getByName(String name, String service) + { + if (map.containsKey(name)) { + return map.get(name); + } else if (map.containsKey(service + "_" + name)) { + return map.get(service + "_" + name); + } else { + return null; + } + } + + public Metrics(String namespace, String path) + { + Map<String, Metric> metrics = readMap(path); + for (Map.Entry<String, Metric> entry : metrics.entrySet()) { + String name = entry.getKey(); + Metric metric = entry.getValue(); + Metric.Type type = metric.type; + String[] dimensions = metric.dimensions.toArray(new String[0]); + String formattedName = pattern.matcher(StringUtils.toLowerCase(name)).replaceAll("_"); + SimpleCollector collector = null; + if (Metric.Type.count.equals(type)) { + collector = new Counter.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.gauge.equals(type)) { + collector = new Gauge.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.timer.equals(type)) { + collector = new Histogram.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .buckets(.1, .25, .5, .75, 1, 2.5, 5, 7.5, 10, 30, 60, 120, 300) + .help(metric.help) + .register(); + } else { + log.error("Unrecognized metric type [%s]", type); + } + + if (collector != null) { + map.put(name, new DimensionsAndCollector(dimensions, collector, metric.conversionFactor)); + } + } + + } + + private Map<String, Metric> readMap(String path) + { + try { + InputStream is; + if (Strings.isNullOrEmpty(path)) { + log.info("Using default metric dimension and types"); Review comment: maybe replace `metric dimensions and types` by `metric configuration`? ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java ########## @@ -0,0 +1,157 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Strings; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.SimpleCollector; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.SortedSet; +import java.util.regex.Pattern; + +public class Metrics +{ + + private static final Logger log = new Logger(Metrics.class); + private final Map<String, DimensionsAndCollector> map = new HashMap<>(); + private final ObjectMapper mapper = new ObjectMapper(); + private final Pattern pattern = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*"); + + public DimensionsAndCollector getByName(String name, String service) + { + if (map.containsKey(name)) { + return map.get(name); + } else if (map.containsKey(service + "_" + name)) { + return map.get(service + "_" + name); + } else { + return null; + } + } + + public Metrics(String namespace, String path) + { + Map<String, Metric> metrics = readMap(path); + for (Map.Entry<String, Metric> entry : metrics.entrySet()) { + String name = entry.getKey(); + Metric metric = entry.getValue(); + Metric.Type type = metric.type; + String[] dimensions = metric.dimensions.toArray(new String[0]); + String formattedName = pattern.matcher(StringUtils.toLowerCase(name)).replaceAll("_"); + SimpleCollector collector = null; + if (Metric.Type.count.equals(type)) { + collector = new Counter.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.gauge.equals(type)) { + collector = new Gauge.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.timer.equals(type)) { + collector = new Histogram.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .buckets(.1, .25, .5, .75, 1, 2.5, 5, 7.5, 10, 30, 60, 120, 300) + .help(metric.help) + .register(); + } else { + log.error("Unrecognized metric type [%s]", type); + } + + if (collector != null) { + map.put(name, new DimensionsAndCollector(dimensions, collector, metric.conversionFactor)); + } + } + + } + + private Map<String, Metric> readMap(String path) + { + try { + InputStream is; + if (Strings.isNullOrEmpty(path)) { + log.info("Using default metric dimension and types"); + is = this.getClass().getClassLoader().getResourceAsStream("defaultMetrics.json"); + } else { + log.info("Using metric dimensions at types at [%s]", path); + is = new FileInputStream(new File(path)); + } + return mapper.readerFor(new TypeReference<Map<String, Metric>>() + { + }).readValue(is); + } + catch (IOException e) { + throw new ISE(e, "Failed to parse metric dimensions and types"); + } + } + + public Map<String, DimensionsAndCollector> getMap() Review comment: maybe we can rename the `map` to `registeredMetrics` and then we could rename this method to `getRegisteredMetrics()`, I feel like this will be easier to read ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java ########## @@ -0,0 +1,157 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Strings; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.SimpleCollector; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.SortedSet; +import java.util.regex.Pattern; + +public class Metrics +{ + + private static final Logger log = new Logger(Metrics.class); + private final Map<String, DimensionsAndCollector> map = new HashMap<>(); + private final ObjectMapper mapper = new ObjectMapper(); + private final Pattern pattern = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*"); + + public DimensionsAndCollector getByName(String name, String service) + { + if (map.containsKey(name)) { + return map.get(name); + } else if (map.containsKey(service + "_" + name)) { + return map.get(service + "_" + name); + } else { + return null; + } + } + + public Metrics(String namespace, String path) + { + Map<String, Metric> metrics = readMap(path); + for (Map.Entry<String, Metric> entry : metrics.entrySet()) { + String name = entry.getKey(); + Metric metric = entry.getValue(); + Metric.Type type = metric.type; + String[] dimensions = metric.dimensions.toArray(new String[0]); + String formattedName = pattern.matcher(StringUtils.toLowerCase(name)).replaceAll("_"); + SimpleCollector collector = null; + if (Metric.Type.count.equals(type)) { + collector = new Counter.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.gauge.equals(type)) { + collector = new Gauge.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .help(metric.help) + .register(); + } else if (Metric.Type.timer.equals(type)) { + collector = new Histogram.Builder() + .namespace(namespace) + .name(formattedName) + .labelNames(dimensions) + .buckets(.1, .25, .5, .75, 1, 2.5, 5, 7.5, 10, 30, 60, 120, 300) + .help(metric.help) + .register(); + } else { + log.error("Unrecognized metric type [%s]", type); + } + + if (collector != null) { + map.put(name, new DimensionsAndCollector(dimensions, collector, metric.conversionFactor)); + } + } + + } + + private Map<String, Metric> readMap(String path) + { + try { + InputStream is; + if (Strings.isNullOrEmpty(path)) { + log.info("Using default metric dimension and types"); + is = this.getClass().getClassLoader().getResourceAsStream("defaultMetrics.json"); + } else { + log.info("Using metric dimensions at types at [%s]", path); + is = new FileInputStream(new File(path)); + } + return mapper.readerFor(new TypeReference<Map<String, Metric>>() + { + }).readValue(is); + } + catch (IOException e) { + throw new ISE(e, "Failed to parse metric dimensions and types"); Review comment: same as above ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitter.java ########## @@ -0,0 +1,153 @@ +/* + * 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.druid.emitter.prometheus; + + +import com.google.common.collect.ImmutableMap; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.exporter.HTTPServer; +import io.prometheus.client.exporter.PushGateway; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.emitter.core.Emitter; +import org.apache.druid.java.util.emitter.core.Event; +import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; + +import java.io.IOException; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * + */ +public class PrometheusEmitter implements Emitter +{ + + private static final Logger log = new Logger(PrometheusEmitter.class); + private final Metrics metrics; + private final PrometheusEmitterConfig config; + private final PrometheusEmitterConfig.Strategy strategy; + private final Pattern pattern = Pattern.compile("[^a-zA-Z0-9_][^a-zA-Z0-9_]*"); + + private HTTPServer server; + private PushGateway pushGateway; + private String identifier; + + static PrometheusEmitter of(PrometheusEmitterConfig config) + { + return new PrometheusEmitter(config); + } + + public PrometheusEmitter(PrometheusEmitterConfig config) + { + this.config = config; + this.strategy = config.getStrategy(); + metrics = new Metrics(config.getNamespace(), config.getDimensionMapPath()); + } + + + @Override + public void start() + { + if (strategy.equals(PrometheusEmitterConfig.Strategy.exporter)) { + if (server == null) { + try { + server = new HTTPServer(config.getPort()); + } + catch (IOException e) { + log.error(e, "Unable to start prometheus HTTPServer"); + } + } else { + log.error("HTTPServer is already started"); + } + } else if (strategy.equals(PrometheusEmitterConfig.Strategy.pushgateway)) { + pushGateway = new PushGateway(config.getPushGatewayAddress()); + } + + } + + @Override + public void emit(Event event) + { + if (event instanceof ServiceMetricEvent) { + emitMetric((ServiceMetricEvent) event); + } + } + + void emitMetric(ServiceMetricEvent metricEvent) + { + String name = metricEvent.getMetric(); + String service = metricEvent.getService(); + Map<String, Object> userDims = metricEvent.getUserDims(); + identifier = (userDims.get("task") == null ? metricEvent.getHost() : (String) userDims.get("task")); + Number value = metricEvent.getValue(); + + DimensionsAndCollector metric = metrics.getByName(name, service); + if (metric != null) { + String[] labelValues = new String[metric.getDimensions().length]; + String[] labelNames = metric.getDimensions(); + for (int i = 0; i < labelValues.length; i++) { + String labelName = labelNames[i]; + //labelName is controlled by the user. Instead of potential NPE on invalid labelName we use "unknown" as the dimension value + Object userDim = userDims.get(labelName); + labelValues[i] = userDim != null ? pattern.matcher(userDim.toString()).replaceAll("_") : "unknown"; + } + + if (metric.getCollector() instanceof Counter) { + ((Counter) metric.getCollector()).labels(labelValues).inc(value.doubleValue()); + } else if (metric.getCollector() instanceof Gauge) { + ((Gauge) metric.getCollector()).labels(labelValues).set(value.doubleValue()); + } else if (metric.getCollector() instanceof Histogram) { + ((Histogram) metric.getCollector()).labels(labelValues).observe(value.doubleValue() / metric.getConversionFactor()); + } else { + log.error("Unrecognized metric type [%s]", metric.getCollector().getClass()); + } + } else { + log.debug("Unmapped metric [%s]", name); + } + } + + @Override + public void flush() + { + Map<String, DimensionsAndCollector> map = metrics.getMap(); + try { + for (DimensionsAndCollector collector : map.values()) { + pushGateway.push(collector.getCollector(), config.getNamespace(), ImmutableMap.of(config.getNamespace(), identifier)); Review comment: potential NPE? if the configured strategy is not pushgateway, then this `pushGateway` wouldn't have been instantiated ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitter.java ########## @@ -0,0 +1,153 @@ +/* + * 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.druid.emitter.prometheus; + + +import com.google.common.collect.ImmutableMap; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.exporter.HTTPServer; +import io.prometheus.client.exporter.PushGateway; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.emitter.core.Emitter; +import org.apache.druid.java.util.emitter.core.Event; +import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; + +import java.io.IOException; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * + */ +public class PrometheusEmitter implements Emitter +{ + + private static final Logger log = new Logger(PrometheusEmitter.class); + private final Metrics metrics; + private final PrometheusEmitterConfig config; + private final PrometheusEmitterConfig.Strategy strategy; + private final Pattern pattern = Pattern.compile("[^a-zA-Z0-9_][^a-zA-Z0-9_]*"); + + private HTTPServer server; + private PushGateway pushGateway; + private String identifier; + + static PrometheusEmitter of(PrometheusEmitterConfig config) + { + return new PrometheusEmitter(config); + } + + public PrometheusEmitter(PrometheusEmitterConfig config) + { + this.config = config; + this.strategy = config.getStrategy(); + metrics = new Metrics(config.getNamespace(), config.getDimensionMapPath()); + } + + + @Override + public void start() Review comment: we should schedule a task to push updates periodically when the strategy is set to `pushgateway` ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/Metrics.java ########## @@ -0,0 +1,157 @@ +/* + * 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.druid.emitter.prometheus; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Strings; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.SimpleCollector; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.logger.Logger; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.SortedSet; +import java.util.regex.Pattern; + +public class Metrics +{ + + private static final Logger log = new Logger(Metrics.class); + private final Map<String, DimensionsAndCollector> map = new HashMap<>(); + private final ObjectMapper mapper = new ObjectMapper(); + private final Pattern pattern = Pattern.compile("[^a-zA-Z_:][^a-zA-Z0-9_:]*"); + + public DimensionsAndCollector getByName(String name, String service) + { + if (map.containsKey(name)) { Review comment: `return Optional.ofNullable(map.get(name)).orElse(map.get(service + "_" + name));` ########## File path: extensions-contrib/prometheus-emitter/src/main/java/org/apache/druid/emitter/prometheus/PrometheusEmitter.java ########## @@ -0,0 +1,153 @@ +/* + * 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.druid.emitter.prometheus; + + +import com.google.common.collect.ImmutableMap; +import io.prometheus.client.Counter; +import io.prometheus.client.Gauge; +import io.prometheus.client.Histogram; +import io.prometheus.client.exporter.HTTPServer; +import io.prometheus.client.exporter.PushGateway; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.emitter.core.Emitter; +import org.apache.druid.java.util.emitter.core.Event; +import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; + +import java.io.IOException; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * + */ +public class PrometheusEmitter implements Emitter +{ + + private static final Logger log = new Logger(PrometheusEmitter.class); + private final Metrics metrics; + private final PrometheusEmitterConfig config; + private final PrometheusEmitterConfig.Strategy strategy; + private final Pattern pattern = Pattern.compile("[^a-zA-Z0-9_][^a-zA-Z0-9_]*"); + + private HTTPServer server; + private PushGateway pushGateway; + private String identifier; + + static PrometheusEmitter of(PrometheusEmitterConfig config) + { + return new PrometheusEmitter(config); + } + + public PrometheusEmitter(PrometheusEmitterConfig config) + { + this.config = config; + this.strategy = config.getStrategy(); + metrics = new Metrics(config.getNamespace(), config.getDimensionMapPath()); + } + + + @Override + public void start() + { + if (strategy.equals(PrometheusEmitterConfig.Strategy.exporter)) { + if (server == null) { + try { + server = new HTTPServer(config.getPort()); + } + catch (IOException e) { + log.error(e, "Unable to start prometheus HTTPServer"); + } + } else { + log.error("HTTPServer is already started"); + } + } else if (strategy.equals(PrometheusEmitterConfig.Strategy.pushgateway)) { + pushGateway = new PushGateway(config.getPushGatewayAddress()); + } + + } + + @Override + public void emit(Event event) + { + if (event instanceof ServiceMetricEvent) { + emitMetric((ServiceMetricEvent) event); + } + } + + void emitMetric(ServiceMetricEvent metricEvent) + { + String name = metricEvent.getMetric(); + String service = metricEvent.getService(); + Map<String, Object> userDims = metricEvent.getUserDims(); + identifier = (userDims.get("task") == null ? metricEvent.getHost() : (String) userDims.get("task")); + Number value = metricEvent.getValue(); + + DimensionsAndCollector metric = metrics.getByName(name, service); + if (metric != null) { + String[] labelValues = new String[metric.getDimensions().length]; + String[] labelNames = metric.getDimensions(); + for (int i = 0; i < labelValues.length; i++) { + String labelName = labelNames[i]; + //labelName is controlled by the user. Instead of potential NPE on invalid labelName we use "unknown" as the dimension value + Object userDim = userDims.get(labelName); + labelValues[i] = userDim != null ? pattern.matcher(userDim.toString()).replaceAll("_") : "unknown"; + } + + if (metric.getCollector() instanceof Counter) { + ((Counter) metric.getCollector()).labels(labelValues).inc(value.doubleValue()); + } else if (metric.getCollector() instanceof Gauge) { + ((Gauge) metric.getCollector()).labels(labelValues).set(value.doubleValue()); + } else if (metric.getCollector() instanceof Histogram) { + ((Histogram) metric.getCollector()).labels(labelValues).observe(value.doubleValue() / metric.getConversionFactor()); + } else { + log.error("Unrecognized metric type [%s]", metric.getCollector().getClass()); + } + } else { + log.debug("Unmapped metric [%s]", name); + } + } + + @Override + public void flush() + { + Map<String, DimensionsAndCollector> map = metrics.getMap(); + try { + for (DimensionsAndCollector collector : map.values()) { + pushGateway.push(collector.getCollector(), config.getNamespace(), ImmutableMap.of(config.getNamespace(), identifier)); Review comment: Also should we use a more meaningful label name for `identifier` instead of using the `config.getNamespace()`? ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
