You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2019/02/13 12:31:09 UTC
[flink] branch master updated: [FLINK-7155][metrics] Add InfluxDB
reporter
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3d39cb0 [FLINK-7155][metrics] Add InfluxDB reporter
3d39cb0 is described below
commit 3d39cb0db939805ed7b86ed8ceabef98a160a97a
Author: ap <48...@users.noreply.github.com>
AuthorDate: Wed Feb 13 13:30:58 2019 +0100
[FLINK-7155][metrics] Add InfluxDB reporter
---
docs/monitoring/metrics.md | 29 ++++
flink-dist/pom.xml | 7 +
flink-dist/src/main/assemblies/opt.xml | 7 +
flink-metrics/flink-metrics-influxdb/pom.xml | 174 +++++++++++++++++++++
.../flink/metrics/influxdb/AbstractReporter.java | 90 +++++++++++
.../flink/metrics/influxdb/InfluxdbReporter.java | 138 ++++++++++++++++
.../metrics/influxdb/InfluxdbReporterOptions.java | 62 ++++++++
.../flink/metrics/influxdb/MeasurementInfo.java | 39 +++++
.../metrics/influxdb/MeasurementInfoProvider.java | 68 ++++++++
.../flink/metrics/influxdb/MetricInfoProvider.java | 31 ++++
.../flink/metrics/influxdb/MetricMapper.java | 80 ++++++++++
.../src/main/resources/META-INF/NOTICE | 18 +++
.../resources/META-INF/licenses/LICENSE.influx | 21 +++
.../metrics/influxdb/InfluxdbReporterTest.java | 135 ++++++++++++++++
.../influxdb/MeasurementInfoProviderTest.java | 70 +++++++++
.../flink/metrics/influxdb/MetricMapperTest.java | 125 +++++++++++++++
.../src/test/resources/log4j-test.properties | 27 ++++
flink-metrics/pom.xml | 1 +
18 files changed, 1122 insertions(+)
diff --git a/docs/monitoring/metrics.md b/docs/monitoring/metrics.md
index 8444c8a..ee7426a 100644
--- a/docs/monitoring/metrics.md
+++ b/docs/monitoring/metrics.md
@@ -644,6 +644,35 @@ metrics.reporter.grph.protocol: TCP
{% endhighlight %}
+### InfluxDB (org.apache.flink.metrics.influxdb.InfluxdbReporter)
+
+In order to use this reporter you must copy `/opt/flink-metrics-influxdb-{{site.version}}.jar` into the `/lib` folder
+of your Flink distribution.
+
+Parameters:
+
+- `host` - the InfluxDB server host
+- `port` - (optional) the InfluxDB server port, defaults to `8086`
+- `db` - the InfluxDB database to store metrics
+- `username` - (optional) InfluxDB username used for authentication
+- `password` - (optional) InfluxDB username's password used for authentication
+
+Example configuration:
+
+{% highlight yaml %}
+
+metrics.reporter.influxdb.class: org.apache.flink.metrics.influxdb.InfluxdbReporter
+metrics.reporter.influxdb.host: localhost
+metrics.reporter.influxdb.port: 8086
+metrics.reporter.influxdb.db: flink
+metrics.reporter.influxdb.username: flink-metrics
+metrics.reporter.influxdb.password: qwerty
+
+{% endhighlight %}
+
+The reporter would send metrics using http protocol with default retention policy defined on InfluxDB server.
+All Flink metrics variables (see [List of all Variables](#list-of-all-variables)) are exported as InfluxDB tags.
+
### Prometheus (org.apache.flink.metrics.prometheus.PrometheusReporter)
In order to use this reporter you must copy `/opt/flink-metrics-prometheus{{site.scala_version_suffix}}-{{site.version}}.jar` into the `/lib` folder
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index 4044db3..48c6ae0 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -235,6 +235,13 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics-influxdb_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-metrics-prometheus_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
diff --git a/flink-dist/src/main/assemblies/opt.xml b/flink-dist/src/main/assemblies/opt.xml
index 10461f3..35a4d27 100644
--- a/flink-dist/src/main/assemblies/opt.xml
+++ b/flink-dist/src/main/assemblies/opt.xml
@@ -92,6 +92,13 @@
</file>
<file>
+ <source>../flink-metrics/flink-metrics-influxdb/target/flink-metrics-influxdb_${scala.binary.version}-${project.version}.jar</source>
+ <outputDirectory>opt/</outputDirectory>
+ <destName>flink-metrics-influxdb-${project.version}.jar</destName>
+ <fileMode>0644</fileMode>
+ </file>
+
+ <file>
<source>../flink-metrics/flink-metrics-prometheus/target/flink-metrics-prometheus_${scala.binary.version}-${project.version}.jar</source>
<outputDirectory>opt/</outputDirectory>
<destName>flink-metrics-prometheus-${project.version}.jar</destName>
diff --git a/flink-metrics/flink-metrics-influxdb/pom.xml b/flink-metrics/flink-metrics-influxdb/pom.xml
new file mode 100644
index 0000000..51ed4a0
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/pom.xml
@@ -0,0 +1,174 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics</artifactId>
+ <version>1.8-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-metrics-influxdb_${scala.binary.version}</artifactId>
+ <name>flink-metrics-influxdb</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.influxdb</groupId>
+ <artifactId>influxdb-java</artifactId>
+ <version>2.14</version>
+ </dependency>
+
+ <!-- test dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils-junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.github.tomakehurst</groupId>
+ <artifactId>wiremock</artifactId>
+ <version>2.19.0</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ </dependencies>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>okhttp</artifactId>
+ <version>3.11.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.squareup.okio</groupId>
+ <artifactId>okio</artifactId>
+ <version>1.14.0</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <version>2.8.11</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.8.11.2</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>2.8.11</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <includes>
+ <include>org.influxdb:*</include>
+ <include>com.squareup.moshi:*</include>
+ <include>com.squareup.okhttp3:*</include>
+ <include>com.squareup.okio:*</include>
+ <include>com.squareup.retrofit2:*</include>
+ <!-- org.msgpack:msgpack-core is not required -->
+ </includes>
+ </artifactSet>
+ <relocations combine.children="append">
+ <relocation>
+ <pattern>org.influxdb</pattern>
+ <shadedPattern>org.apache.flink.metrics.influxdb.shaded.org.influxdb</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.squareup.moshi</pattern>
+ <shadedPattern>org.apache.flink.metrics.influxdb.shaded.com.squareup.moshi</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>okhttp3</pattern>
+ <shadedPattern>org.apache.flink.metrics.influxdb.shaded.okhttp3</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>okio</pattern>
+ <shadedPattern>org.apache.flink.metrics.influxdb.shaded.okio</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>retrofit2</pattern>
+ <shadedPattern>org.apache.flink.metrics.influxdb.shaded.retrofit2</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/AbstractReporter.java b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/AbstractReporter.java
new file mode 100644
index 0000000..c42c3ac
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/AbstractReporter.java
@@ -0,0 +1,90 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.reporter.MetricReporter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * An abstract reporter with registry for metrics.
+ * It's same as {@link org.apache.flink.metrics.reporter.AbstractReporter} but with generalized information of metric.
+ *
+ * @param <MetricInfo> Custom metric information type
+ */
+abstract class AbstractReporter<MetricInfo> implements MetricReporter {
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ protected final Map<Gauge<?>, MetricInfo> gauges = new HashMap<>();
+ protected final Map<Counter, MetricInfo> counters = new HashMap<>();
+ protected final Map<Histogram, MetricInfo> histograms = new HashMap<>();
+ protected final Map<Meter, MetricInfo> meters = new HashMap<>();
+ protected final MetricInfoProvider<MetricInfo> metricInfoProvider;
+
+ protected AbstractReporter(MetricInfoProvider<MetricInfo> metricInfoProvider) {
+ this.metricInfoProvider = metricInfoProvider;
+ }
+
+ @Override
+ public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
+ final MetricInfo metricInfo = metricInfoProvider.getMetricInfo(metricName, group);
+ synchronized (this) {
+ if (metric instanceof Counter) {
+ counters.put((Counter) metric, metricInfo);
+ } else if (metric instanceof Gauge) {
+ gauges.put((Gauge<?>) metric, metricInfo);
+ } else if (metric instanceof Histogram) {
+ histograms.put((Histogram) metric, metricInfo);
+ } else if (metric instanceof Meter) {
+ meters.put((Meter) metric, metricInfo);
+ } else {
+ log.warn("Cannot add unknown metric type {}. This indicates that the reporter " +
+ "does not support this metric type.", metric.getClass().getName());
+ }
+ }
+ }
+
+ @Override
+ public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) {
+ synchronized (this) {
+ if (metric instanceof Counter) {
+ counters.remove(metric);
+ } else if (metric instanceof Gauge) {
+ gauges.remove(metric);
+ } else if (metric instanceof Histogram) {
+ histograms.remove(metric);
+ } else if (metric instanceof Meter) {
+ meters.remove(metric);
+ } else {
+ log.warn("Cannot remove unknown metric type {}. This indicates that the reporter " +
+ "does not support this metric type.", metric.getClass().getName());
+ }
+ }
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporter.java b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporter.java
new file mode 100644
index 0000000..de2ad87
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporter.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.flink.metrics.influxdb;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.metrics.MetricConfig;
+import org.apache.flink.metrics.reporter.MetricReporter;
+import org.apache.flink.metrics.reporter.Scheduled;
+
+import org.influxdb.InfluxDB;
+import org.influxdb.InfluxDBFactory;
+import org.influxdb.dto.BatchPoints;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.ConcurrentModificationException;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import static org.apache.flink.metrics.influxdb.InfluxdbReporterOptions.DB;
+import static org.apache.flink.metrics.influxdb.InfluxdbReporterOptions.HOST;
+import static org.apache.flink.metrics.influxdb.InfluxdbReporterOptions.PASSWORD;
+import static org.apache.flink.metrics.influxdb.InfluxdbReporterOptions.PORT;
+import static org.apache.flink.metrics.influxdb.InfluxdbReporterOptions.USERNAME;
+import static org.apache.flink.metrics.influxdb.InfluxdbReporterOptions.getInteger;
+import static org.apache.flink.metrics.influxdb.InfluxdbReporterOptions.getString;
+
+/**
+ * {@link MetricReporter} that exports {@link Metric Metrics} via InfluxDB.
+ */
+public class InfluxdbReporter extends AbstractReporter<MeasurementInfo> implements Scheduled {
+
+ private String database;
+ private InfluxDB influxDB;
+
+ public InfluxdbReporter() {
+ super(new MeasurementInfoProvider());
+ }
+
+ @Override
+ public void open(MetricConfig config) {
+ String host = getString(config, HOST);
+ int port = getInteger(config, PORT);
+ if (!isValidHost(host) || !isValidPort(port)) {
+ throw new IllegalArgumentException("Invalid host/port configuration. Host: " + host + " Port: " + port);
+ }
+ String database = getString(config, DB);
+ if (database == null) {
+ throw new IllegalArgumentException("'" + DB.key() + "' configuration option is not set");
+ }
+ String url = String.format("http://%s:%d", host, port);
+ String username = getString(config, USERNAME);
+ String password = getString(config, PASSWORD);
+
+ this.database = database;
+ if (username != null && password != null) {
+ influxDB = InfluxDBFactory.connect(url, username, password);
+ } else {
+ influxDB = InfluxDBFactory.connect(url);
+ }
+ log.info("Configured InfluxDBReporter with {host:{}, port:{}, db:{}}", host, port, database);
+ }
+
+ @Override
+ public void close() {
+ if (influxDB != null) {
+ influxDB.close();
+ influxDB = null;
+ }
+ }
+
+ @Override
+ public void report() {
+ BatchPoints report = buildReport();
+ if (report != null) {
+ influxDB.write(report);
+ }
+ }
+
+ @Nullable
+ private BatchPoints buildReport() {
+ Instant timestamp = Instant.now();
+ BatchPoints.Builder report = BatchPoints.database(database);
+ report.retentionPolicy("");
+ try {
+ for (Map.Entry<Gauge<?>, MeasurementInfo> entry : gauges.entrySet()) {
+ report.point(MetricMapper.map(entry.getValue(), timestamp, entry.getKey()));
+ }
+
+ for (Map.Entry<Counter, MeasurementInfo> entry : counters.entrySet()) {
+ report.point(MetricMapper.map(entry.getValue(), timestamp, entry.getKey()));
+ }
+
+ for (Map.Entry<Histogram, MeasurementInfo> entry : histograms.entrySet()) {
+ report.point(MetricMapper.map(entry.getValue(), timestamp, entry.getKey()));
+ }
+
+ for (Map.Entry<Meter, MeasurementInfo> entry : meters.entrySet()) {
+ report.point(MetricMapper.map(entry.getValue(), timestamp, entry.getKey()));
+ }
+ }
+ catch (ConcurrentModificationException | NoSuchElementException e) {
+ // ignore - may happen when metrics are concurrently added or removed
+ // report next time
+ return null;
+ }
+ return report.build();
+ }
+
+ private static boolean isValidHost(String host) {
+ return host != null && !host.isEmpty();
+ }
+
+ private static boolean isValidPort(int port) {
+ return 0 < port && port <= 65535;
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporterOptions.java b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporterOptions.java
new file mode 100644
index 0000000..b22c05b
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/InfluxdbReporterOptions.java
@@ -0,0 +1,62 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.metrics.MetricConfig;
+
+/**
+ * Config options for {@link InfluxdbReporter}.
+ */
+public class InfluxdbReporterOptions {
+
+ public static final ConfigOption<String> HOST = ConfigOptions
+ .key("host")
+ .noDefaultValue()
+ .withDescription("the InfluxDB server host");
+
+ public static final ConfigOption<Integer> PORT = ConfigOptions
+ .key("port")
+ .defaultValue(8086)
+ .withDescription("the InfluxDB server port");
+
+ public static final ConfigOption<String> USERNAME = ConfigOptions
+ .key("username")
+ .noDefaultValue()
+ .withDescription("(optional) InfluxDB username used for authentication");
+
+ public static final ConfigOption<String> PASSWORD = ConfigOptions
+ .key("password")
+ .noDefaultValue()
+ .withDescription("(optional) InfluxDB username's password used for authentication");
+
+ public static final ConfigOption<String> DB = ConfigOptions
+ .key("db")
+ .noDefaultValue()
+ .withDescription("the InfluxDB database to store metrics");
+
+ static String getString(MetricConfig config, ConfigOption<String> key) {
+ return config.getString(key.key(), key.defaultValue());
+ }
+
+ static int getInteger(MetricConfig config, ConfigOption<Integer> key) {
+ return config.getInteger(key.key(), key.defaultValue());
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfo.java b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfo.java
new file mode 100644
index 0000000..da9c6b9
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfo.java
@@ -0,0 +1,39 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import java.util.Map;
+
+final class MeasurementInfo {
+ private final String name;
+ private final Map<String, String> tags;
+
+ MeasurementInfo(String name, Map<String, String> tags) {
+ this.name = name;
+ this.tags = tags;
+ }
+
+ String getName() {
+ return name;
+ }
+
+ Map<String, String> getTags() {
+ return tags;
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfoProvider.java b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfoProvider.java
new file mode 100644
index 0000000..471475f
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MeasurementInfoProvider.java
@@ -0,0 +1,68 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.metrics.CharacterFilter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
+import org.apache.flink.runtime.metrics.groups.FrontMetricGroup;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+class MeasurementInfoProvider implements MetricInfoProvider<MeasurementInfo> {
+ @VisibleForTesting
+ static final char SCOPE_SEPARATOR = '_';
+
+ private static final CharacterFilter CHARACTER_FILTER = new CharacterFilter() {
+ private final Pattern notAllowedCharacters = Pattern.compile("[^a-zA-Z0-9:_]");
+ @Override
+ public String filterCharacters(String input) {
+ return notAllowedCharacters.matcher(input).replaceAll("_");
+ }
+ };
+
+ public MeasurementInfoProvider() {
+ }
+
+ @Override
+ public MeasurementInfo getMetricInfo(String metricName, MetricGroup group) {
+ return new MeasurementInfo(getScopedName(metricName, group), getTags(group));
+ }
+
+ private static Map<String, String> getTags(MetricGroup group) {
+ // Keys are surrounded by brackets: remove them, transforming "<name>" to "name".
+ Map<String, String> tags = new HashMap<>();
+ for (Map.Entry<String, String> variable: group.getAllVariables().entrySet()) {
+ String name = variable.getKey();
+ tags.put(name.substring(1, name.length() - 1), variable.getValue());
+ }
+ return tags;
+ }
+
+ private static String getScopedName(String metricName, MetricGroup group) {
+ return getLogicalScope(group) + SCOPE_SEPARATOR + metricName;
+ }
+
+ private static String getLogicalScope(MetricGroup group) {
+ return ((FrontMetricGroup<AbstractMetricGroup<?>>) group).getLogicalScope(CHARACTER_FILTER, SCOPE_SEPARATOR);
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricInfoProvider.java b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricInfoProvider.java
new file mode 100644
index 0000000..700ece8
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricInfoProvider.java
@@ -0,0 +1,31 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.metrics.MetricGroup;
+
+/**
+ * A generic interface to provide custom information for metrics. See {@link AbstractReporter}.
+ *
+ * @param <MetricInfo> Custom metric information type
+ */
+interface MetricInfoProvider<MetricInfo> {
+
+ MetricInfo getMetricInfo(String metricName, MetricGroup group);
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricMapper.java b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricMapper.java
new file mode 100644
index 0000000..785b839
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/java/org/apache/flink/metrics/influxdb/MetricMapper.java
@@ -0,0 +1,80 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.HistogramStatistics;
+import org.apache.flink.metrics.Meter;
+
+import org.influxdb.dto.Point;
+
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+
+class MetricMapper {
+
+ static Point map(MeasurementInfo info, Instant timestamp, Gauge<?> gauge) {
+ Point.Builder builder = builder(info, timestamp);
+ Object value = gauge.getValue();
+ if (value instanceof Number) {
+ builder.addField("value", (Number) value);
+ } else {
+ builder.addField("value", String.valueOf(value));
+ }
+ return builder.build();
+ }
+
+ static Point map(MeasurementInfo info, Instant timestamp, Counter counter) {
+ return builder(info, timestamp)
+ .addField("count", counter.getCount())
+ .build();
+ }
+
+ static Point map(MeasurementInfo info, Instant timestamp, Histogram histogram) {
+ HistogramStatistics statistics = histogram.getStatistics();
+ return builder(info, timestamp)
+ .addField("count", statistics.size())
+ .addField("min", statistics.getMin())
+ .addField("max", statistics.getMax())
+ .addField("mean", statistics.getMean())
+ .addField("stddev", statistics.getStdDev())
+ .addField("p50", statistics.getQuantile(.50))
+ .addField("p75", statistics.getQuantile(.75))
+ .addField("p95", statistics.getQuantile(.95))
+ .addField("p98", statistics.getQuantile(.98))
+ .addField("p99", statistics.getQuantile(.99))
+ .addField("p999", statistics.getQuantile(.999))
+ .build();
+ }
+
+ static Point map(MeasurementInfo info, Instant timestamp, Meter meter) {
+ return builder(info, timestamp)
+ .addField("count", meter.getCount())
+ .addField("rate", meter.getRate())
+ .build();
+ }
+
+ private static Point.Builder builder(MeasurementInfo info, Instant timestamp) {
+ return Point.measurement(info.getName())
+ .tag(info.getTags())
+ .time(timestamp.toEpochMilli(), TimeUnit.MILLISECONDS);
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/resources/META-INF/NOTICE b/flink-metrics/flink-metrics-influxdb/src/main/resources/META-INF/NOTICE
new file mode 100644
index 0000000..2677c6b
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,18 @@
+flink-metrics-influxdb
+Copyright 2014-2018 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
+
+- com.squareup.moshi:moshi:1.5.0
+- com.squareup.okhttp3:logging-interceptor:3.11.0
+- com.squareup.okhttp3:okhttp:3.11.0
+- com.squareup.okio:okio:1.14.0
+- com.squareup.retrofit2:converter-moshi:2.4.0
+- com.squareup.retrofit2:retrofit:2.4.0
+
+This project bundles the following dependencies under the MIT license. (https://opensource.org/licenses/MIT)
+
+- org.influxdb:influxdb-java:2.14
diff --git a/flink-metrics/flink-metrics-influxdb/src/main/resources/META-INF/licenses/LICENSE.influx b/flink-metrics/flink-metrics-influxdb/src/main/resources/META-INF/licenses/LICENSE.influx
new file mode 100644
index 0000000..f21351c
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/main/resources/META-INF/licenses/LICENSE.influx
@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2014-2017 Stefan Majer
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/InfluxdbReporterTest.java b/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/InfluxdbReporterTest.java
new file mode 100644
index 0000000..e4c6ff7
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/InfluxdbReporterTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.reporter.MetricReporter;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
+import org.apache.flink.runtime.metrics.MetricRegistryImpl;
+import org.apache.flink.runtime.metrics.groups.TaskManagerMetricGroup;
+import org.apache.flink.util.TestLogger;
+
+import com.github.tomakehurst.wiremock.common.ConsoleNotifier;
+import com.github.tomakehurst.wiremock.junit.WireMockRule;
+import org.junit.Rule;
+import org.junit.Test;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.containing;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static org.hamcrest.Matchers.hasEntry;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Integration test for {@link InfluxdbReporter}.
+ */
+public class InfluxdbReporterTest extends TestLogger {
+ private static final String TEST_INFLUXDB_DB = "test-42";
+ private static final String METRIC_HOSTNAME = "task-mgr-1";
+ private static final String METRIC_TM_ID = "tm-id-123";
+
+ @Rule
+ public final WireMockRule wireMockRule = new WireMockRule(
+ wireMockConfig()
+ .dynamicPort()
+ .notifier(new ConsoleNotifier(false)));
+
+ @Test
+ public void testReporterRegistration() throws Exception {
+ MetricRegistryImpl metricRegistry = createMetricRegistry();
+ try {
+ assertEquals(1, metricRegistry.getReporters().size());
+ MetricReporter reporter = metricRegistry.getReporters().get(0);
+ assertTrue(reporter instanceof InfluxdbReporter);
+ } finally {
+ metricRegistry.shutdown().get();
+ }
+ }
+
+ @Test
+ public void testMetricRegistration() throws Exception {
+ MetricRegistryImpl metricRegistry = createMetricRegistry();
+ try {
+ String metricName = "TestCounter";
+ Counter counter = registerTestMetric(metricName, metricRegistry);
+
+ InfluxdbReporter reporter = (InfluxdbReporter) metricRegistry.getReporters().get(0);
+ MeasurementInfo measurementInfo = reporter.counters.get(counter);
+ assertNotNull("test metric must be registered in the reporter", measurementInfo);
+ assertEquals("taskmanager_" + metricName, measurementInfo.getName());
+ assertThat(measurementInfo.getTags(), hasEntry("host", METRIC_HOSTNAME));
+ assertThat(measurementInfo.getTags(), hasEntry("tm_id", METRIC_TM_ID));
+ } finally {
+ metricRegistry.shutdown().get();
+ }
+ }
+
+ @Test
+ public void testMetricReporting() throws Exception {
+ MetricRegistryImpl metricRegistry = createMetricRegistry();
+ try {
+ String metricName = "TestCounter";
+ Counter counter = registerTestMetric(metricName, metricRegistry);
+ counter.inc(42);
+
+ stubFor(post(urlPathEqualTo("/write"))
+ .willReturn(aResponse()
+ .withStatus(200)));
+
+ InfluxdbReporter reporter = (InfluxdbReporter) metricRegistry.getReporters().get(0);
+ reporter.report();
+
+ verify(postRequestedFor(urlPathEqualTo("/write"))
+ .withQueryParam("db", equalTo(TEST_INFLUXDB_DB))
+ .withHeader("Content-Type", containing("text/plain"))
+ .withRequestBody(containing("taskmanager_" + metricName + ",host=" + METRIC_HOSTNAME + ",tm_id=" + METRIC_TM_ID + " count=42i")));
+ } finally {
+ metricRegistry.shutdown().get();
+ }
+ }
+
+ private MetricRegistryImpl createMetricRegistry() {
+ String configPrefix = ConfigConstants.METRICS_REPORTER_PREFIX + "test.";
+ Configuration configuration = new Configuration();
+ configuration.setString(
+ configPrefix + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX,
+ InfluxdbReporter.class.getTypeName());
+ configuration.setString(configPrefix + "host", "localhost");
+ configuration.setString(configPrefix + "port", String.valueOf(wireMockRule.port()));
+ configuration.setString(configPrefix + "db", TEST_INFLUXDB_DB);
+ return new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(configuration));
+ }
+
+ private static Counter registerTestMetric(String metricName, MetricRegistry metricRegistry) {
+ TaskManagerMetricGroup metricGroup = new TaskManagerMetricGroup(metricRegistry, METRIC_HOSTNAME, METRIC_TM_ID);
+ return metricGroup.counter(metricName);
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/MeasurementInfoProviderTest.java b/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/MeasurementInfoProviderTest.java
new file mode 100644
index 0000000..2b360f9
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/MeasurementInfoProviderTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.runtime.metrics.groups.FrontMetricGroup;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.Matchers.hasEntry;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyChar;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Test for {@link MeasurementInfoProvider}.
+ */
+public class MeasurementInfoProviderTest extends TestLogger {
+ private final MeasurementInfoProvider provider = new MeasurementInfoProvider();
+
+ @Test
+ public void simpleTestGetMetricInfo() {
+ String logicalScope = "myService.Status.JVM.ClassLoader";
+ Map<String, String> variables = new HashMap<>();
+ variables.put("<A>", "a");
+ variables.put("<B>", "b");
+ variables.put("<C>", "c");
+ String metricName = "ClassesLoaded";
+ FrontMetricGroup metricGroup = mock(
+ FrontMetricGroup.class,
+ (invocation) -> {
+ throw new UnsupportedOperationException("unexpected method call");
+ });
+ doReturn(variables).when(metricGroup).getAllVariables();
+ doReturn(logicalScope).when(metricGroup).getLogicalScope(any(), anyChar());
+
+ MeasurementInfo info = provider.getMetricInfo(metricName, metricGroup);
+ assertNotNull(info);
+ assertEquals(
+ String.join("" + MeasurementInfoProvider.SCOPE_SEPARATOR, logicalScope, metricName),
+ info.getName());
+ assertThat(info.getTags(), hasEntry("A", "a"));
+ assertThat(info.getTags(), hasEntry("B", "b"));
+ assertThat(info.getTags(), hasEntry("C", "c"));
+ assertEquals(3, info.getTags().size());
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/MetricMapperTest.java b/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/MetricMapperTest.java
new file mode 100644
index 0000000..20afe03
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/test/java/org/apache/flink/metrics/influxdb/MetricMapperTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.flink.metrics.influxdb;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.SimpleCounter;
+import org.apache.flink.metrics.util.TestHistogram;
+import org.apache.flink.metrics.util.TestMeter;
+import org.apache.flink.util.TestLogger;
+
+import org.influxdb.dto.Point;
+import org.junit.Test;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link MetricMapper} checking that metrics are converted to InfluxDB client objects as expected.
+ */
+public class MetricMapperTest extends TestLogger {
+
+ private static final String NAME = "a-metric-name";
+ private static final MeasurementInfo INFO = getMeasurementInfo(NAME);
+ private static final Instant TIMESTAMP = Instant.now();
+
+ @Test
+ public void testMapGauge() {
+ verifyPoint(
+ MetricMapper.map(INFO, TIMESTAMP, (Gauge<Number>) () -> 42),
+ "value=42");
+
+ verifyPoint(
+ MetricMapper.map(INFO, TIMESTAMP, (Gauge<Number>) () -> null),
+ "value=null");
+
+ verifyPoint(
+ MetricMapper.map(INFO, TIMESTAMP, (Gauge<String>) () -> "hello"),
+ "value=hello");
+
+ verifyPoint(
+ MetricMapper.map(INFO, TIMESTAMP, (Gauge<Long>) () -> 42L),
+ "value=42");
+ }
+
+ @Test
+ public void testMapCounter() {
+ Counter counter = new SimpleCounter();
+ counter.inc(42L);
+
+ verifyPoint(
+ MetricMapper.map(INFO, TIMESTAMP, counter),
+ "count=42");
+ }
+
+ @Test
+ public void testMapHistogram() {
+ Histogram histogram = new TestHistogram();
+
+ verifyPoint(
+ MetricMapper.map(INFO, TIMESTAMP, histogram),
+ "count=3",
+ "max=6",
+ "mean=4.0",
+ "min=7",
+ "p50=0.5",
+ "p75=0.75",
+ "p95=0.95",
+ "p98=0.98",
+ "p99=0.99",
+ "p999=0.999",
+ "stddev=5.0");
+ }
+
+ @Test
+ public void testMapMeter() {
+ Meter meter = new TestMeter();
+
+ verifyPoint(
+ MetricMapper.map(INFO, TIMESTAMP, meter),
+ "count=100",
+ "rate=5.0");
+ }
+
+ private void verifyPoint(Point point, String... expectedFields) {
+ // Most methods of Point are package private. We use toString() method to check that values are as expected.
+ // An alternative can be to call lineProtocol() method, which additionally escapes values for InfluxDB format.
+ assertEquals(
+ "Point [name=" + NAME
+ + ", time=" + TIMESTAMP.toEpochMilli()
+ + ", tags={tag-1=42, tag-2=green}"
+ + ", precision=MILLISECONDS"
+ + ", fields={" + String.join(", ", expectedFields) + "}"
+ + "]",
+ point.toString());
+ }
+
+ private static MeasurementInfo getMeasurementInfo(String name) {
+ Map<String, String> tags = new HashMap<>();
+ tags.put("tag-1", "42");
+ tags.put("tag-2", "green");
+ return new MeasurementInfo(name, tags);
+ }
+}
diff --git a/flink-metrics/flink-metrics-influxdb/src/test/resources/log4j-test.properties b/flink-metrics/flink-metrics-influxdb/src/test/resources/log4j-test.properties
new file mode 100644
index 0000000..2226f68
--- /dev/null
+++ b/flink-metrics/flink-metrics-influxdb/src/test/resources/log4j-test.properties
@@ -0,0 +1,27 @@
+#
+# 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.
+#
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+log4j.rootLogger=OFF, testlogger
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
+log4j.appender.testlogger.target = System.err
+log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
+log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-metrics/pom.xml b/flink-metrics/pom.xml
index a3e4fcd..77ea7d1 100644
--- a/flink-metrics/pom.xml
+++ b/flink-metrics/pom.xml
@@ -37,6 +37,7 @@ under the License.
<module>flink-metrics-core</module>
<module>flink-metrics-dropwizard</module>
<module>flink-metrics-graphite</module>
+ <module>flink-metrics-influxdb</module>
<module>flink-metrics-jmx</module>
<module>flink-metrics-prometheus</module>
<module>flink-metrics-statsd</module>