Add ReadOnly status reason and StatusReason display

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetSystemStatusTableIT.java

@@ -64,19 +64,45 @@ public void setSystemStatus() {
               () -> {
                 ResultSet resultSet = statement.executeQuery("SHOW DATANODES");
                 int num = 0;
+                int manualReasonNum = 0;
                 try {
                   while (resultSet.next()) {
                     String status = resultSet.getString("Status");
                     if (status.equals("ReadOnly")) {
                       num++;
                     }
+                    // The ReadOnly requested through SQL is reported with the Manual reason.
+                    // The StatusReason column only appears once at least one node has reported
+                    // its reason, so a missing column means the condition is not satisfied yet.
+                    if ("Manual".equals(resultSet.getString("StatusReason"))) {
+                      manualReasonNum++;
+                    }
                   }
-                } catch (InconsistentDataException e) {
+                } catch (Exception e) {
+                  // The StatusReason column (and the Manual reason behind it) may not be
+                  // propagated yet, or the cluster may be transiently inconsistent while the
+                  // status is changing: treat both as "not satisfied yet".
                   return false;
                 }
-                return num == EnvFactory.getEnv().getDataNodeWrapperList().size();
+                return num == EnvFactory.getEnv().getDataNodeWrapperList().size()
+                    && manualReasonNum == EnvFactory.getEnv().getDataNodeWrapperList().size();
               });
 
+      // The table-model NODES view keeps the status and its reason in two separate columns
+      // instead of merging them into "ReadOnly(Manual)".
+      try (ResultSet nodesResultSet =
+          statement.executeQuery("select * from information_schema.nodes")) {
+        int dataNodeNum = 0;
+        while (nodesResultSet.next()) {
+          if ("DataNode".equals(nodesResultSet.getString("node_type"))) {
+            dataNodeNum++;
+            Assert.assertEquals("ReadOnly", nodesResultSet.getString("status"));
+            Assert.assertEquals("Manual", nodesResultSet.getString("status_reason"));
+          }
+        }
+        Assert.assertEquals(EnvFactory.getEnv().getDataNodeWrapperList().size(), dataNodeNum);
+      }
+
       statement.execute("SET SYSTEM TO RUNNING ON CLUSTER");
       Awaitility.await()
           .atMost(10, TimeUnit.SECONDS)

integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java

@@ -680,6 +680,7 @@ public void testInformationSchema() throws SQLException {
                   "node_id,INT32,TAG,",
                   "node_type,STRING,ATTRIBUTE,",
                   "status,STRING,ATTRIBUTE,",
+                  "status_reason,STRING,ATTRIBUTE,",
                   "internal_address,STRING,ATTRIBUTE,",
                   "internal_port,INT32,ATTRIBUTE,",
                   "version,STRING,ATTRIBUTE,",

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -120,6 +120,7 @@
 import org.apache.iotdb.confignode.manager.externalservice.ExternalServiceManager;
 import org.apache.iotdb.confignode.manager.load.LoadManager;
 import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample;
+import org.apache.iotdb.confignode.manager.load.cache.node.NodeStatistics;
 import org.apache.iotdb.confignode.manager.node.ClusterNodeStartUtils;
 import org.apache.iotdb.confignode.manager.node.NodeManager;
 import org.apache.iotdb.confignode.manager.node.NodeMetrics;
@@ -622,33 +623,55 @@ public TShowClusterResp showCluster() {
               .sorted(Comparator.comparingInt(TDataNodeLocation::getDataNodeId))
               .collect(Collectors.toList());
       Map<Integer, TNodeVersionInfo> nodeVersionInfo = getNodeManager().getNodeVersionInfo();
-      Map<Integer, String> nodeStatus = getLoadManager().getNodeStatusWithReason();
+      Map<Integer, NodeStatistics> nodeStatisticsSnapshot =
+          getLoadManager().getNodeStatisticsSnapshot();
+      Map<Integer, String> nodeStatus = new HashMap<>();
+      Map<Integer, String> nodeStatusReason = new HashMap<>();
+      nodeStatisticsSnapshot.forEach(
+          (nodeId, statistics) -> {
+            nodeStatus.put(nodeId, statistics.getStatus().getStatus());
+            if (statistics.getStatusReason() != null) {
+              nodeStatusReason.put(nodeId, statistics.getStatusReason());
+            }
+          });
       configNodeLocations.forEach(
-          configNodeLocation ->
-              nodeStatus.putIfAbsent(
-                  configNodeLocation.getConfigNodeId(), NodeStatus.Unknown.toString()));
+          configNodeLocation -> {
+            int configNodeId = configNodeLocation.getConfigNodeId();
+            if (!nodeStatus.containsKey(configNodeId)) {
+              // No cache entry means the node has never reported a heartbeat.
+              nodeStatus.put(configNodeId, NodeStatus.Unknown.toString());
+            }
+          });
       dataNodeLocations.forEach(
-          dataNodeLocation ->
-              nodeStatus.putIfAbsent(
-                  dataNodeLocation.getDataNodeId(), NodeStatus.Unknown.toString()));
+          dataNodeLocation -> {
+            int dataNodeId = dataNodeLocation.getDataNodeId();
+            if (!nodeStatus.containsKey(dataNodeId)) {
+              // No cache entry means the node has never reported a heartbeat.
+              nodeStatus.put(dataNodeId, NodeStatus.Unknown.toString());
+            }
+          });
 
       List<TAINodeLocation> aiNodeLocations =
           getNodeManager().getRegisteredAINodes().stream()
               .map(TAINodeConfiguration::getLocation)
               .sorted(Comparator.comparingInt(TAINodeLocation::getAiNodeId))
               .collect(Collectors.toList());
