Implement ExternalService Framework

iotdb-api/external-service-api/pom.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.iotdb</groupId>
+        <artifactId>iotdb-api</artifactId>
+        <version>2.0.7-SNAPSHOT</version>
+    </parent>
+    <artifactId>external-service-api</artifactId>
+    <name>IoTDB: API: External Service API</name>
+    <profiles>
+        <profile>
+            <id>get-jar-with-dependencies</id>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-assembly-plugin</artifactId>
+                        <configuration>
+                            <descriptorRefs>
+                                <descriptorRef>jar-with-dependencies</descriptorRef>
+                            </descriptorRefs>
+                        </configuration>
+                        <executions>
+                            <execution>
+                                <id>make-assembly</id>
+                                <!-- bind to the packaging phase -->
+                                <goals>
+                                    <goal>single</goal>
+                                </goals>
+                                <!-- this is used for inheritance merges -->
+                                <phase>package</phase>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
+</project>

iotdb-api/external-service-api/src/main/java/org/apache/iotdb/externalservice/api/IExternalService.java

@@ -0,0 +1,33 @@
+/*
+ * 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.externalservice.api;
+
+/** An interface to support user-defined service. */
+public interface IExternalService {
+
+  /** Start current service. */
+  void start();
+
+  /**
+   * Stop current service. If current service uses thread or thread pool, current service should
+   * guarantee to putBack thread or thread pool.
+   */
+  void stop();
+}

iotdb-api/pom.xml

@@ -31,6 +31,7 @@
     <name>IoTDB: API</name>
     <modules>
         <module>external-api</module>
