Skip to content

Commit 96d55ed

Browse files
CaideyipiCaideyipi
authored
Improve IT cluster readiness diagnostics (#17903) (#17989)
* Improve IT cluster readiness diagnostics * Increase pipe IT cluster readiness retries (cherry picked from commit ef0d9f8)
1 parent bfb538a commit 96d55ed

3 files changed

Lines changed: 216 additions & 12 deletions

File tree

‎.github/workflows/pipe-it-2cluster.yml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ jobs:
7575
mvn clean verify \
7676
-P with-integration-tests \
7777
-DskipUTs \
78+
-DintegrationTest.clusterReadyRetryCount=90 \
7879
-DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \
7980
-DClusterConfigurations=${{ matrix.cluster }},${{ matrix.cluster }} \
8081
-pl integration-test \
@@ -161,6 +162,7 @@ jobs:
161162
mvn clean verify \
162163
-P with-integration-tests \
163164
-DskipUTs \
165+
-DintegrationTest.clusterReadyRetryCount=90 \
164166
-DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \
165167
-DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \
166168
-pl integration-test \
@@ -245,6 +247,7 @@ jobs:
245247
mvn clean verify \
246248
-P with-integration-tests \
247249
-DskipUTs \
250+
-DintegrationTest.clusterReadyRetryCount=90 \
248251
-DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \
249252
-DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \
250253
-pl integration-test \
@@ -329,6 +332,7 @@ jobs:
329332
mvn clean verify \
330333
-P with-integration-tests \
331334
-DskipUTs \
335+
-DintegrationTest.clusterReadyRetryCount=90 \
332336
-DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \
333337
-DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \
334338
-pl integration-test \
@@ -413,6 +417,7 @@ jobs:
413417
mvn clean verify \
414418
-P with-integration-tests \
415419
-DskipUTs \
420+
-DintegrationTest.clusterReadyRetryCount=90 \
416421
-DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \
417422
-DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }} \
418423
-pl integration-test \
@@ -498,6 +503,7 @@ jobs:
498503
mvn clean verify \
499504
-P with-integration-tests \
500505
-DskipUTs \
506+
-DintegrationTest.clusterReadyRetryCount=90 \
501507
-DintegrationTest.forkCount=1 -DConfigNodeMaxHeapSize=256 -DDataNodeMaxHeapSize=1024 -DDataNodeMaxDirectMemorySize=768 \
502508
-DClusterConfigurations=${{ matrix.cluster1 }},${{ matrix.cluster2 }},${{ matrix.cluster3 }} \
503509
-pl integration-test \

