[MINOR] improvement(server,coordinator): Dynamic conf support report …

…interval of PrometheusPushGatewayMetricReporter (#2232)

### What changes were proposed in this pull request?

support dynamic conf the report interval of PrometheusPushGatewayMetricReporter

### Why are the changes needed?

Adjust the report interval of PrometheusPushGatewayMetricReporter dynamically.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Locally

```
curl  -X POST http://localhost:19948/api/shuffleServer/conf/ops/temp/update -H "Content-Type: application/json" \
-d '{"update":{"rss.metrics.prometheus.pushgateway.report.interval.seconds": "30"}}'
temporarily effective until restart: Update successfully%
```

server.log

```
[2024-11-04 10:55:24.139] [PrometheusPushGatewayMetricReporter-0] [INFO] PrometheusPushGatewayMetricReporter - Pushed metrics to push gateway
[2024-11-04 10:55:54.460] [PrometheusPushGatewayMetricReporter-0] [ERROR] PrometheusPushGatewayMetricReporter - Failed to send metrics to push gateway.
```

common/src/main/java/org/apache/uniffle/common/metrics/prometheus/PrometheusPushGatewayMetricReporter.java

@@ -19,22 +19,26 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Sets;
 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.ReconfigurableRegistry;
 import org.apache.uniffle.common.config.RssConf;
 import org.apache.uniffle.common.exception.RssException;
 import org.apache.uniffle.common.metrics.AbstractMetricReporter;
 import org.apache.uniffle.common.util.ThreadUtils;
 
-public class PrometheusPushGatewayMetricReporter extends AbstractMetricReporter {
+public class PrometheusPushGatewayMetricReporter extends AbstractMetricReporter
+    implements ReconfigurableRegistry.ReconfigureListener {
   private static final Logger LOG =
       LoggerFactory.getLogger(PrometheusPushGatewayMetricReporter.class);
   static final String PUSHGATEWAY_ADDR = "rss.metrics.prometheus.pushgateway.addr";
@@ -50,6 +54,11 @@ public PrometheusPushGatewayMetricReporter(RssConf conf, String instanceId) {
 
   @Override
   public void start() {
+    startInternal();
+    ReconfigurableRegistry.register(Sets.newHashSet(REPORT_INTEVAL), this);
+  }
+
+  private void startInternal() {
     if (pushGateway == null) {
       String address = conf.getString(PUSHGATEWAY_ADDR, null);
       if (StringUtils.isEmpty(address)) {
@@ -83,11 +92,21 @@ public void start() {
 
   @Override
   public void stop() {
+    stopInternal();
+    ReconfigurableRegistry.unregister(this);
+  }
+
+  private void stopInternal() {
     if (scheduledExecutorService != null) {
       scheduledExecutorService.shutdownNow();
     }
   }
 
+  private void restart() {
+    stopInternal();
+    startInternal();
+  }
+
   @VisibleForTesting
   void setPushGateway(PushGateway pushGateway) {
     this.pushGateway = pushGateway;
@@ -121,4 +140,14 @@ static Map<String, String> parseGroupingKey(final String groupingKeyConfig) {
 
     return groupingKey;
   }
+
+  @Override
+  public void update(RssConf conf, Set<String> changedProperties) {
+    if (changedProperties == null) {
+      return;
+    }
+    if (changedProperties.contains(REPORT_INTEVAL)) {
+      restart();
+    }
+  }
 }