Collector: Fixed task not being able to be dropped when the task is stopped & Modify PipeParameters & Supports registration of plugins and reflection usage & Add SQLite support for persistence of plugin and task meta. (#57)

Author: YC27

iotdb-collector/collector-core/pom.xml

@@ -107,6 +107,10 @@
             <groupId>com.google.code.findbugs</groupId>
             <artifactId>jsr305</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.xerial</groupId>
+            <artifactId>sqlite-jdbc</artifactId>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/Application.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.collector.config.Configuration;
 import org.apache.iotdb.collector.service.ApiService;
 import org.apache.iotdb.collector.service.IService;
+import org.apache.iotdb.collector.service.PersistenceService;
 import org.apache.iotdb.collector.service.RuntimeService;
 
 import org.slf4j.Logger;
@@ -39,6 +40,7 @@ public class Application {
   private Application() {
     services.add(new RuntimeService());
     services.add(new ApiService());
+    services.add(new PersistenceService());
   }
 
   public static void main(String[] args) {

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceImpl.java

@@ -20,11 +20,9 @@
 package org.apache.iotdb.collector.api.v1.plugin.impl;
 
 import org.apache.iotdb.collector.api.v1.plugin.PluginApiService;
-import org.apache.iotdb.collector.api.v1.plugin.model.AlterPluginRequest;
 import org.apache.iotdb.collector.api.v1.plugin.model.CreatePluginRequest;
 import org.apache.iotdb.collector.api.v1.plugin.model.DropPluginRequest;
-import org.apache.iotdb.collector.api.v1.plugin.model.StartPluginRequest;
-import org.apache.iotdb.collector.api.v1.plugin.model.StopPluginRequest;
+import org.apache.iotdb.collector.service.RuntimeService;
 
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.SecurityContext;
@@ -34,30 +32,34 @@ public class PluginApiServiceImpl extends PluginApiService {
   @Override
   public Response createPlugin(
       final CreatePluginRequest createPluginRequest, final SecurityContext securityContext) {
-    return Response.ok("create plugin").build();
-  }
+    PluginApiServiceRequestValidationHandler.validateCreatePluginRequest(createPluginRequest);
 
-  @Override
-  public Response alterPlugin(
-      final AlterPluginRequest alterPluginRequest, final SecurityContext securityContext) {
-    return Response.ok("alter plugin").build();
+    return RuntimeService.plugin().isPresent()
+        ? RuntimeService.plugin()
+            .get()
+            .createPlugin(
+                createPluginRequest.getPluginName().toUpperCase(),
+                createPluginRequest.getClassName(),
+                createPluginRequest.getJarName(),
+                null,
+                true)
+        : Response.ok("create plugin").build();
   }
 
   @Override
-  public Response startPlugin(
-      final StartPluginRequest startPluginRequest, final SecurityContext securityContext) {
-    return Response.ok("start plugin").build();
-  }
+  public Response dropPlugin(
+      final DropPluginRequest dropPluginRequest, final SecurityContext securityContext) {
+    PluginApiServiceRequestValidationHandler.validateDropPluginRequest(dropPluginRequest);
 
-  @Override
-  public Response stopPlugin(
-      final StopPluginRequest stopPluginRequest, final SecurityContext securityContext) {
-    return Response.ok("stop plugin").build();
+    return RuntimeService.plugin().isPresent()
+        ? RuntimeService.plugin().get().dropPlugin(dropPluginRequest.getPluginName().toUpperCase())
+        : Response.ok("drop plugin").build();
   }
 
   @Override
-  public Response dropPlugin(
-      final DropPluginRequest dropPluginRequest, final SecurityContext securityContext) {
-    return Response.ok("drop plugin").build();
+  public Response showPlugin(final SecurityContext securityContext) {
+    return RuntimeService.plugin().isPresent()
+        ? RuntimeService.plugin().get().showPlugin()
+        : Response.ok("show plugin").build();
   }
 }

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceRequestValidationHandler.java

@@ -19,4 +19,21 @@
 
 package org.apache.iotdb.collector.api.v1.plugin.impl;
 
-public class PluginApiServiceRequestValidationHandler {}
+import org.apache.iotdb.collector.api.v1.plugin.model.CreatePluginRequest;
+import org.apache.iotdb.collector.api.v1.plugin.model.DropPluginRequest;
+
+import java.util.Objects;
+
+public class PluginApiServiceRequestValidationHandler {
+  private PluginApiServiceRequestValidationHandler() {}
+
+  public static void validateCreatePluginRequest(final CreatePluginRequest createPluginRequest) {
+    Objects.requireNonNull(createPluginRequest.getPluginName(), "plugin name cannot be null");
+    Objects.requireNonNull(createPluginRequest.getClassName(), "class name cannot be null");
+    Objects.requireNonNull(createPluginRequest.getJarName(), "jar name cannot be null");
+  }
+
+  public static void validateDropPluginRequest(final DropPluginRequest dropPluginRequest) {
+    Objects.requireNonNull(dropPluginRequest.getPluginName(), "plugin name cannot be null");
+  }
+}

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceImpl.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.collector.api.v1.task.model.DropTaskRequest;
 import org.apache.iotdb.collector.api.v1.task.model.StartTaskRequest;
 import org.apache.iotdb.collector.api.v1.task.model.StopTaskRequest;
+import org.apache.iotdb.collector.runtime.task.TaskStateEnum;
 import org.apache.iotdb.collector.service.RuntimeService;
 
 import javax.ws.rs.core.Response;
@@ -42,9 +43,11 @@ public Response createTask(
             .get()
             .createTask(
                 createTaskRequest.getTaskId(),
+                TaskStateEnum.RUNNING,
                 createTaskRequest.getSourceAttribute(),
                 createTaskRequest.getProcessorAttribute(),
-                createTaskRequest.getSinkAttribute())
+                createTaskRequest.getSinkAttribute(),
+                true)
         : Response.serverError().entity("Task runtime is down").build();
   }
 

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/Options.java

@@ -36,6 +36,7 @@ public class Options {
     try {
       Class.forName(ApiServiceOptions.class.getName());
       Class.forName(TaskRuntimeOptions.class.getName());
+      Class.forName(PluginRuntimeOptions.class.getName());
     } catch (final ClassNotFoundException e) {
       throw new RuntimeException("Failed to load options", e);
     }

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/PluginRuntimeOptions.java

@@ -0,0 +1,41 @@
+/*
+ * 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.collector.config;
+
+import java.io.File;
+
+public class PluginRuntimeOptions extends Options {
+  public static final Option<String> PLUGIN_LIB_DIR =
+      new Option<String>("plugin_lib_dir", "ext" + File.separator + "plugin") {
+        @Override
+        public void setValue(final String valueString) {
+          value = valueString;
+        }
+      };
+
+  public static final Option<String> PLUGIN_INSTALL_LIB_DIR =
+      new Option<String>(
+          "plugin_install_lib_dir", PLUGIN_LIB_DIR.value() + File.separator + "install") {
+        @Override
+        public void setValue(final String valueString) {
+          value = valueString;
+        }
+      };
+}

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/persistence/DBConstant.java

@@ -0,0 +1,49 @@
+/*
+ * 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.collector.persistence;
+
+public class DBConstant {
+
+  public static final String CREATE_PLUGIN_TABLE_SQL =
+      "CREATE TABLE IF NOT EXISTS plugin\n"
+          + "(\n"
+          + "    plugin_name TEXT PRIMARY KEY,\n"
+          + "    class_name  TEXT NOT NULL,\n"
+          + "    jar_name    TEXT NOT NULL,\n"
+          + "    jar_md5    TEXT NOT NULL,\n"
+          + "    create_time TEXT NOT NULL\n"
+          + ");";
+  public static final String CREATE_TASK_TABLE_SQL =
+      "CREATE TABLE IF NOT EXISTS task\n"
+          + "(\n"
+          + "    task_id             TEXT PRIMARY KEY,\n"
+          + "    task_state          INT  NOT NULL,\n"
+          + "    source_attribute    BLOB NOT NULL,\n"
+          + "    processor_attribute BLOB NOT NULL,\n"
+          + "    sink_attribute      BLOB NOT NULL,\n"
+          + "    create_time         TEXT NOT NULL\n"
+          + ");";
+
+  public static final String PLUGIN_DATABASE_FILE_PATH = "ext/db/plugin.db";
+  public static final String TASK_DATABASE_FILE_PATH = "ext/db/task.db";
+
+  public static final String PLUGIN_DATABASE_URL = "jdbc:sqlite:" + PLUGIN_DATABASE_FILE_PATH;
+  public static final String TASK_DATABASE_URL = "jdbc:sqlite:" + TASK_DATABASE_FILE_PATH;
+}

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/persistence/Persistence.java

@@ -0,0 +1,45 @@
+/*
+ * 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.collector.persistence;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+
+public abstract class Persistence {
+
+  private final String databaseUrl;
+
+  public Persistence(final String databaseUrl) {
+    this.databaseUrl = databaseUrl;
+    initDatabaseFileIfPossible();
+    initTableIfPossible();
+  }
+
+  protected abstract void initDatabaseFileIfPossible();
+
+  protected abstract void initTableIfPossible();
+
+  public abstract void tryResume();
+
+  protected Connection getConnection() throws SQLException {
+    return DriverManager.getConnection(databaseUrl);
+  }
+}