‎integration-test/src/main/java/org/apache/iotdb/it/env/MultiEnvFactory.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ public static BaseEnv getEnv(final int index) throws IndexOutOfBoundsException {
5151
/** Create several environments according to the specific number. */
5252
public static void createEnv(final int num) {
5353
// Not judge EnvType for individual test convenience
54+
envList.clear();
5455
final long startTime = System.currentTimeMillis();
5556
for (int i = 0; i < num; ++i) {
5657
try {

‎integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java‎

Lines changed: 209 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
2323
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
24+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
2425
import org.apache.iotdb.commons.client.ClientPoolFactory;
2526
import org.apache.iotdb.commons.client.IClientManager;
2627
import org.apache.iotdb.commons.client.exception.ClientManagerException;
@@ -79,6 +80,9 @@
7980

8081
public abstract class AbstractEnv implements BaseEnv {
8182
private static final Logger logger = IoTDBTestLogger.logger;
83+
private static final int DEFAULT_CLUSTER_READY_RETRY_COUNT = 30;
84+
private static final String CLUSTER_READY_RETRY_COUNT_PROPERTY =
85+
"integrationTest.clusterReadyRetryCount";
8286

8387
private final Random rand = new Random();
8488
protected List<ConfigNodeWrapper> configNodeWrapperList = Collections.emptyList();
@@ -87,7 +91,7 @@ public abstract class AbstractEnv implements BaseEnv {
8791
protected String testMethodName = null;
8892
protected int index = 0;
8993
protected long startTime;
90-
protected int retryCount = 30;
94+
protected int retryCount = getDefaultClusterReadyRetryCount();
9195
private IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager;
9296
private List<String> configNodeKillPoints = new ArrayList<>();
9397
private List<String> dataNodeKillPoints = new ArrayList<>();
@@ -110,6 +114,12 @@ protected AbstractEnv(final long startTime) {
110114
this.clusterConfig = new MppClusterConfig();
111115
}
112116

117+
private static int getDefaultClusterReadyRetryCount() {
118+
final int configuredRetryCount =
119+
Integer.getInteger(CLUSTER_READY_RETRY_COUNT_PROPERTY, DEFAULT_CLUSTER_READY_RETRY_COUNT);
120+
return configuredRetryCount > 0 ? configuredRetryCount : DEFAULT_CLUSTER_READY_RETRY_COUNT;
121+
}
122+
113123
@Override
114124
public ClusterConfig getConfig() {
115125
return clusterConfig;
@@ -331,12 +341,14 @@ private Map<String, Integer> countNodeStatus(final Map<Integer, String> nodeStat
331341
}
332342

333343
public void checkNodeInStatus(int nodeId, NodeStatus expectation) {
334-
checkClusterStatus(nodeStatusMap -> expectation.getStatus().equals(nodeStatusMap.get(nodeId)));
344+
checkClusterStatus(
345+
nodeStatusMap -> expectation.getStatus().equals(nodeStatusMap.get(nodeId)), m -> true);
335346
}
336347

337348
public void checkClusterStatusWithoutUnknown() {
338349
checkClusterStatus(
339-
nodeStatusMap -> nodeStatusMap.values().stream().noneMatch("Unknown"::equals));
350+
nodeStatusMap -> nodeStatusMap.values().stream().noneMatch("Unknown"::equals),
351+
processStatus -> processStatus.values().stream().noneMatch(i -> i != 0));
340352
testJDBCConnection();
341353
}
342354

@@ -346,6 +358,10 @@ public void checkClusterStatusOneUnknownOtherRunning() {
346358
Map<String, Integer> count = countNodeStatus(nodeStatus);
347359
return count.getOrDefault("Unknown", 0) == 1
348360
&& count.getOrDefault("Running", 0) == nodeStatus.size() - 1;
361+
},
362+
processStatus -> {
363+
long aliveProcessCount = processStatus.values().stream().filter(i -> i == 0).count();
364+
return aliveProcessCount == processStatus.size() - 1;
349365
});
350366
testJDBCConnection();
351367
}
@@ -357,38 +373,89 @@ public void checkClusterStatusOneUnknownOtherRunning() {
357373
* @param statusCheck the predicate to test the status of nodes
358374
*/
359375
public void checkClusterStatus(final Predicate<Map<Integer, String>> statusCheck) {
376+
checkClusterStatus(
377+
statusCheck, processStatus -> processStatus.values().stream().noneMatch(i -> i != 0));
378+
}
379+
380+
/**
381+
* check whether all nodes' status match the provided predicate with RPC. after retryCount times,
382+
* if the status of all nodes still not match the predicate, throw AssertionError.
383+
*
384+
* @param nodeStatusCheck the predicate to test the status of nodes
385+
* @param processStatusCheck the predicate to test the status of processes
386+
*/
387+
public void checkClusterStatus(
388+
final Predicate<Map<Integer, String>> nodeStatusCheck,
389+
final Predicate<Map<AbstractNodeWrapper, Integer>> processStatusCheck) {
360390
logger.info("Testing cluster environment...");
361391
TShowClusterResp showClusterResp;
362392
Exception lastException = null;
363-
boolean flag;
393+
boolean passed;
394+
boolean showClusterPassed = true;
395+
boolean nodeSizePassed = true;
396+
boolean nodeStatusPassed = true;
397+
boolean processStatusPassed = true;
398+
TSStatus showClusterStatus = null;
399+
int actualNodeSize = 0;
400+
Map<Integer, String> lastNodeStatus = null;
401+
Map<AbstractNodeWrapper, Integer> processStatusMap = new HashMap<>();
402+
364403
for (int i = 0; i < retryCount; i++) {
365404
try (final SyncConfigNodeIServiceClient client =
366405
(SyncConfigNodeIServiceClient) getLeaderConfigNodeConnection()) {
367-
flag = true;
406+
passed = true;
407+
showClusterPassed = true;
408+
nodeSizePassed = true;
409+
nodeStatusPassed = true;
410+
processStatusPassed = true;
411+
processStatusMap.clear();
412+
368413
showClusterResp = client.showCluster();
414+
showClusterStatus = showClusterResp.getStatus();
415+
actualNodeSize = showClusterResp.getNodeStatusSize();
416+
lastNodeStatus = showClusterResp.getNodeStatus();
369417

370418
// Check resp status
371419
if (showClusterResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
372-
flag = false;
420+
passed = false;
421+
showClusterPassed = false;
373422
}
374423

375424
// Check the number of nodes
376425
if (showClusterResp.getNodeStatus().size()
377426
!= configNodeWrapperList.size()
378427
+ dataNodeWrapperList.size()
379428
+ aiNodeWrapperList.size()) {
380-
flag = false;
429+
passed = false;
430+
nodeSizePassed = false;
381431
}
382432

383433
// Check the status of nodes
384-
if (flag) {
385-
flag = statusCheck.test(showClusterResp.getNodeStatus());
434+
if (passed) {
435+
passed = nodeStatusCheck.test(showClusterResp.getNodeStatus());
436+
if (!passed) {
437+
nodeStatusPassed = false;
438+
}
439+
}
440+
441+
collectProcessStatus(processStatusMap);
442+
processStatusPassed = processStatusCheck.test(processStatusMap);
443+
if (!processStatusPassed) {
444+
passed = false;
445+
handleProcessStatus(processStatusMap);
386446
}
387447

388-
if (flag) {
448+
if (passed) {
389449
logger.info("The cluster is now ready for testing!");
390450
return;
391451
}
452+
logger.info(
453+
"Retry {}: showClusterPassed={}, nodeSizePassed={}, nodeStatusPassed={}, processStatusPassed={}",
454+
i,
455+
showClusterPassed,
456+
nodeSizePassed,
457+
nodeStatusPassed,
458+
processStatusPassed);
392459
} catch (final Exception e) {
393460
lastException = e;
394461
}
@@ -405,8 +472,132 @@ public void checkClusterStatus(final Predicate<Map<Integer, String>> statusCheck
405472
lastException.getMessage(),
406473
lastException);
407474
}
475+
if (!showClusterPassed) {
476+
logger.error("Show cluster failed: {}", showClusterStatus);
477+
}
478+
if (!nodeSizePassed) {
479+
logger.error("Only {} nodes detected", actualNodeSize);
480+
}
481+
if (!nodeStatusPassed) {
482+
logger.error("Some node status incorrect: {}", lastNodeStatus);
483+
}
484+
if (!processStatusPassed) {
485+
logger.error("Some process status incorrect: {}", formatProcessStatus(processStatusMap));
486+
}
487+
488+
dumpTestJVMSnapshotQuietly("cluster status check failed");
408489
throw new AssertionError(
409-
String.format("After %d times retry, the cluster can't work!", retryCount));
490+
buildClusterStatusFailureMessage(
491+
showClusterPassed,
492+
nodeSizePassed,
493+
nodeStatusPassed,
494+
processStatusPassed,
495+
showClusterStatus,
496+
actualNodeSize,
497+
lastNodeStatus,
498+
processStatusMap,
499+
lastException));
500+
}
501+
502+
private void collectProcessStatus(final Map<AbstractNodeWrapper, Integer> processStatusMap)
503+
throws InterruptedException {
504+
final List<AbstractNodeWrapper> allNodeWrappers = new ArrayList<>();
505+
allNodeWrappers.addAll(configNodeWrapperList);
506+
allNodeWrappers.addAll(dataNodeWrapperList);
507+
allNodeWrappers.addAll(aiNodeWrapperList);
508+
for (final AbstractNodeWrapper nodeWrapper : allNodeWrappers) {
509+
final Process process = nodeWrapper.getInstance();
510+
if (process == null) {
511+
processStatusMap.put(nodeWrapper, -1);
512+
} else {
513+
processStatusMap.put(nodeWrapper, process.isAlive() ? 0 : process.waitFor());
514+
}
515+
}
516+
}
517+
518+
private String buildClusterStatusFailureMessage(
519+
final boolean showClusterPassed,
520+
final boolean nodeSizePassed,
521+
final boolean nodeStatusPassed,
522+
final boolean processStatusPassed,
523+
final TSStatus showClusterStatus,
524+
final int actualNodeSize,
525+
final Map<Integer, String> lastNodeStatus,
526+
final Map<AbstractNodeWrapper, Integer> processStatusMap,
527+
final Exception lastException) {
528+
final StringBuilder builder =
529+
new StringBuilder(
530+
String.format("After %d times retry, the cluster status check failed", retryCount));
531+
builder
532+
.append(": showClusterPassed=")
533+
.append(showClusterPassed)
534+
.append(", nodeSizePassed=")
535+
.append(nodeSizePassed)
536+
.append(", nodeStatusPassed=")
537+
.append(nodeStatusPassed)
538+
.append(", processStatusPassed=")
539+
.append(processStatusPassed)
540+
.append(", expectedNodeSize=")
541+
.append(
542+
configNodeWrapperList.size() + dataNodeWrapperList.size() + aiNodeWrapperList.size())
543+
.append(", actualNodeSize=")
544+
.append(actualNodeSize);
545+
if (showClusterStatus != null) {
546+
builder.append(", showClusterStatus=").append(showClusterStatus);
547+
}
548+
if (lastNodeStatus != null) {
549+
builder.append(", lastNodeStatus=").append(lastNodeStatus);
550+
}
551+
if (!processStatusMap.isEmpty()) {
552+
builder.append(", processStatus=").append(formatProcessStatus(processStatusMap));
553+
}
554+
if (lastException != null) {
555+
builder
556+
.append(", lastException=")
557+
.append(lastException.getClass().getName())
558+
.append(": ")
559+
.append(lastException.getMessage());
560+
}
561+
builder.append(", logDirs=").append(getClusterLogDirs());
562+
return builder.toString();
563+
}
564+
565+
private Map<String, Integer> formatProcessStatus(
566+
final Map<AbstractNodeWrapper, Integer> processStatusMap) {
567+
final Map<String, Integer> result = new LinkedHashMap<>();
568+
processStatusMap.forEach(
569+
(nodeWrapper, statusCode) -> result.put(nodeWrapper.getId(), statusCode));
570+
return result;
571+
}
572+
573+
private List<String> getClusterLogDirs() {
574+
final List<AbstractNodeWrapper> allNodeWrappers = new ArrayList<>();
575+
allNodeWrappers.addAll(configNodeWrapperList);
576+
allNodeWrappers.addAll(dataNodeWrapperList);
577+
allNodeWrappers.addAll(aiNodeWrapperList);
578+
return allNodeWrappers.stream()
579+
.map(AbstractNodeWrapper::getLogDirPath)
580+
.distinct()
581+
.collect(Collectors.toList());
582+
}
583+
584+
private void dumpTestJVMSnapshotQuietly(final String reason) {
585+
try {
586+
logger.info("Dumping test JVM snapshots because {}.", reason);
587+
dumpTestJVMSnapshot();
588+
} catch (final Exception e) {
589+
logger.warn("Failed to dump test JVM snapshots after {}", reason, e);
590+
}
591+
}
592+
593+
private void handleProcessStatus(final Map<AbstractNodeWrapper, Integer> processStatusMap) {
594+
for (final Map.Entry<AbstractNodeWrapper, Integer> entry : processStatusMap.entrySet()) {
595+
final Integer statusCode = entry.getValue();
596+
final AbstractNodeWrapper nodeWrapper = entry.getKey();
597+
if (statusCode != 0) {
598+
logger.info("Node {} is not running due to {}", nodeWrapper.getId(), statusCode);
599+
}
600+
}
410601
}
411602

412603
@Override
@@ -698,6 +889,7 @@ protected void testJDBCConnection() {
698889
.collect(Collectors.toList());
699890
final RequestDelegate<Void> testDelegate =
700891
new ParallelRequestDelegate<>(endpoints, NODE_START_TIMEOUT, this);
892+
final Map<String, String> lastConnectionFailures = Collections.synchronizedMap(new HashMap<>());
701893
for (final DataNodeWrapper dataNode : dataNodeWrapperList) {
702894
final String dataNodeEndpoint = dataNode.getIpAndPortString();
703895
testDelegate.addRequest(
@@ -716,6 +908,8 @@ protected void testJDBCConnection() {
716908
return null;
717909
} catch (final Exception e) {
718910
lastException = e;
911+
lastConnectionFailures.put(
912+
dataNodeEndpoint, e.getClass().getName() + ": " + e.getMessage());
719913
TimeUnit.SECONDS.sleep(1L);
720914
}
721915
}
@@ -729,8 +923,11 @@ protected void testJDBCConnection() {
729923
testDelegate.requestAll();
730924
} catch (final Exception e) {
731925
logger.error("exception in test Cluster with RPC, message: {}", e.getMessage(), e);
926+
dumpTestJVMSnapshotQuietly("JDBC connection check failed");
732927
throw new AssertionError(
733-
String.format("After %d times retry, the cluster can't work!", retryCount));
928+
String.format(
929+
"After %d times retry, JDBC connections to DataNodes are not ready. endpoints=%s, lastConnectionFailures=%s, logDirs=%s",
930+
retryCount, endpoints, lastConnectionFailures, getClusterLogDirs()));
734931
}
735932
}
736933

0 commit comments

Comments
 (0)