feat(pd): add quorum-aware /v1/ready endpoint and raft gauges

docker/README.md

@@ -202,6 +202,32 @@ done
 curl -fsS http://localhost:8088/about
 ```
 
+PD answers two unauthenticated probe endpoints. `/v1/health` is liveness only:
+it returns `200` as soon as the REST listener is up, even when the PD has no
+raft leader. `/v1/ready` returns `200` only while the PD sees a raft leader,
+and `503` otherwise. Each PD answers for itself: a single PD elects itself, and
+in a three-PD group the two that can reach each other elect a leader and turn
+ready, while a partitioned third keeps answering `503` until it sees that
+leader.
+
+The healthchecks in these files still gate on `/v1/health`, because
+`/v1/ready` ships from the next release onwards while the files run published
+images. Two things to know before pointing them at readiness:
+
+- Match on the body, not the status code. As of 1.7.0 PD answers `200` with
+  `{"status":-1,"error":"Unauthorized!"}` on every path its auth interceptor
+  does not exclude, a path that does not exist included, so a status-only
+  probe reads a PD too old to have `/v1/ready` as ready. The body match holds
+  whichever status a refusal carries. Gate with
+  `curl -fsS http://localhost:8620/v1/ready | grep -q '"ready":true'` instead.
+- Pin `HUGEGRAPH_VERSION` to a release that carries the endpoint, or build the
+  images from source with `docker-compose.dev.yml`.
+
+The `HEALTHCHECK` baked into `hugegraph-pd/Dockerfile` is `/v1/health` as well.
+Both compose files override it, so it governs `docker run` and anything else
+inheriting the image probe, and those keep reading a PD without a quorum as
+healthy.
+
 Open `http://localhost:8088` and sign in as `admin` with the password from
 `.env`.
 

hugegraph-pd/README.md

@@ -284,6 +284,8 @@ docker/docker-compose-3pd-3store-3server.yml
 
 PD exposes metrics via REST API at:
 - Health check: `http://<pd-host>:8620/actuator/health`
+- Liveness: `http://<pd-host>:8620/v1/health` (REST listener is up)
+- Readiness: `http://<pd-host>:8620/v1/ready` (`200` only while the PD sees a raft leader)
 - Metrics: `http://<pd-host>:8620/actuator/metrics`
 
 ## Community

hugegraph-pd/docs/api-reference.md

@@ -774,6 +774,56 @@ curl http://localhost:8620/actuator/health
 }
 ```
 
+### Liveness and Readiness
+
+Two unauthenticated endpoints are meant for probes and startup gates:
+
+| Endpoint | Meaning | Status |
+|----------|---------|--------|
+| `GET /v1/health` | Liveness: the REST listener is up. Does not consult raft. | always `200` |
+| `GET /v1/ready` | Readiness: the raft node is active and sees a leader, so this PD is inside a quorum. | `200` when ready, `503` otherwise |
+
+```bash
+curl -i http://localhost:8620/v1/ready
+```
+
+**Response** (leader of a healthy cluster):
+```json
+{
+  "ready": true,
+  "state": "STATE_LEADER",
+  "isLeader": true
+}
+```
+
+A follower reports `"state": "STATE_FOLLOWER"` with `"isLeader": false`. When
+the quorum is lost the PD keeps answering `/v1/health` with `200` but
+`/v1/ready` turns into `503` with `"ready": false`. Being unauthenticated, the
+body carries no cluster addresses; the leader's address stays on `/v1/members`.
+
+The answer is served from state the raft callbacks maintain rather than from
+the raft node, so it stays prompt while an election is running and never waits
+on the node lock. `state` is therefore the last change raft announced. A PD
+reports `STATE_UNINITIALIZED` with `"ready": false` from process start until
+its first raft callback, which is the ordinary startup window before a quorum
+first forms, and jraft emits no callback for candidacy or leadership transfer,
+so a candidate reports `STATE_FOLLOWER` with `"ready": false`.
+
+Point Kubernetes readiness probes, `depends_on` healthchecks and any
+"wait for PD" script at `/v1/ready`; keep liveness probes on `/v1/health`
+so a PD that merely lost its leader is not restarted.
+
+Match on the body rather than on the status code alone. A PD that predates this
+endpoint does not reliably answer `404` for it: `RestAuthentication` refuses a
+request it does not exclude by writing an error envelope, and as of 1.7.0 it
+does so without setting a status, so an unknown path answers `200` with
+`{"status":-1,"error":"Unauthorized!"}`. A status-only probe therefore reads
+such a PD as ready. The body match holds whichever status a refusal carries: a
+shell gate should use
+`curl -fsS http://<pd-host>:8620/v1/ready | grep -q '"ready":true'`, and a
+Kubernetes `httpGet` probe should be paired with a PD image that carries the
+endpoint.
+
 ### Metrics
 
 ```bash
@@ -796,6 +846,30 @@ pd_store_count{state="Offline"} 0.0
 pd_partition_count 36.0
 ```
 
