This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 3c23ae3f39 [IOTDB-3007] Separate DataNode and IoTDB (#5668)
3c23ae3f39 is described below
commit 3c23ae3f39a4cc9754d1a972309e102376cff2df
Author: Haonan <[email protected]>
AuthorDate: Tue Apr 26 11:43:14 2022 +0800
[IOTDB-3007] Separate DataNode and IoTDB (#5668)
---
.../apache/iotdb/commons/conf/IoTDBConstant.java | 1 +
.../java/org/apache/iotdb/db/service/DataNode.java | 155 +++++++++++++++++++--
.../java/org/apache/iotdb/db/service/IoTDB.java | 8 +-
3 files changed, 151 insertions(+), 13 deletions(-)
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
b/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
index f6bad59128..d5c722e8dc 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
@@ -25,6 +25,7 @@ public class IoTDBConstant {
public static final String ENV_FILE_NAME = "iotdb-env";
public static final String IOTDB_CONF = "IOTDB_CONF";
public static final String GLOBAL_DB_NAME = "IoTDB";
+
// when running the program in IDE, we can not get the version info using
// getImplementationVersion()
public static final String VERSION =
diff --git a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
index bedd697089..0c474d6334 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -20,19 +20,43 @@ package org.apache.iotdb.db.service;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.concurrent.IoTDBDefaultThreadExceptionHandler;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.BadNodeUrlException;
import org.apache.iotdb.commons.exception.ConfigurationException;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.service.JMXService;
import org.apache.iotdb.commons.service.RegisterManager;
+import org.apache.iotdb.commons.service.StartupChecks;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp;
import org.apache.iotdb.db.client.ConfigNodeClient;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBConfigCheck;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor;
import org.apache.iotdb.db.consensus.ConsensusImpl;
+import org.apache.iotdb.db.engine.StorageEngineV2;
+import org.apache.iotdb.db.engine.cache.CacheHitRatioMonitor;
+import org.apache.iotdb.db.engine.compaction.CompactionTaskManager;
+import org.apache.iotdb.db.engine.cq.ContinuousQueryService;
+import org.apache.iotdb.db.engine.flush.FlushManager;
+import org.apache.iotdb.db.engine.trigger.service.TriggerRegistrationService;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.mpp.buffer.DataBlockService;
+import org.apache.iotdb.db.mpp.schedule.FragmentInstanceScheduler;
+import org.apache.iotdb.db.protocol.influxdb.meta.InfluxDBMetaManager;
+import org.apache.iotdb.db.protocol.rest.RestService;
+import org.apache.iotdb.db.query.udf.service.TemporaryQueryDataFileService;
+import org.apache.iotdb.db.query.udf.service.UDFClassLoaderManager;
+import org.apache.iotdb.db.query.udf.service.UDFRegistrationService;
+import org.apache.iotdb.db.service.basic.ServiceProvider;
+import org.apache.iotdb.db.service.basic.StandaloneServiceProvider;
+import org.apache.iotdb.db.service.metrics.MetricsService;
+import org.apache.iotdb.db.service.thrift.impl.DataNodeTSIServiceImpl;
+import org.apache.iotdb.db.sync.receiver.ReceiverService;
+import org.apache.iotdb.db.sync.sender.service.SenderService;
+import org.apache.iotdb.db.wal.WALManager;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -62,9 +86,8 @@ public class DataNode implements DataNodeMBean {
// we do not init anything here, so that we can re-initialize the instance
in IT.
}
- private final IoTDB iotdb = IoTDB.getInstance();
-
- private final RegisterManager registerManager = new RegisterManager();
+ private static final RegisterManager registerManager = new RegisterManager();
+ public static ServiceProvider serviceProvider;
// private IClientManager clientManager;
@@ -96,7 +119,7 @@ public class DataNode implements DataNodeMBean {
joinCluster();
active();
} catch (StartupException e) {
- logger.error("Fail to start server", e);
+ logger.error("Fail to start server", e);
stop();
}
}
@@ -172,7 +195,23 @@ public class DataNode implements DataNodeMBean {
IoTDBDescriptor.getInstance().getConfig().setMppMode(true);
IoTDBDescriptor.getInstance().getConfig().setClusterMode(true);
// start iotdb server first
- IoTDB.getInstance().active();
+ StartupChecks checks = new StartupChecks().withDefaultTest();
+ try {
+ checks.verify();
+ } catch (StartupException e) {
+ // TODO: what are some checks
+ logger.error("IoTDB DataNode: failed to start because some checks
failed. ", e);
+ return;
+ }
+ try {
+ setUp();
+ } catch (StartupException | QueryProcessException e) {
+ logger.error("meet error while starting up.", e);
+ deactivate();
+ logger.error("IoTDB DataNode exit");
+ return;
+ }
+ logger.info("IoTDB DataNode has started.");
try {
// TODO: Start consensus layer in some where else
@@ -190,18 +229,116 @@ public class DataNode implements DataNodeMBean {
}
}
+ private void setUp() throws StartupException, QueryProcessException {
+ logger.info("Setting up IoTDB DataNode...");
+
+ Runtime.getRuntime().addShutdownHook(new IoTDBShutdownHook());
+ setUncaughtExceptionHandler();
+ initServiceProvider();
+ registerManager.register(MetricsService.getInstance());
+ logger.info("recover the schema...");
+ initConfigManager();
+ registerManager.register(new JMXService());
+ registerManager.register(FlushManager.getInstance());
+ registerManager.register(CacheHitRatioMonitor.getInstance());
+ registerManager.register(CompactionTaskManager.getInstance());
+ JMXService.registerMBean(getInstance(), mbeanName);
+ registerManager.register(WALManager.getInstance());
+
+ // in mpp mode we need to start some other services
+ registerManager.register(StorageEngineV2.getInstance());
+ registerManager.register(DataBlockService.getInstance());
+ registerManager.register(InternalService.getInstance());
+ registerManager.register(FragmentInstanceScheduler.getInstance());
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setRpcImplClassName(DataNodeTSIServiceImpl.class.getName());
+
+ registerManager.register(TemporaryQueryDataFileService.getInstance());
+ registerManager.register(UDFClassLoaderManager.getInstance());
+ registerManager.register(UDFRegistrationService.getInstance());
+ registerManager.register(ReceiverService.getInstance());
+ registerManager.register(MetricsService.getInstance());
+
+ // in cluster mode, RPC service is not enabled.
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableRpcService()) {
+ registerManager.register(RPCService.getInstance());
+ }
+
+ initProtocols();
+
+ if
(IoTDBDescriptor.getInstance().getConfig().isEnableInfluxDBRpcService()) {
+ InfluxDBMetaManager.getInstance().recover();
+ }
+
+ logger.info(
+ "IoTDB DataNode is setting up, some storage groups may not be ready
now, please wait several seconds...");
+
+ while (!StorageEngineV2.getInstance().isAllSgReady()) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ logger.warn("IoTDB DataNode failed to set up.", e);
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+
+ registerManager.register(SenderService.getInstance());
+ registerManager.register(UpgradeSevice.getINSTANCE());
+ // in mpp mode we temporarily don't start settle service because it uses
StorageEngine directly
+ // in itself, but currently we need to use StorageEngineV2 instead of
StorageEngine in mpp mode.
+ // registerManager.register(SettleService.getINSTANCE());
+ registerManager.register(TriggerRegistrationService.getInstance());
+ registerManager.register(ContinuousQueryService.getInstance());
+
+ // start reporter
+ MetricsService.getInstance().startAllReporter();
+
+ logger.info("Congratulation, IoTDB DataNode is set up successfully. Now,
enjoy yourself!");
+ }
+
+ private void initConfigManager() {
+ long time = System.currentTimeMillis();
+ IoTDB.configManager.init();
+ long end = System.currentTimeMillis() - time;
+ logger.info("spend {}ms to recover schema.", end);
+ logger.info(
+ "After initializing, sequence tsFile threshold is {}, unsequence
tsFile threshold is {}",
+ IoTDBDescriptor.getInstance().getConfig().getSeqTsFileSize(),
+ IoTDBDescriptor.getInstance().getConfig().getUnSeqTsFileSize());
+ }
+
public void stop() {
deactivate();
}
+ private void initServiceProvider() throws QueryProcessException {
+ serviceProvider = new StandaloneServiceProvider();
+ }
+
+ public static void initProtocols() throws StartupException {
+ if
(IoTDBDescriptor.getInstance().getConfig().isEnableInfluxDBRpcService()) {
+ registerManager.register(InfluxDBRPCService.getInstance());
+ }
+ if (IoTDBDescriptor.getInstance().getConfig().isEnableMQTTService()) {
+ registerManager.register(MQTTService.getInstance());
+ }
+ if
(IoTDBRestServiceDescriptor.getInstance().getConfig().isEnableRestService()) {
+ registerManager.register(RestService.getInstance());
+ }
+ }
+
private void deactivate() {
- logger.info("Deactivating data node...");
+ logger.info("Deactivating IoTDB DataNode...");
// stopThreadPools();
registerManager.deregisterAll();
JMXService.deregisterMBean(mbeanName);
- logger.info("Data node is deactivated.");
- // stop the iotdb kernel
- iotdb.stop();
+ logger.info("IoTDB DataNode is deactivated.");
+ }
+
+ private void setUncaughtExceptionHandler() {
+ Thread.setDefaultUncaughtExceptionHandler(new
IoTDBDefaultThreadExceptionHandler());
}
private static class DataNodeHolder {
diff --git a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
index b0324c3a1a..f9e1228893 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/IoTDB.java
@@ -173,7 +173,8 @@ public class IoTDB implements IoTDBMBean {
initInfluxDBMManager();
}
- logger.info("IoTDB is set up, now may some sgs are not ready, please wait
several seconds...");
+ logger.info(
+ "IoTDB is setting up, some storage groups may not be ready now, please
wait several seconds...");
while (IoTDBDescriptor.getInstance().getConfig().isMppMode()
? !StorageEngineV2.getInstance().isAllSgReady()
@@ -238,10 +239,9 @@ public class IoTDB implements IoTDBMBean {
long end = System.currentTimeMillis() - time;
logger.info("spend {}ms to recover schema.", end);
logger.info(
- "After initializing, sequence tsFile threshold is {}, unsequence
tsFile threshold is {}, memtableSize is {}",
+ "After initializing, sequence tsFile threshold is {}, unsequence
tsFile threshold is {}",
IoTDBDescriptor.getInstance().getConfig().getSeqTsFileSize(),
- IoTDBDescriptor.getInstance().getConfig().getUnSeqTsFileSize(),
- IoTDBDescriptor.getInstance().getConfig().getMemtableSizeThreshold());
+ IoTDBDescriptor.getInstance().getConfig().getUnSeqTsFileSize());
}
@Override