+        <module>external-service-api</module>
         <module>pipe-api</module>
         <module>trigger-api</module>
         <module>udf-api</module>

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -338,6 +338,11 @@ public enum TSStatusCode {
   RATIS_READ_UNAVAILABLE(2207),
   PIPE_CONSENSUS_CLOSE_ERROR(2208),
   PIPE_CONSENSUS_WAIT_ORDER_TIMEOUT(2209),
+
+  // ExternalService
+  NO_SUCH_EXTERNAL_SERVICE(2300),
+  EXTERNAL_SERVICE_ALREADY_EXIST(2301),
+  GET_BUILTIN_EXTERNAL_SERVICE_ERROR(2302),
   ;
 
   private final int statusCode;

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4

@@ -115,6 +115,7 @@ keyWords
     | FIRST
     | FLUSH
     | FOR
+    | FORCEDLY
     | FROM
     | FULL
     | FUNCTION
@@ -214,6 +215,8 @@ keyWords
     | SECURITY
     | SELECT
     | SERIESSLOTID
+    | SERVICE
+    | SERVICES
     | SESSION
     | SET
     | SETTLE

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -52,6 +52,8 @@ ddlStatement
     | createFunction | dropFunction | showFunctions
     // Trigger
     | createTrigger | dropTrigger | showTriggers | startTrigger | stopTrigger
+    // ExternalService
+    | createService | startService | stopService | dropService | showService
     // Pipe Task
     | createPipe | alterPipe | dropPipe | startPipe | stopPipe | showPipes
     // Pipe Plugin
@@ -437,6 +439,28 @@ stopTrigger
     : STOP TRIGGER triggerName=identifier
     ;
 
+// ExternalService =========================================================================================
+createService
+    : CREATE SERVICE serviceName=identifier
+        AS className=STRING_LITERAL
+    ;
+
+startService
+    : START SERVICE serviceName=identifier
+    ;
+
+stopService
+    : STOP SERVICE serviceName=identifier
+    ;
+
+dropService
+    : DROP SERVICE serviceName=identifier FORCEDLY?
+
+    ;
+
+showService
+    : SHOW SERVICES (ON targetDataNodeId=INTEGER_LITERAL)?
+    ;
 
 // CQ ==============================================================================================
 // ---- Create Continuous Query

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4

@@ -382,6 +382,11 @@ FOR
     : F O R
     ;
 
+FORCEDLY
+    : F O R C E D L Y
+    ;
+
+
 FROM
     : F R O M
     ;
@@ -786,6 +791,14 @@ SERIESSLOTID
     : S E R I E S S L O T I D
     ;
 
+SERVICE
+    : S E R V I C E
+    ;
+
+SERVICES
+    : S E R V I C E S
+    ;
+
 SESSION
     : S E S S I O N
     ;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java

@@ -60,6 +60,9 @@ public enum CnToDnAsyncRequestType {
   INACTIVE_TRIGGER_INSTANCE,
   UPDATE_TRIGGER_LOCATION,
 
+  // ExternalService
+  GET_BUILTIN_SERVICE,
+
   // Pipe Plugin
   CREATE_PIPE_PLUGIN,
   DROP_PIPE_PLUGIN,

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java

@@ -38,6 +38,7 @@
 import org.apache.iotdb.confignode.client.async.handlers.rpc.DataNodeAsyncRequestRPCHandler;
 import org.apache.iotdb.confignode.client.async.handlers.rpc.DataNodeTSStatusRPCHandler;
 import org.apache.iotdb.confignode.client.async.handlers.rpc.FetchSchemaBlackListRPCHandler;
+import org.apache.iotdb.confignode.client.async.handlers.rpc.GetBuiltInExternalServiceRPCHandler;
 import org.apache.iotdb.confignode.client.async.handlers.rpc.PipeHeartbeatRPCHandler;
 import org.apache.iotdb.confignode.client.async.handlers.rpc.PipePushMetaRPCHandler;
 import org.apache.iotdb.confignode.client.async.handlers.rpc.SchemaUpdateRPCHandler;
@@ -486,6 +487,10 @@ protected void initActionMapBuilder() {
         CnToDnAsyncRequestType.ENABLE_SEPARATION_OF_ADMIN_POWERS,
         (req, client, handler) ->
             client.enableSeparationOfAdminPower((DataNodeTSStatusRPCHandler) handler));
+    actionMapBuilder.put(
+        CnToDnAsyncRequestType.GET_BUILTIN_SERVICE,
+        (req, client, handler) ->
+            client.getBuiltInService((GetBuiltInExternalServiceRPCHandler) handler));
   }
 
   @Override

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.confignode.client.async.handlers.rpc;
 
 import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
 import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.common.rpc.thrift.TTestConnectionResp;
@@ -193,6 +194,14 @@ public static DataNodeAsyncRequestRPCHandler<?> buildHandler(
             dataNodeLocationMap,
             (Map<Integer, TDeviceViewResp>) responseMap,
             countDownLatch);
+      case GET_BUILTIN_SERVICE:
+        return new GetBuiltInExternalServiceRPCHandler(
+            requestType,
+            requestId,
+            targetDataNode,
+            dataNodeLocationMap,
+            (Map<Integer, TExternalServiceListResp>) responseMap,
+            countDownLatch);
       case SET_TTL:
       case CREATE_DATA_REGION:
       case CREATE_SCHEMA_REGION:

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/GetBuiltInExternalServiceRPCHandler.java

@@ -0,0 +1,82 @@
+/*
+ * 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.client.async.handlers.rpc;
+
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp;
+import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+public class GetBuiltInExternalServiceRPCHandler
+    extends DataNodeAsyncRequestRPCHandler<TExternalServiceListResp> {
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(GetBuiltInExternalServiceRPCHandler.class);
+
+  public GetBuiltInExternalServiceRPCHandler(
+      CnToDnAsyncRequestType requestType,
+      int requestId,
+      TDataNodeLocation targetDataNode,
+      Map<Integer, TDataNodeLocation> dataNodeLocationMap,
+      Map<Integer, TExternalServiceListResp> responseMap,
+      CountDownLatch countDownLatch) {
+    super(requestType, requestId, targetDataNode, dataNodeLocationMap, responseMap, countDownLatch);
+  }
+
+  @Override
+  public void onComplete(TExternalServiceListResp response) {
+    // Put response only when success
+    if (response.getStatus().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      responseMap.put(requestId, response);
+    } else {
+      LOGGER.error(
+          "Failed to {} on DataNode: {}, response: {}",
+          requestType,
+          formattedTargetLocation,
+          response);
+    }
+
+    // Always remove to avoid retrying
+    nodeLocationMap.remove(requestId);
+
+    // Always CountDown
+    countDownLatch.countDown();
+  }
+
+  @Override
+  public void onError(Exception e) {
+    String errorMsg =
+        "Failed to "
+            + requestType
+            + " on DataNode: "
+            + formattedTargetLocation
+            + ", exception: "
+            + e.getMessage();
+    LOGGER.error(errorMsg, e);
+
+    // Always CountDown
+    countDownLatch.countDown();
+  }
+}

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java

@@ -46,6 +46,10 @@
 import org.apache.iotdb.confignode.consensus.request.write.datanode.RegisterDataNodePlan;
 import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
 import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
+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.request.write.function.CreateFunctionPlan;
 import org.apache.iotdb.confignode.consensus.request.write.function.DropTableModelFunctionPlan;
 import org.apache.iotdb.confignode.consensus.request.write.function.DropTreeModelFunctionPlan;
@@ -588,6 +592,18 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
         case setThrottleQuota:
           plan = new SetThrottleQuotaPlan();
           break;
+        case CreateExternalService:
+          plan = new CreateExternalServicePlan();
+          break;
+        case StartExternalService:
+          plan = new StartExternalServicePlan();
+          break;
+        case StopExternalService:
+          plan = new StopExternalServicePlan();
+          break;
+        case DropExternalService:
+          plan = new DropExternalServicePlan();
+          break;
         default:
           throw new IOException("unknown PhysicalPlan configPhysicalPlanType: " + planType);
       }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java

@@ -335,6 +335,12 @@ public enum ConfigPhysicalPlanType {
 
   EnableSeparationOfAdminPowers((short) 2200),
 
+  CreateExternalService((short) 2301),
+  StartExternalService((short) 2302),
+  StopExternalService((short) 2303),
+  DropExternalService((short) 2304),
+  ShowExternalService((short) 2305),
+
   /** Test Only. */
   TestOnly((short) 30000),
   ;