+#### Raft membership gauges
+
+Exported on `/actuator/prometheus` for alerting on quorum loss:
+
+| Gauge | Value |
+|-------|-------|
+| `hg_raft_leader` | `1` on the raft leader, `0` elsewhere |
+| `hg_raft_has_leader` | `1` while this PD sees a leader (is inside a quorum), `0` otherwise |
+| `hg_raft_alive_peers` | Number of alive peers on the leader, itself included; `NaN` elsewhere |
+
+`hg_raft_alive_peers` counts the peers the leader has heard from within the
+leader lease timeout, which jraft derives as 90% of the election timeout by
+default.
+
+A cluster has lost its quorum when `sum(hg_raft_leader) == 0` or when
+`hg_raft_has_leader == 0` on every member. Both are briefly true during a
+normal election, so alert on them with a `for:` clause longer than the
+election timeout rather than on the instantaneous value.
+
+Do not aggregate `hg_raft_alive_peers` across instances: it is `NaN` on every
+node but the leader, and one `NaN` sample turns the result of `sum()` or
+`avg()` into `NaN` as well. Select the leader's series instead, for example
+`hg_raft_alive_peers and on(instance) (hg_raft_leader == 1)`.
+
 ### Partition API
 
 #### List Partitions

hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/RaftEngine.java

@@ -27,6 +27,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
@@ -46,6 +48,7 @@
 import com.alipay.sofa.jraft.Status;
 import com.alipay.sofa.jraft.conf.Configuration;
 import com.alipay.sofa.jraft.core.Replicator;
+import com.alipay.sofa.jraft.core.State;
 import com.alipay.sofa.jraft.entity.PeerId;
 import com.alipay.sofa.jraft.entity.Task;
 import com.alipay.sofa.jraft.error.RaftError;
