You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "Wei-hao-Li (via GitHub)" <gi...@apache.org> on 2026/01/12 01:10:30 UTC
[PR] Implement ExternalService Framework [iotdb]
Wei-hao-Li opened a new pull request, #17009:
URL: https://github.com/apache/iotdb/pull/17009
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement ExternalService Framework [iotdb]
Posted by "Wei-hao-Li (via GitHub)" <gi...@apache.org>.
Wei-hao-Li merged PR #17009:
URL: https://github.com/apache/iotdb/pull/17009
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement ExternalService Framework [iotdb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17009:
URL: https://github.com/apache/iotdb/pull/17009#discussion_r2688620076
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ TSStatus status = new TSStatus(TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode());
+ status.setMessage(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+
+ // 2. persist on CN
+ try (ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ TSStatus status = new TSStatus(TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode());
+ status.setMessage(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() == null) {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ serviceInfo.getServiceInstance().start();
+ }
+
+ // 3. persist on CN, rollback if failed
+ try (ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
Review Comment:
The `createExternalServiceInstance` method uses `serviceName` as the class name when calling `Class.forName`, but it should use the `className` from `ServiceInfo`. Change parameter or retrieve className from serviceInfo.
##########
iotdb-protocol/thrift-commons/src/main/thrift/common.thrift:
##########
@@ -261,6 +261,20 @@ struct TNodeLocations {
2: optional list<TDataNodeLocation> dataNodeLocations
}
+struct TExternalServiceListResp {
+ 1: required TSStatus status
+ 2: required list<TExternalServiceEntry> externalServiceInfos
+}
+
+
+struct TExternalServiceEntry {
+ 1: required string serviceName
+ 2: required string className
+ 3: required byte state
+ 4: required i32 dataNodId
Review Comment:
Corrected spelling of 'dataNodId' to 'dataNodeId'.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final byte SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ private static final String SERVICE_NOT_EXISTED =
+ "ExternalService %s is not existed on DataNode %s.";
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message = String.format(SERVICE_NOT_EXISTED, serviceName, plan.getDatanodeId());
Review Comment:
The error message for adding an existing service incorrectly uses `SERVICE_NOT_EXISTED`. It should say 'ExternalService %s already exists on DataNode %s.' instead.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/externalservice/ShowExternalServiceTask.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.queryengine.plan.execution.config.metadata.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.tsfile.block.column.ColumnBuilder;
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.utils.Binary;
+
+public class ShowExternalServiceTask implements IConfigTask {
+ private final int dataNodeId;
+
+ public ShowExternalServiceTask(int dataNodeId) {
+ this.dataNodeId = dataNodeId;
+ }
+
+ @Override
+ public ListenableFuture<ConfigTaskResult> execute(IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.showExternalService(dataNodeId);
+ }
+
+ public static void appendServiceEntry(
+ TExternalServiceEntry externalServiceEntry, ColumnBuilder[] columnBuilders) {
+ columnBuilders[0].writeBinary(
+ new Binary(externalServiceEntry.getServiceName(), TSFileConfig.STRING_CHARSET));
+ columnBuilders[1].writeInt(externalServiceEntry.getDataNodId());
Review Comment:
Corrected method name from 'getDataNodId' to 'getDataNodeId'.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/response/externalservice/ShowExternalServiceResp.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.confignode.consensus.response.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.consensus.common.DataSet;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import javax.validation.constraints.NotNull;
+
+import java.util.Comparator;
+import java.util.List;
+
+public class ShowExternalServiceResp implements DataSet {
+
+ private final List<TExternalServiceEntry> serviceInfoEntryList;
+
+ public ShowExternalServiceResp(@NotNull List<TExternalServiceEntry> serviceInfoEntryList) {
+ this.serviceInfoEntryList = serviceInfoEntryList;
+ }
+
+ public List<TExternalServiceEntry> getServiceInfoEntryList() {
+ return serviceInfoEntryList;
+ }
+
+ public TExternalServiceListResp convertToRpcShowExternalServiceResp() {
+ serviceInfoEntryList.sort(
+ Comparator.comparingInt(TExternalServiceEntry::getDataNodId)
Review Comment:
Corrected method name from 'getDataNodId' to 'getDataNodeId'.
```suggestion
Comparator.comparingInt(TExternalServiceEntry::getDataNodeId)
```
##########
integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/informationschema/IoTDBServicesIT.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.relational.it.query.recent.informationschema;
+
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.apache.iotdb.commons.schema.table.InformationSchema.INFORMATION_DATABASE;
+import static org.apache.iotdb.db.it.utils.TestUtils.createUser;
+import static org.apache.iotdb.db.it.utils.TestUtils.resultSetEqualTest;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest;
+import static org.apache.iotdb.itbase.env.BaseEnv.TABLE_SQL_DIALECT;
+import static org.apache.iotdb.itbase.env.BaseEnv.TREE_SQL_DIALECT;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBServicesIT {
+ private static final String ADMIN_NAME =
+ CommonDescriptor.getInstance().getConfig().getDefaultAdminName();
+ private static final String ADMIN_PWD =
+ CommonDescriptor.getInstance().getConfig().getAdminPassword();
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initClusterEnvironment();
+ createUser("test", "TimechoDB@2021");
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void testQueryResult() {
+ String[] retArray =
+ new String[] {
+ "MQTT,1,STOPPED,org.apache.iotdb.externalservice.Mqtt,built-in,",
+ "REST,1,STOPPED,org.apache.iotdb.externalservice.Rest,built-in,",
+ };
+
+ // TableModel
+ String[] header =
+ new String[] {"service_name", "datanode_id", "state", "class_name", "service_type"};
+
+ String sql = "SELECT * FROM services where datanode_id = 1";
+ tableResultSetEqualTest(sql, header, retArray, INFORMATION_DATABASE);
+ sql = "show services on 1";
+ tableResultSetEqualTest(sql, header, retArray, INFORMATION_DATABASE);
+
+ // TreeModel
+ header = new String[] {"ServiceName", "DataNodeId", "State", "ClassName", "ServiceType"};
+
+ resultSetEqualTest(sql, header, retArray);
+ }
+
+ @Test
+ public void testPrivilege() {
+ testTargetModelPrivilege(TABLE_SQL_DIALECT);
+ testTargetModelPrivilege(TREE_SQL_DIALECT);
+ }
+
+ private void testTargetModelPrivilege(String model) {
+ String sql = "show services";
+ try (Connection connection =
+ EnvFactory.getEnv().getConnection("test", "TimechoDB@2021", model);
+ Statement statement = connection.createStatement()) {
+ statement.executeQuery(sql);
+ } catch (SQLException e) {
+ Assert.assertTrue(
+ e.getMessage()
+ .contains("No permissions for this operation, please add privilege SYSTEM"));
+ }
+
+ try (Connection connection2 = EnvFactory.getEnv().getConnection(ADMIN_NAME, ADMIN_PWD, model)) {
+ ResultSet resultSet = connection2.createStatement().executeQuery(sql);
+ resultSet.close();
Review Comment:
This Statement is not always closed on method exit.
```suggestion
try (Connection connection2 =
EnvFactory.getEnv().getConnection(ADMIN_NAME, ADMIN_PWD, model);
Statement statement = connection2.createStatement();
ResultSet resultSet = statement.executeQuery(sql)) {
// Intentionally no-op: executing the query is sufficient to validate privilege
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ TSStatus status = new TSStatus(TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode());
+ status.setMessage(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+
+ // 2. persist on CN
+ try (ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ TSStatus status = new TSStatus(TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode());
+ status.setMessage(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() == null) {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ serviceInfo.getServiceInstance().start();
+ }
+
+ // 3. persist on CN, rollback if failed
+ try (ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ TSStatus status =
+ new TSStatus(TSStatusCode.EXTERNAL_SERVICE_INSTANCE_CREATE_ERROR.getStatusCode());
+ status.setMessage(
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e));
+ LOGGER.warn(status.getMessage(), e);
+ throw new ExternalServiceManagementException(status);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ TSStatus status = new TSStatus(TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode());
+ status.setMessage(
+ String.format(
+ "Failed to stop External Service %s, because it is not existed!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+
+ // 2. call stop method of ServiceInstance
+ if (serviceInfo.getState() == STOPPED) {
+ return;
+ } else {
+ // The state is RUNNING
+ stopService(serviceInfo);
+ }
+
+ // 3. persist on CN, rollback if failed
+ try (ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID); ) {
+ TSStatus status = client.stopExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().start();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(STOPPED);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private void stopService(ServiceInfo serviceInfo) {
+ checkState(
+ serviceInfo.getServiceInstance() != null,
+ "External Service instance is null when state is RUNNING!",
+ serviceInfo.getServiceName());
+ serviceInfo.getServiceInstance().stop();
+ }
+
+ public void dropService(String serviceName, boolean forcedly)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ TSStatus status = new TSStatus(TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode());
+ status.setMessage(
+ String.format(
+ "Failed to drop External Service %s, because it is not existed!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+ if (serviceInfo.getServiceType() == ServiceInfo.ServiceType.BUILTIN) {
+ TSStatus status =
+ new TSStatus(TSStatusCode.CANNOT_DROP_BUILTIN_EXTERNAL_SERVICE.getStatusCode());
+ status.setMessage(
+ String.format(
+ "Failed to drop External Service %s, because it is BUILT-IN!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+
+ // 2. stop or fail when service are not stopped
+ if (serviceInfo.getState() == STOPPED) {
+ // do nothing
+ } else {
+ // The state is RUNNING
+ if (forcedly) {
+ try {
+ stopService(serviceInfo);
+ } catch (Exception e) {
+ // record errMsg if exception occurs during the stop of service
+ LOGGER.warn(
+ "Failed to stop External Service %s because %s. It will be drop forcedly",
+ serviceName, e.getMessage());
+ }
+ } else {
+ TSStatus status =
+ new TSStatus(TSStatusCode.CANNOT_DROP_RUNNING_EXTERNAL_SERVICE.getStatusCode());
+ status.setMessage(
+ String.format(
+ "Failed to drop External Service %s, because it is RUNNING!", serviceName));
+ throw new ExternalServiceManagementException(status);
+ }
+ }
+
+ // 3. persist on CN
+ try (ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ TSStatus status = client.dropExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+ }
+
+ // 4. modify memory info
+ serviceInfos.remove(serviceName);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void restoreRunningServiceInstance() {
+ // Needn't use lock here, we use this method when active DN and there is no concurrent risk
+ serviceInfos
+ .values()
+ .forEach(
+ serviceInfo -> {
+ // start services with RUNNING state
+ if (serviceInfo.getState() == RUNNING) {
+ IExternalService serviceInstance = serviceInfo.getServiceInstance();
+ serviceInfo.setServiceInstance(serviceInstance);
+ serviceInstance.start();
Review Comment:
Line 308 retrieves the serviceInstance which is potentially null, then line 309 sets it back to the same null value. This should call `createExternalServiceInstance` to create a new instance if null, similar to the pattern in `startService` method.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/metadata/externalservice/StartExternalServiceTask.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.queryengine.plan.execution.config.metadata.externalservice;
+
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.Locale;
+
+public class StartExternalServiceTask implements IConfigTask {
+
+ private final String serviceName;
+
+ public StartExternalServiceTask(String serviceName) {
+ this.serviceName = serviceName.toUpperCase(Locale.ENGLISH);
+ ;
Review Comment:
Remove the extra semicolon on line 36.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,371 @@
+/*
+ * 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.exernalservice;
Review Comment:
Corrected package name from 'exernalservice' to 'externalservice'.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java:
##########
@@ -1093,6 +1097,36 @@ public Statement visitShowTriggers(IoTDBSqlParser.ShowTriggersContext ctx) {
return new ShowTriggersStatement();
}
+ @Override
+ public Statement visitCreateService(IoTDBSqlParser.CreateServiceContext ctx) {
+ throw new UnsupportedOperationException(SERVICE_MANAGEMENT_NOT_SUPPORTED);
+ }
+
+ @Override
+ public Statement visitStartService(IoTDBSqlParser.StartServiceContext ctx) {
+ throw new UnsupportedOperationException(SERVICE_MANAGEMENT_NOT_SUPPORTED);
+ }
+
+ @Override
+ public Statement visitStopService(IoTDBSqlParser.StopServiceContext ctx) {
+ throw new UnsupportedOperationException(SERVICE_MANAGEMENT_NOT_SUPPORTED);
+ }
+
+ @Override
+ public Statement visitDropService(IoTDBSqlParser.DropServiceContext ctx) {
+ throw new UnsupportedOperationException(SERVICE_MANAGEMENT_NOT_SUPPORTED);
+ }
+
+ @Override
+ public Statement visitShowService(IoTDBSqlParser.ShowServiceContext ctx) {
+ // show services on all DNs
+ int dataNodeId = -1;
+ if (ctx.ON() != null) {
+ dataNodeId = Integer.parseInt(ctx.targetDataNodeId.getText());
Review Comment:
Potential uncaught 'java.lang.NumberFormatException'.
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/exernalservice/ShowExternalServicePlan.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.confignode.consensus.request.read.exernalservice;
Review Comment:
Corrected package name from 'exernalservice' to 'externalservice'.
```suggestion
package org.apache.iotdb.confignode.consensus.request.read.externalservice;
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,309 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final byte SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ private static final String SERVICE_NOT_EXISTED =
+ "ExternalService %s is not existed on DataNode %s.";
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message = String.format(SERVICE_NOT_EXISTED, serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ ServiceInfo removed = serviceInfos.remove(serviceName);
+ if (removed == null) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message = String.format(SERVICE_NOT_EXISTED, serviceName, plan.getDataNodeId());
+ } else {
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Start the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus startService(StartExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message = String.format(SERVICE_NOT_EXISTED, serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfo.setState(ServiceInfo.State.RUNNING);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Stop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus stopService(StopExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message = String.format(SERVICE_NOT_EXISTED, serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfo.setState(ServiceInfo.State.STOPPED);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ public ShowExternalServiceResp showService(Set<Integer> dataNodes) {
+ return new ShowExternalServiceResp(
+ datanodeToServiceInfos.entrySet().stream()
+ .filter(entry -> dataNodes.contains(entry.getKey()))
+ .flatMap(
+ entry ->
+ entry.getValue().values().stream()
+ .map(
+ serviceInfo ->
+ new TExternalServiceEntry(
+ serviceInfo.getServiceName(),
+ serviceInfo.getClassName(),
+ serviceInfo.getState().getValue(),
+ entry.getKey(),
+ ServiceInfo.ServiceType.USER_DEFINED.getValue())))
+ .collect(Collectors.toList()));
+ }
+
+ private void serializeInfos(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(SERIALIZATION_VERSION, outputStream);
+ ReadWriteIOUtils.write(datanodeToServiceInfos.size(), outputStream);
+ for (Map.Entry<Integer, Map<String, ServiceInfo>> outerEntry :
+ datanodeToServiceInfos.entrySet()) {
+ ReadWriteIOUtils.write(outerEntry.getKey(), outputStream); // DataNode ID
+
+ Map<String, ServiceInfo> innerMap = outerEntry.getValue();
+ // inner Map
+ ReadWriteIOUtils.write(innerMap.size(), outputStream);
+ for (ServiceInfo innerEntry : innerMap.values()) {
+ serializeServiceInfoWithCRC(innerEntry, outputStream);
+ }
+ }
+ }
+
+ private void serializeServiceInfoWithCRC(ServiceInfo serviceInfo, OutputStream outputStream)
+ throws IOException {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream tempDos = new DataOutputStream(byteArrayOutputStream);
+ serviceInfo.serialize(tempDos);
+ tempDos.flush();
+ byte[] bytes = byteArrayOutputStream.toByteArray();
+
+ crc32.reset();
+ crc32.update(bytes, 0, bytes.length);
+
+ ReadWriteIOUtils.write(bytes.length, outputStream);
+ outputStream.write(bytes);
+ ReadWriteIOUtils.write((int) crc32.getValue(), outputStream);
+ }
+
+ private void deserializeInfos(InputStream inputStream) throws IOException {
+ if (ReadWriteIOUtils.readByte(inputStream) != SERIALIZATION_VERSION) {
+ throw new IOException("Incorrect version of " + SNAPSHOT_FILENAME);
+ }
+
+ int outerSize = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < outerSize; i++) {
+ int dataNodeId = ReadWriteIOUtils.readInt(inputStream);
+ int innerSize = ReadWriteIOUtils.readInt(inputStream);
+
+ Map<String, ServiceInfo> innerMap =
+ datanodeToServiceInfos.computeIfAbsent(
+ dataNodeId, k -> new ConcurrentHashMap<>(innerSize));
+ for (int j = 0; j < innerSize; j++) {
+ ServiceInfo value = deserializeServiceInfoConsiderCRC(inputStream);
+ if (value != null) {
+ innerMap.put(value.getServiceName(), value);
+ }
+ }
+ datanodeToServiceInfos.put(dataNodeId, innerMap);
+ }
+ }
+
+ private ServiceInfo deserializeServiceInfoConsiderCRC(InputStream inputStream)
+ throws IOException {
+ int length = ReadWriteIOUtils.readInt(inputStream);
+ byte[] bytes = new byte[length];
+ inputStream.read(bytes);
Review Comment:
Method deserializeServiceInfoConsiderCRC ignores exceptional return value of InputStream.read.
```suggestion
int offset = 0;
while (offset < length) {
int read = inputStream.read(bytes, offset, length - offset);
if (read == -1) {
throw new IOException("Unexpected end of stream while reading service info.");
}
offset += read;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement ExternalService Framework [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 commented on code in PR #17009:
URL: https://github.com/apache/iotdb/pull/17009#discussion_r2692708885
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java:
##########
@@ -754,6 +761,7 @@ private void sendRestartRequestToConfigNode() throws StartupException {
private void prepareResources() throws StartupException {
prepareUDFResources();
prepareTriggerResources();
+ prepareExternalServiceResources();
Review Comment:
better call this in `initProtocols` method
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java:
##########
@@ -489,6 +492,10 @@ protected void storeRuntimeConfigurations(
/* Store triggerInformationList */
getTriggerInformationList(runtimeConfiguration.getAllTriggerInformation());
+ /* Store externalServiceEntryList */
+ resourcesInformationHolder.setExternalServiceEntryList(
+ runtimeConfiguration.getAllUserDefinedServiceInfo());
Review Comment:
runtimeConfiguration.isSetAllUserDefinedServiceInfo() ? runtimeConfiguration.getAllUserDefinedServiceInfo() : Collections.empty_list();
##########
iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift:
##########
@@ -121,6 +121,7 @@ struct TRuntimeConfiguration {
8: required TAuditConfig auditConfig
9: required string superUserName
10: optional bool enableSeparationOfAdminPowers
+ 11: required list<common.TExternalServiceEntry> allUserDefinedServiceInfo
Review Comment:
```suggestion
11: optional list<common.TExternalServiceEntry> allUserDefinedServiceInfo
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement ExternalService Framework [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 commented on code in PR #17009:
URL: https://github.com/apache/iotdb/pull/17009#discussion_r2681221266
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java:
##########
@@ -546,6 +553,14 @@ private ColumnHeaderConstant() {
new ColumnHeader(CLASS_NAME, TSDataType.TEXT),
new ColumnHeader(NODE_ID, TSDataType.TEXT));
+ public static final List<ColumnHeader> showExternalServiceColumnHeaders =
+ ImmutableList.of(
+ new ColumnHeader(SERVICE_NAME, TSDataType.TEXT),
+ new ColumnHeader(DATA_NODE_ID, TSDataType.INT32),
+ new ColumnHeader(STATE, TSDataType.TEXT),
+ new ColumnHeader(CLASS_NAME, TSDataType.TEXT),
+ new ColumnHeader(SERVICE_TYPE, TSDataType.TEXT));
Review Comment:
```suggestion
new ColumnHeader(SERVICE_NAME, TSDataType.STRING),
new ColumnHeader(DATA_NODE_ID, TSDataType.INT32),
new ColumnHeader(STATE, TSDataType.STRING),
new ColumnHeader(CLASS_NAME, TSDataType.STRING),
new ColumnHeader(SERVICE_TYPE, TSDataType.STRING));
```
##########
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java:
##########
@@ -26,6 +26,7 @@ public enum ServiceType {
RPC_SERVICE("RPC ServerService", "RPCService"),
INFLUX_SERVICE("InfluxDB Protocol Service", "InfluxDB Protocol"),
MQTT_SERVICE("MQTTService", "MqttService"),
+ EXTERNAL_SERVICE("ExternalService Service", "ExternalService"),
Review Comment:
It seems that this enum is never used.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java:
##########
@@ -936,6 +941,38 @@ private TSStatus checkTriggerManagement(IAuditEntity auditEntity, Supplier<Strin
return checkGlobalAuth(auditEntity, PrivilegeType.USE_TRIGGER, auditObject);
}
+ // ======================= externalService related ================================
+ @Override
+ public TSStatus visitCreateExternalService(
+ CreateExternalServiceStatement createExternalServiceStatement,
+ TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitStartExternalService(
+ StartExternalServiceStatement startExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitStopExternalService(
+ StopExternalServiceStatement stopExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitDropExternalService(
+ DropExternalServiceStatement dropExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java:
##########
@@ -936,6 +941,38 @@ private TSStatus checkTriggerManagement(IAuditEntity auditEntity, Supplier<Strin
return checkGlobalAuth(auditEntity, PrivilegeType.USE_TRIGGER, auditObject);
}
+ // ======================= externalService related ================================
+ @Override
+ public TSStatus visitCreateExternalService(
+ CreateExternalServiceStatement createExternalServiceStatement,
+ TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitStartExternalService(
+ StartExternalServiceStatement startExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitStopExternalService(
+ StopExternalServiceStatement stopExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitDropExternalService(
+ DropExternalServiceStatement dropExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitShowExternalService(
+ ShowExternalServiceStatement showExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
Review Comment:
add TsStatusCode
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call stop method of ServiceInstance
+ if (serviceInfo.getState() == STOPPED) {
+ return;
+ } else {
+ // The state is RUNNING
+ stopService(serviceInfo);
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.stopExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().start();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(STOPPED);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private void stopService(ServiceInfo serviceInfo) {
+ checkState(
+ serviceInfo.getServiceInstance() != null,
+ "External Service instance is null when state is RUNNING!",
+ serviceInfo.getServiceName());
+ serviceInfo.getServiceInstance().stop();
+ }
+
+ public void dropService(String serviceName, boolean forcedly)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is not existed!", serviceName));
+ }
+ if (serviceInfo.getServiceType() == ServiceInfo.ServiceType.BUILTIN) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is BUILT-IN!", serviceName));
+ }
+
+ // 2. stop or fail when service are not stopped
+ if (serviceInfo.getState() == STOPPED) {
+ // do nothing
+ } else {
+ // The state is RUNNING
+ if (forcedly) {
+ try {
+ stopService(serviceInfo);
+ } catch (Exception e) {
+ // record errMsg if exception occurs during the stop of service
+ LOGGER.warn(
+ "Failed to stop External Service %s because %s. It will be drop forcedly",
+ serviceName, e.getMessage());
+ }
+ } else {
+ throw new ExternalServiceManagementException(
Review Comment:
dd TsStatusCode
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java:
##########
@@ -4839,4 +4849,82 @@ public void handlePipeConfigClientExit(final String clientId) {
LOGGER.warn("Failed to handlePipeConfigClientExit.", e);
}
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createExternalService(
+ String serviceName, String className) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().createService(serviceName, className);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> startExternalService(String serviceName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().startService(serviceName);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> stopExternalService(String serviceName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().stopService(serviceName);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> dropExternalService(
+ String serviceName, boolean forcedly) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().dropService(serviceName, forcedly);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> showExternalService(int dataNodeId) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ Iterator<TExternalServiceEntry> iterator =
+ ExternalServiceManagementService.getInstance().showService(dataNodeId);
+
+ List<TSDataType> outputDataTypes =
+ ColumnHeaderConstant.showExternalServiceColumnHeaders.stream()
+ .map(ColumnHeader::getColumnType)
+ .collect(Collectors.toList());
+ TsBlockBuilder builder = new TsBlockBuilder(outputDataTypes);
+ TimeColumnBuilder timeColumnBuilder = builder.getTimeColumnBuilder();
+ while (iterator.hasNext()) {
+ timeColumnBuilder.writeLong(0L);
+ appendServiceEntry(iterator.next(), builder.getValueColumnBuilders());
+ builder.declarePosition();
+ }
+
+ future.set(
+ new ConfigTaskResult(
+ TSStatusCode.SUCCESS_STATUS, builder.build(), getShowExternalServiceHeader()));
+ } catch (Exception e) {
+ future.setException(e);
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/BuiltinExternalServices.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor;
+
+import java.util.function.Supplier;
+
+public enum BuiltinExternalServices {
+ MQTT(
+ "MQTT",
+ "org.apache.iotdb.externalservice.Mqtt",
+ IoTDBDescriptor.getInstance().getConfig()::isEnableMQTTService),
+ REST(
+ "REST",
+ "org.apache.iotdb.externalservice.Rest",
+ IoTDBRestServiceDescriptor.getInstance().getConfig()::isEnableRestService);
Review Comment:
```suggestion
MQTT(
"MQTT",
"org.apache.iotdb.externalservice.Mqtt",
// IoTDBDescriptor.getInstance().getConfig()::isEnableMQTTService),
() -> false),
REST(
"REST",
"org.apache.iotdb.externalservice.Rest",
// IoTDBRestServiceDescriptor.getInstance().getConfig()::isEnableRestService);
() -> false);
```
better always return false before you really seperate the rest and mqtt service outside. Otherwise, mqtt and rest ITs will fail because `org.apache.iotdb.externalservice.Mqtt` and `org.apache.iotdb.externalservice.Rest` is not found
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java:
##########
@@ -4839,4 +4849,82 @@ public void handlePipeConfigClientExit(final String clientId) {
LOGGER.warn("Failed to handlePipeConfigClientExit.", e);
}
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createExternalService(
+ String serviceName, String className) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().createService(serviceName, className);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> startExternalService(String serviceName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().startService(serviceName);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> stopExternalService(String serviceName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().stopService(serviceName);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> dropExternalService(
+ String serviceName, boolean forcedly) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().dropService(serviceName, forcedly);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java:
##########
@@ -936,6 +941,38 @@ private TSStatus checkTriggerManagement(IAuditEntity auditEntity, Supplier<Strin
return checkGlobalAuth(auditEntity, PrivilegeType.USE_TRIGGER, auditObject);
}
+ // ======================= externalService related ================================
+ @Override
+ public TSStatus visitCreateExternalService(
+ CreateExternalServiceStatement createExternalServiceStatement,
+ TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitStartExternalService(
+ StartExternalServiceStatement startExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java:
##########
@@ -1087,6 +1092,52 @@ public Node visitShowFunctionsStatement(RelationalSqlParser.ShowFunctionsStateme
return new ShowFunctions();
}
+ @Override
+ public Node visitCreateServiceStatement(RelationalSqlParser.CreateServiceStatementContext ctx) {
+ String serviceName = ((Identifier) visit(ctx.serviceName)).getValue();
+ String className = ((StringLiteral) visit(ctx.className)).getValue();
+ return new CreateExternalService(getLocation(ctx), serviceName, className);
+ }
+
+ @Override
+ public Node visitStartServiceStatement(RelationalSqlParser.StartServiceStatementContext ctx) {
+ return new StartExternalService(
+ getLocation(ctx), ((Identifier) visit(ctx.serviceName)).getValue());
+ }
+
+ @Override
+ public Node visitStopServiceStatement(RelationalSqlParser.StopServiceStatementContext ctx) {
+ return new StopExternalService(
+ getLocation(ctx), ((Identifier) visit(ctx.serviceName)).getValue());
+ }
+
+ @Override
+ public Node visitDropServiceStatement(RelationalSqlParser.DropServiceStatementContext ctx) {
+ return new DropExternalService(
+ getLocation(ctx), ((Identifier) visit(ctx.serviceName)).getValue(), ctx.FORCEDLY() != null);
+ }
+
+ @Override
+ public Node visitShowServiceStatement(RelationalSqlParser.ShowServiceStatementContext ctx) {
+ // show services on all DNs
+ Optional<Expression> where = Optional.empty();
+ if (ctx.ON() != null) {
+ where =
+ Optional.of(
+ new ComparisonExpression(
+ ComparisonExpression.Operator.EQUAL,
+ new Identifier("datanode_id"),
Review Comment:
```suggestion
new Identifier(ColumnHeaderConstant.DATA_NODE_ID_TABLE_MODEL),
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java:
##########
@@ -1287,6 +1287,7 @@ private void initProtocols() throws StartupException {
if (IoTDBRestServiceDescriptor.getInstance().getConfig().isEnableRestService()) {
registerManager.register(RestService.getInstance());
}
+ // registerManager.register(ExternalServiceManagementService.getInstance());
Review Comment:
need this? or should be just deleted.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
Review Comment:
use try-with-resource to close the client
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java:
##########
@@ -936,6 +941,38 @@ private TSStatus checkTriggerManagement(IAuditEntity auditEntity, Supplier<Strin
return checkGlobalAuth(auditEntity, PrivilegeType.USE_TRIGGER, auditObject);
}
+ // ======================= externalService related ================================
+ @Override
+ public TSStatus visitCreateExternalService(
+ CreateExternalServiceStatement createExternalServiceStatement,
+ TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
Review Comment:
() -> "" should be replaced with exact sql which can derived from `CreateExternalServiceStatement`.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java:
##########
@@ -1511,6 +1519,37 @@ protected IConfigTask visitDropFunction(DropFunction node, MPPQueryContext conte
return new DropFunctionTask(Model.TABLE, node.getUdfName());
}
+ @Override
+ protected IConfigTask visitCreateExternalService(
+ CreateExternalService node, MPPQueryContext context) {
+ context.setQueryType(QueryType.WRITE);
Review Comment:
```suggestion
context.setQueryType(QueryType.WRITE);
accessControl.checkUserGlobalSysPrivilege(context);
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java:
##########
@@ -936,6 +941,38 @@ private TSStatus checkTriggerManagement(IAuditEntity auditEntity, Supplier<Strin
return checkGlobalAuth(auditEntity, PrivilegeType.USE_TRIGGER, auditObject);
}
+ // ======================= externalService related ================================
+ @Override
+ public TSStatus visitCreateExternalService(
+ CreateExternalServiceStatement createExternalServiceStatement,
+ TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitStartExternalService(
+ StartExternalServiceStatement startExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
+ }
+
+ @Override
+ public TSStatus visitStopExternalService(
+ StopExternalServiceStatement stopExternalServiceStatement, TreeAccessCheckContext context) {
+ return checkGlobalAuth(context, PrivilegeType.SYSTEM, () -> "");
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
Review Comment:
add status code
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementException.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.exernalservice;
+
+public class ExternalServiceManagementException extends RuntimeException {
Review Comment:
```suggestion
public class ExternalServiceManagementException extends IoTDBRuntimeException {
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call stop method of ServiceInstance
+ if (serviceInfo.getState() == STOPPED) {
+ return;
+ } else {
+ // The state is RUNNING
+ stopService(serviceInfo);
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.stopExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().start();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(STOPPED);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private void stopService(ServiceInfo serviceInfo) {
+ checkState(
+ serviceInfo.getServiceInstance() != null,
+ "External Service instance is null when state is RUNNING!",
+ serviceInfo.getServiceName());
+ serviceInfo.getServiceInstance().stop();
+ }
+
+ public void dropService(String serviceName, boolean forcedly)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
Review Comment:
add TsStatusCode
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
Review Comment:
use try-with-resource to close the client
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
Review Comment:
add TsStatusCode
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
Review Comment:
close it
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
Review Comment:
```suggestion
} else { // The state is STOPPED
if (serviceInfo.getServiceInstance() == null) {
// lazy create Instance
serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
}
serviceInfo.getServiceInstance().start();
}
```
After creating the instance, you still need to create it.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call stop method of ServiceInstance
+ if (serviceInfo.getState() == STOPPED) {
+ return;
+ } else {
+ // The state is RUNNING
+ stopService(serviceInfo);
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.stopExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().start();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(STOPPED);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private void stopService(ServiceInfo serviceInfo) {
+ checkState(
+ serviceInfo.getServiceInstance() != null,
+ "External Service instance is null when state is RUNNING!",
+ serviceInfo.getServiceName());
+ serviceInfo.getServiceInstance().stop();
+ }
+
+ public void dropService(String serviceName, boolean forcedly)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is not existed!", serviceName));
+ }
+ if (serviceInfo.getServiceType() == ServiceInfo.ServiceType.BUILTIN) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is BUILT-IN!", serviceName));
+ }
+
+ // 2. stop or fail when service are not stopped
+ if (serviceInfo.getState() == STOPPED) {
+ // do nothing
+ } else {
+ // The state is RUNNING
+ if (forcedly) {
+ try {
+ stopService(serviceInfo);
+ } catch (Exception e) {
+ // record errMsg if exception occurs during the stop of service
+ LOGGER.warn(
+ "Failed to stop External Service %s because %s. It will be drop forcedly",
+ serviceName, e.getMessage());
+ }
+ } else {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is RUNNING!", serviceName));
+ }
+ }
+
+ // 3. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.dropExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfos.remove(serviceName);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void restoreRunningServices() {
Review Comment:
not called?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call stop method of ServiceInstance
+ if (serviceInfo.getState() == STOPPED) {
+ return;
+ } else {
+ // The state is RUNNING
+ stopService(serviceInfo);
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.stopExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().start();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(STOPPED);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private void stopService(ServiceInfo serviceInfo) {
+ checkState(
+ serviceInfo.getServiceInstance() != null,
+ "External Service instance is null when state is RUNNING!",
+ serviceInfo.getServiceName());
+ serviceInfo.getServiceInstance().stop();
+ }
+
+ public void dropService(String serviceName, boolean forcedly)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is not existed!", serviceName));
+ }
+ if (serviceInfo.getServiceType() == ServiceInfo.ServiceType.BUILTIN) {
+ throw new ExternalServiceManagementException(
Review Comment:
Add TsStatusCode
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call stop method of ServiceInstance
+ if (serviceInfo.getState() == STOPPED) {
+ return;
+ } else {
+ // The state is RUNNING
+ stopService(serviceInfo);
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.stopExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().start();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(STOPPED);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private void stopService(ServiceInfo serviceInfo) {
+ checkState(
+ serviceInfo.getServiceInstance() != null,
+ "External Service instance is null when state is RUNNING!",
+ serviceInfo.getServiceName());
+ serviceInfo.getServiceInstance().stop();
+ }
+
+ public void dropService(String serviceName, boolean forcedly)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is not existed!", serviceName));
+ }
+ if (serviceInfo.getServiceType() == ServiceInfo.ServiceType.BUILTIN) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is BUILT-IN!", serviceName));
+ }
+
+ // 2. stop or fail when service are not stopped
+ if (serviceInfo.getState() == STOPPED) {
+ // do nothing
+ } else {
+ // The state is RUNNING
+ if (forcedly) {
+ try {
+ stopService(serviceInfo);
+ } catch (Exception e) {
+ // record errMsg if exception occurs during the stop of service
+ LOGGER.warn(
+ "Failed to stop External Service %s because %s. It will be drop forcedly",
+ serviceName, e.getMessage());
+ }
+ } else {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to drop External Service %s, because it is RUNNING!", serviceName));
+ }
+ }
+
+ // 3. persist on CN
+ ConfigNodeClient client =
Review Comment:
close it.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
+ String.format("Failed to create External Service %s, it already exists!", serviceName));
+ }
+
+ // 2. persist on CN
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status =
+ client.createExternalService(
+ new TCreateExternalServiceReq(QueryId.getDataNodeId(), serviceName, className));
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 3. modify memory info
+ serviceInfos.put(
+ serviceName,
+ new ServiceInfo(serviceName, className, ServiceInfo.ServiceType.USER_DEFINED));
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ public void startService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call start method of ServiceInstance, create if Instance was not created
+ if (serviceInfo.getState() == RUNNING) {
+ return;
+ } else {
+ // The state is STOPPED
+ if (serviceInfo.getServiceInstance() != null) {
+ serviceInfo.getServiceInstance().start();
+ } else {
+ // lazy create Instance
+ serviceInfo.setServiceInstance(createExternalServiceInstance(serviceName));
+ }
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TSStatus status = client.startExternalService(QueryId.getDataNodeId(), serviceName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ serviceInfo.getServiceInstance().stop();
+ throw new IoTDBRuntimeException(status.message, status.code);
+ }
+
+ // 4. modify memory info
+ serviceInfo.setState(RUNNING);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private IExternalService createExternalServiceInstance(String serviceName) {
+ // close ClassLoader automatically to release the file handle
+ try (ExternalServiceClassLoader classLoader = new ExternalServiceClassLoader(libRoot); ) {
+ return (IExternalService)
+ Class.forName(serviceName, true, classLoader).getDeclaredConstructor().newInstance();
+ } catch (IOException
+ | InstantiationException
+ | InvocationTargetException
+ | NoSuchMethodException
+ | IllegalAccessException
+ | ClassNotFoundException
+ | ClassCastException e) {
+ String errorMessage =
+ String.format(
+ "Failed to start External Service %s, because its instance can not be constructed successfully. Exception: %s",
+ serviceName, e);
+ LOGGER.warn(errorMessage, e);
+ throw new ExternalServiceManagementException(errorMessage);
+ }
+ }
+
+ public void stopService(String serviceName) throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ if (serviceInfo == null) {
+ throw new ExternalServiceManagementException(
+ String.format(
+ "Failed to start External Service %s, because it is not existed!", serviceName));
+ }
+
+ // 2. call stop method of ServiceInstance
+ if (serviceInfo.getState() == STOPPED) {
+ return;
+ } else {
+ // The state is RUNNING
+ stopService(serviceInfo);
+ }
+
+ // 3. persist on CN, rollback if failed
+ ConfigNodeClient client =
Review Comment:
close it.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java:
##########
@@ -4839,4 +4849,82 @@ public void handlePipeConfigClientExit(final String clientId) {
LOGGER.warn("Failed to handlePipeConfigClientExit.", e);
}
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createExternalService(
+ String serviceName, String className) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().createService(serviceName, className);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> startExternalService(String serviceName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().startService(serviceName);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
Review Comment:
same as above
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java:
##########
@@ -4839,4 +4849,82 @@ public void handlePipeConfigClientExit(final String clientId) {
LOGGER.warn("Failed to handlePipeConfigClientExit.", e);
}
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createExternalService(
+ String serviceName, String className) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().createService(serviceName, className);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
Review Comment:
```suggestion
} catch (IoTDBRuntimeException e) {
future.setException(e);
} catch (Exception e) {
future.setException(new ExternalServiceManagementException(301?, e.getMessage()));
}
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/ExternalServiceManagementService.java:
##########
@@ -0,0 +1,342 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
+import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
+import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.externalservice.api.IExternalService;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.RUNNING;
+import static org.apache.iotdb.commons.externalservice.ServiceInfo.State.STOPPED;
+
+public class ExternalServiceManagementService {
+ @GuardedBy("lock")
+ private final Map<String, ServiceInfo> serviceInfos;
+
+ private final String libRoot;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ExternalServiceManagementService.class);
+
+ private ExternalServiceManagementService(String libRoot) {
+ this.serviceInfos = new HashMap<>();
+ restoreBuiltInServices();
+ this.libRoot = libRoot;
+ }
+
+ public Iterator<TExternalServiceEntry> showService(int dataNodeId)
+ throws ClientManagerException, TException {
+ try {
+ lock.readLock().lock();
+
+ ConfigNodeClient client =
+ ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID);
+ TExternalServiceListResp resp = client.showExternalService(dataNodeId);
+ if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ throw new IoTDBRuntimeException(resp.getStatus().message, resp.getStatus().code);
+ }
+ return resp.getExternalServiceInfos().iterator();
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ public void createService(String serviceName, String className)
+ throws ClientManagerException, TException {
+ try {
+ lock.writeLock().lock();
+
+ // 1. validate
+ if (serviceInfos.containsKey(serviceName)) {
+ throw new ExternalServiceManagementException(
Review Comment:
let ExternalServiceManagementException extend IoTDBRuntimeException
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/exernalservice/BuiltinExternalServices.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.exernalservice;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor;
+
+import java.util.function.Supplier;
+
+public enum BuiltinExternalServices {
+ MQTT(
+ "MQTT",
+ "org.apache.iotdb.externalservice.Mqtt",
+ IoTDBDescriptor.getInstance().getConfig()::isEnableMQTTService),
+ REST(
+ "REST",
+ "org.apache.iotdb.externalservice.Rest",
+ IoTDBRestServiceDescriptor.getInstance().getConfig()::isEnableRestService);
Review Comment:
```suggestion
MQTT(
"MQTT",
"org.apache.iotdb.externalservice.Mqtt",
// IoTDBDescriptor.getInstance().getConfig()::isEnableMQTTService),
() -> false),
REST(
"REST",
"org.apache.iotdb.externalservice.Rest",
// IoTDBRestServiceDescriptor.getInstance().getConfig()::isEnableRestService);
() -> false);
```
better always return false before you really seperate the rest and mqtt service outside. Otherwise, mqtt and rest ITs will fail because `org.apache.iotdb.externalservice.Mqtt` and `org.apache.iotdb.externalservice.Rest` is not found
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java:
##########
@@ -4839,4 +4849,82 @@ public void handlePipeConfigClientExit(final String clientId) {
LOGGER.warn("Failed to handlePipeConfigClientExit.", e);
}
}
+
+ @Override
+ public SettableFuture<ConfigTaskResult> createExternalService(
+ String serviceName, String className) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().createService(serviceName, className);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> startExternalService(String serviceName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().startService(serviceName);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
+ }
+ return future;
+ }
+
+ @Override
+ public SettableFuture<ConfigTaskResult> stopExternalService(String serviceName) {
+ SettableFuture<ConfigTaskResult> future = SettableFuture.create();
+ try {
+ ExternalServiceManagementService.getInstance().stopService(serviceName);
+ future.set(new ConfigTaskResult(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ } catch (Exception e) {
+ future.setException(new ExternalServiceManagementException(e.getMessage()));
Review Comment:
same as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Implement ExternalService Framework [iotdb]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 commented on code in PR #17009:
URL: https://github.com/apache/iotdb/pull/17009#discussion_r2681095062
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java:
##########
@@ -270,6 +270,10 @@ public class IoTDBConfig {
private String triggerTemporaryLibDir =
triggerDir + File.separator + IoTDBConstant.TMP_FOLDER_NAME;
+ /** External lib directory for UDF, stores user-uploaded JAR files */
Review Comment:
for UDF?
##########
iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4:
##########
@@ -1764,6 +1793,8 @@ SECURITY: 'SECURITY';
SEEK: 'SEEK';
SELECT: 'SELECT';
SERIALIZABLE: 'SERIALIZABLE';
+SERVICE: 'SERVICE';
+SERVICES: 'SERVICES';
Review Comment:
need to add these three in nonReserved section, otherwise, they will be reserved words, but actually, they shouldn't be.
<img width="2702" height="1282" alt="Image" src="https://github.com/user-attachments/assets/9f5b76f6-d077-49d0-af93-45594087bab6" />
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Start the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus startService(StartExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.RUNNING);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Stop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus stopService(StopExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.STOPPED);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ public ShowExternalServiceResp showService(Set<Integer> dataNodes) {
+ return new ShowExternalServiceResp(
+ datanodeToServiceInfos.entrySet().stream()
+ .filter(entry -> dataNodes.contains(entry.getKey()))
+ .flatMap(
+ entry ->
+ entry.getValue().values().stream()
+ .map(
+ serviceInfo ->
+ new TExternalServiceEntry(
+ serviceInfo.getServiceName(),
+ serviceInfo.getClassName(),
+ serviceInfo.getState().getValue(),
+ entry.getKey(),
+ ServiceInfo.ServiceType.USER_DEFINED.getValue())))
+ .collect(Collectors.toList()));
+ }
+
+ private void serializeInfos(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(SERIALIZATION_VERSION, outputStream);
+ ReadWriteIOUtils.write(datanodeToServiceInfos.size(), outputStream);
+ for (Map.Entry<Integer, Map<String, ServiceInfo>> outerEntry :
+ datanodeToServiceInfos.entrySet()) {
+ ReadWriteIOUtils.write(outerEntry.getKey(), outputStream); // DataNode ID
+
+ Map<String, ServiceInfo> innerMap = outerEntry.getValue();
+ // inner Map
+ ReadWriteIOUtils.write(innerMap.size(), outputStream);
+ for (ServiceInfo innerEntry : innerMap.values()) {
+ serializeServiceInfoWithCRC(innerEntry, outputStream);
+ }
+ }
+ }
+
+ private void serializeServiceInfoWithCRC(ServiceInfo serviceInfo, OutputStream outputStream)
+ throws IOException {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream tempDos = new DataOutputStream(byteArrayOutputStream);
+ serviceInfo.serialize(tempDos);
+ tempDos.flush();
+ byte[] bytes = byteArrayOutputStream.toByteArray();
+
+ crc32.reset();
+ crc32.update(bytes, 0, bytes.length);
+
+ ReadWriteIOUtils.write(bytes.length, outputStream);
+ outputStream.write(bytes);
+ ReadWriteIOUtils.write(crc32.getValue(), outputStream);
+ }
+
+ private void deserializeInfos(InputStream inputStream) throws IOException {
+ if (ReadWriteIOUtils.readInt(inputStream) != SERIALIZATION_VERSION) {
+ throw new IOException("Incorrect version of " + SNAPSHOT_FILENAME);
+ }
+
+ int outerSize = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < outerSize; i++) {
+ int dataNodeId = ReadWriteIOUtils.readInt(inputStream);
+ int innerSize = ReadWriteIOUtils.readInt(inputStream);
+
+ Map<String, ServiceInfo> innerMap =
+ datanodeToServiceInfos.computeIfAbsent(
+ dataNodeId, k -> new ConcurrentHashMap<>(innerSize));
+ for (int j = 0; j < innerSize; j++) {
+ ServiceInfo value = deserializeServiceInfoConsiderCRC(inputStream);
+ if (value != null) {
+ innerMap.put(value.getServiceName(), value);
+ }
+ }
+ datanodeToServiceInfos.put(dataNodeId, innerMap);
+ }
+ }
+
+ private ServiceInfo deserializeServiceInfoConsiderCRC(InputStream inputStream)
+ throws IOException {
+ int length = ReadWriteIOUtils.readInt(inputStream);
+ byte[] bytes = new byte[length];
+ inputStream.read(bytes);
+
+ crc32.reset();
+ crc32.update(bytes, 0, length);
+
+ long expectedCRC = ReadWriteIOUtils.readLong(inputStream);
+ if (crc32.getValue() != expectedCRC) {
+ LOGGER.error("Mismatched CRC32 code when deserializing service info.");
+ return null;
+ }
+
+ return ServiceInfo.deserialize(inputStream);
+ }
+
+ @Override
+ public boolean processTakeSnapshot(File snapshotDir) throws IOException {
+ File snapshotFile = new File(snapshotDir, SNAPSHOT_FILENAME);
+ if (snapshotFile.exists() && snapshotFile.isFile()) {
+ LOGGER.error(
+ "Failed to take snapshot, because snapshot file [{}] is already exist.",
+ snapshotFile.getAbsolutePath());
+ return false;
+ }
+
+ try (FileOutputStream fileOutputStream = new FileOutputStream(snapshotFile)) {
+
+ serializeInfos(fileOutputStream);
+
+ // fsync
+ fileOutputStream.getFD().sync();
+
+ return true;
+ }
+ }
+
+ @Override
+ public void processLoadSnapshot(File snapshotDir) throws IOException {
+ File snapshotFile = new File(snapshotDir, SNAPSHOT_FILENAME);
+ if (!snapshotFile.exists() || !snapshotFile.isFile()) {
+ LOGGER.error(
+ "Failed to load snapshot,snapshot file [{}] is not exist.",
+ snapshotFile.getAbsolutePath());
+ return;
+ }
+
+ try (FileInputStream fileInputStream = new FileInputStream(snapshotFile)) {
+
+ clear();
+
+ deserializeInfos(fileInputStream);
+ }
+ }
+
+ public void clear() {
+ datanodeToServiceInfos.values().forEach(subMap -> subMap.clear());
Review Comment:
```suggestion
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
Review Comment:
```suggestion
ServiceInfo removed = serviceInfos.remove(serviceName);
if (removed == null) {
res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
res.message =
String.format(
"ExternalService %s is not existed on DataNode %s.",
serviceName, plan.getDataNodeId());
} else {
res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
}
```
##########
iotdb-core/confignode/src/test/resources/confignode1conf/iotdb-system.properties:
##########
@@ -33,8 +33,8 @@ cn_metric_prometheus_reporter_port=9091
timestamp_precision=ms
data_region_consensus_protocol_class=org.apache.iotdb.consensus.iot.IoTConsensus
schema_region_consensus_protocol_class=org.apache.iotdb.consensus.ratis.RatisConsensus
-schema_replication_factor=3
-data_replication_factor=3
+schema_replication_factor=1
+data_replication_factor=1
Review Comment:
change them back
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Start the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus startService(StartExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.RUNNING);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Stop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus stopService(StopExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.STOPPED);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ public ShowExternalServiceResp showService(Set<Integer> dataNodes) {
+ return new ShowExternalServiceResp(
+ datanodeToServiceInfos.entrySet().stream()
+ .filter(entry -> dataNodes.contains(entry.getKey()))
+ .flatMap(
+ entry ->
+ entry.getValue().values().stream()
+ .map(
+ serviceInfo ->
+ new TExternalServiceEntry(
+ serviceInfo.getServiceName(),
+ serviceInfo.getClassName(),
+ serviceInfo.getState().getValue(),
+ entry.getKey(),
+ ServiceInfo.ServiceType.USER_DEFINED.getValue())))
+ .collect(Collectors.toList()));
+ }
+
+ private void serializeInfos(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(SERIALIZATION_VERSION, outputStream);
+ ReadWriteIOUtils.write(datanodeToServiceInfos.size(), outputStream);
+ for (Map.Entry<Integer, Map<String, ServiceInfo>> outerEntry :
+ datanodeToServiceInfos.entrySet()) {
+ ReadWriteIOUtils.write(outerEntry.getKey(), outputStream); // DataNode ID
+
+ Map<String, ServiceInfo> innerMap = outerEntry.getValue();
+ // inner Map
+ ReadWriteIOUtils.write(innerMap.size(), outputStream);
+ for (ServiceInfo innerEntry : innerMap.values()) {
+ serializeServiceInfoWithCRC(innerEntry, outputStream);
+ }
+ }
+ }
+
+ private void serializeServiceInfoWithCRC(ServiceInfo serviceInfo, OutputStream outputStream)
+ throws IOException {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream tempDos = new DataOutputStream(byteArrayOutputStream);
+ serviceInfo.serialize(tempDos);
+ tempDos.flush();
+ byte[] bytes = byteArrayOutputStream.toByteArray();
+
+ crc32.reset();
+ crc32.update(bytes, 0, bytes.length);
+
+ ReadWriteIOUtils.write(bytes.length, outputStream);
+ outputStream.write(bytes);
+ ReadWriteIOUtils.write(crc32.getValue(), outputStream);
Review Comment:
```suggestion
ReadWriteIOUtils.write((int) crc32.getValue(), outputStream);
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Start the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus startService(StartExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.RUNNING);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Stop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus stopService(StopExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.STOPPED);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ public ShowExternalServiceResp showService(Set<Integer> dataNodes) {
+ return new ShowExternalServiceResp(
+ datanodeToServiceInfos.entrySet().stream()
+ .filter(entry -> dataNodes.contains(entry.getKey()))
+ .flatMap(
+ entry ->
+ entry.getValue().values().stream()
+ .map(
+ serviceInfo ->
+ new TExternalServiceEntry(
+ serviceInfo.getServiceName(),
+ serviceInfo.getClassName(),
+ serviceInfo.getState().getValue(),
+ entry.getKey(),
+ ServiceInfo.ServiceType.USER_DEFINED.getValue())))
+ .collect(Collectors.toList()));
+ }
+
+ private void serializeInfos(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(SERIALIZATION_VERSION, outputStream);
+ ReadWriteIOUtils.write(datanodeToServiceInfos.size(), outputStream);
+ for (Map.Entry<Integer, Map<String, ServiceInfo>> outerEntry :
+ datanodeToServiceInfos.entrySet()) {
+ ReadWriteIOUtils.write(outerEntry.getKey(), outputStream); // DataNode ID
+
+ Map<String, ServiceInfo> innerMap = outerEntry.getValue();
+ // inner Map
+ ReadWriteIOUtils.write(innerMap.size(), outputStream);
+ for (ServiceInfo innerEntry : innerMap.values()) {
+ serializeServiceInfoWithCRC(innerEntry, outputStream);
+ }
+ }
+ }
+
+ private void serializeServiceInfoWithCRC(ServiceInfo serviceInfo, OutputStream outputStream)
+ throws IOException {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream tempDos = new DataOutputStream(byteArrayOutputStream);
+ serviceInfo.serialize(tempDos);
+ tempDos.flush();
+ byte[] bytes = byteArrayOutputStream.toByteArray();
+
+ crc32.reset();
+ crc32.update(bytes, 0, bytes.length);
+
+ ReadWriteIOUtils.write(bytes.length, outputStream);
+ outputStream.write(bytes);
+ ReadWriteIOUtils.write(crc32.getValue(), outputStream);
+ }
+
+ private void deserializeInfos(InputStream inputStream) throws IOException {
+ if (ReadWriteIOUtils.readInt(inputStream) != SERIALIZATION_VERSION) {
+ throw new IOException("Incorrect version of " + SNAPSHOT_FILENAME);
+ }
+
+ int outerSize = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < outerSize; i++) {
+ int dataNodeId = ReadWriteIOUtils.readInt(inputStream);
+ int innerSize = ReadWriteIOUtils.readInt(inputStream);
+
+ Map<String, ServiceInfo> innerMap =
+ datanodeToServiceInfos.computeIfAbsent(
+ dataNodeId, k -> new ConcurrentHashMap<>(innerSize));
+ for (int j = 0; j < innerSize; j++) {
+ ServiceInfo value = deserializeServiceInfoConsiderCRC(inputStream);
+ if (value != null) {
+ innerMap.put(value.getServiceName(), value);
+ }
+ }
+ datanodeToServiceInfos.put(dataNodeId, innerMap);
+ }
+ }
+
+ private ServiceInfo deserializeServiceInfoConsiderCRC(InputStream inputStream)
+ throws IOException {
+ int length = ReadWriteIOUtils.readInt(inputStream);
+ byte[] bytes = new byte[length];
+ inputStream.read(bytes);
+
+ crc32.reset();
+ crc32.update(bytes, 0, length);
+
+ long expectedCRC = ReadWriteIOUtils.readLong(inputStream);
Review Comment:
```suggestion
int expectedCRC = ReadWriteIOUtils.readInt(inputStream);
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
Review Comment:
define `"ExternalService %s is not existed on DataNode %s."` as a constant
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Start the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus startService(StartExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.RUNNING);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Stop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus stopService(StopExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.STOPPED);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
Review Comment:
```suggestion
ServiceInfo serviceInfo = serviceInfos.get(serviceName);
if (serviceInfo == null) {
res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
res.message =
String.format(
"ExternalService %s is not existed on DataNode %s.",
serviceName, plan.getDataNodeId());
} else {
serviceInfo.setState(ServiceInfo.State.STOPPED);
res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
}
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Start the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus startService(StartExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.RUNNING);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
Review Comment:
```suggestion
ServiceInfo serviceInfo = serviceInfos.get(serviceName);
if (serviceInfo == null) {
res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
res.message =
String.format(
"ExternalService %s is not existed on DataNode %s.",
serviceName, plan.getDataNodeId());
} else {
serviceInfo.setState(ServiceInfo.State.RUNNING);
res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
}
```
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java:
##########
@@ -0,0 +1,313 @@
+/*
+ * 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.confignode.manager.externalservice;
+
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceEntry;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.externalservice.ServiceInfo;
+import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.CreateExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.DropExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StartExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.request.write.externalservice.StopExternalServicePlan;
+import org.apache.iotdb.confignode.consensus.response.externalservice.ShowExternalServiceResp;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.zip.CRC32;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public class ExternalServiceInfo implements SnapshotProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ExternalServiceInfo.class);
+
+ private final Map<Integer, Map<String, ServiceInfo>> datanodeToServiceInfos;
+
+ private static final String SNAPSHOT_FILENAME = "service_info.bin";
+ private static final int SERIALIZATION_VERSION = 1;
+ private final CRC32 crc32 = new CRC32();
+
+ public ExternalServiceInfo() {
+ datanodeToServiceInfos = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * Add a new ExternalService on target DataNode.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was not existed on target DataNode, otherwise
+ * EXTERNAL_SERVICE_AlREADY_EXIST
+ */
+ public TSStatus addService(CreateExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDatanodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceInfo().getServiceName();
+ if (serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.EXTERNAL_SERVICE_ALREADY_EXIST.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s has already been created on DataNode %s.",
+ serviceName, plan.getDatanodeId());
+ } else {
+ serviceInfos.put(serviceName, plan.getServiceInfo());
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Drop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus dropService(DropExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ serviceInfos.remove(serviceName);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Start the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus startService(StartExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.RUNNING);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ /**
+ * Stop the ExternalService whose name is same as <tt>serviceName</tt> in plan.
+ *
+ * @return SUCCESS_STATUS if <tt>this service</tt> was existed on target DataNode, otherwise
+ * NO_SUCH_EXTERNAL_SERVICE
+ */
+ public TSStatus stopService(StopExternalServicePlan plan) {
+ TSStatus res = new TSStatus();
+ Map<String, ServiceInfo> serviceInfos =
+ datanodeToServiceInfos.computeIfAbsent(
+ plan.getDataNodeId(), k -> new ConcurrentHashMap<>());
+ String serviceName = plan.getServiceName();
+ if (!serviceInfos.containsKey(serviceName)) {
+ res.code = TSStatusCode.NO_SUCH_EXTERNAL_SERVICE.getStatusCode();
+ res.message =
+ String.format(
+ "ExternalService %s is not existed on DataNode %s.",
+ serviceName, plan.getDataNodeId());
+ } else {
+ ServiceInfo serviceInfo = serviceInfos.get(serviceName);
+ // The WRITE operations of StateMachine are not concurrent
+ checkState(serviceInfo != null, "Target serviceInfo should not be null.");
+ serviceInfo.setState(ServiceInfo.State.STOPPED);
+ res.code = TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+ return res;
+ }
+
+ public ShowExternalServiceResp showService(Set<Integer> dataNodes) {
+ return new ShowExternalServiceResp(
+ datanodeToServiceInfos.entrySet().stream()
+ .filter(entry -> dataNodes.contains(entry.getKey()))
+ .flatMap(
+ entry ->
+ entry.getValue().values().stream()
+ .map(
+ serviceInfo ->
+ new TExternalServiceEntry(
+ serviceInfo.getServiceName(),
+ serviceInfo.getClassName(),
+ serviceInfo.getState().getValue(),
+ entry.getKey(),
+ ServiceInfo.ServiceType.USER_DEFINED.getValue())))
+ .collect(Collectors.toList()));
+ }
+
+ private void serializeInfos(OutputStream outputStream) throws IOException {
+ ReadWriteIOUtils.write(SERIALIZATION_VERSION, outputStream);
+ ReadWriteIOUtils.write(datanodeToServiceInfos.size(), outputStream);
+ for (Map.Entry<Integer, Map<String, ServiceInfo>> outerEntry :
+ datanodeToServiceInfos.entrySet()) {
+ ReadWriteIOUtils.write(outerEntry.getKey(), outputStream); // DataNode ID
+
+ Map<String, ServiceInfo> innerMap = outerEntry.getValue();
+ // inner Map
+ ReadWriteIOUtils.write(innerMap.size(), outputStream);
+ for (ServiceInfo innerEntry : innerMap.values()) {
+ serializeServiceInfoWithCRC(innerEntry, outputStream);
+ }
+ }
+ }
+
+ private void serializeServiceInfoWithCRC(ServiceInfo serviceInfo, OutputStream outputStream)
+ throws IOException {
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+ DataOutputStream tempDos = new DataOutputStream(byteArrayOutputStream);
+ serviceInfo.serialize(tempDos);
+ tempDos.flush();
+ byte[] bytes = byteArrayOutputStream.toByteArray();
+
+ crc32.reset();
+ crc32.update(bytes, 0, bytes.length);
+
+ ReadWriteIOUtils.write(bytes.length, outputStream);
+ outputStream.write(bytes);
+ ReadWriteIOUtils.write(crc32.getValue(), outputStream);
+ }
+
+ private void deserializeInfos(InputStream inputStream) throws IOException {
+ if (ReadWriteIOUtils.readInt(inputStream) != SERIALIZATION_VERSION) {
+ throw new IOException("Incorrect version of " + SNAPSHOT_FILENAME);
+ }
+
+ int outerSize = ReadWriteIOUtils.readInt(inputStream);
+ for (int i = 0; i < outerSize; i++) {
+ int dataNodeId = ReadWriteIOUtils.readInt(inputStream);
+ int innerSize = ReadWriteIOUtils.readInt(inputStream);
+
+ Map<String, ServiceInfo> innerMap =
+ datanodeToServiceInfos.computeIfAbsent(
+ dataNodeId, k -> new ConcurrentHashMap<>(innerSize));
+ for (int j = 0; j < innerSize; j++) {
+ ServiceInfo value = deserializeServiceInfoConsiderCRC(inputStream);
+ if (value != null) {
+ innerMap.put(value.getServiceName(), value);
+ }
+ }
+ datanodeToServiceInfos.put(dataNodeId, innerMap);
+ }
+ }
+
+ private ServiceInfo deserializeServiceInfoConsiderCRC(InputStream inputStream)
+ throws IOException {
+ int length = ReadWriteIOUtils.readInt(inputStream);
+ byte[] bytes = new byte[length];
+ inputStream.read(bytes);
+
+ crc32.reset();
+ crc32.update(bytes, 0, length);
+
+ long expectedCRC = ReadWriteIOUtils.readLong(inputStream);
+ if (crc32.getValue() != expectedCRC) {
Review Comment:
```suggestion
if (((int) crc32.getValue()) != expectedCRC) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org