draft

Signed-off-by: Weihao Li <[email protected]>

Author: Wei-hao-Li

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

@@ -341,8 +341,8 @@ public enum TSStatusCode {
 
   // ExternalService
   NO_SUCH_EXTERNAL_SERVICE(2300),
-  EXTERNAL_SERVICE_ALREADY_ACTIVE(2301),
-  EXTERNAL_SERVICE_ALREADY_EXIST(2302),
+  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

@@ -214,6 +214,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,27 @@ 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
+    ;
+
+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

@@ -786,6 +786,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

@@ -48,6 +48,8 @@
 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;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/read/exernalservice/ShowExternalServicePlan.java

@@ -22,23 +22,21 @@
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
 import org.apache.iotdb.confignode.consensus.request.read.ConfigPhysicalReadPlan;
 
+import java.util.List;
 import java.util.Objects;
 
-/**
- * Get infos of ExternalService by the specific DataNode's id. And return all when dataNodeID is set
- * to -1.
- */
+/** Get infos of ExternalService by the DataNode's id. */
 public class ShowExternalServicePlan extends ConfigPhysicalReadPlan {
 
-  private final int dataNodeId;
+  private final List<Integer> dataNodeIds;
 
-  public ShowExternalServicePlan(final int dataNodeId) {
+  public ShowExternalServicePlan(List<Integer> dataNodeIds) {
     super(ConfigPhysicalPlanType.GetDataNodeConfiguration);
-    this.dataNodeId = dataNodeId;
+    this.dataNodeIds = dataNodeIds;
   }
 
-  public Integer getDataNodeId() {
-    return dataNodeId;
+  public List<Integer> getDataNodeIds() {
+    return dataNodeIds;
   }
 
   @Override
@@ -50,11 +48,11 @@ public boolean equals(final Object o) {
       return false;
     }
     final ShowExternalServicePlan that = (ShowExternalServicePlan) o;
-    return dataNodeId == that.dataNodeId;
+    return dataNodeIds.equals(that.dataNodeIds);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(dataNodeId);
+    return Objects.hash(dataNodeIds);
   }
 }