You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2022/04/26 01:52:33 UTC
[iotdb] branch IOTDB-3007 created (now 10e7e25828)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a change to branch IOTDB-3007
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 10e7e25828 [IOTDB-3007] Separate DataNode and IoTDB
This branch includes the following new commits:
new 10e7e25828 [IOTDB-3007] Separate DataNode and IoTDB
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: [IOTDB-3007] Separate DataNode and IoTDB
Posted by ha...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch IOTDB-3007
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 10e7e258281d388004120232c6c6bea9dcaa07c8
Author: HTHou <hh...@outlook.com>
AuthorDate: Tue Apr 26 09:52:15 2022 +0800
[IOTDB-3007] Separate DataNode and IoTDB
---
.../apache/iotdb/commons/conf/IoTDBConstant.java | 2 +
.../java/org/apache/iotdb/db/service/DataNode.java | 156 +++++++++++++++++++--
.../java/org/apache/iotdb/db/service/IoTDB.java | 8 +-
3 files changed, 154 insertions(+), 12 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..07bb394ff1 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,8 @@ 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";
+ public static final String GLOBAL_DATANODE_NAME = "DataNode";
+
// 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..c80503ee6c 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;
@@ -172,7 +195,26 @@ 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(
+ "{}: failed to start because some checks failed. ",
+ IoTDBConstant.GLOBAL_DATANODE_NAME,
+ e);
+ return;
+ }
+ try {
+ setUp();
+ } catch (StartupException | QueryProcessException e) {
+ logger.error("meet error while starting up.", e);
+ deactivate();
+ logger.error("{} exit", IoTDBConstant.GLOBAL_DATANODE_NAME);
+ return;
+ }
+ logger.info("{} has started.", IoTDBConstant.GLOBAL_DATANODE_NAME);
try {
// TODO: Start consensus layer in some where else
@@ -190,18 +232,116 @@ public class DataNode implements DataNodeMBean {
}
}
+ private void setUp() throws StartupException, QueryProcessException {
+ logger.info("Setting up 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(
+ "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("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, 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 DataNode...");
// stopThreadPools();
registerManager.deregisterAll();
JMXService.deregisterMBean(mbeanName);
- logger.info("Data node is deactivated.");
- // stop the iotdb kernel
- iotdb.stop();
+ logger.info("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