[server] Add rack label for tablet server metrics (#1468)

Co-authored-by: Liebing <[email protected]>

Author: LB-Yu

fluss-server/src/main/java/com/alibaba/fluss/server/metrics/ServerMetricUtils.java

@@ -31,6 +31,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
 import javax.management.AttributeNotFoundException;
 import javax.management.InstanceNotFoundException;
 import javax.management.MBeanException;
@@ -79,9 +80,13 @@ public static CoordinatorMetricGroup createCoordinatorGroup(
     }
 
     public static TabletServerMetricGroup createTabletServerGroup(
-            MetricRegistry registry, String clusterId, String hostname, int serverId) {
+            MetricRegistry registry,
+            String clusterId,
+            @Nullable String rack,
+            String hostname,
+            int serverId) {
         TabletServerMetricGroup tabletServerMetricGroup =
-                new TabletServerMetricGroup(registry, clusterId, hostname, serverId);
+                new TabletServerMetricGroup(registry, clusterId, rack, hostname, serverId);
         createAndInitializeStatusMetricGroup(tabletServerMetricGroup);
         return tabletServerMetricGroup;
     }

fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/TabletServerMetricGroup.java

@@ -38,6 +38,7 @@ public class TabletServerMetricGroup extends AbstractMetricGroup {
             MapUtils.newConcurrentHashMap();
 
     protected final String clusterId;
+    protected final String rack;
     protected final String hostname;
     protected final int serverId;
 
@@ -49,9 +50,10 @@ public class TabletServerMetricGroup extends AbstractMetricGroup {
     private final Counter delayedFetchFromClientExpireCount;
 
     public TabletServerMetricGroup(
-            MetricRegistry registry, String clusterId, String hostname, int serverId) {
+            MetricRegistry registry, String clusterId, String rack, String hostname, int serverId) {
         super(registry, new String[] {clusterId, hostname, NAME}, null);
         this.clusterId = clusterId;
+        this.rack = rack;
         this.hostname = hostname;
         this.serverId = serverId;
 
@@ -75,6 +77,12 @@ public TabletServerMetricGroup(
     @Override
     protected final void putVariables(Map<String, String> variables) {
         variables.put("cluster_id", clusterId);
+        if (rack != null) {
+            variables.put("rack", rack);
+        } else {
+            // The value of an empty string indicates no rack
+            variables.put("rack", "");
+        }
         variables.put("host", hostname);
         variables.put("server_id", String.valueOf(serverId));
     }

fluss-server/src/main/java/com/alibaba/fluss/server/tablet/TabletServer.java

@@ -177,6 +177,7 @@ protected void startServices() throws Exception {
                     ServerMetricUtils.createTabletServerGroup(
                             metricRegistry,
                             ServerMetricUtils.validateAndGetClusterId(conf),
+                            rack,
                             endpoints.get(0).getHost(),
                             serverId);
 

fluss-server/src/test/java/com/alibaba/fluss/server/metrics/group/TestingMetricGroups.java

@@ -25,7 +25,7 @@
 public class TestingMetricGroups {
 
     public static final TabletServerMetricGroup TABLET_SERVER_METRICS =
-            new TabletServerMetricGroup(NOPMetricRegistry.INSTANCE, "fluss", "host", 0);
+            new TabletServerMetricGroup(NOPMetricRegistry.INSTANCE, "fluss", "host", "rack", 0);
 
     public static final CoordinatorMetricGroup COORDINATOR_METRICS =
             new CoordinatorMetricGroup(NOPMetricRegistry.INSTANCE, "cluster1", "host", "0");