Details

    • Type: Improvement
    • Status: Closed
    • Priority: Critical
    • Resolution: Fixed
    • Affects Version/s: 1.3.0
    • Fix Version/s: 1.3.0, 1.4.0
    • Component/s: Metrics
    • Labels:
      None

      Description

      We at OfferUp use Datadog a lot for metrics and dashboards, and I believe a lot other companies also do.

      Flink right now only has a StatsD metrics reporter, and users have to set up Datadog Agent in order to receive metrics from StatsD and transport them to Datadog. We don't like this approach.

      We prefer to have a Datadog metrics reporter directly contacting Datadog http endpoint.

      I'll take this ticket myself.

        Issue Links

          Activity

          Hide
          githubbot ASF GitHub Bot added a comment -

          GitHub user bowenli86 opened a pull request:

          https://github.com/apache/flink/pull/3736

          [Flink-6013][metrics] Add Datadog HTTP metrics reporter

          I'm adding a DatadogHttpReporter for Flink metrics system.

          The implementation, including making parameters in Flink's metrics as Datadog tags, is a best practice based on our long time working experience and understanding of Datadog. It might be a bit different than how other metrics reporters work, but it truly helps developers to find and filter metrics quickly, better categorize metrics, and visualize them on Datadog dashboards, especially when users (like OfferUp) have a dozen individual Flink clusters.

          ------

          Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
          If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html).
          In addition to going through the list, please provide a meaningful description of your changes.

          • [x] General
          • The pull request references the related JIRA issue ("FLINK-6013 Add Datadog HTTP metrics reporter")
          • The pull request addresses only one issue
          • Each commit in the PR has a meaningful commit message (including the JIRA id)
          • [x] Documentation
          • Documentation has been added for new functionality
          • Old documentation affected by the pull request has been updated
          • JavaDoc for public methods has been added
          • [x] Tests & Build
          • Functionality added by the pull request is covered by tests
          • `mvn clean verify` has been executed successfully locally or a Travis build has passed

          You can merge this pull request into a Git repository by running:

          $ git pull https://github.com/bowenli86/flink FLINK-6013

          Alternatively you can review and apply these changes as the patch at:

          https://github.com/apache/flink/pull/3736.patch

          To close this pull request, make a commit to your master/trunk branch
          with (at least) the following in the commit message:

          This closes #3736


          commit 72037bde640258bde618ddc10b8bd10645bbaf8d
          Author: Bowen Li <bowenli86@gmail.com>
          Date: 2017-04-18T17:27:17Z

          FLINK-6013[metrics] Add Datadog HTTP metrics reporter

          commit e8ced6d03eac47150648401566afce6f12ea03d0
          Author: Bowen Li <bowenli86@gmail.com>
          Date: 2017-04-18T17:27:54Z

          Merge branch 'master' into FLINK-6013

          commit 27ae0584eb79bd1339934c06d4a4266be9264fb2
          Author: Bowen Li <bowenli86@gmail.com>
          Date: 2017-04-18T18:23:10Z

          move okhttp dependencies to flink-metrics

          commit 4b48f4d32a8b122f35dfd6322174e469ff0a5a89
          Author: Bowen Li <bowenli86@gmail.com>
          Date: 2017-04-18T18:57:11Z

          add Apache License file header

          commit a9ca61a92e4f0beb37652da361acfdcd50d11523
          Author: Bowen Li <bowenli86@gmail.com>
          Date: 2017-04-18T19:12:24Z

          add more code comments

          commit cfe2fdf8d7456d657bd35e937e0f7618086af024
          Author: Bowen Li <bowenli86@gmail.com>
          Date: 2017-04-18T19:47:15Z

          remove okhttp from flink-metrics

          commit 76b54b8ab7fc6eaad8c2bd7d54e79791110d9690
          Author: Bowen Li <bowenli86@gmail.com>
          Date: 2017-04-18T19:53:14Z

          add more doc


          Show
          githubbot ASF GitHub Bot added a comment - GitHub user bowenli86 opened a pull request: https://github.com/apache/flink/pull/3736 [Flink-6013] [metrics] Add Datadog HTTP metrics reporter I'm adding a DatadogHttpReporter for Flink metrics system. The implementation, including making parameters in Flink's metrics as Datadog tags, is a best practice based on our long time working experience and understanding of Datadog. It might be a bit different than how other metrics reporters work, but it truly helps developers to find and filter metrics quickly, better categorize metrics, and visualize them on Datadog dashboards, especially when users (like OfferUp) have a dozen individual Flink clusters. ------ Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration. If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide] ( http://flink.apache.org/how-to-contribute.html ). In addition to going through the list, please provide a meaningful description of your changes. [x] General The pull request references the related JIRA issue (" FLINK-6013 Add Datadog HTTP metrics reporter") The pull request addresses only one issue Each commit in the PR has a meaningful commit message (including the JIRA id) [x] Documentation Documentation has been added for new functionality Old documentation affected by the pull request has been updated JavaDoc for public methods has been added [x] Tests & Build Functionality added by the pull request is covered by tests `mvn clean verify` has been executed successfully locally or a Travis build has passed You can merge this pull request into a Git repository by running: $ git pull https://github.com/bowenli86/flink FLINK-6013 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3736.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3736 commit 72037bde640258bde618ddc10b8bd10645bbaf8d Author: Bowen Li <bowenli86@gmail.com> Date: 2017-04-18T17:27:17Z FLINK-6013 [metrics] Add Datadog HTTP metrics reporter commit e8ced6d03eac47150648401566afce6f12ea03d0 Author: Bowen Li <bowenli86@gmail.com> Date: 2017-04-18T17:27:54Z Merge branch 'master' into FLINK-6013 commit 27ae0584eb79bd1339934c06d4a4266be9264fb2 Author: Bowen Li <bowenli86@gmail.com> Date: 2017-04-18T18:23:10Z move okhttp dependencies to flink-metrics commit 4b48f4d32a8b122f35dfd6322174e469ff0a5a89 Author: Bowen Li <bowenli86@gmail.com> Date: 2017-04-18T18:57:11Z add Apache License file header commit a9ca61a92e4f0beb37652da361acfdcd50d11523 Author: Bowen Li <bowenli86@gmail.com> Date: 2017-04-18T19:12:24Z add more code comments commit cfe2fdf8d7456d657bd35e937e0f7618086af024 Author: Bowen Li <bowenli86@gmail.com> Date: 2017-04-18T19:47:15Z remove okhttp from flink-metrics commit 76b54b8ab7fc6eaad8c2bd7d54e79791110d9690 Author: Bowen Li <bowenli86@gmail.com> Date: 2017-04-18T19:53:14Z add more doc
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          Could you change the PR title to "FLINK-6013[metrics] Add Datadog HTTP metrics reporter"? The comments aren't being mirrored to JIRA.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 Could you change the PR title to " FLINK-6013 [metrics] Add Datadog HTTP metrics reporter"? The comments aren't being mirrored to JIRA.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          Where is the title that you're referring to? This PR's title at the very top is already "FLINK-6013[metrics] Add Datadog HTTP metrics reporter" (without double quote)

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 Where is the title that you're referring to? This PR's title at the very top is already " FLINK-6013 [metrics] Add Datadog HTTP metrics reporter" (without double quote)
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          Aha! Updated! Apparently Chrome search is case insensitive

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 Aha! Updated! Apparently Chrome search is case insensitive
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          Green build! @zentol

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 Green build! @zentol
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112613733

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java —
          @@ -0,0 +1,197 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.apache.flink.metrics.Histogram;
          +import org.apache.flink.metrics.Metric;
          +import org.apache.flink.metrics.MetricConfig;
          +import org.apache.flink.metrics.MetricGroup;
          +import org.apache.flink.metrics.reporter.MetricReporter;
          +import org.apache.flink.metrics.reporter.Scheduled;
          +import org.slf4j.Logger;
          +import org.slf4j.LoggerFactory;
          +
          +import java.util.ArrayList;
          +import java.util.Arrays;
          +import java.util.List;
          +import java.util.Map;
          +import java.util.concurrent.ConcurrentHashMap;
          +
          +
          +/**
          + * Metric Reporter for Datadog
          + *
          + * Variables in metrics scope will be sent to Datadog as tags
          + * */
          +public class DatadogHttpReporter implements MetricReporter, Scheduled {
          + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class);
          +
          + // Both Flink's Gauge and Meter values are taken as gauge in Datadog
          + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>();
          + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>();
          + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>();
          +
          + private DatadogHttpClient client;
          + private List<String> configTags;
          +
          + public static final String API_KEY = "apikey";
          + public static final String TAGS = "tags";
          +
          + @Override
          + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
          + final String name = group.getMetricIdentifier(metricName);
          +
          + List<String> tags = new ArrayList<>(configTags);
          + tags.addAll(getTagsFromMetricGroup(group));
          +
          + if (metric instanceof Counter)

          { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + }

          else if (metric instanceof Gauge)

          { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + }

          else if(metric instanceof Meter)

          { + Meter m = (Meter) metric; + // Only consider rate + meters.put(m, new DMeter(m, name, tags)); + }

          else if (metric instanceof Histogram) {
          + LOGGER.warn("Cannot add {} because Datadog HTTP API doesn't support Histogram", metricName);
          + } else {
          + LOGGER.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) {
          + if (metric instanceof Counter)

          { + counters.remove(metric); + }

          else if (metric instanceof Gauge)

          { + gauges.remove(metric); + }

          else if (metric instanceof Meter)

          { + meters.remove(metric); + }

          else if (metric instanceof Histogram)

          { + // No Histogram is registered + }

          else {
          + LOGGER.warn("Cannot remove unknown metric type {}. This indicates that the reporter " +
          + "does not support this metric type.", metric.getClass().getName());
          + }
          + }
          +
          + @Override
          + public void open(MetricConfig config)

          { + client = new DatadogHttpClient(config.getString(API_KEY, null)); + LOGGER.info("Configured DatadogHttpReporter"); + + configTags = getTagsFromConfig(config.getString(TAGS, "")); + }

          +
          + @Override
          + public void close()

          { + client.close(); + LOGGER.info("Shut down DatadogHttpReporter"); + }

          +
          + @Override
          + public void report() {
          + DatadogHttpRequest request = new DatadogHttpRequest();
          +
          + for (DGauge g : gauges.values()) {
          + try

          { + // Will throw exception if the Gauge is not of Number type + // Flink uses Gauge to store many types other than Number + g.getMetricValue(); + request.addGauge(g); + }

          catch (Exception e) {
          + // ignore if the Gauge is not of Number type
          — End diff –

          You can safely remove the metric in this case.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112613733 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java — @@ -0,0 +1,197 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.reporter.MetricReporter; +import org.apache.flink.metrics.reporter.Scheduled; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + + +/** + * Metric Reporter for Datadog + * + * Variables in metrics scope will be sent to Datadog as tags + * */ +public class DatadogHttpReporter implements MetricReporter, Scheduled { + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class); + + // Both Flink's Gauge and Meter values are taken as gauge in Datadog + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>(); + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>(); + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>(); + + private DatadogHttpClient client; + private List<String> configTags; + + public static final String API_KEY = "apikey"; + public static final String TAGS = "tags"; + + @Override + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) { + final String name = group.getMetricIdentifier(metricName); + + List<String> tags = new ArrayList<>(configTags); + tags.addAll(getTagsFromMetricGroup(group)); + + if (metric instanceof Counter) { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + } else if (metric instanceof Gauge) { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + } else if(metric instanceof Meter) { + Meter m = (Meter) metric; + // Only consider rate + meters.put(m, new DMeter(m, name, tags)); + } else if (metric instanceof Histogram) { + LOGGER.warn("Cannot add {} because Datadog HTTP API doesn't support Histogram", metricName); + } else { + LOGGER.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) { + if (metric instanceof Counter) { + counters.remove(metric); + } else if (metric instanceof Gauge) { + gauges.remove(metric); + } else if (metric instanceof Meter) { + meters.remove(metric); + } else if (metric instanceof Histogram) { + // No Histogram is registered + } else { + LOGGER.warn("Cannot remove unknown metric type {}. This indicates that the reporter " + + "does not support this metric type.", metric.getClass().getName()); + } + } + + @Override + public void open(MetricConfig config) { + client = new DatadogHttpClient(config.getString(API_KEY, null)); + LOGGER.info("Configured DatadogHttpReporter"); + + configTags = getTagsFromConfig(config.getString(TAGS, "")); + } + + @Override + public void close() { + client.close(); + LOGGER.info("Shut down DatadogHttpReporter"); + } + + @Override + public void report() { + DatadogHttpRequest request = new DatadogHttpRequest(); + + for (DGauge g : gauges.values()) { + try { + // Will throw exception if the Gauge is not of Number type + // Flink uses Gauge to store many types other than Number + g.getMetricValue(); + request.addGauge(g); + } catch (Exception e) { + // ignore if the Gauge is not of Number type — End diff – You can safely remove the metric in this case.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112614175

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java —
          @@ -0,0 +1,197 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.apache.flink.metrics.Histogram;
          +import org.apache.flink.metrics.Metric;
          +import org.apache.flink.metrics.MetricConfig;
          +import org.apache.flink.metrics.MetricGroup;
          +import org.apache.flink.metrics.reporter.MetricReporter;
          +import org.apache.flink.metrics.reporter.Scheduled;
          +import org.slf4j.Logger;
          +import org.slf4j.LoggerFactory;
          +
          +import java.util.ArrayList;
          +import java.util.Arrays;
          +import java.util.List;
          +import java.util.Map;
          +import java.util.concurrent.ConcurrentHashMap;
          +
          +
          +/**
          + * Metric Reporter for Datadog
          + *
          + * Variables in metrics scope will be sent to Datadog as tags
          + * */
          +public class DatadogHttpReporter implements MetricReporter, Scheduled {
          + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class);
          +
          + // Both Flink's Gauge and Meter values are taken as gauge in Datadog
          + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>();
          + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>();
          + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>();
          +
          + private DatadogHttpClient client;
          + private List<String> configTags;
          +
          + public static final String API_KEY = "apikey";
          + public static final String TAGS = "tags";
          +
          + @Override
          + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
          + final String name = group.getMetricIdentifier(metricName);
          +
          + List<String> tags = new ArrayList<>(configTags);
          + tags.addAll(getTagsFromMetricGroup(group));
          +
          + if (metric instanceof Counter)

          { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + }

          else if (metric instanceof Gauge)

          { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + }

          else if(metric instanceof Meter) {
          — End diff –

          missing space after if

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112614175 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java — @@ -0,0 +1,197 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.reporter.MetricReporter; +import org.apache.flink.metrics.reporter.Scheduled; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + + +/** + * Metric Reporter for Datadog + * + * Variables in metrics scope will be sent to Datadog as tags + * */ +public class DatadogHttpReporter implements MetricReporter, Scheduled { + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class); + + // Both Flink's Gauge and Meter values are taken as gauge in Datadog + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>(); + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>(); + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>(); + + private DatadogHttpClient client; + private List<String> configTags; + + public static final String API_KEY = "apikey"; + public static final String TAGS = "tags"; + + @Override + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) { + final String name = group.getMetricIdentifier(metricName); + + List<String> tags = new ArrayList<>(configTags); + tags.addAll(getTagsFromMetricGroup(group)); + + if (metric instanceof Counter) { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + } else if (metric instanceof Gauge) { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + } else if(metric instanceof Meter) { — End diff – missing space after if
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112613661

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java —
          @@ -0,0 +1,97 @@
          +/*
          + * 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.datadog;
          +
          +import com.fasterxml.jackson.core.JsonProcessingException;
          +import com.fasterxml.jackson.databind.ObjectMapper;
          +import okhttp3.MediaType;
          +import okhttp3.OkHttpClient;
          +import okhttp3.Request;
          +import okhttp3.Response;
          +import okhttp3.RequestBody;
          +
          +import java.io.IOException;
          +import java.util.concurrent.TimeUnit;
          +
          +/**
          + * Http client talking to Datadog
          + * */
          +public class DatadogHttpClient{
          + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s";
          + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s";
          + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8");
          + private static final int TIMEOUT = 3;
          + private static final ObjectMapper MAPPER = new ObjectMapper();
          +
          + private final String seriesUrl;
          + private final String validateUrl;
          + private final OkHttpClient client;
          + private final String apiKey;
          +
          + public DatadogHttpClient(String dgApiKey) {
          + if(dgApiKey == null || dgApiKey.isEmpty()) {
          + throw new IllegalArgumentException(
          — End diff –

          please remove the line-break here

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112613661 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java — @@ -0,0 +1,97 @@ +/* + * 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.datadog; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.RequestBody; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +/** + * Http client talking to Datadog + * */ +public class DatadogHttpClient{ + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s"; + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s"; + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8"); + private static final int TIMEOUT = 3; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private final String seriesUrl; + private final String validateUrl; + private final OkHttpClient client; + private final String apiKey; + + public DatadogHttpClient(String dgApiKey) { + if(dgApiKey == null || dgApiKey.isEmpty()) { + throw new IllegalArgumentException( — End diff – please remove the line-break here
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112613622

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java —
          @@ -0,0 +1,97 @@
          +/*
          + * 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.datadog;
          +
          +import com.fasterxml.jackson.core.JsonProcessingException;
          +import com.fasterxml.jackson.databind.ObjectMapper;
          +import okhttp3.MediaType;
          +import okhttp3.OkHttpClient;
          +import okhttp3.Request;
          +import okhttp3.Response;
          +import okhttp3.RequestBody;
          +
          +import java.io.IOException;
          +import java.util.concurrent.TimeUnit;
          +
          +/**
          + * Http client talking to Datadog
          + * */
          +public class DatadogHttpClient{
          + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s";
          + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s";
          + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8");
          + private static final int TIMEOUT = 3;
          + private static final ObjectMapper MAPPER = new ObjectMapper();
          +
          + private final String seriesUrl;
          + private final String validateUrl;
          + private final OkHttpClient client;
          + private final String apiKey;
          +
          + public DatadogHttpClient(String dgApiKey) {
          + if(dgApiKey == null || dgApiKey.isEmpty()) {
          — End diff –

          missing space after if

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112613622 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java — @@ -0,0 +1,97 @@ +/* + * 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.datadog; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.RequestBody; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +/** + * Http client talking to Datadog + * */ +public class DatadogHttpClient{ + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s"; + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s"; + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8"); + private static final int TIMEOUT = 3; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private final String seriesUrl; + private final String validateUrl; + private final OkHttpClient client; + private final String apiKey; + + public DatadogHttpClient(String dgApiKey) { + if(dgApiKey == null || dgApiKey.isEmpty()) { — End diff – missing space after if
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112613477

          — Diff: flink-dist/src/main/assemblies/opt.xml —
          @@ -95,5 +95,12 @@
          <destName>flink-metrics-statsd-$

          {project.version}.jar</destName>
          <fileMode>0644</fileMode>
          </file>
          +
          + <file>
          + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version}

          .jar</source>
          — End diff –

          Since you're using the maven-jar-plugin the jar containing the dependencies has the "-jar-with-dependencies" suffix, like `flink-metrics-graphite-1.2-SNAPSHOT-jar-with-dependencies.jar`.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112613477 — Diff: flink-dist/src/main/assemblies/opt.xml — @@ -95,5 +95,12 @@ <destName>flink-metrics-statsd-$ {project.version}.jar</destName> <fileMode>0644</fileMode> </file> + + <file> + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version} .jar</source> — End diff – Since you're using the maven-jar-plugin the jar containing the dependencies has the "-jar-with-dependencies" suffix, like `flink-metrics-graphite-1.2-SNAPSHOT-jar-with-dependencies.jar`.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112613595

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java —
          @@ -0,0 +1,65 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.datadog.utils.TimestampUtils;
          +
          +import java.util.ArrayList;
          +import java.util.List;
          +
          +/**
          + * Abstract metric of Datadog for serialization
          + * */
          +public abstract class DMetric {
          + private final String metric; // Metric name
          + private final MetricType type;
          + private final List<String> tags;
          +
          + public DMetric(MetricType metricType, String metric, List<String> tags)

          { + this.type = metricType; + this.metric = metric; + this.tags = tags; + }

          +
          + public MetricType getType() {
          + return type;
          — End diff –

          you missed adjusting the DGauge, DCounter and DSeries javadocs.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112613595 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java — @@ -0,0 +1,65 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.datadog.utils.TimestampUtils; + +import java.util.ArrayList; +import java.util.List; + +/** + * Abstract metric of Datadog for serialization + * */ +public abstract class DMetric { + private final String metric; // Metric name + private final MetricType type; + private final List<String> tags; + + public DMetric(MetricType metricType, String metric, List<String> tags) { + this.type = metricType; + this.metric = metric; + this.tags = tags; + } + + public MetricType getType() { + return type; — End diff – you missed adjusting the DGauge, DCounter and DSeries javadocs.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112727098

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java —
          @@ -0,0 +1,97 @@
          +/*
          + * 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.datadog;
          +
          +import com.fasterxml.jackson.core.JsonProcessingException;
          +import com.fasterxml.jackson.databind.ObjectMapper;
          +import okhttp3.MediaType;
          +import okhttp3.OkHttpClient;
          +import okhttp3.Request;
          +import okhttp3.Response;
          +import okhttp3.RequestBody;
          +
          +import java.io.IOException;
          +import java.util.concurrent.TimeUnit;
          +
          +/**
          + * Http client talking to Datadog
          + * */
          +public class DatadogHttpClient{
          + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s";
          + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s";
          + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8");
          + private static final int TIMEOUT = 3;
          + private static final ObjectMapper MAPPER = new ObjectMapper();
          +
          + private final String seriesUrl;
          + private final String validateUrl;
          + private final OkHttpClient client;
          + private final String apiKey;
          +
          + public DatadogHttpClient(String dgApiKey) {
          + if(dgApiKey == null || dgApiKey.isEmpty()) {
          + throw new IllegalArgumentException(
          — End diff –

          done

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112727098 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java — @@ -0,0 +1,97 @@ +/* + * 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.datadog; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.RequestBody; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +/** + * Http client talking to Datadog + * */ +public class DatadogHttpClient{ + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s"; + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s"; + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8"); + private static final int TIMEOUT = 3; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private final String seriesUrl; + private final String validateUrl; + private final OkHttpClient client; + private final String apiKey; + + public DatadogHttpClient(String dgApiKey) { + if(dgApiKey == null || dgApiKey.isEmpty()) { + throw new IllegalArgumentException( — End diff – done
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112727145

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java —
          @@ -0,0 +1,197 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.apache.flink.metrics.Histogram;
          +import org.apache.flink.metrics.Metric;
          +import org.apache.flink.metrics.MetricConfig;
          +import org.apache.flink.metrics.MetricGroup;
          +import org.apache.flink.metrics.reporter.MetricReporter;
          +import org.apache.flink.metrics.reporter.Scheduled;
          +import org.slf4j.Logger;
          +import org.slf4j.LoggerFactory;
          +
          +import java.util.ArrayList;
          +import java.util.Arrays;
          +import java.util.List;
          +import java.util.Map;
          +import java.util.concurrent.ConcurrentHashMap;
          +
          +
          +/**
          + * Metric Reporter for Datadog
          + *
          + * Variables in metrics scope will be sent to Datadog as tags
          + * */
          +public class DatadogHttpReporter implements MetricReporter, Scheduled {
          + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class);
          +
          + // Both Flink's Gauge and Meter values are taken as gauge in Datadog
          + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>();
          + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>();
          + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>();
          +
          + private DatadogHttpClient client;
          + private List<String> configTags;
          +
          + public static final String API_KEY = "apikey";
          + public static final String TAGS = "tags";
          +
          + @Override
          + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
          + final String name = group.getMetricIdentifier(metricName);
          +
          + List<String> tags = new ArrayList<>(configTags);
          + tags.addAll(getTagsFromMetricGroup(group));
          +
          + if (metric instanceof Counter)

          { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + }

          else if (metric instanceof Gauge)

          { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + }

          else if(metric instanceof Meter) {
          — End diff –

          done

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112727145 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java — @@ -0,0 +1,197 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.reporter.MetricReporter; +import org.apache.flink.metrics.reporter.Scheduled; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + + +/** + * Metric Reporter for Datadog + * + * Variables in metrics scope will be sent to Datadog as tags + * */ +public class DatadogHttpReporter implements MetricReporter, Scheduled { + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class); + + // Both Flink's Gauge and Meter values are taken as gauge in Datadog + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>(); + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>(); + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>(); + + private DatadogHttpClient client; + private List<String> configTags; + + public static final String API_KEY = "apikey"; + public static final String TAGS = "tags"; + + @Override + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) { + final String name = group.getMetricIdentifier(metricName); + + List<String> tags = new ArrayList<>(configTags); + tags.addAll(getTagsFromMetricGroup(group)); + + if (metric instanceof Counter) { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + } else if (metric instanceof Gauge) { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + } else if(metric instanceof Meter) { — End diff – done
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112727162

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java —
          @@ -0,0 +1,197 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.apache.flink.metrics.Histogram;
          +import org.apache.flink.metrics.Metric;
          +import org.apache.flink.metrics.MetricConfig;
          +import org.apache.flink.metrics.MetricGroup;
          +import org.apache.flink.metrics.reporter.MetricReporter;
          +import org.apache.flink.metrics.reporter.Scheduled;
          +import org.slf4j.Logger;
          +import org.slf4j.LoggerFactory;
          +
          +import java.util.ArrayList;
          +import java.util.Arrays;
          +import java.util.List;
          +import java.util.Map;
          +import java.util.concurrent.ConcurrentHashMap;
          +
          +
          +/**
          + * Metric Reporter for Datadog
          + *
          + * Variables in metrics scope will be sent to Datadog as tags
          + * */
          +public class DatadogHttpReporter implements MetricReporter, Scheduled {
          + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class);
          +
          + // Both Flink's Gauge and Meter values are taken as gauge in Datadog
          + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>();
          + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>();
          + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>();
          +
          + private DatadogHttpClient client;
          + private List<String> configTags;
          +
          + public static final String API_KEY = "apikey";
          + public static final String TAGS = "tags";
          +
          + @Override
          + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
          + final String name = group.getMetricIdentifier(metricName);
          +
          + List<String> tags = new ArrayList<>(configTags);
          + tags.addAll(getTagsFromMetricGroup(group));
          +
          + if (metric instanceof Counter)

          { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + }

          else if (metric instanceof Gauge)

          { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + }

          else if(metric instanceof Meter)

          { + Meter m = (Meter) metric; + // Only consider rate + meters.put(m, new DMeter(m, name, tags)); + }

          else if (metric instanceof Histogram) {
          + LOGGER.warn("Cannot add {} because Datadog HTTP API doesn't support Histogram", metricName);
          + } else {
          + LOGGER.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) {
          + if (metric instanceof Counter)

          { + counters.remove(metric); + }

          else if (metric instanceof Gauge)

          { + gauges.remove(metric); + }

          else if (metric instanceof Meter)

          { + meters.remove(metric); + }

          else if (metric instanceof Histogram)

          { + // No Histogram is registered + }

          else {
          + LOGGER.warn("Cannot remove unknown metric type {}. This indicates that the reporter " +
          + "does not support this metric type.", metric.getClass().getName());
          + }
          + }
          +
          + @Override
          + public void open(MetricConfig config)

          { + client = new DatadogHttpClient(config.getString(API_KEY, null)); + LOGGER.info("Configured DatadogHttpReporter"); + + configTags = getTagsFromConfig(config.getString(TAGS, "")); + }

          +
          + @Override
          + public void close()

          { + client.close(); + LOGGER.info("Shut down DatadogHttpReporter"); + }

          +
          + @Override
          + public void report() {
          + DatadogHttpRequest request = new DatadogHttpRequest();
          +
          + for (DGauge g : gauges.values()) {
          + try

          { + // Will throw exception if the Gauge is not of Number type + // Flink uses Gauge to store many types other than Number + g.getMetricValue(); + request.addGauge(g); + }

          catch (Exception e) {
          + // ignore if the Gauge is not of Number type
          — End diff –

          nice catch! done

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112727162 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java — @@ -0,0 +1,197 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.reporter.MetricReporter; +import org.apache.flink.metrics.reporter.Scheduled; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + + +/** + * Metric Reporter for Datadog + * + * Variables in metrics scope will be sent to Datadog as tags + * */ +public class DatadogHttpReporter implements MetricReporter, Scheduled { + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class); + + // Both Flink's Gauge and Meter values are taken as gauge in Datadog + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>(); + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>(); + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>(); + + private DatadogHttpClient client; + private List<String> configTags; + + public static final String API_KEY = "apikey"; + public static final String TAGS = "tags"; + + @Override + public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) { + final String name = group.getMetricIdentifier(metricName); + + List<String> tags = new ArrayList<>(configTags); + tags.addAll(getTagsFromMetricGroup(group)); + + if (metric instanceof Counter) { + Counter c = (Counter) metric; + counters.put(c, new DCounter(c, name, tags)); + } else if (metric instanceof Gauge) { + Gauge g = (Gauge) metric; + gauges.put(g, new DGauge(g, name, tags)); + } else if(metric instanceof Meter) { + Meter m = (Meter) metric; + // Only consider rate + meters.put(m, new DMeter(m, name, tags)); + } else if (metric instanceof Histogram) { + LOGGER.warn("Cannot add {} because Datadog HTTP API doesn't support Histogram", metricName); + } else { + LOGGER.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) { + if (metric instanceof Counter) { + counters.remove(metric); + } else if (metric instanceof Gauge) { + gauges.remove(metric); + } else if (metric instanceof Meter) { + meters.remove(metric); + } else if (metric instanceof Histogram) { + // No Histogram is registered + } else { + LOGGER.warn("Cannot remove unknown metric type {}. This indicates that the reporter " + + "does not support this metric type.", metric.getClass().getName()); + } + } + + @Override + public void open(MetricConfig config) { + client = new DatadogHttpClient(config.getString(API_KEY, null)); + LOGGER.info("Configured DatadogHttpReporter"); + + configTags = getTagsFromConfig(config.getString(TAGS, "")); + } + + @Override + public void close() { + client.close(); + LOGGER.info("Shut down DatadogHttpReporter"); + } + + @Override + public void report() { + DatadogHttpRequest request = new DatadogHttpRequest(); + + for (DGauge g : gauges.values()) { + try { + // Will throw exception if the Gauge is not of Number type + // Flink uses Gauge to store many types other than Number + g.getMetricValue(); + request.addGauge(g); + } catch (Exception e) { + // ignore if the Gauge is not of Number type — End diff – nice catch! done
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112728271

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java —
          @@ -0,0 +1,65 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.datadog.utils.TimestampUtils;
          +
          +import java.util.ArrayList;
          +import java.util.List;
          +
          +/**
          + * Abstract metric of Datadog for serialization
          + * */
          +public abstract class DMetric {
          + private final String metric; // Metric name
          + private final MetricType type;
          + private final List<String> tags;
          +
          + public DMetric(MetricType metricType, String metric, List<String> tags)

          { + this.type = metricType; + this.metric = metric; + this.tags = tags; + }

          +
          + public MetricType getType() {
          + return type;
          — End diff –

          In DGauge and DCounter, their fields 'gauge' and 'counter' don't impact the json serialization.

          I commented on DCounter#getMetricValue() and DGauge#getMetricValue()

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112728271 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DMetric.java — @@ -0,0 +1,65 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.datadog.utils.TimestampUtils; + +import java.util.ArrayList; +import java.util.List; + +/** + * Abstract metric of Datadog for serialization + * */ +public abstract class DMetric { + private final String metric; // Metric name + private final MetricType type; + private final List<String> tags; + + public DMetric(MetricType metricType, String metric, List<String> tags) { + this.type = metricType; + this.metric = metric; + this.tags = tags; + } + + public MetricType getType() { + return type; — End diff – In DGauge and DCounter, their fields 'gauge' and 'counter' don't impact the json serialization. I commented on DCounter#getMetricValue() and DGauge#getMetricValue()
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112729036

          — Diff: flink-dist/src/main/assemblies/opt.xml —
          @@ -95,5 +95,12 @@
          <destName>flink-metrics-statsd-$

          {project.version}.jar</destName>
          <fileMode>0644</fileMode>
          </file>
          +
          + <file>
          + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version}

          .jar</source>
          — End diff –

          Can you elaborate more, and give me more guidance on how to do it correctly?

          I copied this piece of code, and saw it exists in flink-metrics-dropwizard/pom.xml, flink-metrics-ganglia/pom.xml, and flink-metrics-graphite/pom.xml.

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112729036 — Diff: flink-dist/src/main/assemblies/opt.xml — @@ -95,5 +95,12 @@ <destName>flink-metrics-statsd-$ {project.version}.jar</destName> <fileMode>0644</fileMode> </file> + + <file> + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version} .jar</source> — End diff – Can you elaborate more, and give me more guidance on how to do it correctly? I copied this piece of code, and saw it exists in flink-metrics-dropwizard/pom.xml, flink-metrics-ganglia/pom.xml, and flink-metrics-graphite/pom.xml.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112729232

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java —
          @@ -0,0 +1,97 @@
          +/*
          + * 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.datadog;
          +
          +import com.fasterxml.jackson.core.JsonProcessingException;
          +import com.fasterxml.jackson.databind.ObjectMapper;
          +import okhttp3.MediaType;
          +import okhttp3.OkHttpClient;
          +import okhttp3.Request;
          +import okhttp3.Response;
          +import okhttp3.RequestBody;
          +
          +import java.io.IOException;
          +import java.util.concurrent.TimeUnit;
          +
          +/**
          + * Http client talking to Datadog
          + * */
          +public class DatadogHttpClient{
          + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s";
          + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s";
          + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8");
          + private static final int TIMEOUT = 3;
          + private static final ObjectMapper MAPPER = new ObjectMapper();
          +
          + private final String seriesUrl;
          + private final String validateUrl;
          + private final OkHttpClient client;
          + private final String apiKey;
          +
          + public DatadogHttpClient(String dgApiKey) {
          + if(dgApiKey == null || dgApiKey.isEmpty()) {
          — End diff –

          done

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112729232 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpClient.java — @@ -0,0 +1,97 @@ +/* + * 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.datadog; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.RequestBody; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +/** + * Http client talking to Datadog + * */ +public class DatadogHttpClient{ + private static final String SERIES_URL_FORMAT = "https://app.datadoghq.com/api/v1/series?api_key=%s"; + private static final String VALIDATE_URL_FORMAT = "https://app.datadoghq.com/api/v1/validate?api_key=%s"; + private static final MediaType MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8"); + private static final int TIMEOUT = 3; + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private final String seriesUrl; + private final String validateUrl; + private final OkHttpClient client; + private final String apiKey; + + public DatadogHttpClient(String dgApiKey) { + if(dgApiKey == null || dgApiKey.isEmpty()) { — End diff – done
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r112985930

          — Diff: flink-dist/src/main/assemblies/opt.xml —
          @@ -95,5 +95,12 @@
          <destName>flink-metrics-statsd-$

          {project.version}.jar</destName>
          <fileMode>0644</fileMode>
          </file>
          +
          + <file>
          + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version}

          .jar</source>
          — End diff –

          @zentol any details?

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r112985930 — Diff: flink-dist/src/main/assemblies/opt.xml — @@ -95,5 +95,12 @@ <destName>flink-metrics-statsd-$ {project.version}.jar</destName> <fileMode>0644</fileMode> </file> + + <file> + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version} .jar</source> — End diff – @zentol any details?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          @zentol any further details and suggestions on the maven-jar-plugin part?

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 @zentol any further details and suggestions on the maven-jar-plugin part?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r113167440

          — Diff: flink-dist/src/main/assemblies/opt.xml —
          @@ -95,5 +95,12 @@
          <destName>flink-metrics-statsd-$

          {project.version}.jar</destName>
          <fileMode>0644</fileMode>
          </file>
          +
          + <file>
          + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version}

          .jar</source>
          — End diff –

          The `maven-jar-plugin` creates 2 jars. For the datadog reporter these are, right now,

          1. `flink-metrics-datadog-1.3-SNAPSHOT.jar`
          2. `flink-metrics-datadog-1.3-SNAPSHOT-jav-with-dependencies.jar`

          The first jar doesn't contain any dependencies, whereas the second one contains all dependencies declared in the pom.xml, as long as the aren't set to `provided`. We want the latter to be placed in `/opt`, as such the `<source>` tag must point to that jar.

          Basically replace the existing `<source>` configuration with `<source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-$

          {project.version}

          -jar-with-dependencies.jar</source>`

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r113167440 — Diff: flink-dist/src/main/assemblies/opt.xml — @@ -95,5 +95,12 @@ <destName>flink-metrics-statsd-$ {project.version}.jar</destName> <fileMode>0644</fileMode> </file> + + <file> + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version} .jar</source> — End diff – The `maven-jar-plugin` creates 2 jars. For the datadog reporter these are, right now, 1. `flink-metrics-datadog-1.3-SNAPSHOT.jar` 2. `flink-metrics-datadog-1.3-SNAPSHOT-jav-with-dependencies.jar` The first jar doesn't contain any dependencies, whereas the second one contains all dependencies declared in the pom.xml, as long as the aren't set to `provided`. We want the latter to be placed in `/opt`, as such the `<source>` tag must point to that jar. Basically replace the existing `<source>` configuration with `<source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-$ {project.version} -jar-with-dependencies.jar</source>`
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r113259225

          — Diff: flink-dist/src/main/assemblies/opt.xml —
          @@ -95,5 +95,12 @@
          <destName>flink-metrics-statsd-$

          {project.version}.jar</destName>
          <fileMode>0644</fileMode>
          </file>
          +
          + <file>
          + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version}

          .jar</source>
          — End diff –

          Thank you for clarifying!

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r113259225 — Diff: flink-dist/src/main/assemblies/opt.xml — @@ -95,5 +95,12 @@ <destName>flink-metrics-statsd-$ {project.version}.jar</destName> <fileMode>0644</fileMode> </file> + + <file> + <source>../flink-metrics/flink-metrics-datadog/target/flink-metrics-datadog-${project.version} .jar</source> — End diff – Thank you for clarifying!
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/3736

          I think this is starting to look very good!

          Given that this introduces new libraries as dependencies (okhttp, okio), should we pro-actively shade those away to avoid dependency conflicts?
          Admittedly, it would only impact users that explicitly drop in the datadog reporter, but it might still be nice for those users. Given that we build a jr-with-dependencies anyways, the step to shading is small...

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3736 I think this is starting to look very good! Given that this introduces new libraries as dependencies (okhttp, okio), should we pro-actively shade those away to avoid dependency conflicts? Admittedly, it would only impact users that explicitly drop in the datadog reporter, but it might still be nice for those users. Given that we build a jr-with-dependencies anyways, the step to shading is small...
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          @StephanEwen @zentol I shaded okhttp3 and okio from flink-metrics-datadog. I didn't use '<if>shade-flink</id>' because I found it somehow prevents me from building a uber jar. Let me know if it's ok to shade them this way.

          ```
          $ jar -tf target/flink-metrics-datadog-1.3-SNAPSHOT-jar-with-dependencies.jar

          META-INF/
          META-INF/MANIFEST.MF
          META-INF/DEPENDENCIES
          META-INF/LICENSE
          META-INF/NOTICE
          org/
          org/apache/
          org/apache/flink/
          org/apache/flink/metrics/
          org/apache/flink/metrics/datadog/
          org/apache/flink/metrics/datadog/DatadogHttpClient.class
          org/apache/flink/metrics/datadog/DatadogHttpReporter$DatadogHttpRequest.class
          org/apache/flink/metrics/datadog/DatadogHttpReporter.class
          org/apache/flink/metrics/datadog/DCounter.class
          org/apache/flink/metrics/datadog/DGauge.class
          org/apache/flink/metrics/datadog/DMeter.class
          org/apache/flink/metrics/datadog/DMetric.class
          org/apache/flink/metrics/datadog/DSeries.class
          org/apache/flink/metrics/datadog/MetricType.class
          META-INF/maven/
          META-INF/maven/org.apache.flink/
          META-INF/maven/org.apache.flink/flink-metrics-datadog/
          META-INF/maven/org.apache.flink/flink-metrics-datadog/pom.xml
          META-INF/maven/org.apache.flink/flink-metrics-datadog/pom.properties
          META-INF/maven/org.apache.flink/force-shading/
          META-INF/maven/org.apache.flink/force-shading/pom.xml
          META-INF/maven/org.apache.flink/force-shading/pom.properties
          org/apache/flink/shaded/
          org/apache/flink/shaded/okhttp3/
          org/apache/flink/shaded/okhttp3/Address.class
          .....
          org/apache/flink/shaded/okhttp3/WebSocketListener.class
          META-INF/maven/com.squareup.okhttp3/
          META-INF/maven/com.squareup.okhttp3/okhttp/
          META-INF/maven/com.squareup.okhttp3/okhttp/pom.xml
          META-INF/maven/com.squareup.okhttp3/okhttp/pom.properties
          org/apache/flink/shaded/okio/
          org/apache/flink/shaded/okio/AsyncTimeout$1.class
          ...
          org/apache/flink/shaded/okio/Util.class
          META-INF/maven/com.squareup.okio/
          META-INF/maven/com.squareup.okio/okio/
          META-INF/maven/com.squareup.okio/okio/pom.xml
          META-INF/maven/com.squareup.okio/okio/pom.properties
          ```

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 @StephanEwen @zentol I shaded okhttp3 and okio from flink-metrics-datadog. I didn't use '<if>shade-flink</id>' because I found it somehow prevents me from building a uber jar. Let me know if it's ok to shade them this way. ``` $ jar -tf target/flink-metrics-datadog-1.3-SNAPSHOT-jar-with-dependencies.jar META-INF/ META-INF/MANIFEST.MF META-INF/DEPENDENCIES META-INF/LICENSE META-INF/NOTICE org/ org/apache/ org/apache/flink/ org/apache/flink/metrics/ org/apache/flink/metrics/datadog/ org/apache/flink/metrics/datadog/DatadogHttpClient.class org/apache/flink/metrics/datadog/DatadogHttpReporter$DatadogHttpRequest.class org/apache/flink/metrics/datadog/DatadogHttpReporter.class org/apache/flink/metrics/datadog/DCounter.class org/apache/flink/metrics/datadog/DGauge.class org/apache/flink/metrics/datadog/DMeter.class org/apache/flink/metrics/datadog/DMetric.class org/apache/flink/metrics/datadog/DSeries.class org/apache/flink/metrics/datadog/MetricType.class META-INF/maven/ META-INF/maven/org.apache.flink/ META-INF/maven/org.apache.flink/flink-metrics-datadog/ META-INF/maven/org.apache.flink/flink-metrics-datadog/pom.xml META-INF/maven/org.apache.flink/flink-metrics-datadog/pom.properties META-INF/maven/org.apache.flink/force-shading/ META-INF/maven/org.apache.flink/force-shading/pom.xml META-INF/maven/org.apache.flink/force-shading/pom.properties org/apache/flink/shaded/ org/apache/flink/shaded/okhttp3/ org/apache/flink/shaded/okhttp3/Address.class ..... org/apache/flink/shaded/okhttp3/WebSocketListener.class META-INF/maven/com.squareup.okhttp3/ META-INF/maven/com.squareup.okhttp3/okhttp/ META-INF/maven/com.squareup.okhttp3/okhttp/pom.xml META-INF/maven/com.squareup.okhttp3/okhttp/pom.properties org/apache/flink/shaded/okio/ org/apache/flink/shaded/okio/AsyncTimeout$1.class ... org/apache/flink/shaded/okio/Util.class META-INF/maven/com.squareup.okio/ META-INF/maven/com.squareup.okio/okio/ META-INF/maven/com.squareup.okio/okio/pom.xml META-INF/maven/com.squareup.okio/okio/pom.properties ```
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/3736

          +1 from my side
          @zentol any further comments/concerns?

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3736 +1 from my side @zentol any further comments/concerns?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r113405541

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java —
          @@ -0,0 +1,199 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.apache.flink.metrics.Histogram;
          +import org.apache.flink.metrics.Metric;
          +import org.apache.flink.metrics.MetricConfig;
          +import org.apache.flink.metrics.MetricGroup;
          +import org.apache.flink.metrics.reporter.MetricReporter;
          +import org.apache.flink.metrics.reporter.Scheduled;
          +import org.slf4j.Logger;
          +import org.slf4j.LoggerFactory;
          +
          +import java.util.ArrayList;
          +import java.util.Arrays;
          +import java.util.List;
          +import java.util.Map;
          +import java.util.concurrent.ConcurrentHashMap;
          +
          +
          +/**
          + * Metric Reporter for Datadog
          + *
          + * Variables in metrics scope will be sent to Datadog as tags
          + * */
          +public class DatadogHttpReporter implements MetricReporter, Scheduled {
          + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class);
          +
          + // Both Flink's Gauge and Meter values are taken as gauge in Datadog
          + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>();
          + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>();
          + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>();
          +
          + private DatadogHttpClient client;
          + private List<String> configTags;
          +
          + public static final String API_KEY = "apikey";
          — End diff –

          You could define these as a `ConfigOption` which would make it more obvious what the default value is.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r113405541 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java — @@ -0,0 +1,199 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.reporter.MetricReporter; +import org.apache.flink.metrics.reporter.Scheduled; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + + +/** + * Metric Reporter for Datadog + * + * Variables in metrics scope will be sent to Datadog as tags + * */ +public class DatadogHttpReporter implements MetricReporter, Scheduled { + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class); + + // Both Flink's Gauge and Meter values are taken as gauge in Datadog + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>(); + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>(); + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>(); + + private DatadogHttpClient client; + private List<String> configTags; + + public static final String API_KEY = "apikey"; — End diff – You could define these as a `ConfigOption` which would make it more obvious what the default value is.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r113405034

          — Diff: flink-metrics/flink-metrics-datadog/pom.xml —
          @@ -0,0 +1,109 @@
          +<?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/xsd/maven-4.0.0.xsd">
          + <modelVersion>4.0.0</modelVersion>
          +
          + <parent>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-metrics</artifactId>
          + <version>1.3-SNAPSHOT</version>
          + <relativePath>..</relativePath>
          + </parent>
          +
          + <artifactId>flink-metrics-datadog</artifactId>
          + <name>flink-metrics-datadog</name>
          +
          + <dependencies>
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-metrics-core</artifactId>
          + <version>$

          {project.version}</version>
          + <scope>provided</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.fasterxml.jackson.core</groupId>
          + <artifactId>jackson-databind</artifactId>
          + <scope>provided</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.squareup.okhttp3</groupId>
          + <artifactId>okhttp</artifactId>
          + <version>3.6.0</version>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.squareup.okio</groupId>
          + <artifactId>okio</artifactId>
          + <version>1.11.0</version>
          + </dependency>
          +
          +
          + <!-- test dependencies -->
          +
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-runtime_2.10</artifactId>
          + <version>${project.version}

          </version>
          + <scope>test</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-test-utils-junit</artifactId>
          + <version>$

          {project.version}

          </version>
          + <scope>test</scope>
          + </dependency>
          + </dependencies>
          +
          + <build>
          + <plugins>
          + <plugin>
          + <groupId>org.apache.maven.plugins</groupId>
          + <artifactId>maven-shade-plugin</artifactId>
          + <executions>
          + <execution>
          + <phase>package</phase>
          + <goals>
          + <goal>shade</goal>
          + </goals>
          + <configuration>
          + <shadedClassifierName>jar-with-dependencies</shadedClassifierName>
          — End diff –

          I would remove this and adjust the `opt.xml` again; this makes it more consistent with other shaded jars that we produce.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r113405034 — Diff: flink-metrics/flink-metrics-datadog/pom.xml — @@ -0,0 +1,109 @@ +<?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/xsd/maven-4.0.0.xsd "> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-metrics</artifactId> + <version>1.3-SNAPSHOT</version> + <relativePath>..</relativePath> + </parent> + + <artifactId>flink-metrics-datadog</artifactId> + <name>flink-metrics-datadog</name> + + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-metrics-core</artifactId> + <version>$ {project.version}</version> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>okhttp</artifactId> + <version>3.6.0</version> + </dependency> + + <dependency> + <groupId>com.squareup.okio</groupId> + <artifactId>okio</artifactId> + <version>1.11.0</version> + </dependency> + + + <!-- test dependencies --> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-runtime_2.10</artifactId> + <version>${project.version} </version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-test-utils-junit</artifactId> + <version>$ {project.version} </version> + <scope>test</scope> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <shadedClassifierName>jar-with-dependencies</shadedClassifierName> — End diff – I would remove this and adjust the `opt.xml` again; this makes it more consistent with other shaded jars that we produce.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r113510125

          — Diff: flink-metrics/flink-metrics-datadog/pom.xml —
          @@ -0,0 +1,109 @@
          +<?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/xsd/maven-4.0.0.xsd">
          + <modelVersion>4.0.0</modelVersion>
          +
          + <parent>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-metrics</artifactId>
          + <version>1.3-SNAPSHOT</version>
          + <relativePath>..</relativePath>
          + </parent>
          +
          + <artifactId>flink-metrics-datadog</artifactId>
          + <name>flink-metrics-datadog</name>
          +
          + <dependencies>
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-metrics-core</artifactId>
          + <version>$

          {project.version}</version>
          + <scope>provided</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.fasterxml.jackson.core</groupId>
          + <artifactId>jackson-databind</artifactId>
          + <scope>provided</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.squareup.okhttp3</groupId>
          + <artifactId>okhttp</artifactId>
          + <version>3.6.0</version>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.squareup.okio</groupId>
          + <artifactId>okio</artifactId>
          + <version>1.11.0</version>
          + </dependency>
          +
          +
          + <!-- test dependencies -->
          +
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-runtime_2.10</artifactId>
          + <version>${project.version}

          </version>
          + <scope>test</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-test-utils-junit</artifactId>
          + <version>$

          {project.version}

          </version>
          + <scope>test</scope>
          + </dependency>
          + </dependencies>
          +
          + <build>
          + <plugins>
          + <plugin>
          + <groupId>org.apache.maven.plugins</groupId>
          + <artifactId>maven-shade-plugin</artifactId>
          + <executions>
          + <execution>
          + <phase>package</phase>
          + <goals>
          + <goal>shade</goal>
          + </goals>
          + <configuration>
          + <shadedClassifierName>jar-with-dependencies</shadedClassifierName>
          — End diff –

          After removing this line, the shaded jar will be named 'flink-metrics-datadog-1.3-SNAPSHOT-shaded.jar'. Is it what you want?

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r113510125 — Diff: flink-metrics/flink-metrics-datadog/pom.xml — @@ -0,0 +1,109 @@ +<?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/xsd/maven-4.0.0.xsd "> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-metrics</artifactId> + <version>1.3-SNAPSHOT</version> + <relativePath>..</relativePath> + </parent> + + <artifactId>flink-metrics-datadog</artifactId> + <name>flink-metrics-datadog</name> + + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-metrics-core</artifactId> + <version>$ {project.version}</version> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>okhttp</artifactId> + <version>3.6.0</version> + </dependency> + + <dependency> + <groupId>com.squareup.okio</groupId> + <artifactId>okio</artifactId> + <version>1.11.0</version> + </dependency> + + + <!-- test dependencies --> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-runtime_2.10</artifactId> + <version>${project.version} </version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-test-utils-junit</artifactId> + <version>$ {project.version} </version> + <scope>test</scope> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <shadedClassifierName>jar-with-dependencies</shadedClassifierName> — End diff – After removing this line, the shaded jar will be named 'flink-metrics-datadog-1.3-SNAPSHOT-shaded.jar'. Is it what you want?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r113512176

          — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java —
          @@ -0,0 +1,199 @@
          +/*
          + * 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.datadog;
          +
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.apache.flink.metrics.Histogram;
          +import org.apache.flink.metrics.Metric;
          +import org.apache.flink.metrics.MetricConfig;
          +import org.apache.flink.metrics.MetricGroup;
          +import org.apache.flink.metrics.reporter.MetricReporter;
          +import org.apache.flink.metrics.reporter.Scheduled;
          +import org.slf4j.Logger;
          +import org.slf4j.LoggerFactory;
          +
          +import java.util.ArrayList;
          +import java.util.Arrays;
          +import java.util.List;
          +import java.util.Map;
          +import java.util.concurrent.ConcurrentHashMap;
          +
          +
          +/**
          + * Metric Reporter for Datadog
          + *
          + * Variables in metrics scope will be sent to Datadog as tags
          + * */
          +public class DatadogHttpReporter implements MetricReporter, Scheduled {
          + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class);
          +
          + // Both Flink's Gauge and Meter values are taken as gauge in Datadog
          + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>();
          + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>();
          + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>();
          +
          + private DatadogHttpClient client;
          + private List<String> configTags;
          +
          + public static final String API_KEY = "apikey";
          — End diff –

          I wouldn't worry too much about since 1) the current code works well, and changing it doesn't provide extra readability 2) it requires dependency on flink-core, which is completely unnecessary just because of a config

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r113512176 — Diff: flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporter.java — @@ -0,0 +1,199 @@ +/* + * 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.datadog; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.apache.flink.metrics.Histogram; +import org.apache.flink.metrics.Metric; +import org.apache.flink.metrics.MetricConfig; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.reporter.MetricReporter; +import org.apache.flink.metrics.reporter.Scheduled; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + + +/** + * Metric Reporter for Datadog + * + * Variables in metrics scope will be sent to Datadog as tags + * */ +public class DatadogHttpReporter implements MetricReporter, Scheduled { + private static final Logger LOGGER = LoggerFactory.getLogger(DatadogHttpReporter.class); + + // Both Flink's Gauge and Meter values are taken as gauge in Datadog + private final Map<Gauge, DGauge> gauges = new ConcurrentHashMap<>(); + private final Map<Counter, DCounter> counters = new ConcurrentHashMap<>(); + private final Map<Meter, DMeter> meters = new ConcurrentHashMap<>(); + + private DatadogHttpClient client; + private List<String> configTags; + + public static final String API_KEY = "apikey"; — End diff – I wouldn't worry too much about since 1) the current code works well, and changing it doesn't provide extra readability 2) it requires dependency on flink-core, which is completely unnecessary just because of a config
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r113519219

          — Diff: flink-metrics/flink-metrics-datadog/pom.xml —
          @@ -0,0 +1,109 @@
          +<?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/xsd/maven-4.0.0.xsd">
          + <modelVersion>4.0.0</modelVersion>
          +
          + <parent>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-metrics</artifactId>
          + <version>1.3-SNAPSHOT</version>
          + <relativePath>..</relativePath>
          + </parent>
          +
          + <artifactId>flink-metrics-datadog</artifactId>
          + <name>flink-metrics-datadog</name>
          +
          + <dependencies>
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-metrics-core</artifactId>
          + <version>$

          {project.version}</version>
          + <scope>provided</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.fasterxml.jackson.core</groupId>
          + <artifactId>jackson-databind</artifactId>
          + <scope>provided</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.squareup.okhttp3</groupId>
          + <artifactId>okhttp</artifactId>
          + <version>3.6.0</version>
          + </dependency>
          +
          + <dependency>
          + <groupId>com.squareup.okio</groupId>
          + <artifactId>okio</artifactId>
          + <version>1.11.0</version>
          + </dependency>
          +
          +
          + <!-- test dependencies -->
          +
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-runtime_2.10</artifactId>
          + <version>${project.version}

          </version>
          + <scope>test</scope>
          + </dependency>
          +
          + <dependency>
          + <groupId>org.apache.flink</groupId>
          + <artifactId>flink-test-utils-junit</artifactId>
          + <version>$

          {project.version}

          </version>
          + <scope>test</scope>
          + </dependency>
          + </dependencies>
          +
          + <build>
          + <plugins>
          + <plugin>
          + <groupId>org.apache.maven.plugins</groupId>
          + <artifactId>maven-shade-plugin</artifactId>
          + <executions>
          + <execution>
          + <phase>package</phase>
          + <goals>
          + <goal>shade</goal>
          + </goals>
          + <configuration>
          + <shadedClassifierName>jar-with-dependencies</shadedClassifierName>
          — End diff –

          yes.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r113519219 — Diff: flink-metrics/flink-metrics-datadog/pom.xml — @@ -0,0 +1,109 @@ +<?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/xsd/maven-4.0.0.xsd "> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-metrics</artifactId> + <version>1.3-SNAPSHOT</version> + <relativePath>..</relativePath> + </parent> + + <artifactId>flink-metrics-datadog</artifactId> + <name>flink-metrics-datadog</name> + + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-metrics-core</artifactId> + <version>$ {project.version}</version> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>okhttp</artifactId> + <version>3.6.0</version> + </dependency> + + <dependency> + <groupId>com.squareup.okio</groupId> + <artifactId>okio</artifactId> + <version>1.11.0</version> + </dependency> + + + <!-- test dependencies --> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-runtime_2.10</artifactId> + <version>${project.version} </version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-test-utils-junit</artifactId> + <version>$ {project.version} </version> + <scope>test</scope> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <shadedClassifierName>jar-with-dependencies</shadedClassifierName> — End diff – yes.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          @StephanEwen @zentol
          Hi guys, I just tested it on my machine and it works as expected. What's your plan on this PR? I found on email thread that there'll be a feature freeze of release 1.3 on May 1st, and I'd really desire to get this into 1.3

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 @StephanEwen @zentol Hi guys, I just tested it on my machine and it works as expected. What's your plan on this PR? I found on email thread that there'll be a feature freeze of release 1.3 on May 1st, and I'd really desire to get this into 1.3
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          Sorry, I'm rather busy at the moment myself due to the feature freeze and may not be able to try this out in time.

          To be honest, given that if we merge it for 1.4-SNAPSHOT the jar can be easily ported to 1.3 (just replace the versions in the pom.xml and rebuild it) I'm prioritizing this less than other changes that don't have this luxury.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 Sorry, I'm rather busy at the moment myself due to the feature freeze and may not be able to try this out in time. To be honest, given that if we merge it for 1.4-SNAPSHOT the jar can be easily ported to 1.3 (just replace the versions in the pom.xml and rebuild it) I'm prioritizing this less than other changes that don't have this luxury.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/3736

          @bowenli86 Can you build and test this? Then I would be +1 to merge it to 1.3
          It would not be committer tested, but contributor tested, and it is an optional component that is not part of Flink's core functionality, so less critical.

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3736 @bowenli86 Can you build and test this? Then I would be +1 to merge it to 1.3 It would not be committer tested, but contributor tested, and it is an optional component that is not part of Flink's core functionality, so less critical.
          Hide
          phoenixjiangnan Bowen Li added a comment -

          Stephan Ewen Chesnay Schepler While testing the reporter, I found Datadog is unstable in filtering metrics by host names. Contacted Datadog engineering team. They got back to me this morning saying they found the issue, and instead of putting host names as tags, Datadog opens a metric format with extra field of ‘host’.

          Example. Rather than
          "metric='my.series2', points=150, tags=["host:localhost, version:1"])", they changed to "metric='my.series', points=100, host="localhost", tags=["version:1"]"

          I'll make corresponding changes to my reporter

          Show
          phoenixjiangnan Bowen Li added a comment - Stephan Ewen Chesnay Schepler While testing the reporter, I found Datadog is unstable in filtering metrics by host names. Contacted Datadog engineering team. They got back to me this morning saying they found the issue, and instead of putting host names as tags, Datadog opens a metric format with extra field of ‘host’. Example. Rather than "metric='my.series2', points=150, tags= ["host:localhost, version:1"] )", they changed to "metric='my.series', points=100, host="localhost", tags= ["version:1"] " I'll make corresponding changes to my reporter
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          @StephanEwen @zentol I've made the following changes:

          1) Datadog (DD) itself has a bug of being unstable when users filter metrics by 'host' in dashboards if 'host' is sent as tags, details in FLINK-6013(https://issues.apache.org/jira/browse/FLINK-6013). Communicated with DD engineers, they fixed the issue, and now, as advised by them, I'm making 'host' a separate field in metric serialization. Tested the new approach successfully, and we are able to filter metrics by 'host' consistently correctly on DD dashboard.
          2) Added more unit tests for the above change. Unit tests cover both cases when 'host' is present in MetricGroup and when it isn't
          3) Found Maven can only recognize and run unit tests when file name ends with 'Test' rather than 'Tests'. So I renamed my test file to 'DatadogHttpClientTest' from 'DatadogHttpClientTests'
          4) Upgraded okhttp and okio to newer versions
          5) Found mocking system millisec will impact the new okhttp. So I separated unit tests to two enclosed test sets in 'DatadogHttpClientTest'

          Having successfully tested this across my company in the past couple days, I'm now pretty confident this is production ready.

          I agree on @zentol 's proposal if you guys are currently busy with other things, as long as we make sure this ends up in 1.3 Thanks, guys!

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 @StephanEwen @zentol I've made the following changes: 1) Datadog (DD) itself has a bug of being unstable when users filter metrics by 'host' in dashboards if 'host' is sent as tags, details in FLINK-6013 ( https://issues.apache.org/jira/browse/FLINK-6013 ). Communicated with DD engineers, they fixed the issue, and now, as advised by them, I'm making 'host' a separate field in metric serialization. Tested the new approach successfully, and we are able to filter metrics by 'host' consistently correctly on DD dashboard. 2) Added more unit tests for the above change. Unit tests cover both cases when 'host' is present in MetricGroup and when it isn't 3) Found Maven can only recognize and run unit tests when file name ends with 'Test' rather than 'Tests'. So I renamed my test file to 'DatadogHttpClientTest' from 'DatadogHttpClientTests' 4) Upgraded okhttp and okio to newer versions 5) Found mocking system millisec will impact the new okhttp. So I separated unit tests to two enclosed test sets in 'DatadogHttpClientTest' Having successfully tested this across my company in the past couple days, I'm now pretty confident this is production ready. I agree on @zentol 's proposal if you guys are currently busy with other things, as long as we make sure this ends up in 1.3 Thanks, guys!
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r114501062

          — Diff: flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java —
          @@ -0,0 +1,194 @@
          +/*
          + * 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.datadog;
          +
          +import com.fasterxml.jackson.core.JsonProcessingException;
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.junit.Before;
          +import org.junit.Test;
          +import org.junit.experimental.runners.Enclosed;
          +import org.junit.runner.RunWith;
          +import org.powermock.api.mockito.PowerMockito;
          +import org.powermock.core.classloader.annotations.PrepareForTest;
          +import org.powermock.modules.junit4.PowerMockRunner;
          +
          +import java.util.Arrays;
          +import java.util.List;
          +
          +import static org.junit.Assert.assertEquals;
          +
          +@RunWith(Enclosed.class)
          +public class DatadogHttpClientTest {
          + public static class TestApiKey {
          + @Test(expected = IllegalArgumentException.class)
          + public void testValidateApiKey() {
          + new DatadogHttpClient("fake_key");
          — End diff –

          This always sends a request to the Datadog servers correct? I'm a bit skeptical since this means if the servers are down we end up with a failing test. In general though, throwing out requests at them even though we know they will fail anyway doesn't seem ehhh.... nice?

          Just checking that it fails when supplying null or an empty string is enough.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r114501062 — Diff: flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java — @@ -0,0 +1,194 @@ +/* + * 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.datadog; + +import com.fasterxml.jackson.core.JsonProcessingException; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.runners.Enclosed; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.util.Arrays; +import java.util.List; + +import static org.junit.Assert.assertEquals; + +@RunWith(Enclosed.class) +public class DatadogHttpClientTest { + public static class TestApiKey { + @Test(expected = IllegalArgumentException.class) + public void testValidateApiKey() { + new DatadogHttpClient("fake_key"); — End diff – This always sends a request to the Datadog servers correct? I'm a bit skeptical since this means if the servers are down we end up with a failing test. In general though, throwing out requests at them even though we know they will fail anyway doesn't seem ehhh.... nice? Just checking that it fails when supplying null or an empty string is enough.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3736#discussion_r114595396

          — Diff: flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java —
          @@ -0,0 +1,194 @@
          +/*
          + * 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.datadog;
          +
          +import com.fasterxml.jackson.core.JsonProcessingException;
          +import org.apache.flink.metrics.Counter;
          +import org.apache.flink.metrics.Gauge;
          +import org.apache.flink.metrics.Meter;
          +import org.junit.Before;
          +import org.junit.Test;
          +import org.junit.experimental.runners.Enclosed;
          +import org.junit.runner.RunWith;
          +import org.powermock.api.mockito.PowerMockito;
          +import org.powermock.core.classloader.annotations.PrepareForTest;
          +import org.powermock.modules.junit4.PowerMockRunner;
          +
          +import java.util.Arrays;
          +import java.util.List;
          +
          +import static org.junit.Assert.assertEquals;
          +
          +@RunWith(Enclosed.class)
          +public class DatadogHttpClientTest {
          + public static class TestApiKey {
          + @Test(expected = IllegalArgumentException.class)
          + public void testValidateApiKey() {
          + new DatadogHttpClient("fake_key");
          — End diff –

          Make sense. There're lots of uncertainty why testing integration with 3rd party - you don't want to depend to much on it, but you have to test it somehow. It's hard to find a balance somewhere in between.

          I'll make changes as you recommended.

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on a diff in the pull request: https://github.com/apache/flink/pull/3736#discussion_r114595396 — Diff: flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpClientTest.java — @@ -0,0 +1,194 @@ +/* + * 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.datadog; + +import com.fasterxml.jackson.core.JsonProcessingException; +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.Meter; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.runners.Enclosed; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.util.Arrays; +import java.util.List; + +import static org.junit.Assert.assertEquals; + +@RunWith(Enclosed.class) +public class DatadogHttpClientTest { + public static class TestApiKey { + @Test(expected = IllegalArgumentException.class) + public void testValidateApiKey() { + new DatadogHttpClient("fake_key"); — End diff – Make sense. There're lots of uncertainty why testing integration with 3rd party - you don't want to depend to much on it, but you have to test it somehow. It's hard to find a balance somewhere in between. I'll make changes as you recommended.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          @zentol do you have any other concerns?

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 @zentol do you have any other concerns?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          No, I think we addressed all issues

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 No, I think we addressed all issues
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          @zentol cool! Let me know how to get this into 1.3

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 @zentol cool! Let me know how to get this into 1.3
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          This won't make it into 1.3. For one the feature freeze is in effect since today, so it is too late to merge it now.

          That said it may be possible to merge it for 1.4 and maintain a separate repository for a 1.3 version that we can also link in the docs. But I'm not entirely sure.

          I tried it out this morning, and it worked very well. There was an odd thing where the host field in the metric identifier wasn't shown for the taskmanager metrics but for the jobmanager, but that may hidden by Datadog automatically when the hosts match. (i.e. the first part of the metric identifier)

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 This won't make it into 1.3. For one the feature freeze is in effect since today, so it is too late to merge it now. That said it may be possible to merge it for 1.4 and maintain a separate repository for a 1.3 version that we can also link in the docs. But I'm not entirely sure. I tried it out this morning, and it worked very well. There was an odd thing where the host field in the metric identifier wasn't shown for the taskmanager metrics but for the jobmanager, but that may hidden by Datadog automatically when the hosts match. (i.e. the first part of the metric identifier)
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          I see - by looking at the code, I saw Flink has switched to version 1.4-SNAPSHOT 12h ago. Getting it into 1.4 is fine, our company has a open ticket for getting this code into Flink, and I want to get it done.

          We haven't seen the issue you described on our side yet. I'll keep an eye on it if it comes up.

          What's next for this PR? Anyone from Data Artisan will take a look at this PR, and merge it in?

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 I see - by looking at the code, I saw Flink has switched to version 1.4-SNAPSHOT 12h ago. Getting it into 1.4 is fine, our company has a open ticket for getting this code into Flink, and I want to get it done. We haven't seen the issue you described on our side yet. I'll keep an eye on it if it comes up. What's next for this PR? Anyone from Data Artisan will take a look at this PR, and merge it in?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          I can merge it sometime this week, once i found a few more things to merge alongside it.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 I can merge it sometime this week, once i found a few more things to merge alongside it.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user StephanEwen commented on the issue:

          https://github.com/apache/flink/pull/3736

          @bowenli86 Th good thing is that should be able to use the reporter version 1.4-SNAPSHOT also with a Flink 1.3 release, because the reporter API is quite stable by now.

          Show
          githubbot ASF GitHub Bot added a comment - Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3736 @bowenli86 Th good thing is that should be able to use the reporter version 1.4-SNAPSHOT also with a Flink 1.3 release, because the reporter API is quite stable by now.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          Sure.

          BTW, that reminds me of one thing - I need to update the project version in my code from 1.3-SNAPSHOT to 1.4-SNAPSHOT.

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 Sure. BTW, that reminds me of one thing - I need to update the project version in my code from 1.3-SNAPSHOT to 1.4-SNAPSHOT.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          merging.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 merging.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/3736

          @zentol @StephanEwen why not also merge this into 1.3? There are literally no modified or removed lines of code.

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/3736 @zentol @StephanEwen why not also merge this into 1.3? There are literally no modified or removed lines of code.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user asfgit closed the pull request at:

          https://github.com/apache/flink/pull/3736

          Show
          githubbot ASF GitHub Bot added a comment - Github user asfgit closed the pull request at: https://github.com/apache/flink/pull/3736
          Hide
          Zentol Chesnay Schepler added a comment -

          1.4: 54ceec16c11655da4181c0816a3b12d1c4bab465

          Show
          Zentol Chesnay Schepler added a comment - 1.4: 54ceec16c11655da4181c0816a3b12d1c4bab465
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          @greghogan My simple answer is that its feature freeze. There has to be some point in time where new features, regardless of size, stability etc. can't make it into a release. My understanding is that we decided to make the feature freeze date this point in time. When something like this is put into place it naturally follows that some contributions that are ready around the time of the FF will not be merged. Given that the FF was already delayed by a week, implicitly giving every contribution a 1 week grace period, I don't see a reason to make an exception here. It wasn't merged before the FF; it wasn't merged before the delayed FF, so it's not going into the release.

          That's my take on the whole thing.

          PS: Me closing this PR was not meant to imply "and that's the end of it". I just amended the commit before this discussion started and didn't remove it.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 @greghogan My simple answer is that its feature freeze. There has to be some point in time where new features, regardless of size, stability etc. can't make it into a release. My understanding is that we decided to make the feature freeze date this point in time. When something like this is put into place it naturally follows that some contributions that are ready around the time of the FF will not be merged. Given that the FF was already delayed by a week, implicitly giving every contribution a 1 week grace period, I don't see a reason to make an exception here. It wasn't merged before the FF; it wasn't merged before the delayed FF, so it's not going into the release. That's my take on the whole thing. PS: Me closing this PR was not meant to imply "and that's the end of it". I just amended the commit before this discussion started and didn't remove it.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/3736

          @zentol, perhaps "feature freeze" is better named "major feature freeze" and next week's "code freeze" as "minor feature freeze". My understanding is that the first cutoff is to prevent half-merged features (with "major" described as requiring multiple PRs), though master has always been quasi-releasable.

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/3736 @zentol, perhaps "feature freeze" is better named "major feature freeze" and next week's "code freeze" as "minor feature freeze". My understanding is that the first cutoff is to prevent half-merged features (with "major" described as requiring multiple PRs), though master has always been quasi-releasable.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          According to https://cwiki.apache.org/confluence/display/FLINK/Time-based+releases the release of an RC happens when code-freeze is in effect.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 According to https://cwiki.apache.org/confluence/display/FLINK/Time-based+releases the release of an RC happens when code-freeze is in effect.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user greghogan commented on the issue:

          https://github.com/apache/flink/pull/3736

          @zentol yes, and the release release 16 days from now.

          Show
          githubbot ASF GitHub Bot added a comment - Github user greghogan commented on the issue: https://github.com/apache/flink/pull/3736 @zentol yes, and the release release 16 days from now.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          Well we already messed with the timings of the process by extending the feature freeze for a week, so I'm going with the next best thing which is the order of events. (Feature freeze -> Code freeze -> RC)

          But let's cut this short and ask the source of truth, aka the release manager.

          @rmetzger Is the code-freeze in effect? If not, when will it start?

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 Well we already messed with the timings of the process by extending the feature freeze for a week, so I'm going with the next best thing which is the order of events. (Feature freeze -> Code freeze -> RC) But let's cut this short and ask the source of truth, aka the release manager. @rmetzger Is the code-freeze in effect? If not, when will it start?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user zentol commented on the issue:

          https://github.com/apache/flink/pull/3736

          Merging to 1.3 now.

          Show
          githubbot ASF GitHub Bot added a comment - Github user zentol commented on the issue: https://github.com/apache/flink/pull/3736 Merging to 1.3 now.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user bowenli86 commented on the issue:

          https://github.com/apache/flink/pull/3736

          Cool! 1.3 here we go!

          Show
          githubbot ASF GitHub Bot added a comment - Github user bowenli86 commented on the issue: https://github.com/apache/flink/pull/3736 Cool! 1.3 here we go!
          Hide
          Zentol Chesnay Schepler added a comment -

          1.3: 625da00a5797c29969b947f69ccda91ef87eb5c4

          Show
          Zentol Chesnay Schepler added a comment - 1.3: 625da00a5797c29969b947f69ccda91ef87eb5c4

            People

            • Assignee:
              phoenixjiangnan Bowen Li
              Reporter:
              phoenixjiangnan Bowen Li
            • Votes:
              0 Vote for this issue
              Watchers:
              2 Start watching this issue

              Dates

              • Created:
                Updated:
                Resolved:

                Development