refactor(metrics): phase 1 deprecation of the legacy monitor stack

METRICS_CHANGELOG.md

@@ -3,6 +3,17 @@ Metrics Changelog
 
 This file tracks Prometheus metric additions, changes, and removals in java-tron. For the full set of metrics emitted today, see the references at the bottom.
 
+**4.8.3**
+
+### New Metrics
+
+#### Core
+
+- `tron:node_info` (Info, labels `version`, `genesis_block_id`) — static node identity: the running node version string plus the full genesis block hash as the canonical chain identifier, for fleet dashboards and version/chain alert rules. ([#6923](https://github.com/tronprotocol/java-tron/issues/6923))
+- `tron:block_fetch_armed` (Counter) — incremented when `FetchBlockService` arms a fetch tracking for a block; denominator for secondary-fetch rates. ([#6923](https://github.com/tronprotocol/java-tron/issues/6923))
+- `tron:block_fetch_secondary` (Counter) — incremented when a secondary fetch request is sent to an alternate peer. ([#6923](https://github.com/tronprotocol/java-tron/issues/6923))
+- `tron:block_already_known` (Counter) — incremented when a block response matches an outstanding adv request whose exact block ID is already known before that response is processed (best-effort signal; concurrent arrivals may be missed). ([#6923](https://github.com/tronprotocol/java-tron/issues/6923))
+
 **4.8.2**
 
 ### New Metrics

common/src/main/java/org/tron/common/prometheus/MetricKeys.java

@@ -21,6 +21,10 @@ public static class Counter {
     public static final String P2P_ERROR = "tron:p2p_error";
     public static final String P2P_DISCONNECT = "tron:p2p_disconnect";
     public static final String INTERNAL_SERVICE_FAIL = "tron:internal_service_fail";
+    // verification counters for the bounded fetch latency estimator rollout
+    public static final String BLOCK_FETCH_ARMED = "tron:block_fetch_armed";
+    public static final String BLOCK_FETCH_SECONDARY = "tron:block_fetch_secondary";
+    public static final String BLOCK_ALREADY_KNOWN = "tron:block_already_known";
 
     private Counter() {
       throw new IllegalStateException("Counter");
@@ -44,6 +48,16 @@ private Gauge() {
 
   }
 
+  // Info
+  public static class Info {
+    public static final String NODE_INFO = "tron:node";
+
+    private Info() {
+      throw new IllegalStateException("Info");
+    }
+
+  }
+
   // Histogram
   public static class Histogram {
     public static final String HTTP_SERVICE_LATENCY = "tron:http_service_latency_seconds";

common/src/main/java/org/tron/common/prometheus/MetricLabels.java

@@ -78,4 +78,17 @@ private Histogram() {
 
   }
 
+  // Info
+  public static class Info {
+    public static final String VERSION = "version";
+    // identifies the genesis block: the label value is the chain id derived from the
+    // genesis block hash, so the label is named genesis_block_id
+    public static final String CHAIN_ID = "genesis_block_id";
+
+    private Info() {
+      throw new IllegalStateException("Info");
+    }
+
+  }
+
 }

common/src/main/java/org/tron/common/prometheus/Metrics.java

@@ -65,4 +65,8 @@ public static void histogramObserve(Histogram.Timer startTimer) {
   public static void histogramObserve(String key, double amt, String... labels) {
     MetricsHistogram.observe(key, amt, labels);
   }
+
+  public static void info(String key, String... labels) {
+    MetricsInfo.set(key, labels);
+  }
 }

common/src/main/java/org/tron/common/prometheus/MetricsCounter.java

@@ -19,6 +19,15 @@ class MetricsCounter {
     init(MetricKeys.Counter.P2P_DISCONNECT, "tron p2p disconnect .", "type");
     init(MetricKeys.Counter.INTERNAL_SERVICE_FAIL, "internal Service fail.",
         "class", "method");
+    init(MetricKeys.Counter.BLOCK_FETCH_ARMED,
+        "in-flight fetch requests armed by the fetch-block service.");
+    init(MetricKeys.Counter.BLOCK_FETCH_SECONDARY,
+        "secondary fetch requests issued by the fetch-block failover estimator.");
+    init(MetricKeys.Counter.BLOCK_ALREADY_KNOWN,
+        "adv block responses matched to an outstanding request whose exact block id "
+            + "was already known when the response was handled (best-effort: concurrent "
+            + "arrivals may be missed; a duplicate is not attributed to secondary "
+            + "fetches).");
   }
 
   private MetricsCounter() {

common/src/main/java/org/tron/common/prometheus/MetricsInfo.java

@@ -0,0 +1,40 @@
+package org.tron.common.prometheus;
+
+import io.prometheus.client.Info;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j(topic = "metrics")
+class MetricsInfo {
+
+  private static final Map<String, Info> container = new ConcurrentHashMap<>();
+
+  static {
+    init(MetricKeys.Info.NODE_INFO, "tron node info.",
+        MetricLabels.Info.VERSION, MetricLabels.Info.CHAIN_ID);
+  }
+
+  private MetricsInfo() {
+    throw new IllegalStateException("MetricsInfo");
+  }
+
+  private static void init(String name, String help, String... labels) {
+    container.put(name, Info.build()
+        .name(name)
+        .help(help)
+        .labelNames(labels)
+        .register());
+  }
+
+  static void set(String key, String... labels) {
+    if (Metrics.enabled()) {
+      Info info = container.get(key);
+      if (info == null) {
+        logger.info("{} not exist", key);
+        return;
+      }
+      info.labels(labels);
+    }
+  }
+}

framework/src/main/java/org/tron/core/metrics/MetricsKey.java

@@ -23,6 +23,5 @@ public class MetricsKey {
   public static final String NET_API_DETAIL_QPS = "net.api.detail.qps.";
   public static final String NET_API_DETAIL_FAIL_QPS = "net.api.detail.failQps.";
   public static final String NET_API_DETAIL_OUT_TRAFFIC = "net.api.detail.outTraffic.";
-  public static final String NET_LATENCY_FETCH_BLOCK = "net.latency.fetch.block.";
 
 }

framework/src/main/java/org/tron/core/net/messagehandler/BlockMsgHandler.java

@@ -15,8 +15,6 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.exception.P2pException;
 import org.tron.core.exception.P2pException.TypeEnum;
-import org.tron.core.metrics.MetricsKey;
-import org.tron.core.metrics.MetricsUtil;
 import org.tron.core.net.TronNetDelegate;
 import org.tron.core.net.message.TronMessage;
 import org.tron.core.net.message.adv.BlockMessage;
@@ -91,10 +89,19 @@ public void processMessage(PeerConnection peer, TronMessage msg) throws P2pExcep
       }
       Long time = peer.getAdvInvRequest().remove(item);
       if (null != time) {
-        MetricsUtil.histogramUpdateUnCheck(MetricsKey.NET_LATENCY_FETCH_BLOCK
-                + peer.getInetAddress(), now - time);
+        peer.updateFetchLatency(now - time);
         Metrics.histogramObserve(MetricKeys.Histogram.BLOCK_FETCH_LATENCY,
             (now - time) / Metrics.MILLISECONDS_PER_SECOND);
+        // Best-effort duplicate signal: only responses matched to an outstanding adv
+        // request whose exact block id was already known before this response is
+        // processed (a concurrent or redundant arrival) are counted. The lookup is
+        // exact-id (block store + khaos), not a height comparison: an unknown fork
+        // block below head must not count. Concurrency can still let a simultaneous
+        // arrival slip through, and a duplicate is not attributed to a secondary
+        // fetch, so this is a lower-bound indicator rather than an exact count.
+        if (tronNetDelegate.containBlock(blockId)) {
+          Metrics.counterInc(MetricKeys.Counter.BLOCK_ALREADY_KNOWN, 1);
+        }
       }
       Metrics.histogramObserve(MetricKeys.Histogram.BLOCK_RECEIVE_DELAY,
           (now - blockMessage.getBlockCapsule().getTimeStamp()) / Metrics.MILLISECONDS_PER_SECOND);

framework/src/main/java/org/tron/core/net/peer/PeerConnection.java

@@ -24,7 +24,9 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Scope;
 import org.springframework.stereotype.Component;
+import org.tron.common.math.StrictMathWrapper;
 import org.tron.common.overlay.message.Message;
+import org.tron.common.parameter.CommonParameter;
 import org.tron.common.prometheus.MetricKeys;
 import org.tron.common.prometheus.Metrics;
 import org.tron.common.utils.Pair;
@@ -92,6 +94,21 @@ public class PeerConnection {
   @Getter
   private volatile long blockRcvTime;
 
+  /**
+   * EWMA smoothing divisor for the fetch latency estimator: the previous estimate is
+   * weighted (EWMA_DIVISOR - 1) / EWMA_DIVISOR and the new sample 1 / EWMA_DIVISOR,
+   * i.e. alpha = 0.1. This trades off smoothing against responsiveness and sits in the
+   * same order of magnitude as TCP's SRTT gain (1/8, RFC 6298). Under a large
+   * degradation the relative ordering of two peers can flip within 1-2 samples, while
+   * the absolute value converges smoothly (e.g. seeded at 100, ten 500ms samples walk
+   * 140, 176, 208, 237, 263, 286, 307, 326, 343, 358 without ever hitting the clamp).
+   */
+  private static final int EWMA_DIVISOR = 10;
+
+  private volatile long fetchLatency;
+
+  private volatile boolean fetchLatencySeeded;
+
   @Getter
   @Setter
   private volatile TronState tronState = TronState.INIT;
@@ -184,6 +201,49 @@ public void setChannel(Channel channel) {
         Args.getInstance().getRateLimiterDisconnect());
   }
 
+  /**
+   * Bounded fetch latency estimator with an explicit unsampled state.
+   *
+   * <p>The channel's average latency is never part of the sample sequence; it is only a
+   * read fallback while the estimator is unsampled (see {@link #getFetchLatency()}). The
+   * first measured fetch latency directly replaces the unsampled state (isomorphic to
+   * RFC 6298 SRTT initialization), and subsequent samples are blended with an EWMA of
+   * alpha = 1 / EWMA_DIVISOR = 0.1. With integer division the EWMA has a fixed point, e.g.
+   * (499 * 9 + 500) / 10 = 499, which damps jitter around the saturation bound.
+   *
+   * <p>A single fetch worker reads this value while the channel event loop writes it;
+   * volatile is sufficient for this benign race and no lock should be added.
+   *
+   * @param latencyMillis measured fetch latency in milliseconds
+   */
+  public void updateFetchLatency(long latencyMillis) {
+    if (!fetchLatencySeeded) {
+      fetchLatency = clampFetchLatency(latencyMillis);
+      fetchLatencySeeded = true;
+    } else {
+      fetchLatency = clampFetchLatency(
+          (fetchLatency * (EWMA_DIVISOR - 1) + latencyMillis) / EWMA_DIVISOR);
+    }
+  }
+
+  /**
+   * Returns the bounded fetch latency estimate. While the estimator has not observed a
+   * real fetch sample yet, the channel's average latency is returned as a read fallback
+   * (an unknown peer is treated via its transport-level estimate instead of 0).
+   */
+  public long getFetchLatency() {
+    if (!fetchLatencySeeded) {
+      return channel.getAvgLatency();
+    }
+    return fetchLatency;
+  }
+
+  private long clampFetchLatency(long latency) {
+    // Saturation intentionally makes the >= timeout gate in FetchBlockService trigger.
+    return StrictMathWrapper.max(0,
+        StrictMathWrapper.min(CommonParameter.getInstance().fetchBlockTimeout, latency));
+  }
+
   public void setBlockBothHave(BlockId blockId) {
     this.blockBothHave = blockId;
     this.blockBothHaveUpdateTime = System.currentTimeMillis();

framework/src/main/java/org/tron/core/net/service/fetchblock/FetchBlockService.java

@@ -13,11 +13,11 @@
 import org.springframework.stereotype.Component;
 import org.tron.common.es.ExecutorServiceManager;
 import org.tron.common.parameter.CommonParameter;
+import org.tron.common.prometheus.MetricKeys;
+import org.tron.common.prometheus.Metrics;
 import org.tron.common.utils.Sha256Hash;
 import org.tron.core.ChainBaseManager;
 import org.tron.core.capsule.BlockCapsule;
-import org.tron.core.metrics.MetricsKey;
-import org.tron.core.metrics.MetricsUtil;
 import org.tron.core.net.TronNetDelegate;
 import org.tron.core.net.message.adv.FetchInvDataMessage;
 import org.tron.core.net.peer.Item;
@@ -73,6 +73,7 @@ public void fetchBlock(List<Sha256Hash> sha256HashList, PeerConnection peer) {
         .findFirst().ifPresent(sha256Hash -> {
           long now = System.currentTimeMillis();
           fetchBlockInfo = new FetchBlockInfo(sha256Hash, peer, now);
+          Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_ARMED, 1);
           logger.info("Set fetchBlockInfo, block: {}, peer: {}, time: {}", sha256Hash,
               peer.getInetAddress(), now);
         });
@@ -97,16 +98,17 @@ private void fetchBlockProcess(FetchBlockInfo fetchBlock) {
         .filter(PeerConnection::isIdle)
         .filter(filterPeer -> !filterPeer.equals(fetchBlock.getPeer()))
         .filter(filterPeer -> filterPeer.getAdvInvReceive().getIfPresent(item) != null)
-        .filter(filterPeer -> getPeerTop75(filterPeer)
-            <= CommonParameter.getInstance().fetchBlockTimeout)
-        .min(Comparator.comparingDouble(this::getPeerTop75));
+        // Seeded estimates are clamped to the fetch timeout; the unseeded channel-latency
+        // fallback is not, but min() ordering and the saturation gate keep it safe.
+        .min(Comparator.comparingDouble(this::getPeerLatency));
 
     if (optionalPeerConnection.isPresent()) {
       optionalPeerConnection.ifPresent(firstPeer -> {
         if (shouldFetchBlock(firstPeer, fetchBlock)
             && firstPeer.checkAndPutAdvInvRequest(item, System.currentTimeMillis())) {
           firstPeer.sendMessage(new FetchInvDataMessage(Collections.singletonList(item.getHash()),
               item.getType()));
+          Metrics.counterInc(MetricKeys.Counter.BLOCK_FETCH_SECONDARY, 1);
           this.fetchBlockInfo = null;
         }
       });
@@ -120,21 +122,27 @@ private void fetchBlockProcess(FetchBlockInfo fetchBlock) {
   }
 
   private boolean shouldFetchBlock(PeerConnection newPeer, FetchBlockInfo fetchBlock) {
-    double newPeerTop75 = getPeerTop75(newPeer);
-    double oldPeerTop75 = getPeerTop75(fetchBlock.getPeer());
+    double newPeerLatency = getPeerLatency(newPeer);
+    double oldPeerLatency = getPeerLatency(fetchBlock.getPeer());
     long oldPeerSpendTime = System.currentTimeMillis() - fetchBlock.getTime();
-    if (oldPeerTop75 > fetchTimeOut || oldPeerSpendTime >= fetchTimeOut) {
+    // Switch unconditionally on a hard timeout: an unseeded or saturated old peer must not
+    // permanently wedge fetchBlockInfo.
+    if (oldPeerSpendTime >= fetchTimeOut) {
       return true;
     }
 
-    double oldPeerLeftTime = oldPeerTop75 - oldPeerSpendTime;
-    return newPeerTop75 < oldPeerLeftTime * BLOCK_FETCH_LEFT_TIME_PERCENT
-        && oldPeerSpendTime + newPeerTop75 < fetchTimeOut;
+    // Require a strictly better peer for the latency saturation gate to prevent 500v500 flapping.
+    if (oldPeerLatency >= fetchTimeOut && newPeerLatency < oldPeerLatency) {
+      return true;
+    }
+
+    double oldPeerLeftTime = oldPeerLatency - oldPeerSpendTime;
+    return newPeerLatency < oldPeerLeftTime * BLOCK_FETCH_LEFT_TIME_PERCENT
+        && oldPeerSpendTime + newPeerLatency < fetchTimeOut;
   }
 
-  private double getPeerTop75(PeerConnection peerConnection) {
-    return MetricsUtil.getHistogram(MetricsKey.NET_LATENCY_FETCH_BLOCK
-        + peerConnection.getInetAddress()).getSnapshot().get75thPercentile();
+  private double getPeerLatency(PeerConnection peerConnection) {
+    return peerConnection.getFetchLatency();
   }
 
   private static class FetchBlockInfo {
@@ -159,4 +167,4 @@ public FetchBlockInfo(Sha256Hash hash, PeerConnection peer, long time) {
 
   }
 
-}
+}

framework/src/main/java/org/tron/core/services/RpcApiService.java

@@ -10,6 +10,7 @@
 import io.grpc.netty.NettyServerBuilder;
 import io.grpc.stub.StreamObserver;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -2655,9 +2656,15 @@ public void getBlock(GrpcAPI.BlockReq  request,
 
   public class MonitorApi extends MonitorGrpc.MonitorImplBase {
 
+    private final AtomicBoolean deprecatedWarned = new AtomicBoolean(false);
+
     @Override
     public void getStatsInfo(EmptyMessage request,
         StreamObserver<Protocol.MetricsInfo> responseObserver) {
+      if (deprecatedWarned.compareAndSet(false, true)) {
+        logger.warn("rpc Monitor.GetStatsInfo is deprecated and will be removed in a "
+            + "future major release; migrate to the prometheus metrics endpoint");
+      }
       responseObserver.onNext(metricsApiService.getMetricProtoInfo());
       responseObserver.onCompleted();
     }

framework/src/main/java/org/tron/core/services/http/MetricsServlet.java

@@ -1,5 +1,6 @@
 package org.tron.core.services.http;
 
+import java.util.concurrent.atomic.AtomicBoolean;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import lombok.extern.slf4j.Slf4j;
@@ -13,10 +14,16 @@
 @Slf4j(topic = "API")
 public class MetricsServlet extends RateLimiterServlet {
 
+  private static final AtomicBoolean deprecatedWarned = new AtomicBoolean(false);
+
   @Autowired
   private MetricsApiService metricsApiService;
 
   protected void doGet(HttpServletRequest request, HttpServletResponse response) {
+    if (deprecatedWarned.compareAndSet(false, true)) {
+      logger.warn("HTTP /monitor/getstatsinfo is deprecated and will be removed in a "
+          + "future major release; migrate to the prometheus metrics endpoint");
+    }
     try {
       MetricsInfo metricsInfo = metricsApiService.getMetricsInfo();