[DRAFT] IoTDB Collector Node

iotdb-collector/collector-core/pom.xml

@@ -0,0 +1,62 @@
+<?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/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.iotdb</groupId>
+        <artifactId>iotdb-collector</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>collector-core</artifactId>
+    <name>IoTDB: Collector: Core</name>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.iotdb</groupId>
+            <artifactId>collector-openapi</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-http</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-util</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-servlet</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.glassfish.jersey.containers</groupId>
+            <artifactId>jersey-container-servlet-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.glassfish.jersey.inject</groupId>
+            <artifactId>jersey-hk2</artifactId>
+        </dependency>
+    </dependencies>
+</project>

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

@@ -0,0 +1,47 @@
+/*
+ * 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;
+
+import org.apache.iotdb.collector.config.CollectorConfig;
+import org.apache.iotdb.collector.config.CollectorDescriptor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.Charset;
+
+public class Application {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(Application.class);
+
+  private static final CollectorConfig COLLECTOR_CONFIG =
+      CollectorDescriptor.getInstance().getConfig();
+
+  public static void main(String[] args) {
+    LOGGER.info("Environment variables: {}", CollectorConfig.getEnvironmentVariables());
+    LOGGER.info("Default charset is: {}", Charset.defaultCharset().displayName());
+
+    final Application application = new Application();
+
+    LOGGER.info("IoTDB-CollectorNode configuration: {}", COLLECTOR_CONFIG.getConfigMessage());
+    LOGGER.info(
+        "Congratulations, IoTDB CollectorNode is set up successfully. Now, enjoy yourself!");
+  }
+}

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/CollectorAgent.java

@@ -0,0 +1,50 @@
+/*
+ * 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.agent;
+
+import org.apache.iotdb.collector.agent.executor.CollectorTaskExecutorAgent;
+import org.apache.iotdb.collector.agent.plugin.CollectorPluginAgent;
+import org.apache.iotdb.collector.agent.task.CollectorTaskAgent;
+
+public class CollectorAgent {
+
+  private final CollectorTaskAgent collectorTaskAgent = CollectorTaskAgent.instance();
+  private final CollectorTaskExecutorAgent collectorTaskExecutorAgent =
+      CollectorTaskExecutorAgent.instance();
+  private final CollectorPluginAgent collectorPluginAgent = CollectorPluginAgent.instance();
+
+  private CollectorAgent() {}
+
+  public static CollectorTaskAgent task() {
+    return CollectorAgentHolder.INSTANCE.collectorTaskAgent;
+  }
+
+  public static CollectorTaskExecutorAgent executor() {
+    return CollectorAgentHolder.INSTANCE.collectorTaskExecutorAgent;
+  }
+
+  public static CollectorPluginAgent plugin() {
+    return CollectorAgentHolder.INSTANCE.collectorPluginAgent;
+  }
+
+  private static class CollectorAgentHolder {
+    private static final CollectorAgent INSTANCE = new CollectorAgent();
+  }
+}

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/collect/CollectorEventCollector.java

@@ -0,0 +1,48 @@
+/*
+ * 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.agent.collect;
+
+import org.apache.iotdb.pipe.api.collector.EventCollector;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.BlockingQueue;
+
+public class CollectorEventCollector implements EventCollector {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(CollectorEventCollector.class);
+
+  private final BlockingQueue<Event> pendingQueue;
+
+  public CollectorEventCollector(final BlockingQueue<Event> pendingQueue) {
+    this.pendingQueue = pendingQueue;
+  }
+
+  @Override
+  public void collect(final Event event) {
+    try {
+      pendingQueue.put(event);
+    } catch (final InterruptedException e) {
+      LOGGER.warn("collect event failed because {}", e.getMessage(), e);
+    }
+  }
+}

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorProcessorTaskExecutor.java

@@ -0,0 +1,68 @@
+/*
+ * 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.agent.executor;
+
+import org.apache.iotdb.collector.agent.task.CollectorTask;
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+public class CollectorProcessorTaskExecutor extends CollectorTaskExecutor {
+
+  private static final Logger LOGGER =
+      LoggerFactory.getLogger(CollectorProcessorTaskExecutor.class);
+
+  private static final Map<String, ExecutorService> PROCESSOR_EXECUTOR = new ConcurrentHashMap<>();
+  private static final Map<String, CollectorTask> PROCESSOR_TASK_MAP = new ConcurrentHashMap<>();
+
+  public boolean validateIfAbsent(final String taskId) {
+    return !PROCESSOR_EXECUTOR.containsKey(taskId) && !PROCESSOR_TASK_MAP.containsKey(taskId);
+  }
+
+  @Override
+  public Optional<ExecutorService> getExecutor(final String taskId) {
+    return Optional.of(
+        IoTDBThreadPoolFactory.newSingleThreadExecutor("collector-processor-executor-" + taskId));
+  }
+
+  @Override
+  public void recordExecution(
+      final CollectorTask collectorTask, final ExecutorService executorService) {
+    final String taskId = collectorTask.getTaskId();
+    PROCESSOR_EXECUTOR.putIfAbsent(taskId, executorService);
+    PROCESSOR_TASK_MAP.putIfAbsent(taskId, collectorTask);
+
+    LOGGER.info("register collector processor task {}", taskId);
+  }
+
+  @Override
+  public void eraseExecution(final String taskId) {
+    PROCESSOR_TASK_MAP.remove(taskId).stop();
+    PROCESSOR_EXECUTOR.remove(taskId).shutdownNow();
+
+    LOGGER.info("deregister collector processor task {}", taskId);
+  }
+}

iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorSinkTaskExecutor.java

@@ -0,0 +1,67 @@
+/*
+ * 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.agent.executor;
+
+import org.apache.iotdb.collector.agent.task.CollectorTask;
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+public class CollectorSinkTaskExecutor extends CollectorTaskExecutor {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(CollectorSinkTaskExecutor.class);
+
+  private static final Map<String, ExecutorService> SINK_EXECUTOR = new ConcurrentHashMap<>();
+  private static final Map<String, CollectorTask> SINK_TASK_MAP = new ConcurrentHashMap<>();
+
+  public boolean validateIfAbsent(final String taskId) {
+    return !SINK_EXECUTOR.containsKey(taskId) && !SINK_TASK_MAP.containsKey(taskId);
+  }
+
+  @Override
+  public Optional<ExecutorService> getExecutor(final String taskId) {
+    return Optional.of(
+        IoTDBThreadPoolFactory.newSingleThreadExecutor("collector-sink-executor-" + taskId));
+  }
+
+  @Override
+  public void recordExecution(
+      final CollectorTask collectorTask, final ExecutorService executorService) {
+    final String taskId = collectorTask.getTaskId();
+    SINK_EXECUTOR.putIfAbsent(taskId, executorService);
+    SINK_TASK_MAP.putIfAbsent(taskId, collectorTask);
+
+    LOGGER.info("register collector sink task {}", taskId);
+  }
+
+  @Override
+  public void eraseExecution(final String taskId) {
+    SINK_TASK_MAP.remove(taskId).stop();
+    SINK_EXECUTOR.remove(taskId).shutdownNow();
+
+    LOGGER.info("deregister collector sink task {}", taskId);
+  }
+}