This is an automated email from the ASF dual-hosted git repository. haonan pushed a commit to branch standalone_write in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 13a6a3d2dda5f9dfe4f472cc4d95fc2333289810 Author: HTHou <[email protected]> AuthorDate: Sun Jun 5 00:17:14 2022 +0800 implement set storage group for new standalone IoTDB --- .../apache/iotdb/db/auth/AuthorizerManager.java | 14 ++ .../iotdb/db/localconfignode/LocalConfigNode.java | 11 - .../plan/execution/config/SetStorageGroupTask.java | 10 +- .../java/org/apache/iotdb/db/service/NewIoTDB.java | 249 +++++++++++++++++++++ .../org/apache/iotdb/db/service/NewIoTDBMBean.java | 27 +++ 5 files changed, 296 insertions(+), 15 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java b/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java index 057f9f9d57..867cac0135 100644 --- a/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java +++ b/server/src/main/java/org/apache/iotdb/db/auth/AuthorizerManager.java @@ -427,6 +427,20 @@ public class AuthorizerManager implements IAuthorizer { /** Check the user */ public TSStatus checkUser(String username, String password) throws ConfigNodeConnectionException { + // TODO:(Local AuthCheck) + if (!conf.getConfig().isClusterMode()) { + try { + if (login(username, password)) { + return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); + } else { + return RpcUtils.getStatus( + TSStatusCode.WRONG_LOGIN_PASSWORD_ERROR, "Authentication failed."); + } + } catch (AuthException e) { + return RpcUtils.getStatus( + TSStatusCode.WRONG_LOGIN_PASSWORD_ERROR, "Authentication failed."); + } + } authReadWriteLock.readLock().lock(); try { User user = userCache.getIfPresent(username); diff --git a/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java b/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java index a48887b687..1f1ef7afc4 100644 --- a/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java +++ b/server/src/main/java/org/apache/iotdb/db/localconfignode/LocalConfigNode.java @@ -217,17 +217,6 @@ public class LocalConfigNode { if (!config.isEnableMemControl()) { MemTableManager.getInstance().addOrDeleteStorageGroup(1); } - - if (config.isMppMode() && !config.isClusterMode()) { - for (DataRegionId dataRegionId : dataPartitionTable.setStorageGroup(storageGroup)) { - try { - storageEngine.createDataRegion(dataRegionId, storageGroup.getFullPath(), Long.MAX_VALUE); - } catch (DataRegionException e) { - // TODO (Fix exception type) - throw new MetadataException(e); - } - } - } } public void deleteStorageGroup(PartialPath storageGroup) throws MetadataException { diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java index a06c54f774..433b68d0f0 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/execution/config/SetStorageGroupTask.java @@ -22,7 +22,6 @@ package org.apache.iotdb.db.mpp.plan.execution.config; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.IClientManager; import org.apache.iotdb.commons.consensus.PartitionRegionId; -import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.confignode.rpc.thrift.TSetStorageGroupReq; import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema; import org.apache.iotdb.db.client.ConfigNodeClient; @@ -84,10 +83,13 @@ public class SetStorageGroupTask implements IConfigTask { try { LocalConfigNode localConfigNode = LocalConfigNode.getInstance(); localConfigNode.setStorageGroup(setStorageGroupStatement.getStorageGroupPath()); - localConfigNode.setTTL( - setStorageGroupStatement.getStorageGroupPath(), setStorageGroupStatement.getTTL()); + if (setStorageGroupStatement.getTTL() != null) { + localConfigNode.setTTL( + setStorageGroupStatement.getStorageGroupPath(), setStorageGroupStatement.getTTL()); + } // schemaReplicationFactor, dataReplicationFactor, timePartitionInterval are ignored - } catch (MetadataException | IOException e) { + } catch (Exception e) { + e.printStackTrace(); future.setException(e); } future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS)); diff --git a/server/src/main/java/org/apache/iotdb/db/service/NewIoTDB.java b/server/src/main/java/org/apache/iotdb/db/service/NewIoTDB.java new file mode 100644 index 0000000000..44da688233 --- /dev/null +++ b/server/src/main/java/org/apache/iotdb/db/service/NewIoTDB.java @@ -0,0 +1,249 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.service; + +import org.apache.iotdb.commons.concurrent.IoTDBDefaultThreadExceptionHandler; +import org.apache.iotdb.commons.conf.IoTDBConstant; +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.commons.udf.service.UDFClassLoaderManager; +import org.apache.iotdb.commons.udf.service.UDFRegistrationService; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.conf.IoTDBStartCheck; +import org.apache.iotdb.db.conf.rest.IoTDBRestServiceCheck; +import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor; +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.localconfignode.LocalConfigNode; +import org.apache.iotdb.db.metadata.LocalSchemaProcessor; +import org.apache.iotdb.db.mpp.execution.datatransfer.DataBlockService; +import org.apache.iotdb.db.mpp.execution.schedule.DriverScheduler; +import org.apache.iotdb.db.protocol.rest.RestService; +import org.apache.iotdb.db.rescon.PrimitiveArrayManager; +import org.apache.iotdb.db.rescon.SystemInfo; +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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; + +public class NewIoTDB implements NewIoTDBMBean { + + private static final Logger logger = LoggerFactory.getLogger(NewIoTDB.class); + private final String mbeanName = + String.format("%s:%s=%s", IoTDBConstant.IOTDB_PACKAGE, IoTDBConstant.JMX_TYPE, "IoTDB"); + private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + private static final RegisterManager registerManager = new RegisterManager(); + public static LocalSchemaProcessor schemaProcessor = LocalSchemaProcessor.getInstance(); + public static LocalConfigNode configManager = LocalConfigNode.getInstance(); + + public static NewIoTDB getInstance() { + return IoTDBHolder.INSTANCE; + } + + public static void main(String[] args) { + try { + IoTDBStartCheck.getInstance().checkConfig(); + IoTDBRestServiceCheck.getInstance().checkConfig(); + } catch (ConfigurationException | IOException e) { + logger.error("meet error when doing start checking", e); + System.exit(1); + } + NewIoTDB daemon = NewIoTDB.getInstance(); + IoTDBDescriptor.getInstance().getConfig().setMppMode(true); + daemon.active(); + } + + public void 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_DB_NAME, e); + return; + } + + // set recover config, avoid creating deleted time series when recovering wal + boolean prevIsAutoCreateSchemaEnabled = config.isAutoCreateSchemaEnabled(); + config.setAutoCreateSchemaEnabled(false); + boolean prevIsEnablePartialInsert = config.isEnablePartialInsert(); + config.setEnablePartialInsert(true); + // Set datanodeId to 0 for standalone IoTDB + config.setDataNodeId(0); + + try { + setUp(); + } catch (StartupException | QueryProcessException e) { + logger.error("meet error while starting up.", e); + deactivate(); + logger.error("{} exit", IoTDBConstant.GLOBAL_DB_NAME); + return; + } + + // reset config + config.setAutoCreateSchemaEnabled(prevIsAutoCreateSchemaEnabled); + config.setEnablePartialInsert(prevIsEnablePartialInsert); + + logger.info("{} has started.", IoTDBConstant.GLOBAL_DB_NAME); + } + + private void setUp() throws StartupException, QueryProcessException { + logger.info("Setting up IoTDB..."); + + Runtime.getRuntime().addShutdownHook(new IoTDBShutdownHook()); + setUncaughtExceptionHandler(); + // init rpc service + IoTDBDescriptor.getInstance() + .getConfig() + .setRpcImplClassName(DataNodeTSIServiceImpl.class.getName()); + // in cluster mode, RPC service is not enabled. + if (IoTDBDescriptor.getInstance().getConfig().isEnableRpcService()) { + registerManager.register(RPCService.getInstance()); + } + 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(SenderService.getInstance()); + registerManager.register(WALManager.getInstance()); + + registerManager.register(StorageEngineV2.getInstance()); + registerManager.register(DataBlockService.getInstance()); + registerManager.register(InternalService.getInstance()); + registerManager.register(DriverScheduler.getInstance()); + + registerManager.register(TemporaryQueryDataFileService.getInstance()); + registerManager.register( + UDFClassLoaderManager.setupAndGetInstance( + IoTDBDescriptor.getInstance().getConfig().getUdfDir())); + registerManager.register( + UDFRegistrationService.setupAndGetInstance( + IoTDBDescriptor.getInstance().getConfig().getSystemDir() + + File.separator + + "udf" + + File.separator)); + registerManager.register(ReceiverService.getInstance()); + + initProtocols(); + + logger.info( + "IoTDB 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 failed to set up.", e); + Thread.currentThread().interrupt(); + return; + } + } + + 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. + if (!IoTDBDescriptor.getInstance().getConfig().isMppMode()) { + registerManager.register(SettleService.getINSTANCE()); + } + registerManager.register(TriggerRegistrationService.getInstance()); + registerManager.register(ContinuousQueryService.getInstance()); + + // start reporter + MetricsService.getInstance().startAllReporter(); + + logger.info("Congratulation, IoTDB is set up successfully. Now, enjoy yourself!"); + } + + 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 IoTDB..."); + registerManager.deregisterAll(); + JMXService.deregisterMBean(mbeanName); + logger.info("IoTDB is deactivated."); + } + + 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()); + } + + @Override + public void stop() { + deactivate(); + } + + public void shutdown() throws Exception { + // TODO shutdown is not equal to stop() + logger.info("Deactivating IoTDB..."); + registerManager.shutdownAll(); + PrimitiveArrayManager.close(); + SystemInfo.getInstance().close(); + JMXService.deregisterMBean(mbeanName); + logger.info("IoTDB is deactivated."); + } + + private void setUncaughtExceptionHandler() { + Thread.setDefaultUncaughtExceptionHandler(new IoTDBDefaultThreadExceptionHandler()); + } + + private static class IoTDBHolder { + + private static final NewIoTDB INSTANCE = new NewIoTDB(); + + private IoTDBHolder() {} + } +} diff --git a/server/src/main/java/org/apache/iotdb/db/service/NewIoTDBMBean.java b/server/src/main/java/org/apache/iotdb/db/service/NewIoTDBMBean.java new file mode 100644 index 0000000000..a2e85a77ff --- /dev/null +++ b/server/src/main/java/org/apache/iotdb/db/service/NewIoTDBMBean.java @@ -0,0 +1,27 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.service; + +import org.apache.iotdb.commons.exception.ShutdownException; + +@FunctionalInterface +public interface NewIoTDBMBean { + + void stop() throws ShutdownException; +}