-      Map<Integer, String> nodeStatusMap = getLoadManager().getNodeStatusWithReason();
       aiNodeLocations.forEach(
-          aiNodeLocation ->
-              nodeStatusMap.putIfAbsent(
-                  aiNodeLocation.getAiNodeId(), NodeStatus.Unknown.toString()));
+          aiNodeLocation -> {
+            int aiNodeId = aiNodeLocation.getAiNodeId();
+            if (!nodeStatus.containsKey(aiNodeId)) {
+              // No cache entry means the node has never reported a heartbeat.
+              nodeStatus.put(aiNodeId, NodeStatus.Unknown.toString());
+            }
+          });
 
       return new TShowClusterResp()
           .setStatus(status)
           .setConfigNodeList(configNodeLocations)
           .setDataNodeList(dataNodeLocations)
           .setAiNodeList(aiNodeLocations)
           .setNodeStatus(nodeStatus)
+          .setNodeStatusReason(nodeStatusReason)
           .setNodeVersionInfo(nodeVersionInfo);
     } else {
       return new TShowClusterResp()
@@ -657,6 +680,7 @@ public TShowClusterResp showCluster() {
           .setDataNodeList(Collections.emptyList())
           .setAiNodeList(Collections.emptyList())
           .setNodeStatus(Collections.emptyMap())
+          .setNodeStatusReason(Collections.emptyMap())
           .setNodeVersionInfo(Collections.emptyMap());
     }
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java

@@ -40,6 +40,7 @@
 import org.apache.iotdb.confignode.manager.load.cache.LoadCache;
 import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusGroupHeartbeatSample;
 import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample;
+import org.apache.iotdb.confignode.manager.load.cache.node.NodeStatistics;
 import org.apache.iotdb.confignode.manager.load.cache.region.RegionHeartbeatSample;
 import org.apache.iotdb.confignode.manager.load.service.EventService;
 import org.apache.iotdb.confignode.manager.load.service.HeartbeatService;
@@ -259,22 +260,24 @@ public NodeStatus getNodeStatus(int nodeId) {
   }
 
   /**
-   * Safely get the specified Node's current status with reason.
+   * Safely get the specified Node's current status reason.
    *
    * @param nodeId The specified NodeId
-   * @return The specified Node's current status if the nodeCache contains it, Unknown otherwise
+   * @return The reason why the Node is in its current status, null if the node has no reason or the
+   *     cache doesn't exist
    */
-  public String getNodeStatusWithReason(int nodeId) {
-    return loadCache.getNodeStatusWithReason(nodeId);
+  public String getNodeStatusReason(int nodeId) {
+    return loadCache.getNodeStatusReason(nodeId);
   }
 
   /**
-   * Get all Node's current status with reason.
+   * Get all Nodes' current statistics in a single traversal of the node cache, so that each node's
+   * status and reason come from the same statistics snapshot.
    *
-   * @return Map<NodeId, NodeStatus with reason>
+   * @return Map<NodeId, NodeStatistics>
    */
-  public Map<Integer, String> getNodeStatusWithReason() {
-    return loadCache.getNodeStatusWithReason();
+  public Map<Integer, NodeStatistics> getNodeStatisticsSnapshot() {
+    return loadCache.getNodeStatisticsSnapshot();
   }
 
   /**

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java

@@ -536,25 +536,29 @@ public NodeStatus getNodeStatus(int nodeId) {
   }
 
   /**
-   * Safely get the specified Node's current status with reason.
+   * Safely get the specified Node's current status reason.
    *
    * @param nodeId The specified NodeId
-   * @return The specified Node's current status if the nodeCache contains it, Unknown otherwise
+   * @return The reason why the Node is in its current status, null if the node has no reason or the
+   *     cache doesn't exist
    */
-  public String getNodeStatusWithReason(int nodeId) {
-    return Optional.ofNullable(nodeCacheMap.get(nodeId))
-        .map(BaseNodeCache::getNodeStatusWithReason)
-        .orElseGet(() -> NodeStatus.Unknown.getStatus() + "(NoHeartbeat)");
+  public String getNodeStatusReason(int nodeId) {
+    BaseNodeCache nodeCache = nodeCacheMap.get(nodeId);
+    return nodeCache == null ? null : nodeCache.getNodeStatusReason();
   }
 
   /**
-   * Get all Node's current status with reason.
+   * Get all Nodes' current statistics in a single traversal. Each node's status and reason are read
+   * from the same immutable {@link NodeStatistics} snapshot, so a concurrent heartbeat update
+   * cannot produce a status/reason pair from different heartbeats.
    *
-   * @return Map<NodeId, NodeStatus with reason>
+   * @return Map<NodeId, NodeStatistics>
    */
-  public Map<Integer, String> getNodeStatusWithReason() {
+  public Map<Integer, NodeStatistics> getNodeStatisticsSnapshot() {
     return nodeCacheMap.entrySet().stream()
-        .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getNodeStatusWithReason()));
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey, e -> (NodeStatistics) e.getValue().getCurrentStatistics()));
   }
 
   /**

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java

@@ -58,12 +58,9 @@ public NodeStatus getNodeStatus() {
   }
 
   /**
-   * @return The reason why lead to current NodeStatus.
+   * @return The reason why lead to current NodeStatus, null if there is none.
    */
-  public String getNodeStatusWithReason() {
-    NodeStatistics statistics = (NodeStatistics) this.currentStatistics.get();
-    return statistics.getStatusReason() == null
-        ? statistics.getStatus().getStatus()
-        : statistics.getStatus().getStatus() + "(" + statistics.getStatusReason() + ")";
+  public String getNodeStatusReason() {
+    return ((NodeStatistics) currentStatistics.get()).getStatusReason();
   }
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java

@@ -501,7 +501,7 @@ public List<TAINodeInfo> getRegisteredAINodeInfoList() {
     for (TAINodeConfiguration aiNodeConfiguration : getRegisteredAINodes()) {
       TAINodeInfo aiNodeInfo = new TAINodeInfo();
       aiNodeInfo.setAiNodeId(aiNodeConfiguration.getLocation().getAiNodeId());
-      aiNodeInfo.setStatus(getLoadManager().getNodeStatusWithReason(aiNodeInfo.getAiNodeId()));
+      aiNodeInfo.setStatus(getLoadManager().getNodeStatus(aiNodeInfo.getAiNodeId()).getStatus());
       aiNodeInfo.setInternalAddress(aiNodeConfiguration.getLocation().getInternalEndPoint().ip);
       aiNodeInfo.setInternalPort(aiNodeConfiguration.getLocation().getInternalEndPoint().port);
       aiNodeInfoList.add(aiNodeInfo);
@@ -720,7 +720,8 @@ public List<TDataNodeInfo> getRegisteredDataNodeInfoList() {
             TDataNodeInfo dataNodeInfo = new TDataNodeInfo();
             int dataNodeId = registeredDataNode.getLocation().getDataNodeId();
             dataNodeInfo.setDataNodeId(dataNodeId);
-            dataNodeInfo.setStatus(getLoadManager().getNodeStatusWithReason(dataNodeId));
+            dataNodeInfo.setStatus(getLoadManager().getNodeStatus(dataNodeId).getStatus());
+            dataNodeInfo.setStatusReason(getLoadManager().getNodeStatusReason(dataNodeId));
             dataNodeInfo.setRpcAddresss(
                 registeredDataNode.getLocation().getClientRpcEndPoint().getIp());
             dataNodeInfo.setRpcPort(
@@ -856,7 +857,8 @@ public List<TConfigNodeInfo> getRegisteredConfigNodeInfoList() {
             TConfigNodeInfo info = new TConfigNodeInfo();
             int configNodeId = configNodeLocation.getConfigNodeId();
             info.setConfigNodeId(configNodeId);
-            info.setStatus(getLoadManager().getNodeStatusWithReason(configNodeId));
+            info.setStatus(getLoadManager().getNodeStatus(configNodeId).getStatus());
+            info.setStatusReason(getLoadManager().getNodeStatusReason(configNodeId));
             info.setInternalAddress(configNodeLocation.getInternalEndPoint().getIp());
             info.setInternalPort(configNodeLocation.getInternalEndPoint().getPort());
             info.setRoleType(

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java

@@ -19,13 +19,18 @@
 package org.apache.iotdb.confignode.manager.load.cache;
 
 import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.cluster.NodeType;
 import org.apache.iotdb.confignode.manager.load.cache.node.ConfigNodeHeartbeatCache;
 import org.apache.iotdb.confignode.manager.load.cache.node.DataNodeHeartbeatCache;
 import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample;
+import org.apache.iotdb.confignode.manager.load.cache.node.NodeStatistics;
+import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatResp;
 
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.Map;
+
 public class NodeCacheTest {
 
   @Test
@@ -48,4 +53,65 @@ public void updateStatisticsTest() {
     Assert.assertEquals(NodeStatus.Running, configNodeHeartbeatCache.getNodeStatus());
     Assert.assertEquals(0, configNodeHeartbeatCache.getLoadScore());
   }
+
+  @Test
+  public void statusReasonPropagationTest() {
+    DataNodeHeartbeatCache dataNodeHeartbeatCache = new DataNodeHeartbeatCache(1);
+
+    // A heartbeat response carrying a status reason (e.g. ReadOnly + DiskFull from the DataNode)
+    // publishes the reason into the node statistics.
+    TDataNodeHeartbeatResp heartbeatResp =
+        new TDataNodeHeartbeatResp()
+            .setHeartbeatTimestamp(System.nanoTime())
+            .setStatus(NodeStatus.ReadOnly.getStatus())
+            .setStatusReason(NodeStatus.DISK_FULL);
+    dataNodeHeartbeatCache.cacheHeartbeatSample(new NodeHeartbeatSample(heartbeatResp));
+    dataNodeHeartbeatCache.updateCurrentStatistics(false);
+    Assert.assertEquals(NodeStatus.ReadOnly, dataNodeHeartbeatCache.getNodeStatus());
+    Assert.assertEquals(NodeStatus.DISK_FULL, dataNodeHeartbeatCache.getNodeStatusReason());
+
+    // A heartbeat response without a status reason clears the previous reason.
+    heartbeatResp =
+        new TDataNodeHeartbeatResp()
+            .setHeartbeatTimestamp(System.nanoTime())
+            .setStatus(NodeStatus.Running.getStatus());
+    dataNodeHeartbeatCache.cacheHeartbeatSample(new NodeHeartbeatSample(heartbeatResp));
+    dataNodeHeartbeatCache.updateCurrentStatistics(false);
+    Assert.assertEquals(NodeStatus.Running, dataNodeHeartbeatCache.getNodeStatus());
+    Assert.assertNull(dataNodeHeartbeatCache.getNodeStatusReason());
+
+    // An Unknown decided by the failure detector (a stale heartbeat) never carries a reason. The
+    // stale sample is accepted because the fresh cache has an empty sliding window.
+    DataNodeHeartbeatCache staleCache = new DataNodeHeartbeatCache(3);
+    staleCache.cacheHeartbeatSample(
+        new NodeHeartbeatSample(System.nanoTime() - 60_000_000_000L, NodeStatus.ReadOnly));
+    staleCache.updateCurrentStatistics(false);
+    Assert.assertEquals(NodeStatus.Unknown, staleCache.getNodeStatus());
+    Assert.assertNull(staleCache.getNodeStatusReason());
+  }
+
+  @Test
+  public void loadCacheSnapshotKeepsStatusAndReasonFromSameStatistics() {
+    LoadCache loadCache = new LoadCache();
+    loadCache.createNodeHeartbeatCache(NodeType.DataNode, 1);
+    loadCache.cacheDataNodeHeartbeatSample(
+        1,
+        new NodeHeartbeatSample(
+            new TDataNodeHeartbeatResp()
+                .setHeartbeatTimestamp(System.nanoTime())
+                .setStatus(NodeStatus.ReadOnly.getStatus())
+                .setStatusReason(NodeStatus.MANUAL)));
+    loadCache.updateNodeStatistics(false);
+
+    // The snapshot used by SHOW CLUSTER reads status and reason from one statistics object.
+    Map<Integer, NodeStatistics> snapshot = loadCache.getNodeStatisticsSnapshot();
+    Assert.assertEquals(1, snapshot.size());
+    Assert.assertEquals(NodeStatus.ReadOnly, snapshot.get(1).getStatus());
+    Assert.assertEquals(NodeStatus.MANUAL, snapshot.get(1).getStatusReason());
+    Assert.assertEquals(NodeStatus.MANUAL, loadCache.getNodeStatusReason(1));
+
+    // A missing cache yields no reason and Unknown status, without a fake merged string.
+    Assert.assertNull(loadCache.getNodeStatusReason(2));
+    Assert.assertEquals(NodeStatus.Unknown, loadCache.getNodeStatus(2));
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java

@@ -2581,8 +2581,7 @@ void sampleDiskLoad(TLoadSample loadSample) {
             RamUsageEstimator.humanReadableUnits((long) totalDisk),
             freeDiskRatio,
             commonConfig.getDiskSpaceWarningThreshold());
-        commonConfig.setNodeStatus(NodeStatus.ReadOnly);
-        commonConfig.setStatusReason(NodeStatus.DISK_FULL);
+        commonConfig.setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.DISK_FULL);
       } else if (NodeStatus.ReadOnly.equals(commonConfig.getNodeStatus())
           && NodeStatus.DISK_FULL.equals(commonConfig.getStatusReason())) {
         commonConfig.setNodeStatus(NodeStatus.Running);
@@ -2761,7 +2760,12 @@ public TShowAppliedConfigurationsResp showAppliedConfigurations() throws TExcept
   @Override
   public TSStatus setSystemStatus(String status) throws TException {
     try {
-      commonConfig.setNodeStatus(NodeStatus.parse(status));
+      NodeStatus nodeStatus = NodeStatus.parse(status);
+      if (nodeStatus == NodeStatus.ReadOnly) {
+        commonConfig.setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.MANUAL);
+      } else {
+        commonConfig.setNodeStatus(nodeStatus);
+      }
     } catch (Exception e) {
       return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage());
     }