@@ -63,14 +66,24 @@
 @Slf4j
 public class RaftEngine {
 
+    /**
+     * Refresh period of the alive peer count behind {@code hg_raft_alive_peers}. jraft's own
+     * step-down timer walks the same peer set every half election timeout, so a one second
+     * poll adds nothing next to the work the node already does, and it keeps the gauge fresh
+     * well inside any scrape interval.
+     */
+    private static final long ALIVE_PEERS_REFRESH_MS = 1000L;
+
     private volatile static RaftEngine instance = new RaftEngine();
     private RaftStateMachine stateMachine;
     private String groupId = "pd_raft";
     private PDConfig.Raft config;
     private RaftGroupService raftGroupService;
     private RpcServer rpcServer;
-    private Node raftNode;
+    private volatile Node raftNode;
     private RaftRpcClient raftRpcClient;
+    private volatile int alivePeerCount = -1;
+    private ScheduledExecutorService alivePeersRefresher;
 
     public RaftEngine() {
         this.stateMachine = new RaftStateMachine();
@@ -133,6 +146,7 @@ public synchronized boolean init(PDConfig.Raft config) {
         this.raftGroupService =
                 new RaftGroupService(groupId, serverId, nodeOptions, rpcServer, true);
         this.raftNode = raftGroupService.start(false);
+        startAlivePeersRefresher();
         log.info("RaftEngine start successfully: id = {}, peers list = {}", groupId,
                  nodeOptions.getInitialConf().getPeers());
         return this.raftNode != null;
@@ -182,6 +196,24 @@ public List<ChannelHandler> backChannelHandlers() {
     }
 
     public void shutDown() {
+        if (this.alivePeersRefresher != null) {
+            this.alivePeersRefresher.shutdownNow();
+            try {
+                // Best effort: shutdownNow only interrupts, and a refresh parked in
+                // listAlivePeers waits on a lock acquire the interrupt does not break,
+                // for up to the raft rpc connect timeout per unreachable peer. A refresh
+                // that outlives this wait may publish one stale count over the reset
+                // below; acceptable while shutDown has no production caller.
+                if (!this.alivePeersRefresher.awaitTermination(1, TimeUnit.SECONDS)) {
+                    log.warn("Raft alive-peers refresher still running after shutdown; " +
+                             "hg_raft_alive_peers may briefly report a stale value");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+            this.alivePeersRefresher = null;
+        }
+        this.alivePeerCount = -1;
         if (this.raftGroupService != null) {
             this.raftGroupService.shutdown();
             try {
@@ -203,7 +235,139 @@ public void shutDown() {
     }
 
     public boolean isLeader() {
-        return this.raftNode.isLeader(true);
+        Node node = this.raftNode;
+        return node != null && node.isLeader(true);
+    }
+
+    /**
+     * Whether this PD is ready in the sense {@code GET /v1/ready} answers: the raft node
+     * is active and sees a leader.
+     * <p>
+     * A follower only keeps its leader while heartbeats keep arriving inside the election
+     * timeout, and a leader only keeps its role while it can reach a quorum. Seeing a leader
+     * therefore means this node is part of a quorum from its own point of view, which is the
+     * signal a readiness probe needs. Served from the state machine callbacks, not from the
+     * raft node, so it never waits on the node lock.
+     * <p>
+     * Derived from {@link #getRaftStatus()} so {@code hg_raft_has_leader} cannot drift from
+     * the endpoint it is documented to mirror.
+     */
+    public boolean hasLeader() {
+        return getRaftStatus().isReady();
+    }
+
+    /**
+     * Take a view of the local raft state from the volatile copies the state machine
+     * callbacks maintain, never from the raft node itself. During an election jraft holds
+     * the node lock while it reconnects to peers, so a probe that read the node stalled for
+     * the connect timeout instead of answering its 503 promptly. All fields derive from one
+     * read of one immutable view, swapped in whole per callback, so they cannot contradict
+     * each other.
+     * <p>
+     * The state reported is the last one a callback announced: leader, follower, error or
+     * shutdown, and {@code STATE_UNINITIALIZED} until the first callback runs. jraft emits
+     * no callback for candidacy or leadership transfer, so a candidate reads as a follower
+     * that sees no leader, which yields the same not-ready answer. The view trails the node
+     * by whatever sits in the FSM queue ahead of the announcement, which is the price of
+     * never waiting on the node lock.
+     * <p>
+     * A missing raft node needs no separate branch: before {@link #init} the view still
+     * holds its initial {@code STATE_UNINITIALIZED}, and {@code shutDown()} drops the node
+     * only after {@code join()} has let the shutdown callback announce
+     * {@code STATE_SHUTDOWN}. Neither state is active, so neither reads as ready, and a
+     * branch on the node would report an uninitialized PD where the callback said error or
+     * shutdown.
+     */
+    public RaftStatus getRaftStatus() {
+        RaftStateMachine.ProbeView view = this.stateMachine.getProbeView();
+        return new RaftStatus(view.state.isActive() && view.seesLeader,
+                              view.state.name(), State.STATE_LEADER == view.state);
+    }
+
+    /**
+     * Immutable view of the raft state behind {@code GET /v1/ready}. It carries no cluster
+     * addresses: the endpoint is unauthenticated, and the leader's address stays on the
+     * authenticated {@code /v1/members}.
+     */
+    public static final class RaftStatus {
+
+        private final boolean ready;
+        private final String state;
+        private final boolean localLeader;
+
+        RaftStatus(boolean ready, String state, boolean localLeader) {
+            this.ready = ready;
+            this.state = state;
+            this.localLeader = localLeader;
+        }
+
+        public boolean isReady() {
+            return this.ready;
+        }
+
+        /**
+         * @return the jraft node state name, never null
+         */
+        public String getState() {
+            return this.state;
+        }
+
+        public boolean isLocalLeader() {
+            return this.localLeader;
+        }
+    }
+
+    /**
+     * Number of raft peers, this node included, that the leader has heard from within the
+     * leader lease timeout, which jraft derives as 90% of the election timeout by default.
+     * Only the leader tracks replication state, so any other node reports -1.
+     * <p>
+     * This is the value the refresher last published, at most one refresh period old, and -1
+     * until the first refresh runs. The count cannot be read here: {@code listAlivePeers}
+     * takes the node read lock before it checks for leadership, so a caller that read the
+     * node would wait out whoever holds the write lock, which during an election is a
+     * per-peer reconnect bounded only by the rpc connect timeout.
+     */
+    public int getAlivePeerCount() {
+        return this.alivePeerCount;
+    }
+
+    private void startAlivePeersRefresher() {
+        ScheduledExecutorService refresher =
+                Executors.newSingleThreadScheduledExecutor(runnable -> {
+                    Thread thread = new Thread(runnable, "pd-raft-alive-peers");
+                    thread.setDaemon(true);
+                    return thread;
+                });
+        refresher.scheduleWithFixedDelay(this::refreshAlivePeerCount, 0,
+                                         ALIVE_PEERS_REFRESH_MS, TimeUnit.MILLISECONDS);
+        this.alivePeersRefresher = refresher;
+    }
+
+    /**
+     * Read the alive peer count from the raft node and publish it for the gauge. Runs on the
+     * refresher thread, never on a request thread, because the read can block for as long as
+     * an election holds the node write lock. A fixed delay schedule means a blocked refresh
+     * only delays the next one, and the gauge keeps reporting the last value meanwhile.
+     */
+    void refreshAlivePeerCount() {
+        Node node = this.raftNode;
+        if (node == null || !this.stateMachine.isLeader()) {
+            this.alivePeerCount = -1;
+            return;
+        }
+        try {
+            this.alivePeerCount = node.listAlivePeers().size();
+        } catch (IllegalStateException e) {
+            // Lost leadership between the check and the call
+            this.alivePeerCount = -1;
+        } catch (Throwable e) {
+            // scheduleWithFixedDelay cancels every later run if the task throws,
+            // and an Error escaping here would leave the gauge serving its last
+            // value forever, so this catch has to be wider than Exception.
+            log.warn("Failed to refresh the raft alive peer count", e);
+            this.alivePeerCount = -1;
+        }
     }
 
     /**
@@ -232,7 +396,8 @@ public PDConfig.Raft getConfig() {
     }
 
     public PeerId getLeader() {
-        return raftNode.getLeaderId();
+        Node node = this.raftNode;
+        return node == null ? null : node.getLeaderId();
     }
 
     /**