[ISSUE-169] Support metric reporter and Support promethues push gateway #415

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

roryqi merged 10 commits into apache:master from xianjingfeng:issue_169

Dec 21, 2022

common/pom.xml

-Original file line number
+Diff line change
@@ Expand Up / @@ -60,6 +60,10 @@ @@
           <groupId>io.prometheus</groupId>
           <artifactId>simpleclient_servlet</artifactId>
         </dependency>
+        <dependency>
+          <groupId>io.prometheus</groupId>
+          <artifactId>simpleclient_pushgateway</artifactId>
+        </dependency>
         <dependency>
           <groupId>com.fasterxml.jackson.core</groupId>
           <artifactId>jackson-databind</artifactId>
@@ Expand Down @@

common/src/main/java/org/apache/uniffle/common/config/RssBaseConf.java

-Original file line number
+Diff line change
@@ Expand Up / @@ -199,6 +199,12 @@ public class RssBaseConf extends RssConf { @@
           .defaultValue(false)
           .withDescription("Whether enable test mode for the shuffle server.");
+      public static final ConfigOption<String> RSS_METRICS_REPORTER_CLASS = ConfigOptions
+          .key("rss.metrics.reporter.class")
+          .stringType()
+          .noDefaultValue()
+          .withDescription("The class of metrics reporter.");
       public boolean loadCommonConf(Map<String, String> properties) {
         if (properties == null) {
           return false;
@@ Expand Down @@

common/src/main/java/org/apache/uniffle/common/metrics/AbstractMetricReporter.java

-Original file line number
+Diff line change
@@ -0,0 +1,41 @@
+    /*
+     * 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.uniffle.common.metrics;
+    import java.util.ArrayList;
+    import java.util.List;
+    import io.prometheus.client.CollectorRegistry;
+    import org.apache.uniffle.common.config.RssConf;
+    public abstract class AbstractMetricReporter implements MetricReporter {
+      protected final RssConf conf;
+      protected final String instanceId;
+      protected List<CollectorRegistry> registryList = new ArrayList<>();
+      public AbstractMetricReporter(RssConf conf, String instanceId) {
+        this.conf = conf;
+        this.instanceId = instanceId;
+      }
+      @Override
+      public void addCollectorRegistry(CollectorRegistry registry) {
+        registryList.add(registry);
+      }
+    }

common/src/main/java/org/apache/uniffle/common/metrics/MetricReporter.java

-Original file line number
+Diff line change
@@ -0,0 +1,29 @@
+    /*
+     * 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.uniffle.common.metrics;
+    import io.prometheus.client.CollectorRegistry;
+    public interface MetricReporter {
+      void start();
+      void stop();
+      void addCollectorRegistry(CollectorRegistry registry);
+    }

common/src/main/java/org/apache/uniffle/common/metrics/MetricReporterFactory.java

-Original file line number
+Diff line change
@@ -0,0 +1,39 @@
+    /*
+     * Licensed to the Apache Software Foundation (ASF) under one or more
+     * contributor license agreements.  See the NOTICE file distributed with
+     * this work for additional information regarding copyright ownership.
+     * The ASF licenses this file to You under the Apache License, Version 2.0
+     * (the "License"); you may not use this file except in compliance with
+     * the License.  You may obtain a copy of the License at
+     *
+     *    http://www.apache.org/licenses/LICENSE-2.0
+     *
+     * Unless required by applicable law or agreed to in writing, software
+     * distributed under the License is distributed on an "AS IS" BASIS,
+     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+     * See the License for the specific language governing permissions and
+     * limitations under the License.
+     */
+    package org.apache.uniffle.common.metrics;
+    import java.lang.reflect.Constructor;
+    import org.apache.commons.lang3.StringUtils;
+    import org.apache.uniffle.common.config.RssBaseConf;
+    import org.apache.uniffle.common.config.RssConf;
+    public class MetricReporterFactory {
+      public static MetricReporter getMetricReporter(RssConf conf, String instanceId) throws Exception {
+        String name = conf.get(RssBaseConf.RSS_METRICS_REPORTER_CLASS);
+        if (StringUtils.isEmpty(name)) {
+          return null;
+        }
+        Class<?> klass = Class.forName(name);
+        Constructor<?> constructor;
+        constructor = klass.getConstructor(conf.getClass(), instanceId.getClass());
+        return (AbstractMetricReporter) constructor.newInstance(conf, instanceId);
+      }
+    }

...ava/org/apache/uniffle/common/metrics/prometheus/PrometheusPushGatewayMetricReporter.java

-Original file line number
+Diff line change
@@ -0,0 +1,120 @@
+    /*
+     * 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.uniffle.common.metrics.prometheus;
+    import java.util.HashMap;
+    import java.util.Map;
+    import java.util.concurrent.Executors;
+    import java.util.concurrent.ScheduledExecutorService;
+    import java.util.concurrent.TimeUnit;
+    import com.google.common.annotations.VisibleForTesting;
+    import io.prometheus.client.CollectorRegistry;
+    import io.prometheus.client.exporter.PushGateway;
+    import org.apache.commons.lang3.StringUtils;
+    import org.slf4j.Logger;
+    import org.slf4j.LoggerFactory;
+    import org.apache.uniffle.common.config.RssConf;
+    import org.apache.uniffle.common.metrics.AbstractMetricReporter;
+    import org.apache.uniffle.common.util.ThreadUtils;
+    public class PrometheusPushGatewayMetricReporter extends AbstractMetricReporter {
+      private static final Logger LOG = LoggerFactory.getLogger(PrometheusPushGatewayMetricReporter.class);
+      static final String PUSHGATEWAY_ADDR = "rss.metrics.prometheus.pushgateway.addr";
+      static final String GROUPING_KEY = "rss.metrics.prometheus.pushgateway.groupingkey";
+      static final String JOB_NAME = "rss.metrics.prometheus.pushgateway.jobname";
+      static final String REPORT_INTEVAL = "rss.metrics.prometheus.pushgateway.report.interval.seconds";
+      private ScheduledExecutorService scheduledExecutorService;
+      private PushGateway pushGateway;
+      public PrometheusPushGatewayMetricReporter(RssConf conf, String instanceId)  {
+        super(conf, instanceId);
+      }
+      @Override
+      public void start() {
+        if (pushGateway == null) {
+          String address = conf.getString(PUSHGATEWAY_ADDR, null);
+          if (StringUtils.isEmpty(address)) {
+            throw new RuntimeException(PUSHGATEWAY_ADDR + " should not be empty!");
+          }
+          pushGateway = new PushGateway(address);
+        }
+        String jobName = conf.getString(JOB_NAME, null);
+        if (StringUtils.isEmpty(jobName)) {
+          throw new RuntimeException(JOB_NAME + " should not be empty!");
+        }
+        Map<String, String> groupingKey = parseGroupingKey(conf.getString(GROUPING_KEY, ""));
+        groupingKey.put("instance", instanceId);
+        int reportInterval = conf.getInteger(REPORT_INTEVAL, 10);
+        scheduledExecutorService = Executors.newScheduledThreadPool(1,
+            ThreadUtils.getThreadFactory("PrometheusPushGatewayMetricReporter-%d"));
+        scheduledExecutorService.scheduleWithFixedDelay(() -> {
+          for (CollectorRegistry registry : registryList) {
+            try {
+              pushGateway.push(registry, jobName, groupingKey);
+            } catch (Throwable e) {
+              LOG.error("Failed to send metrics to push gateway.", e);
+            }
+          }
+        }, 0, reportInterval, TimeUnit.SECONDS);
+      }
+      @Override
+      public void stop() {
+        if (scheduledExecutorService != null) {
+          scheduledExecutorService.shutdownNow();
+        }
+      }
+      @VisibleForTesting
+      void setPushGateway(PushGateway pushGateway) {
+        this.pushGateway = pushGateway;
+      }
+      static Map<String, String> parseGroupingKey(final String groupingKeyConfig) {
+        Map<String, String> groupingKey = new HashMap<>();
+        if (!groupingKeyConfig.isEmpty()) {
+          String[] kvs = groupingKeyConfig.split(";");
+          for (String kv : kvs) {
+            int idx = kv.indexOf("=");
+            if (idx < 0) {
+              LOG.warn("Invalid prometheusPushGateway groupingKey:{}, will be ignored", kv);
+              continue;
+            }
+            String labelKey = kv.substring(0, idx);
+            String labelValue = kv.substring(idx + 1);
+            if (StringUtils.isEmpty(labelKey)
+                || StringUtils.isEmpty(labelValue)) {
+              LOG.warn(
+                  "Invalid groupingKey {labelKey:{}, labelValue:{}} must not be empty",
+                  labelKey,
+                  labelValue);
+              continue;
+            }
+            groupingKey.put(labelKey, labelValue);
+          }
+          return groupingKey;
+        }
+        return groupingKey;
+      }
+    }

...org/apache/uniffle/common/metrics/prometheus/PrometheusPushGatewayMetricReporterTest.java

-Original file line number
+Diff line change
@@ -0,0 +1,114 @@
+    /*
+     * 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.uniffle.common.metrics.prometheus;
+    import java.io.IOException;
+    import java.util.Map;
+    import java.util.concurrent.CountDownLatch;
+    import java.util.concurrent.TimeUnit;
+    import io.prometheus.client.CollectorRegistry;
+    import io.prometheus.client.Counter;
+    import io.prometheus.client.exporter.PushGateway;
+    import org.junit.jupiter.api.Test;
+    import org.apache.uniffle.common.config.RssBaseConf;
+    import org.apache.uniffle.common.config.RssConf;
+    import org.apache.uniffle.common.metrics.MetricReporter;
+    import org.apache.uniffle.common.metrics.MetricReporterFactory;
+    import org.apache.uniffle.common.metrics.MetricsManager;
+    import static org.junit.jupiter.api.Assertions.assertEquals;
+    import static org.junit.jupiter.api.Assertions.assertNotNull;
+    import static org.junit.jupiter.api.Assertions.assertTrue;
+    public class PrometheusPushGatewayMetricReporterTest {
+      @Test
+      public void testParseGroupingKey() {
+        Map<String, String> groupingKey =
+            PrometheusPushGatewayMetricReporter.parseGroupingKey("k1=v1;k2=v2");
+        assertNotNull(groupingKey);
+        assertEquals("v1", groupingKey.get("k1"));
+        assertEquals("v2", groupingKey.get("k2"));
+      }
+      @Test
+      public void testParseIncompleteGroupingKey() {
+        Map<String, String> groupingKey =
+            PrometheusPushGatewayMetricReporter.parseGroupingKey("k1=");
+        assertTrue(groupingKey.isEmpty());
+        groupingKey = PrometheusPushGatewayMetricReporter.parseGroupingKey("=v1");
+        assertTrue(groupingKey.isEmpty());
+        groupingKey = PrometheusPushGatewayMetricReporter.parseGroupingKey("k1");
+        assertTrue(groupingKey.isEmpty());
+      }
+      @Test
+      public void test() throws Exception {
+        RssConf conf = new RssConf();
+        conf.setString(RssBaseConf.RSS_METRICS_REPORTER_CLASS,
+            PrometheusPushGatewayMetricReporter.class.getCanonicalName());
+        conf.setString(PrometheusPushGatewayMetricReporter.PUSHGATEWAY_ADDR, "");
+        conf.setString(PrometheusPushGatewayMetricReporter.GROUPING_KEY, "a=1;b=2");
+        String jobName = "jobname";
+        conf.setString(PrometheusPushGatewayMetricReporter.JOB_NAME, jobName);
+        String instanceId = "127.0.0.1-19999";
+        MetricReporter metricReporter = MetricReporterFactory.getMetricReporter(conf, instanceId);
+        assertTrue(metricReporter instanceof PrometheusPushGatewayMetricReporter);
+        MetricsManager metricsManager = new MetricsManager();
+        CollectorRegistry collectorRegistry = metricsManager.getCollectorRegistry();
+        metricReporter.addCollectorRegistry(collectorRegistry);
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        Counter counter1 = metricsManager.addCounter("counter1");
+        counter1.inc();
+        PushGateway pushGateway = new CustomPushGateway((registry, job, groupingKey) -> {
+          countDownLatch.countDown();
+          assertEquals(jobName, job);
+          assertEquals(3, groupingKey.size());
+          assertEquals(instanceId, groupingKey.get("instance"));
+          assertEquals(1, counter1.get());
+        });
+        ((PrometheusPushGatewayMetricReporter) metricReporter).setPushGateway(pushGateway);
+        metricReporter.start();
+        countDownLatch.await(20, TimeUnit.SECONDS);
+        metricReporter.stop();
+      }
+      class CustomPushGateway extends PushGateway {
+        private final CustomCallback<CollectorRegistry, String, Map<String, String>> callback;
+        CustomPushGateway(CustomCallback<CollectorRegistry, String, Map<String, String>> callback) {
+          super("localhost");
+          this.callback = callback;
+        }
+        @Override
+        public void push(CollectorRegistry registry, String job, Map<String, String> groupingKey) throws IOException {
+          callback.apply(registry, job, groupingKey);
+        }
+      }
+      @FunctionalInterface
+      interface CustomCallback<P1, P2, P3> {
+        void apply(P1 p1, P2 p2, P3 p3);
+      }
+    }

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[ISSUE-169] Support metric reporter and Support promethues push gateway #415

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!