[IOTDB-2880]Add procedure framework (#5477)

[IOTDB-2880]Add procedure framework (#5477)

node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java

@@ -81,7 +81,8 @@ public enum ThreadName {
   DATA_BLOCK_MANAGER_RPC_CLIENT("DataBlockManagerRPC-Client"),
   INTERNAL_SERVICE_RPC_SERVER("InternalServiceRPC"),
   INTERNAL_SERVICE_RPC_CLIENT("InternalServiceRPC-Client"),
-  ;
+  PROCEDURE_NODE_SERVER("ProcedureNode-Server"),
+  PROCEDURE_NODE_CLIENT("ProcedureNode-Client");
 
   private final String name;
 

node-commons/src/main/java/org/apache/iotdb/commons/service/ServiceType.java

@@ -71,7 +71,8 @@ public enum ServiceType {
   DATA_NODE_MANAGEMENT_SERVICE("Data Node management service", "DataNodeManagementServer"),
   FRAGMENT_INSTANCE_MANAGER_SERVICE("Fragment instance manager", "FragmentInstanceManager"),
   DATA_BLOCK_MANAGER_SERVICE("Data block manager", "DataBlockManager"),
-  INTERNAL_SERVICE("Internal Service", "InternalService");
+  INTERNAL_SERVICE("Internal Service", "InternalService"),
+  PROCEDURE_SERVICE("Procedure  Service", "ProcedureService");
 
   private final String name;
   private final String jmxName;

pom.xml

@@ -87,6 +87,7 @@
         <module>thrift-cluster</module>
         <module>thrift-sync</module>
         <module>thrift-influxdb</module>
+        <module>thrift-procedure</module>
         <module>service-rpc</module>
         <module>jdbc</module>
         <module>influxdb-protocol</module>
@@ -115,6 +116,7 @@
         <module>metrics</module>
         <module>integration</module>
         <module>consensus</module>
+        <module>procedure</module>
         <!--        <module>library-udf</module>-->
     </modules>
     <!-- Properties Management -->

procedure/pom.xml

@@ -0,0 +1,93 @@
+<?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-parent</artifactId>
+        <version>0.14.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>iotdb-procedure</artifactId>
+    <name>procedure</name>
+    <dependencies>
+        <!-- The version of thrift is overridden because using 0.13.0 in the cluster module
+        will cause unclear bugs -->
+        <dependency>
+            <groupId>org.apache.thrift</groupId>
+            <artifactId>libthrift</artifactId>
+            <version>0.14.1</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iotdb</groupId>
+            <artifactId>iotdb-thrift-procedure</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.thrift</groupId>
+                    <artifactId>libthrift</artifactId>
+                </exclusion>
+            </exclusions>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.iotdb</groupId>
+            <artifactId>node-commons</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>commons-cli</groupId>
+            <artifactId>commons-cli</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>4.0.2</version>
+            <scope>test</scope>
+        </dependency>
+        <!-- for mocked test-->
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-module-junit4</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.powermock</groupId>
+            <artifactId>powermock-api-mockito2</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-pool2</artifactId>
+        </dependency>
+    </dependencies>
+</project>

procedure/src/main/java/org/apache/iotdb/procedure/CompletedProcedureCleaner.java

@@ -0,0 +1,106 @@
+/*
+ * 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.procedure;
+
+import org.apache.iotdb.procedure.conf.ProcedureNodeConfigDescriptor;
+import org.apache.iotdb.procedure.store.IProcedureStore;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Internal cleaner that removes the completed procedure results after a TTL.
+ *
+ * <p>NOTE: This is a special case handled in timeoutLoop().
+ *
+ * <p>Since the client code looks more or less like:
+ *
+ * <pre>
+ *   procId = master.doOperation()
+ *   while (master.getProcResult(procId) == ProcInProgress);
+ * </pre>
+ *
+ * The master should not throw away the proc result as soon as the procedure is done but should wait
+ * a result request from the client (see executor.removeResult(procId)) The client will call
+ * something like master.isProcDone() or master.getProcResult() which will return the result/state
+ * to the client, and it will mark the completed proc as ready to delete. note that the client may
+ * not receive the response from the master (e.g. master failover) so, if we delay a bit the real
+ * deletion of the proc result the client will be able to get the result the next try.
+ */
+public class CompletedProcedureCleaner<Env> extends InternalProcedure<Env> {
+  private static final Logger LOG = LoggerFactory.getLogger(CompletedProcedureCleaner.class);
+
+  static final long CLEANER_INTERVAL =
+      ProcedureNodeConfigDescriptor.getInstance().getConf().getCompletedCleanInterval();
+  private static final int DEFAULT_BATCH_SIZE = 32;
+
+  private final Map<Long, CompletedProcedureRetainer<Env>> completed;
+  private final IProcedureStore store;
+
+  public CompletedProcedureCleaner(
+      IProcedureStore store, Map<Long, CompletedProcedureRetainer<Env>> completedMap) {
+    super(TimeUnit.SECONDS.toMillis(CLEANER_INTERVAL));
+    this.completed = completedMap;
+    this.store = store;
+  }
+
+  @Override
+  protected void periodicExecute(final Env env) {
+    if (completed.isEmpty()) {
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("No completed procedures to cleanup.");
+      }
+      return;
+    }
+
+    final long evictTtl = ProcedureExecutor.EVICT_TTL;
+    final long[] batchIds = new long[DEFAULT_BATCH_SIZE];
+    int batchCount = 0;
+
+    final long now = System.currentTimeMillis();
+    final Iterator<Map.Entry<Long, CompletedProcedureRetainer<Env>>> it =
+        completed.entrySet().iterator();
+    while (it.hasNext() && store.isRunning()) {
+      final Map.Entry<Long, CompletedProcedureRetainer<Env>> entry = it.next();
+      final CompletedProcedureRetainer<Env> retainer = entry.getValue();
+      final Procedure<?> proc = retainer.getProcedure();
+      if (retainer.isExpired(now, evictTtl)) {
+        // Failed procedures aren't persisted in WAL.
+        batchIds[batchCount++] = entry.getKey();
+        if (batchCount == batchIds.length) {
+          store.delete(batchIds, 0, batchCount);
+          batchCount = 0;
+        }
+        it.remove();
+        LOG.trace("Evict completed {}", proc);
+      }
+    }
+    if (batchCount > 0) {
+      store.delete(batchIds, 0, batchCount);
+    }
+    // let the store do some cleanup works, i.e, delete the place marker for preserving the max
+    // procedure id.
+    store.cleanup();
+  }
+}

procedure/src/main/java/org/apache/iotdb/procedure/CompletedProcedureRetainer.java

@@ -0,0 +1,36 @@
+/*
+ * 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.procedure;
+
+public class CompletedProcedureRetainer<Env> {
+  private final Procedure<Env> procedure;
+
+  public CompletedProcedureRetainer(Procedure<Env> procedure) {
+    this.procedure = procedure;
+  }
+
+  public Procedure<Env> getProcedure() {
+    return procedure;
+  }
+
+  public boolean isExpired(long now, long evictTtl) {
+    return (now - procedure.getLastUpdate()) >= evictTtl;
+  }
+}

procedure/src/main/java/org/apache/iotdb/procedure/InternalProcedure.java

@@ -0,0 +1,61 @@
+/*
+ * 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.procedure;
+
+import org.apache.iotdb.procedure.exception.ProcedureSuspendedException;
+import org.apache.iotdb.procedure.exception.ProcedureYieldException;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Internal Procedure, do some preiodic job for framework
+ *
+ * @param <Env>
+ */
+public abstract class InternalProcedure<Env> extends Procedure<Env> {
+  public InternalProcedure(long toMillis) {
+    setTimeout(toMillis);
+  }
+
+  protected abstract void periodicExecute(final Env env);
+
+  @Override
+  protected Procedure<Env>[] execute(Env env)
+      throws ProcedureYieldException, ProcedureSuspendedException, InterruptedException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  protected void rollback(Env env) throws IOException, InterruptedException {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  protected boolean abort(Env env) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public void serialize(ByteBuffer byteBuffer) throws IOException {}
+
+  @Override
+  public void deserialize(ByteBuffer byteBuffer) throws IOException {}
+}