Collector: Refactoring the three phases of data collection tasks using disruptor (#48)
diff --git a/.gitignore b/.gitignore index e8ef7a9..dced3d9 100644 --- a/.gitignore +++ b/.gitignore
@@ -4,4 +4,7 @@ # IntelliJ relaged .idea/ -*.iml \ No newline at end of file +*.iml + +# log +logs/ \ No newline at end of file
diff --git a/iotdb-collector/collector-core/pom.xml b/iotdb-collector/collector-core/pom.xml index ba372d0..f6ed16d 100644 --- a/iotdb-collector/collector-core/pom.xml +++ b/iotdb-collector/collector-core/pom.xml
@@ -99,6 +99,10 @@ <groupId>org.glassfish.jersey.inject</groupId> <artifactId>jersey-hk2</artifactId> </dependency> + <dependency> + <groupId>com.lmax</groupId> + <artifactId>disruptor</artifactId> + </dependency> </dependencies> <build> <plugins>
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/Application.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/Application.java index 6cf6456..96ecdaf 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/Application.java +++ b/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.RuntimeService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +37,7 @@ private final LinkedList<IService> services = new LinkedList<>(); private Application() { + services.add(new RuntimeService()); services.add(new ApiService()); }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/CollectorAgent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/CollectorAgent.java deleted file mode 100644 index bca4106..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/CollectorAgent.java +++ /dev/null
@@ -1,50 +0,0 @@ -/* - * 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(); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorProcessorTaskExecutor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorProcessorTaskExecutor.java deleted file mode 100644 index 385fd60..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorProcessorTaskExecutor.java +++ /dev/null
@@ -1,68 +0,0 @@ -/* - * 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); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorSinkTaskExecutor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorSinkTaskExecutor.java deleted file mode 100644 index a490e61..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorSinkTaskExecutor.java +++ /dev/null
@@ -1,67 +0,0 @@ -/* - * 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); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorSourceTaskExecutor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorSourceTaskExecutor.java deleted file mode 100644 index fde8236..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorSourceTaskExecutor.java +++ /dev/null
@@ -1,67 +0,0 @@ -/* - * 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 CollectorSourceTaskExecutor extends CollectorTaskExecutor { - - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorSourceTaskExecutor.class); - - private static final Map<String, ExecutorService> SOURCE_EXECUTOR = new ConcurrentHashMap<>(); - private static final Map<String, CollectorTask> SOURCE_TASK_MAP = new ConcurrentHashMap<>(); - - public boolean validateIfAbsent(final String taskId) { - return !SOURCE_EXECUTOR.containsKey(taskId) && !SOURCE_TASK_MAP.containsKey(taskId); - } - - @Override - public Optional<ExecutorService> getExecutor(final String taskId) { - return Optional.of( - IoTDBThreadPoolFactory.newSingleThreadExecutor("collector-source-executor-" + taskId)); - } - - @Override - public void recordExecution( - final CollectorTask collectorTask, final ExecutorService executorService) { - final String taskId = collectorTask.getTaskId(); - SOURCE_EXECUTOR.put(taskId, executorService); - SOURCE_TASK_MAP.putIfAbsent(taskId, collectorTask); - - LOGGER.info("register collector source task {}", taskId); - } - - @Override - public void eraseExecution(String taskId) { - SOURCE_TASK_MAP.remove(taskId).stop(); - SOURCE_EXECUTOR.remove(taskId).shutdownNow(); - - LOGGER.info("deregister collector source task {}", taskId); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorTaskExecutor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorTaskExecutor.java deleted file mode 100644 index 8100766..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorTaskExecutor.java +++ /dev/null
@@ -1,64 +0,0 @@ -/* - * 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.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Optional; -import java.util.concurrent.ExecutorService; - -public abstract class CollectorTaskExecutor { - - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorTaskExecutor.class); - - public void register(final CollectorTask collectorTask) { - if (validateIfAbsent(collectorTask.getTaskId())) { - getExecutor(collectorTask.getTaskId()) - .ifPresent( - executor -> { - executor.submit(collectorTask); - LOGGER.info("register success {}", collectorTask.getTaskId()); - recordExecution(collectorTask, executor); - }); - } else { - LOGGER.warn("task {} has existed", collectorTask.getTaskId()); - } - } - - public abstract boolean validateIfAbsent(final String taskId); - - public abstract Optional<ExecutorService> getExecutor(final String taskId); - - public abstract void recordExecution( - final CollectorTask collectorTask, final ExecutorService executorService); - - public void deregister(final String taskId) { - if (!validateIfAbsent(taskId)) { - eraseExecution(taskId); - } else { - LOGGER.warn("task {} has not existed", taskId); - } - } - - public abstract void eraseExecution(final String taskId); -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorTaskExecutorAgent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorTaskExecutorAgent.java deleted file mode 100644 index 5adcace..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/executor/CollectorTaskExecutorAgent.java +++ /dev/null
@@ -1,53 +0,0 @@ -/* - * 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; - -public class CollectorTaskExecutorAgent { - - private final CollectorSourceTaskExecutor sourceTaskExecutor; - private final CollectorProcessorTaskExecutor processorTaskExecutor; - private final CollectorSinkTaskExecutor sinkTaskExecutor; - - private CollectorTaskExecutorAgent() { - sourceTaskExecutor = new CollectorSourceTaskExecutor(); - processorTaskExecutor = new CollectorProcessorTaskExecutor(); - sinkTaskExecutor = new CollectorSinkTaskExecutor(); - } - - public CollectorSourceTaskExecutor getSourceTaskExecutor() { - return CollectorTaskExecutorAgentHolder.INSTANCE.sourceTaskExecutor; - } - - public CollectorProcessorTaskExecutor getProcessorTaskExecutor() { - return CollectorTaskExecutorAgentHolder.INSTANCE.processorTaskExecutor; - } - - public CollectorSinkTaskExecutor getSinkTaskExecutor() { - return CollectorTaskExecutorAgentHolder.INSTANCE.sinkTaskExecutor; - } - - public static CollectorTaskExecutorAgent instance() { - return CollectorTaskExecutorAgentHolder.INSTANCE; - } - - private static class CollectorTaskExecutorAgentHolder { - private static final CollectorTaskExecutorAgent INSTANCE = new CollectorTaskExecutorAgent(); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/plugin/CollectorPluginAgent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/plugin/CollectorPluginAgent.java deleted file mode 100644 index 33ee5ca..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/plugin/CollectorPluginAgent.java +++ /dev/null
@@ -1,39 +0,0 @@ -/* - * 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.plugin; - -public class CollectorPluginAgent { - private final CollectorPluginConstructor collectorPluginConstructor = - CollectorPluginConstructor.instance(); - - private CollectorPluginAgent() {} - - public CollectorPluginConstructor constructor() { - return CollectorPluginAgentHolder.INSTANCE.collectorPluginConstructor; - } - - public static CollectorPluginAgent instance() { - return new CollectorPluginAgent(); - } - - private static class CollectorPluginAgentHolder { - private static final CollectorPluginAgent INSTANCE = new CollectorPluginAgent(); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/plugin/CollectorPluginConstructor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/plugin/CollectorPluginConstructor.java deleted file mode 100644 index cce94f6..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/plugin/CollectorPluginConstructor.java +++ /dev/null
@@ -1,77 +0,0 @@ -/* - * 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.plugin; - -import org.apache.iotdb.collector.plugin.BuiltinCollectorPlugin; -import org.apache.iotdb.collector.plugin.builtin.processor.DoNothingProcessor; -import org.apache.iotdb.collector.plugin.builtin.sink.SessionSink; -import org.apache.iotdb.collector.plugin.builtin.source.HttpSource; -import org.apache.iotdb.pipe.api.PipePlugin; -import org.apache.iotdb.pipe.api.PipeProcessor; -import org.apache.iotdb.pipe.api.PipeSink; -import org.apache.iotdb.pipe.api.PipeSource; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.HashMap; -import java.util.Map; -import java.util.function.Supplier; - -public class CollectorPluginConstructor { - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorPluginConstructor.class); - - protected final Map<String, Supplier<PipePlugin>> pluginConstructors = new HashMap<>(); - - private CollectorPluginConstructor() { - initConstructors(); - } - - private void initConstructors() { - pluginConstructors.put( - BuiltinCollectorPlugin.HTTP_SOURCE.getCollectorPluginName(), HttpSource::new); - pluginConstructors.put( - BuiltinCollectorPlugin.DO_NOTHING_PROCESSOR.getCollectorPluginName(), - DoNothingProcessor::new); - pluginConstructors.put( - BuiltinCollectorPlugin.IOTDB_SESSION_SINK.getCollectorPluginName(), SessionSink::new); - LOGGER.info("builtin plugin has been initialized"); - } - - public PipeSource getSource(final String pluginName) { - return (PipeSource) pluginConstructors.get(pluginName).get(); - } - - public PipeProcessor getProcessor(final String pluginName) { - return (PipeProcessor) pluginConstructors.get(pluginName).get(); - } - - public PipeSink getSink(final String pluginName) { - return (PipeSink) pluginConstructors.get(pluginName).get(); - } - - public static CollectorPluginConstructor instance() { - return CollectorPluginConstructorHolder.INSTANCE; - } - - private static class CollectorPluginConstructorHolder { - private static final CollectorPluginConstructor INSTANCE = new CollectorPluginConstructor(); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorProcessorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorProcessorTask.java deleted file mode 100644 index 8d1acfc..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorProcessorTask.java +++ /dev/null
@@ -1,88 +0,0 @@ -/* - * 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.task; - -import org.apache.iotdb.collector.agent.collect.CollectorEventCollector; -import org.apache.iotdb.commons.pipe.task.EventSupplier; -import org.apache.iotdb.pipe.api.PipeProcessor; -import org.apache.iotdb.pipe.api.event.Event; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Map; -import java.util.concurrent.BlockingQueue; - -public class CollectorProcessorTask extends CollectorTask { - - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorProcessorTask.class); - - private final Map<String, String> processorAttribute; - private final PipeProcessor pipeProcessor; - private final EventSupplier eventSupplier; - private final BlockingQueue<Event> pendingQueue; - private final CollectorEventCollector collectorEventCollector; - private boolean isStarted = true; - - public CollectorProcessorTask( - final String taskId, - final Map<String, String> processorAttribute, - final PipeProcessor pipeProcessor, - final EventSupplier eventSupplier, - final BlockingQueue<Event> pendingQueue) { - super(taskId); - this.processorAttribute = processorAttribute; - this.pipeProcessor = pipeProcessor; - this.eventSupplier = eventSupplier; - this.pendingQueue = pendingQueue; - this.collectorEventCollector = new CollectorEventCollector(pendingQueue); - } - - @Override - public void runMayThrow() { - while (isStarted) { - try { - pipeProcessor.process(eventSupplier.supply(), collectorEventCollector); - } catch (final Exception e) { - LOGGER.warn("error occur while processing event because {}", e.getMessage()); - } - } - } - - public Map<String, String> getProcessorAttribute() { - return processorAttribute; - } - - public PipeProcessor getPipeProcessor() { - return pipeProcessor; - } - - public EventSupplier getEventSupplier() { - return eventSupplier; - } - - public BlockingQueue<Event> getPendingQueue() { - return pendingQueue; - } - - public void stop() { - isStarted = false; - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorSinkTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorSinkTask.java deleted file mode 100644 index 77e65b4..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorSinkTask.java +++ /dev/null
@@ -1,87 +0,0 @@ -/* - * 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.task; - -import org.apache.iotdb.pipe.api.PipeSink; -import org.apache.iotdb.pipe.api.event.Event; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Map; -import java.util.concurrent.BlockingQueue; - -public class CollectorSinkTask extends CollectorTask { - - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorSinkTask.class); - - private final Map<String, String> sinkAttribute; - private final PipeSink pipeSink; - private final BlockingQueue<Event> pendingQueue; - private boolean isStarted = true; - - public CollectorSinkTask( - final String taskId, - final Map<String, String> sinkAttribute, - final PipeSink pipeSink, - final BlockingQueue<Event> pendingQueue) { - super(taskId); - this.sinkAttribute = sinkAttribute; - this.pipeSink = pipeSink; - this.pendingQueue = pendingQueue; - } - - @Override - public void runMayThrow() { - try { - pipeSink.handshake(); - } catch (final Exception e) { - LOGGER.warn("handshake fail because {}", e.getMessage()); - } - isStarted = true; - while (isStarted) { - try { - final Event event = pendingQueue.take(); - pipeSink.transfer(event); - LOGGER.info("transfer event {} success, remain number is {}", event, pendingQueue.size()); - } catch (final InterruptedException e) { - LOGGER.warn("interrupted while waiting for take a event"); - } catch (final Exception e) { - LOGGER.warn("error occur while transfer event to endpoint"); - } - } - } - - public Map<String, String> getSinkAttribute() { - return sinkAttribute; - } - - public PipeSink getPipeSink() { - return pipeSink; - } - - public void stop() { - isStarted = false; - } - - public BlockingQueue<Event> getPendingQueue() { - return pendingQueue; - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorSourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorSourceTask.java deleted file mode 100644 index 26269c3..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorSourceTask.java +++ /dev/null
@@ -1,69 +0,0 @@ -/* - * 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.task; - -import org.apache.iotdb.commons.pipe.task.EventSupplier; -import org.apache.iotdb.pipe.api.PipeSource; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Map; - -public class CollectorSourceTask extends CollectorTask { - - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorSourceTask.class); - - private final Map<String, String> sourceAttribute; - private final PipeSource pipeSource; - - public CollectorSourceTask( - final String taskId, final Map<String, String> sourceAttribute, final PipeSource pipeSource) { - super(taskId); - this.sourceAttribute = sourceAttribute; - this.pipeSource = pipeSource; - } - - @Override - public void runMayThrow() throws Throwable { - pipeSource.start(); - } - - public Map<String, String> getSourceAttribute() { - return sourceAttribute; - } - - public PipeSource getPipeSource() { - return pipeSource; - } - - public EventSupplier getEventSupplier() { - return pipeSource::supply; - } - - @Override - public void stop() { - try { - pipeSource.close(); - } catch (final Exception e) { - LOGGER.warn("failed to close pipe source {}", pipeSource, e); - } - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTaskAgent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTaskAgent.java deleted file mode 100644 index 9cd787d..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTaskAgent.java +++ /dev/null
@@ -1,117 +0,0 @@ -/* - * 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.task; - -import org.apache.iotdb.collector.agent.executor.CollectorProcessorTaskExecutor; -import org.apache.iotdb.collector.agent.executor.CollectorSinkTaskExecutor; -import org.apache.iotdb.collector.agent.executor.CollectorSourceTaskExecutor; -import org.apache.iotdb.collector.agent.executor.CollectorTaskExecutorAgent; -import org.apache.iotdb.collector.agent.plugin.CollectorPluginAgent; -import org.apache.iotdb.collector.agent.plugin.CollectorPluginConstructor; -import org.apache.iotdb.collector.plugin.BuiltinCollectorPlugin; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Map; -import java.util.concurrent.LinkedBlockingQueue; - -public class CollectorTaskAgent { - - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorTaskAgent.class); - - private static final CollectorPluginConstructor CONSTRUCTOR = - CollectorPluginAgent.instance().constructor(); - private static final CollectorSourceTaskExecutor SOURCE_TASK_EXECUTOR = - CollectorTaskExecutorAgent.instance().getSourceTaskExecutor(); - private static final CollectorProcessorTaskExecutor PROCESSOR_TASK_EXECUTOR = - CollectorTaskExecutorAgent.instance().getProcessorTaskExecutor(); - private static final CollectorSinkTaskExecutor SINK_TASK_EXECUTOR = - CollectorTaskExecutorAgent.instance().getSinkTaskExecutor(); - - private CollectorTaskAgent() {} - - public boolean createCollectorTask( - final Map<String, String> sourceAttribute, - final Map<String, String> processorAttribute, - final Map<String, String> sinkAttribute, - final String taskId) { - try { - final CollectorSourceTask collectorSourceTask = - new CollectorSourceTask( - taskId, - sourceAttribute, - CONSTRUCTOR.getSource( - sourceAttribute.getOrDefault( - "source-plugin", - BuiltinCollectorPlugin.HTTP_SOURCE.getCollectorPluginName()))); - SOURCE_TASK_EXECUTOR.register(collectorSourceTask); - - final CollectorProcessorTask collectorProcessorTask = - new CollectorProcessorTask( - taskId, - processorAttribute, - CONSTRUCTOR.getProcessor( - processorAttribute.getOrDefault( - "processor-plugin", - BuiltinCollectorPlugin.DO_NOTHING_PROCESSOR.getCollectorPluginName())), - collectorSourceTask.getEventSupplier(), - new LinkedBlockingQueue<>()); - PROCESSOR_TASK_EXECUTOR.register(collectorProcessorTask); - - final CollectorSinkTask collectorSinkTask = - new CollectorSinkTask( - taskId, - sinkAttribute, - CONSTRUCTOR.getSink( - sinkAttribute.getOrDefault( - "sink-plugin", - BuiltinCollectorPlugin.IOTDB_SESSION_SINK.getCollectorPluginName())), - collectorProcessorTask.getPendingQueue()); - SINK_TASK_EXECUTOR.register(collectorSinkTask); - } catch (final Exception e) { - LOGGER.warn("create collector task error", e); - return false; - } - - return true; - } - - public boolean stopCollectorTask(final String taskId) { - try { - SOURCE_TASK_EXECUTOR.deregister(taskId); - PROCESSOR_TASK_EXECUTOR.deregister(taskId); - SINK_TASK_EXECUTOR.deregister(taskId); - } catch (final Exception e) { - LOGGER.warn("stop collector task error", e); - return false; - } - - return true; - } - - public static CollectorTaskAgent instance() { - return CollectorTaskAgentHolder.INSTANCE; - } - - private static class CollectorTaskAgentHolder { - private static final CollectorTaskAgent INSTANCE = new CollectorTaskAgent(); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/impl/PingApiServiceImpl.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/ping/impl/PingApiServiceImpl.java similarity index 87% rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/impl/PingApiServiceImpl.java rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/ping/impl/PingApiServiceImpl.java index 7d73e38..13b49ca 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/impl/PingApiServiceImpl.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/ping/impl/PingApiServiceImpl.java
@@ -15,10 +15,10 @@ * limitations under the License. */ -package org.apache.iotdb.collector.api.impl; +package org.apache.iotdb.collector.api.ping.impl; -import org.apache.iotdb.collector.api.PingApiService; -import org.apache.iotdb.collector.api.v1.model.ExecutionStatus; +import org.apache.iotdb.collector.api.ping.PingApiService; +import org.apache.iotdb.collector.api.ping.model.ExecutionStatus; import org.apache.iotdb.rpc.TSStatusCode; import javax.ws.rs.core.Response;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/handler/RequestValidationHandler.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/handler/RequestValidationHandler.java deleted file mode 100644 index 9456c2f..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/handler/RequestValidationHandler.java +++ /dev/null
@@ -1,37 +0,0 @@ -/* - * 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.api.v1.handler; - -import org.apache.iotdb.collector.api.v1.model.CreatePipeRequest; -import org.apache.iotdb.collector.api.v1.model.StopPipeRequest; - -import java.util.Objects; - -public class RequestValidationHandler { - private RequestValidationHandler() {} - - public static void validateCreateRequest(final CreatePipeRequest createPipeRequest) { - Objects.requireNonNull(createPipeRequest.getTaskId(), "taskId cannot be null"); - } - - public static void validateStopRequest(final StopPipeRequest stopPipeRequest) { - Objects.requireNonNull(stopPipeRequest.getTaskId(), "taskId cannot be null"); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/impl/AdminApiServiceImpl.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/impl/AdminApiServiceImpl.java deleted file mode 100644 index 07d1f3e..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/impl/AdminApiServiceImpl.java +++ /dev/null
@@ -1,98 +0,0 @@ -/* - * 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.api.v1.impl; - -import org.apache.iotdb.collector.agent.CollectorAgent; -import org.apache.iotdb.collector.api.v1.AdminApiService; -import org.apache.iotdb.collector.api.v1.NotFoundException; -import org.apache.iotdb.collector.api.v1.handler.RequestValidationHandler; -import org.apache.iotdb.collector.api.v1.model.AlterPipeRequest; -import org.apache.iotdb.collector.api.v1.model.CreatePipeRequest; -import org.apache.iotdb.collector.api.v1.model.DropPipeRequest; -import org.apache.iotdb.collector.api.v1.model.StartPipeRequest; -import org.apache.iotdb.collector.api.v1.model.StopPipeRequest; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.ws.rs.core.Response; -import javax.ws.rs.core.SecurityContext; - -public class AdminApiServiceImpl extends AdminApiService { - - private static final Logger LOGGER = LoggerFactory.getLogger(AdminApiServiceImpl.class); - - @Override - public Response alterPipe( - final AlterPipeRequest alterPipeRequest, final SecurityContext securityContext) - throws NotFoundException { - return Response.ok("alterPipe").build(); - } - - @Override - public Response createPipe( - final CreatePipeRequest createPipeRequest, final SecurityContext securityContext) - throws NotFoundException { - RequestValidationHandler.validateCreateRequest(createPipeRequest); - - final boolean createdResult = - CollectorAgent.task() - .createCollectorTask( - createPipeRequest.getSourceAttribute(), - createPipeRequest.getProcessorAttribute(), - createPipeRequest.getSinkAttribute(), - createPipeRequest.getTaskId()); - if (createdResult) { - LOGGER.info("Create task successful"); - return Response.status(Response.Status.OK).entity("create task success").build(); - } - LOGGER.warn("Create task failed"); - return Response.status(Response.Status.BAD_REQUEST).entity("create task fail").build(); - } - - @Override - public Response dropPipe( - final DropPipeRequest dropPipeRequest, final SecurityContext securityContext) - throws NotFoundException { - return Response.ok("dropPipe").build(); - } - - @Override - public Response startPipe( - final StartPipeRequest startPipeRequest, final SecurityContext securityContext) - throws NotFoundException { - return Response.ok("startPipe").build(); - } - - @Override - public Response stopPipe( - final StopPipeRequest stopPipeRequest, final SecurityContext securityContext) - throws NotFoundException { - RequestValidationHandler.validateStopRequest(stopPipeRequest); - - final boolean stopResult = CollectorAgent.task().stopCollectorTask(stopPipeRequest.getTaskId()); - if (stopResult) { - LOGGER.info("Stop task: {} successful", stopPipeRequest.getTaskId()); - return Response.ok().entity("stop task: " + stopPipeRequest.getTaskId() + " success").build(); - } - LOGGER.warn("Stop task: {} failed", stopPipeRequest.getTaskId()); - return Response.status(Response.Status.BAD_REQUEST).entity("stop task fail").build(); - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceImpl.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceImpl.java new file mode 100644 index 0000000..61daa03 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceImpl.java
@@ -0,0 +1,64 @@ +/* + * 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.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; + +public class PluginApiServiceImpl extends PluginApiService { + + @Override + public Response createPlugin( + final CreatePluginRequest createPluginRequest, final SecurityContext securityContext) { + return Response.ok("create plugin").entity(RuntimeService.plugin().createPlugin()).build(); + } + + @Override + public Response alterPlugin( + final AlterPluginRequest alterPluginRequest, final SecurityContext securityContext) { + return Response.ok("alter plugin").entity(RuntimeService.plugin().alterPlugin()).build(); + } + + @Override + public Response startPlugin( + final StartPluginRequest startPluginRequest, final SecurityContext securityContext) { + return Response.ok("start plugin").entity(RuntimeService.plugin().startPlugin()).build(); + } + + @Override + public Response stopPlugin( + final StopPluginRequest stopPluginRequest, final SecurityContext securityContext) { + return Response.ok("stop plugin").entity(RuntimeService.plugin().stopPlugin()).build(); + } + + @Override + public Response dropPlugin( + final DropPluginRequest dropPluginRequest, final SecurityContext securityContext) { + return Response.ok("drop plugin").entity(RuntimeService.plugin().dropPlugin()).build(); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceRequestValidationHandler.java similarity index 68% copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceRequestValidationHandler.java index 4f13ff9..36adce3 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/plugin/impl/PluginApiServiceRequestValidationHandler.java
@@ -17,21 +17,6 @@ * under the License. */ -package org.apache.iotdb.collector.agent.task; +package org.apache.iotdb.collector.api.v1.plugin.impl; -import org.apache.iotdb.commons.concurrent.WrappedRunnable; - -public abstract class CollectorTask extends WrappedRunnable { - - protected final String taskId; - - protected CollectorTask(final String taskId) { - this.taskId = taskId; - } - - public String getTaskId() { - return taskId; - } - - public abstract void stop(); -} +public class PluginApiServiceRequestValidationHandler {}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceImpl.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceImpl.java new file mode 100644 index 0000000..94a0aca --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceImpl.java
@@ -0,0 +1,77 @@ +/* + * 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.api.v1.task.impl; + +import org.apache.iotdb.collector.api.v1.task.TaskApiService; +import org.apache.iotdb.collector.api.v1.task.model.AlterTaskRequest; +import org.apache.iotdb.collector.api.v1.task.model.CreateTaskRequest; +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.service.RuntimeService; + +import javax.ws.rs.core.Response; +import javax.ws.rs.core.SecurityContext; + +public class TaskApiServiceImpl extends TaskApiService { + + @Override + public Response createTask( + final CreateTaskRequest createTaskRequest, final SecurityContext securityContext) { + TaskApiServiceRequestValidationHandler.validateCreateRequest(createTaskRequest); + + return RuntimeService.task() + .createTask( + createTaskRequest.getTaskId(), + createTaskRequest.getSourceAttribute(), + createTaskRequest.getProcessorAttribute(), + createTaskRequest.getSinkAttribute()); + } + + @Override + public Response alterTask( + final AlterTaskRequest alterTaskRequest, final SecurityContext securityContext) { + return Response.ok("alter task").build(); + } + + @Override + public Response startTask( + final StartTaskRequest startTaskRequest, final SecurityContext securityContext) { + TaskApiServiceRequestValidationHandler.validateStartRequest(startTaskRequest); + + return RuntimeService.task().startTask(startTaskRequest.getTaskId()); + } + + @Override + public Response stopTask( + final StopTaskRequest stopTaskRequest, final SecurityContext securityContext) { + TaskApiServiceRequestValidationHandler.validateStopRequest(stopTaskRequest); + + return RuntimeService.task().stopTask(stopTaskRequest.getTaskId()); + } + + @Override + public Response dropTask( + final DropTaskRequest dropTaskRequest, final SecurityContext securityContext) { + TaskApiServiceRequestValidationHandler.validateDropRequest(dropTaskRequest); + + return RuntimeService.task().dropTask(dropTaskRequest.getTaskId()); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceRequestValidationHandler.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceRequestValidationHandler.java new file mode 100644 index 0000000..7397f0c --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/api/v1/task/impl/TaskApiServiceRequestValidationHandler.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.api.v1.task.impl; + +import org.apache.iotdb.collector.api.v1.task.model.CreateTaskRequest; +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 java.util.Objects; + +public class TaskApiServiceRequestValidationHandler { + private TaskApiServiceRequestValidationHandler() {} + + public static void validateCreateRequest(final CreateTaskRequest createTaskRequest) { + Objects.requireNonNull(createTaskRequest.getTaskId(), "taskId cannot be null"); + } + + public static void validateStopRequest(final StopTaskRequest stopTaskRequest) { + Objects.requireNonNull(stopTaskRequest.getTaskId(), "taskId cannot be null"); + } + + public static void validateStartRequest(final StartTaskRequest startTaskRequest) { + Objects.requireNonNull(startTaskRequest.getTaskId(), "taskId cannot be null"); + } + + public static void validateDropRequest(final DropTaskRequest dropTaskRequest) { + Objects.requireNonNull(dropTaskRequest.getTaskId(), "taskId cannot be null"); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/ApiServiceOptions.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/ApiServiceOptions.java index 2e3ee55..8a66cbb 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/ApiServiceOptions.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/ApiServiceOptions.java
@@ -24,7 +24,7 @@ public static final Option<Integer> PORT = new Option<Integer>("api_service_port", 17070) { @Override - public void setValue(String valueString) { + public void setValue(final String valueString) { value = Integer.parseInt(valueString); } };
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/Options.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/Options.java index 8ad2887..9360248 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/Options.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/Options.java
@@ -32,6 +32,15 @@ private static final Map<String, Option<?>> OPTIONS = new ConcurrentHashMap<>(); + static { + try { + Class.forName("org.apache.iotdb.collector.config.ApiServiceOptions"); + Class.forName("org.apache.iotdb.collector.config.TaskRuntimeOptions"); + } catch (final ClassNotFoundException e) { + throw new RuntimeException("Failed to load ApiServiceOptions", e); + } + } + public abstract static class Option<T> { private final String key;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/TaskRuntimeOptions.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/TaskRuntimeOptions.java new file mode 100644 index 0000000..70efcc9 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/config/TaskRuntimeOptions.java
@@ -0,0 +1,63 @@ +/* + * 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; + +public class TaskRuntimeOptions extends Options { + + public static final Option<Integer> TASK_SOURCE_PARALLELISM_NUM = + new Option<Integer>("task_source_parallelism_num", 4) { + @Override + public void setValue(final String valueString) { + value = Integer.parseInt(valueString); + } + }; + + public static final Option<Integer> TASK_PROCESS_PARALLELISM_NUM = + new Option<Integer>("task_process_parallelism_num", 4) { + @Override + public void setValue(final String valueString) { + value = Integer.parseInt(valueString); + } + }; + + public static final Option<Integer> TASK_SINK_PARALLELISM_NUM = + new Option<Integer>("task_sink_parallelism_num", 4) { + @Override + public void setValue(final String valueString) { + value = Integer.parseInt(valueString); + } + }; + + public static final Option<Integer> TASK_PROCESSOR_RING_BUFFER_SIZE = + new Option<Integer>("task_processor_ring_buffer_size", 1024) { + @Override + public void setValue(String valueString) { + value = Integer.parseInt(valueString); + } + }; + + public static final Option<Integer> TASK_SINK_RING_BUFFER_SIZE = + new Option<Integer>("task_sink_ring_buffer_size", 1024) { + @Override + public void setValue(String valueString) { + value = Integer.parseInt(valueString); + } + }; +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinCollectorPlugin.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java similarity index 72% rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinCollectorPlugin.java rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java index 2d676e5..683a2c4 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinCollectorPlugin.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/BuiltinPlugin.java
@@ -19,19 +19,23 @@ package org.apache.iotdb.collector.plugin; -import org.apache.iotdb.collector.plugin.builtin.processor.DoNothingProcessor; -import org.apache.iotdb.collector.plugin.builtin.sink.SessionSink; -import org.apache.iotdb.collector.plugin.builtin.source.HttpSource; +import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor; +import org.apache.iotdb.collector.plugin.sink.SessionSink; +import org.apache.iotdb.collector.plugin.source.HttpPullSource; +import org.apache.iotdb.collector.plugin.source.HttpPushSource; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.Set; -public enum BuiltinCollectorPlugin { +public enum BuiltinPlugin { - // Sources - HTTP_SOURCE("http-source", HttpSource.class), + // PushSources + HTTP_PUSH_SOURCE("http-push-source", HttpPushSource.class), + + // PullSources + HTTP_PULL_SOURCE("http-pull-source", HttpPullSource.class), // Processors DO_NOTHING_PROCESSOR("do-nothing-processor", DoNothingProcessor.class), @@ -43,7 +47,7 @@ private final Class<?> collectorPluginClass; private final String className; - BuiltinCollectorPlugin(String collectorPluginName, Class<?> collectorPluginClass) { + BuiltinPlugin(final String collectorPluginName, final Class<?> collectorPluginClass) { this.collectorPluginName = collectorPluginName; this.collectorPluginClass = collectorPluginClass; this.className = collectorPluginClass.getName(); @@ -65,8 +69,10 @@ Collections.unmodifiableSet( new HashSet<>( Arrays.asList( - // Sources - HTTP_SOURCE.getCollectorPluginName().toUpperCase(), + // PushSources + HTTP_PUSH_SOURCE.getCollectorPluginName().toUpperCase(), + // PullSources + HTTP_PULL_SOURCE.getCollectorPluginName().toUpperCase(), // Processors DO_NOTHING_PROCESSOR.getCollectorPluginName().toUpperCase(), // Sinks
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPullSource.java similarity index 68% copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPullSource.java index 4f13ff9..097111a 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPullSource.java
@@ -17,21 +17,6 @@ * under the License. */ -package org.apache.iotdb.collector.agent.task; +package org.apache.iotdb.collector.plugin.api; -import org.apache.iotdb.commons.concurrent.WrappedRunnable; - -public abstract class CollectorTask extends WrappedRunnable { - - protected final String taskId; - - protected CollectorTask(final String taskId) { - this.taskId = taskId; - } - - public String getTaskId() { - return taskId; - } - - public abstract void stop(); -} +public abstract class CollectorPullSource implements CollectorSource {}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/event/SourceEvent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPushSource.java similarity index 64% copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/event/SourceEvent.java copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPushSource.java index 813e25e..622ed52 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/event/SourceEvent.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorPushSource.java
@@ -17,29 +17,25 @@ * under the License. */ -package org.apache.iotdb.collector.plugin.builtin.source.event; +package org.apache.iotdb.collector.plugin.api; +import org.apache.iotdb.pipe.api.collector.EventCollector; import org.apache.iotdb.pipe.api.event.Event; -public class SourceEvent implements Event { - private String name; +public abstract class CollectorPushSource implements CollectorSource { - public SourceEvent() {} + protected final EventCollector collector; - public SourceEvent(final String name) { - this.name = name; - } - - public String getName() { - return name; - } - - public void setName(final String name) { - this.name = name; + public CollectorPushSource(final EventCollector collector) { + this.collector = collector; } @Override - public String toString() { - return "SourceEvent [name=" + name + "]"; + public final Event supply() { + throw new UnsupportedOperationException(); + } + + public final void supply(final Event event) throws Exception { + collector.collect(event); } }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorSource.java similarity index 68% copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorSource.java index 4f13ff9..2e483e1 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/CollectorSource.java
@@ -17,21 +17,8 @@ * under the License. */ -package org.apache.iotdb.collector.agent.task; +package org.apache.iotdb.collector.plugin.api; -import org.apache.iotdb.commons.concurrent.WrappedRunnable; +import org.apache.iotdb.pipe.api.PipeSource; -public abstract class CollectorTask extends WrappedRunnable { - - protected final String taskId; - - protected CollectorTask(final String taskId) { - this.taskId = taskId; - } - - public String getTaskId() { - return taskId; - } - - public abstract void stop(); -} +public interface CollectorSource extends StoppablePlugin, PipeSource {}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/RuntimeConfig.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/RuntimeConfig.java new file mode 100644 index 0000000..1da84bc --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/RuntimeConfig.java
@@ -0,0 +1,63 @@ +/* + * 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.plugin.api; + +import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeRuntimeEnvironment; + +public class RuntimeConfig implements PipeExtractorRuntimeConfiguration { + + public static class RuntimeEnvironment implements PipeRuntimeEnvironment { + + public int getParallelism() { + return 0; + } + + public int getParallelismIndex() { + return 0; + } + + @Override + public String getPipeName() { + return ""; + } + + @Override + public long getCreationTime() { + return 0; + } + } + + @Override + public PipeRuntimeEnvironment getRuntimeEnvironment() { + return new PipeRuntimeEnvironment() { + + @Override + public String getPipeName() { + return ""; + } + + @Override + public long getCreationTime() { + return 0; + } + }; + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/StoppablePlugin.java similarity index 68% copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/StoppablePlugin.java index 4f13ff9..8a97564 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/api/StoppablePlugin.java
@@ -17,21 +17,8 @@ * under the License. */ -package org.apache.iotdb.collector.agent.task; +package org.apache.iotdb.collector.plugin.api; -import org.apache.iotdb.commons.concurrent.WrappedRunnable; - -public abstract class CollectorTask extends WrappedRunnable { - - protected final String taskId; - - protected CollectorTask(final String taskId) { - this.taskId = taskId; - } - - public String getTaskId() { - return taskId; - } - - public abstract void stop(); +public interface StoppablePlugin { + void stop() throws Exception; }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpSource.java deleted file mode 100644 index 65ba79b..0000000 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/HttpSource.java +++ /dev/null
@@ -1,79 +0,0 @@ -/* - * 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.plugin.builtin.source; - -import org.apache.iotdb.collector.plugin.builtin.source.event.SourceEvent; -import org.apache.iotdb.pipe.api.PipeSource; -import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration; -import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration; -import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; -import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; -import org.apache.iotdb.pipe.api.event.Event; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.Random; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; - -public class HttpSource implements PipeSource { - - private static final Logger LOGGER = LoggerFactory.getLogger(HttpSource.class); - - private static final BlockingQueue<Event> queue = new LinkedBlockingQueue<>(); - private boolean isStarted = true; - - @Override - public void validate(PipeParameterValidator validator) throws Exception {} - - @Override - public void customize(PipeParameters parameters, PipeExtractorRuntimeConfiguration configuration) - throws Exception {} - - @Override - public void customize(PipeParameters parameters, PipeSourceRuntimeConfiguration configuration) - throws Exception {} - - @Override - public void start() { - isStarted = true; - while (isStarted) { - Event event = new SourceEvent(String.valueOf(new Random().nextInt(1000))); - try { - queue.put(event); - Thread.sleep(1000); - LOGGER.info("event: {} created success", event); - } catch (final InterruptedException e) { - LOGGER.warn("failed to create event because {}", e.getMessage()); - } - } - } - - @Override - public Event supply() throws Exception { - return queue.take(); - } - - @Override - public void close() throws Exception { - isStarted = false; - } -}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/event/SourceEvent.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java similarity index 94% rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/event/SourceEvent.java rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java index 813e25e..0c097ce 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/source/event/SourceEvent.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/event/SourceEvent.java
@@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.collector.plugin.builtin.source.event; +package org.apache.iotdb.collector.plugin.event; import org.apache.iotdb.pipe.api.event.Event;
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/processor/DoNothingProcessor.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/processor/DoNothingProcessor.java similarity index 74% rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/processor/DoNothingProcessor.java rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/processor/DoNothingProcessor.java index 436fc20..2bf2ee9 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/processor/DoNothingProcessor.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/processor/DoNothingProcessor.java
@@ -17,7 +17,7 @@ * under the License. */ -package org.apache.iotdb.collector.plugin.builtin.processor; +package org.apache.iotdb.collector.plugin.processor; import org.apache.iotdb.pipe.api.PipeProcessor; import org.apache.iotdb.pipe.api.collector.EventCollector; @@ -35,24 +35,27 @@ private static final Logger LOGGER = LoggerFactory.getLogger(DoNothingProcessor.class); @Override - public void validate(PipeParameterValidator validator) throws Exception { + public void validate(final PipeParameterValidator validator) throws Exception { // Do Nothing } @Override - public void customize(PipeParameters parameters, PipeProcessorRuntimeConfiguration configuration) + public void customize( + final PipeParameters parameters, final PipeProcessorRuntimeConfiguration configuration) throws Exception { // Do Nothing } @Override - public void process(TabletInsertionEvent tabletInsertionEvent, EventCollector eventCollector) + public void process( + final TabletInsertionEvent tabletInsertionEvent, final EventCollector eventCollector) throws Exception { - // Do Nothing + LOGGER.info("DoNothingProcessor process tabletInsertionEvent: {}", tabletInsertionEvent); + eventCollector.collect(tabletInsertionEvent); } @Override - public void process(Event event, EventCollector eventCollector) throws Exception { + public void process(final Event event, final EventCollector eventCollector) throws Exception { LOGGER.info("DoNothingProcessor process event: {}", event); eventCollector.collect(event); }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/sink/SessionSink.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/sink/SessionSink.java similarity index 77% rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/sink/SessionSink.java rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/sink/SessionSink.java index e2a5260..64966db 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/builtin/sink/SessionSink.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/sink/SessionSink.java
@@ -17,9 +17,9 @@ * under the License. */ -package org.apache.iotdb.collector.plugin.builtin.sink; +package org.apache.iotdb.collector.plugin.sink; -import org.apache.iotdb.collector.plugin.builtin.source.event.SourceEvent; +import org.apache.iotdb.collector.plugin.event.SourceEvent; import org.apache.iotdb.pipe.api.PipeSink; import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration; import org.apache.iotdb.pipe.api.customizer.configuration.PipeSinkRuntimeConfiguration; @@ -36,14 +36,16 @@ private static final Logger LOGGER = LoggerFactory.getLogger(SessionSink.class); @Override - public void validate(PipeParameterValidator validator) throws Exception {} + public void validate(final PipeParameterValidator validator) throws Exception {} @Override - public void customize(PipeParameters parameters, PipeConnectorRuntimeConfiguration configuration) + public void customize( + final PipeParameters parameters, final PipeConnectorRuntimeConfiguration configuration) throws Exception {} @Override - public void customize(PipeParameters parameters, PipeSinkRuntimeConfiguration configuration) + public void customize( + final PipeParameters parameters, final PipeSinkRuntimeConfiguration configuration) throws Exception {} @Override @@ -55,10 +57,10 @@ public void heartbeat() throws Exception {} @Override - public void transfer(TabletInsertionEvent tabletInsertionEvent) throws Exception {} + public void transfer(final TabletInsertionEvent tabletInsertionEvent) throws Exception {} @Override - public void transfer(Event event) throws Exception { + public void transfer(final Event event) throws Exception { final SourceEvent sourceEvent = (SourceEvent) event; LOGGER.info("SessionSink transfer successfully {}", sourceEvent); }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java new file mode 100644 index 0000000..83881a3 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPullSource.java
@@ -0,0 +1,70 @@ +/* + * 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.plugin.source; + +import org.apache.iotdb.collector.plugin.api.CollectorPullSource; +import org.apache.iotdb.collector.plugin.event.SourceEvent; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; +import org.apache.iotdb.pipe.api.event.Event; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Random; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; + +public class HttpPullSource extends CollectorPullSource { + + private static final Logger LOGGER = LoggerFactory.getLogger(HttpPullSource.class); + + @Override + public void validate(PipeParameterValidator pipeParameterValidator) {} + + @Override + public void customize( + PipeParameters pipeParameters, + PipeExtractorRuntimeConfiguration pipeExtractorRuntimeConfiguration) {} + + @Override + public void customize( + PipeParameters pipeParameters, + PipeSourceRuntimeConfiguration pipeSourceRuntimeConfiguration) {} + + @Override + public void start() throws Exception {} + + @Override + public Event supply() { + final Event event = new SourceEvent(String.valueOf(new Random().nextInt(1000))); + LOGGER.info("event: {} created success", event); + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + return event; + } + + @Override + public void stop() throws Exception {} + + @Override + public void close() {} +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java new file mode 100644 index 0000000..96d03d0 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/plugin/source/HttpPushSource.java
@@ -0,0 +1,101 @@ +/* + * 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.plugin.source; + +import org.apache.iotdb.collector.plugin.api.CollectorPushSource; +import org.apache.iotdb.collector.plugin.event.SourceEvent; +import org.apache.iotdb.pipe.api.collector.EventCollector; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration; +import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; +import org.apache.iotdb.pipe.api.event.Event; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Random; +import java.util.concurrent.TimeUnit; + +public class HttpPushSource extends CollectorPushSource { + + private static final Logger LOGGER = LoggerFactory.getLogger(HttpPushSource.class); + + private volatile boolean isStarted = true; + private Thread workerThread; + + public HttpPushSource(final EventCollector collector) { + super(collector); + } + + @Override + public void validate(final PipeParameterValidator validator) {} + + @Override + public void customize( + final PipeParameters parameters, final PipeExtractorRuntimeConfiguration configuration) {} + + @Override + public void customize( + final PipeParameters parameters, final PipeSourceRuntimeConfiguration configuration) {} + + @Override + public void start() { + if (workerThread == null || !workerThread.isAlive()) { + isStarted = true; + workerThread = new Thread(this::doWork); + workerThread.start(); + } + } + + private void doWork() { + try { + while (isStarted && !Thread.currentThread().isInterrupted()) { + final Event event = new SourceEvent(String.valueOf(new Random().nextInt(1000))); + LOGGER.info("event: {} created success", event); + supply(event); + TimeUnit.SECONDS.sleep(2); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (Exception e) { + LOGGER.error("Error in push source", e); + } + } + + @Override + public void stop() throws Exception { + this.isStarted = false; + } + + @Override + public void close() { + isStarted = false; + if (workerThread != null) { + workerThread.interrupt(); + try { + workerThread.join(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + workerThread = null; + } + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginFactory.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginFactory.java new file mode 100644 index 0000000..89bdca4 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginFactory.java
@@ -0,0 +1,53 @@ +/* + * 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.runtime.plugin; + +import org.apache.iotdb.collector.plugin.BuiltinPlugin; +import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor; +import org.apache.iotdb.collector.plugin.sink.SessionSink; +import org.apache.iotdb.collector.plugin.source.HttpPullSource; +import org.apache.iotdb.pipe.api.PipePlugin; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.Map; +import java.util.function.Supplier; + +public class PluginFactory { + private static final Logger LOGGER = LoggerFactory.getLogger(PluginFactory.class); + + protected final Map<String, Supplier<PipePlugin>> pluginConstructors = new HashMap<>(); + + public PluginFactory() { + initFactory(); + } + + private void initFactory() { + pluginConstructors.put( + BuiltinPlugin.HTTP_PULL_SOURCE.getCollectorPluginName(), HttpPullSource::new); + pluginConstructors.put( + BuiltinPlugin.DO_NOTHING_PROCESSOR.getCollectorPluginName(), DoNothingProcessor::new); + pluginConstructors.put( + BuiltinPlugin.IOTDB_SESSION_SINK.getCollectorPluginName(), SessionSink::new); + LOGGER.info("builtin plugin has been initialized"); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginRuntime.java similarity index 69% copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginRuntime.java index 4f13ff9..52f49e1 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/plugin/PluginRuntime.java
@@ -17,21 +17,27 @@ * under the License. */ -package org.apache.iotdb.collector.agent.task; +package org.apache.iotdb.collector.runtime.plugin; -import org.apache.iotdb.commons.concurrent.WrappedRunnable; +public class PluginRuntime { -public abstract class CollectorTask extends WrappedRunnable { - - protected final String taskId; - - protected CollectorTask(final String taskId) { - this.taskId = taskId; + public boolean createPlugin() { + return true; } - public String getTaskId() { - return taskId; + public boolean alterPlugin() { + return true; } - public abstract void stop(); + public boolean startPlugin() { + return true; + } + + public boolean stopPlugin() { + return true; + } + + public boolean dropPlugin() { + return true; + } }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java new file mode 100644 index 0000000..3754f5f --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/TaskRuntime.java
@@ -0,0 +1,138 @@ +/* + * 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.runtime.task; + +import org.apache.iotdb.collector.runtime.task.def.ProcessorTask; +import org.apache.iotdb.collector.runtime.task.def.PushSourceTask; +import org.apache.iotdb.collector.runtime.task.def.SinkTask; +import org.apache.iotdb.collector.runtime.task.def.SourceTask; +import org.apache.iotdb.collector.runtime.task.def.TaskRepository; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.ws.rs.core.Response; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class TaskRuntime { + + private static final Logger LOGGER = LoggerFactory.getLogger(TaskRuntime.class); + + private static final Map<String, TaskRepository> TASK_REPOSITORY_MAP = new ConcurrentHashMap<>(); + + public Response createTask( + final String taskId, + final Map<String, String> sourceAttribute, + final Map<String, String> processorAttribute, + final Map<String, String> sinkAttribute) { + try { + if (validateTaskIsExist(taskId)) { + return Response.status(Response.Status.CONFLICT) + .entity(String.format("task %s has existed", taskId)) + .build(); + } + + final SinkTask sinkTask = new SinkTask(sinkAttribute); + final ProcessorTask processorTask = new ProcessorTask(processorAttribute, sinkTask); + final SourceTask sourceTask = new PushSourceTask(taskId, sourceAttribute, processorTask); + final TaskRepository taskRepository = new TaskRepository(sourceTask, processorTask, sinkTask); + + TASK_REPOSITORY_MAP.put(taskId, taskRepository); + taskRepository.create(); + + LOGGER.info("Successfully created task {}", taskId); + return Response.status(Response.Status.CREATED) + .entity(String.format("Successfully created task %s", taskId)) + .build(); + } catch (final Exception e) { + LOGGER.warn("Failed to create task", e); + return Response.serverError() + .entity(String.format("Failed to create task %s, because %s", taskId, e.getMessage())) + .build(); + } + } + + public boolean alterTask() { + return true; + } + + public Response startTask(final String taskId) { + if (!validateTaskIsExist(taskId)) { + return Response.status(Response.Status.NOT_FOUND) + .entity(String.format("task %s not found", taskId)) + .build(); + } + + TASK_REPOSITORY_MAP.get(taskId).start(); + LOGGER.info("Task {} started successfully", taskId); + return Response.status(Response.Status.OK) + .entity(String.format("task %s start successfully", taskId)) + .build(); + } + + public Response stopTask(final String taskId) { + if (!validateTaskIsExist(taskId)) { + return Response.status(Response.Status.NOT_FOUND) + .entity(String.format("task %s not found", taskId)) + .build(); + } + + try { + final TaskRepository taskRepository = TASK_REPOSITORY_MAP.get(taskId); + if (taskRepository != null) { + taskRepository.stop(); + } + } catch (final Exception e) { + LOGGER.warn("Failed to stop task", e); + return Response.serverError() + .entity(String.format("Failed to stop task %s, because %s", taskId, e.getMessage())) + .build(); + } + + LOGGER.info("Task {} stopped successfully", taskId); + return Response.status(Response.Status.OK) + .entity(String.format("task %s stop successfully", taskId)) + .build(); + } + + public Response dropTask(final String taskId) { + if (!validateTaskIsExist(taskId)) { + return Response.status(Response.Status.NOT_FOUND) + .entity(String.format("task %s not found", taskId)) + .build(); + } + + TASK_REPOSITORY_MAP.remove(taskId).drop(); + LOGGER.info("Task {} dropped successfully", taskId); + return Response.status(Response.Status.OK) + .entity(String.format("task %s drop successfully", taskId)) + .build(); + } + + private boolean validateTaskIsExist(final String taskId) { + if (TASK_REPOSITORY_MAP.containsKey(taskId)) { + return true; + } + LOGGER.warn("Task {} not found", taskId); + return false; + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/collect/CollectorEventCollector.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventCollector.java similarity index 63% rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/collect/CollectorEventCollector.java rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventCollector.java index 056685e..43886a5 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/collect/CollectorEventCollector.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventCollector.java
@@ -17,32 +17,27 @@ * under the License. */ -package org.apache.iotdb.collector.agent.collect; +package org.apache.iotdb.collector.runtime.task.datastructure; import org.apache.iotdb.pipe.api.collector.EventCollector; import org.apache.iotdb.pipe.api.event.Event; +import com.lmax.disruptor.RingBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.concurrent.BlockingQueue; +public class TaskEventCollector implements EventCollector { -public class CollectorEventCollector implements EventCollector { + private static final Logger LOGGER = LoggerFactory.getLogger(TaskEventCollector.class); + private final RingBuffer<TaskEventContainer> ringBuffer; - private static final Logger LOGGER = LoggerFactory.getLogger(CollectorEventCollector.class); - - private final BlockingQueue<Event> pendingQueue; - - public CollectorEventCollector(final BlockingQueue<Event> pendingQueue) { - this.pendingQueue = pendingQueue; + public TaskEventCollector(final RingBuffer<TaskEventContainer> ringBuffer) { + this.ringBuffer = ringBuffer; } @Override public void collect(final Event event) { - try { - pendingQueue.put(event); - } catch (final InterruptedException e) { - LOGGER.warn("collect event failed because {}", e.getMessage(), e); - } + ringBuffer.publishEvent((container, sequence, o) -> container.setEvent(event), event); + LOGGER.info("successfully publish event {}", event); } }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventConsumer.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventConsumer.java new file mode 100644 index 0000000..7f94487 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventConsumer.java
@@ -0,0 +1,63 @@ +/* + * 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.runtime.task.datastructure; + +import org.apache.iotdb.pipe.api.PipePlugin; +import org.apache.iotdb.pipe.api.PipeProcessor; +import org.apache.iotdb.pipe.api.PipeSink; + +import com.lmax.disruptor.WorkHandler; + +public class TaskEventConsumer implements WorkHandler<TaskEventContainer> { + + private final PipePlugin plugin; + private final TaskEventCollector collector; + private final TaskEventConsumerController consumerController; + + public TaskEventConsumer( + final PipePlugin plugin, + final TaskEventCollector collector, + final TaskEventConsumerController consumerController) { + this.plugin = plugin; + this.collector = collector; + this.consumerController = consumerController; + } + + public TaskEventConsumer( + final PipePlugin plugin, final TaskEventConsumerController consumerController) { + this(plugin, null, consumerController); + } + + @Override + public void onEvent(final TaskEventContainer taskEventContainer) throws Exception { + if (!consumerController.shouldRun()) { + return; + } + if (plugin instanceof PipeProcessor) { + ((PipeProcessor) plugin).process(taskEventContainer.getEvent(), this.collector); + } else if (plugin instanceof PipeSink) { + ((PipeSink) plugin).transfer(taskEventContainer.getEvent()); + } + } + + public TaskEventConsumerController getConsumerController() { + return consumerController; + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventConsumerController.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventConsumerController.java new file mode 100644 index 0000000..7506a09 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventConsumerController.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.runtime.task.datastructure; + +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.LockSupport; + +public class TaskEventConsumerController { + + private final AtomicBoolean running = new AtomicBoolean(true); + + private static final long PARK_NANOS = 100_000_000L; + + public void pause() { + running.set(false); + } + + public void resume() { + running.set(true); + } + + public boolean shouldRun() { + while (!running.get()) { + LockSupport.parkNanos(PARK_NANOS); + } + return running.get(); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventContainer.java similarity index 69% rename from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java rename to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventContainer.java index 4f13ff9..8ac42fd 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/datastructure/TaskEventContainer.java
@@ -17,21 +17,20 @@ * under the License. */ -package org.apache.iotdb.collector.agent.task; +package org.apache.iotdb.collector.runtime.task.datastructure; -import org.apache.iotdb.commons.concurrent.WrappedRunnable; +import org.apache.iotdb.pipe.api.event.Event; -public abstract class CollectorTask extends WrappedRunnable { +public class TaskEventContainer implements Event { + private Event event; - protected final String taskId; + public TaskEventContainer() {} - protected CollectorTask(final String taskId) { - this.taskId = taskId; + public Event getEvent() { + return event; } - public String getTaskId() { - return taskId; + public void setEvent(final Event event) { + this.event = event; } - - public abstract void stop(); }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/ProcessorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/ProcessorTask.java new file mode 100644 index 0000000..c498280 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/ProcessorTask.java
@@ -0,0 +1,114 @@ +/* + * 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.runtime.task.def; + +import org.apache.iotdb.collector.config.TaskRuntimeOptions; +import org.apache.iotdb.collector.plugin.processor.DoNothingProcessor; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventCollector; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventConsumer; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventContainer; +import org.apache.iotdb.collector.runtime.task.exception.DisruptorTaskExceptionHandler; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; + +import com.lmax.disruptor.BlockingWaitStrategy; +import com.lmax.disruptor.RingBuffer; +import com.lmax.disruptor.dsl.Disruptor; +import com.lmax.disruptor.dsl.ProducerType; +import com.lmax.disruptor.util.DaemonThreadFactory; + +import java.util.Map; +import java.util.Optional; + +public class ProcessorTask extends Task implements TaskComponent { + + private Disruptor<TaskEventContainer> processorDisruptor; + private TaskEventConsumer[] eventConsumers; + + private final PipeParameters parameters; + private final int processParallelismNum; + private final TaskEventCollector collector; + + public ProcessorTask(final Map<String, String> processorAttributes, final SinkTask sinkTask) { + this.parameters = new PipeParameters(processorAttributes); + this.processParallelismNum = + this.parameters.getIntOrDefault( + TaskRuntimeOptions.TASK_PROCESS_PARALLELISM_NUM.key(), + TaskRuntimeOptions.TASK_PROCESS_PARALLELISM_NUM.value()); + this.collector = new TaskEventCollector(sinkTask.getSinkRingBuffer()); + + this.initProcessorDisruptor(); + } + + @Override + public void create() { + if (this.processorDisruptor == null) { + this.initProcessorDisruptor(); + } + + this.eventConsumers = + this.getConsumer( + this.createInstance(DoNothingProcessor.class), processParallelismNum, collector); + + this.processorDisruptor.setDefaultExceptionHandler(new DisruptorTaskExceptionHandler()); + this.processorDisruptor.handleEventsWithWorkerPool(this.eventConsumers); + this.processorDisruptor.start(); + } + + private void initProcessorDisruptor() { + this.processorDisruptor = + new Disruptor<>( + TaskEventContainer::new, + this.parameters.getIntOrDefault( + TaskRuntimeOptions.TASK_PROCESSOR_RING_BUFFER_SIZE.key(), + TaskRuntimeOptions.TASK_PROCESSOR_RING_BUFFER_SIZE.value()), + DaemonThreadFactory.INSTANCE, + ProducerType.MULTI, + new BlockingWaitStrategy()); + } + + @Override + public void start() throws Exception { + for (final TaskEventConsumer consumer : this.eventConsumers) { + consumer.getConsumerController().resume(); + } + } + + @Override + public void stop() { + for (final TaskEventConsumer consumer : this.eventConsumers) { + consumer.getConsumerController().pause(); + } + } + + @Override + public void drop() { + if (this.processorDisruptor != null) { + this.processorDisruptor.shutdown(); + this.processorDisruptor = null; + } + } + + public Optional<RingBuffer<TaskEventContainer>> getProcessorRingBuffer() { + if (this.processorDisruptor != null) { + return Optional.of(this.processorDisruptor.getRingBuffer()); + } + return Optional.empty(); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/PullSourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/PullSourceTask.java new file mode 100644 index 0000000..0e0c19c --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/PullSourceTask.java
@@ -0,0 +1,80 @@ +/* + * 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.runtime.task.def; + +import org.apache.iotdb.collector.plugin.source.HttpPullSource; +import org.apache.iotdb.pipe.api.event.Event; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.LockSupport; + +public class PullSourceTask extends SourceTask { + + private static final Logger LOGGER = LoggerFactory.getLogger(PullSourceTask.class); + + public PullSourceTask( + final String taskId, + final Map<String, String> sourceParams, + final ProcessorTask processorTask) { + super(taskId, sourceParams, processorTask); + } + + @Override + public void create() { + createSourceTask(); + for (int i = 0; i < sourceParallelismNum; i++) { + // use sourceAttribute later + try (final HttpPullSource source = createInstance(HttpPullSource.class)) { + addSourceTask(source); + SOURCE_EXECUTOR_SERVICE + .get(taskId) + .submit( + () -> { + try { + source.start(); + while (SOURCE_TASK_STATUS.containsKey(taskId)) { + if (SOURCE_TASK_STATUS.get(taskId) == TaskState.Stopped) { + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1)); + continue; + } + + final Event event = source.supply(); + processorTask + .getProcessorRingBuffer() + .ifPresent( + ringBuffer -> + ringBuffer.publishEvent( + ((container, sequence, o) -> container.setEvent(event)), + event)); + } + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } catch (final Exception e) { + LOGGER.warn("Pull source task failed", e); + } + } + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/PushSourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/PushSourceTask.java new file mode 100644 index 0000000..176f671 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/PushSourceTask.java
@@ -0,0 +1,72 @@ +/* + * 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.runtime.task.def; + +import org.apache.iotdb.collector.plugin.source.HttpPushSource; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventCollector; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; + +public class PushSourceTask extends SourceTask { + + private static final Logger LOGGER = LoggerFactory.getLogger(PushSourceTask.class); + + private final TaskEventCollector collector; + + public PushSourceTask( + final String taskId, + final Map<String, String> sourceParams, + final ProcessorTask processorTask) { + super(taskId, sourceParams, processorTask); + + this.collector = + new TaskEventCollector( + processorTask.getProcessorRingBuffer().isPresent() + ? processorTask.getProcessorRingBuffer().get() + : null); + } + + @Override + public void create() { + createSourceTask(); + for (int i = 0; i < sourceParallelismNum; i++) { + // use sourceAttribute later + try (final HttpPushSource source = new HttpPushSource(collector)) { + addSourceTask(source); + SOURCE_EXECUTOR_SERVICE + .get(taskId) + .submit( + () -> { + try { + source.start(); + } catch (final Exception e) { + LOGGER.warn("Failed to start push source", e); + throw new RuntimeException(e); + } + }); + } catch (Exception e) { + LOGGER.warn("failed to create instance of HttpSource", e); + } + } + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/SinkTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/SinkTask.java new file mode 100644 index 0000000..e04b230 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/SinkTask.java
@@ -0,0 +1,105 @@ +/* + * 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.runtime.task.def; + +import org.apache.iotdb.collector.config.TaskRuntimeOptions; +import org.apache.iotdb.collector.plugin.sink.SessionSink; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventConsumer; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventContainer; +import org.apache.iotdb.collector.runtime.task.exception.DisruptorTaskExceptionHandler; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; + +import com.lmax.disruptor.BlockingWaitStrategy; +import com.lmax.disruptor.RingBuffer; +import com.lmax.disruptor.dsl.Disruptor; +import com.lmax.disruptor.dsl.ProducerType; +import com.lmax.disruptor.util.DaemonThreadFactory; + +import java.util.Map; + +public class SinkTask extends Task implements TaskComponent { + + private Disruptor<TaskEventContainer> sinkDisruptor; + private TaskEventConsumer[] eventConsumers; + private final PipeParameters parameters; + private final int sinkParallelismNum; + + public SinkTask(final Map<String, String> processorAttributes) { + this.parameters = new PipeParameters(processorAttributes); + this.sinkParallelismNum = + this.parameters.getIntOrDefault( + TaskRuntimeOptions.TASK_SINK_PARALLELISM_NUM.key(), + TaskRuntimeOptions.TASK_SINK_PARALLELISM_NUM.value()); + + this.initSinkDisruptor(); + } + + @Override + public void create() { + if (this.sinkDisruptor == null) { + this.initSinkDisruptor(); + } + + this.eventConsumers = + this.getConsumer(this.createInstance(SessionSink.class), sinkParallelismNum, null); + + this.sinkDisruptor.setDefaultExceptionHandler(new DisruptorTaskExceptionHandler()); + this.sinkDisruptor.handleEventsWithWorkerPool(this.eventConsumers); + this.sinkDisruptor.start(); + } + + private void initSinkDisruptor() { + this.sinkDisruptor = + new Disruptor<>( + TaskEventContainer::new, + this.parameters.getIntOrDefault( + TaskRuntimeOptions.TASK_SINK_RING_BUFFER_SIZE.key(), + TaskRuntimeOptions.TASK_SINK_RING_BUFFER_SIZE.value()), + DaemonThreadFactory.INSTANCE, + ProducerType.MULTI, + new BlockingWaitStrategy()); + } + + @Override + public void start() throws Exception { + for (final TaskEventConsumer consumer : this.eventConsumers) { + consumer.getConsumerController().resume(); + } + } + + @Override + public void stop() { + for (final TaskEventConsumer consumer : this.eventConsumers) { + consumer.getConsumerController().pause(); + } + } + + @Override + public void drop() { + if (this.sinkDisruptor != null) { + this.sinkDisruptor.shutdown(); + this.sinkDisruptor = null; + } + } + + public RingBuffer<TaskEventContainer> getSinkRingBuffer() { + return this.sinkDisruptor.getRingBuffer(); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/SourceTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/SourceTask.java new file mode 100644 index 0000000..2a822ca --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/SourceTask.java
@@ -0,0 +1,130 @@ +/* + * 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.runtime.task.def; + +import org.apache.iotdb.collector.config.TaskRuntimeOptions; +import org.apache.iotdb.commons.concurrent.IoTThreadFactory; +import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor; +import org.apache.iotdb.pipe.api.PipeSource; +import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +public abstract class SourceTask extends Task implements TaskComponent { + + private static final Logger LOGGER = LoggerFactory.getLogger(SourceTask.class); + + // Source needs the processor's RingBuffer to publish event. + protected final ProcessorTask processorTask; + protected final int sourceParallelismNum; + protected final String taskId; + // Store the status of the tasks, Running or Stopped + protected static final Map<String, TaskState> SOURCE_TASK_STATUS = new ConcurrentHashMap<>(); + protected static final Map<String, ExecutorService> SOURCE_EXECUTOR_SERVICE = + new ConcurrentHashMap<>(); + // Source tasks list + protected static final Map<String, List<PipeSource>> SOURCE_TASK = new ConcurrentHashMap<>(); + + public SourceTask( + final String taskId, + final Map<String, String> sourceParams, + final ProcessorTask processorTask) { + this.taskId = taskId; + final PipeParameters params = new PipeParameters(sourceParams); + this.sourceParallelismNum = + params.getIntOrDefault( + TaskRuntimeOptions.TASK_SOURCE_PARALLELISM_NUM.key(), + TaskRuntimeOptions.TASK_SOURCE_PARALLELISM_NUM.value()); + this.processorTask = processorTask; + } + + @Override + public void start() throws Exception { + SOURCE_TASK_STATUS.computeIfPresent(taskId, (taskId, status) -> TaskState.Running); + SOURCE_TASK + .get(taskId) + .forEach( + source -> { + try { + source.start(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Override + public synchronized void stop() { + SOURCE_TASK_STATUS.computeIfPresent(taskId, (taskId, status) -> TaskState.Stopped); + SOURCE_TASK + .get(taskId) + .forEach( + source -> { + try { + source.close(); + } catch (Exception e) { + LOGGER.warn("Failed to close source", e); + } + }); + } + + @Override + public void drop() { + stop(); + SOURCE_TASK.remove(taskId); + SOURCE_TASK_STATUS.remove(taskId); + SOURCE_EXECUTOR_SERVICE.remove(taskId).shutdown(); + } + + protected void addSourceTask(final PipeSource source) { + if (!SOURCE_TASK.containsKey(taskId)) { + SOURCE_TASK.put(taskId, new CopyOnWriteArrayList<>()); + } + SOURCE_TASK.get(taskId).add(source); + } + + protected void createSourceTask() { + SOURCE_EXECUTOR_SERVICE.putIfAbsent( + taskId, + new WrappedThreadPoolExecutor( + sourceParallelismNum, + sourceParallelismNum, + 0L, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + new IoTThreadFactory("source-executor"), + "source-executor")); + SOURCE_TASK_STATUS.putIfAbsent(taskId, TaskState.Running); + } + + protected enum TaskState { + Running, + Stopped + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/Task.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/Task.java new file mode 100644 index 0000000..8468795 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/Task.java
@@ -0,0 +1,69 @@ +/* + * 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.runtime.task.def; + +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventCollector; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventConsumer; +import org.apache.iotdb.collector.runtime.task.datastructure.TaskEventConsumerController; +import org.apache.iotdb.pipe.api.PipePlugin; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Constructor; +import java.lang.reflect.InvocationTargetException; +import java.util.Objects; +import java.util.stream.Stream; + +public abstract class Task { + + private static final Logger LOGGER = LoggerFactory.getLogger(Task.class); + + protected TaskEventConsumer[] getConsumer( + final PipePlugin plugin, final int consumerNum, final TaskEventCollector collector) { + return Stream.generate(() -> createConsumer(plugin, collector)) + .limit(consumerNum) + .toArray(TaskEventConsumer[]::new); + } + + private TaskEventConsumer createConsumer( + final PipePlugin plugin, final TaskEventCollector collector) { + return Objects.nonNull(collector) + ? new TaskEventConsumer(plugin, collector, new TaskEventConsumerController()) + : new TaskEventConsumer(plugin, new TaskEventConsumerController()); + } + + protected <T> T createInstance(final Class<T> clazz) { + try { + final Constructor<T> constructor = clazz.getDeclaredConstructor(); + constructor.setAccessible(true); + return constructor.newInstance(); + } catch (final NoSuchMethodException e) { + LOGGER.warn("class {} is abstract class.", clazz, e); + } catch (final IllegalAccessException e) { + LOGGER.warn("failed to visit class {} constructor method.", clazz, e); + } catch (final InstantiationException e) { + LOGGER.warn("failed to instantiate class {}.", clazz, e); + } catch (final InvocationTargetException e) { + LOGGER.warn("the constructor threw an exception.", e); + } + throw new RuntimeException(); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskComponent.java similarity index 68% copy from iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java copy to iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskComponent.java index 4f13ff9..729bc6d 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/agent/task/CollectorTask.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskComponent.java
@@ -16,22 +16,14 @@ * specific language governing permissions and limitations * under the License. */ +package org.apache.iotdb.collector.runtime.task.def; -package org.apache.iotdb.collector.agent.task; +public interface TaskComponent { + void create() throws Exception; -import org.apache.iotdb.commons.concurrent.WrappedRunnable; + void start() throws Exception; -public abstract class CollectorTask extends WrappedRunnable { + void stop(); - protected final String taskId; - - protected CollectorTask(final String taskId) { - this.taskId = taskId; - } - - public String getTaskId() { - return taskId; - } - - public abstract void stop(); + void drop(); }
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskRepository.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskRepository.java new file mode 100644 index 0000000..c5a1626 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/def/TaskRepository.java
@@ -0,0 +1,83 @@ +/* + * 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.runtime.task.def; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class TaskRepository { + + private static final Logger LOGGER = LoggerFactory.getLogger(TaskRepository.class); + + private final TaskComponent sourceComponent; + private final TaskComponent processorComponent; + private final TaskComponent sinkComponent; + + public TaskRepository( + final TaskComponent sourceComponent, + final TaskComponent processorComponent, + final TaskComponent sinkComponent) { + this.sourceComponent = sourceComponent; + this.processorComponent = processorComponent; + this.sinkComponent = sinkComponent; + } + + // Disruptor consumers must be started before producers + public void create() { + try { + sinkComponent.create(); + processorComponent.create(); + sourceComponent.create(); + } catch (final Exception e) { + LOGGER.warn("Failed to create task", e); + } + } + + // Disruptor consumers must be started before producers + public void start() { + try { + sinkComponent.start(); + processorComponent.start(); + sourceComponent.start(); + } catch (final Exception e) { + LOGGER.warn("Failed to start task", e); + } + } + + public void stop() { + try { + sourceComponent.stop(); + processorComponent.stop(); + sinkComponent.stop(); + } catch (final Exception e) { + LOGGER.warn("Failed to stop task", e); + } + } + + public void drop() { + try { + sourceComponent.drop(); + processorComponent.drop(); + sinkComponent.drop(); + } catch (final Exception e) { + LOGGER.warn("Failed to drop task", e); + } + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/exception/DisruptorTaskExceptionHandler.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/exception/DisruptorTaskExceptionHandler.java new file mode 100644 index 0000000..8716d84 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/runtime/task/exception/DisruptorTaskExceptionHandler.java
@@ -0,0 +1,43 @@ +/* + * 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.runtime.task.exception; + +import com.lmax.disruptor.ExceptionHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DisruptorTaskExceptionHandler implements ExceptionHandler<Object> { + private static final Logger LOGGER = LoggerFactory.getLogger(DisruptorTaskExceptionHandler.class); + + @Override + public void handleEventException(Throwable ex, long sequence, Object event) { + LOGGER.error("Event processing failed [seq={}, event={}]", sequence, event, ex); + } + + @Override + public void handleOnStartException(Throwable ex) { + LOGGER.error("Failed to start disruptor", ex); + } + + @Override + public void handleOnShutdownException(Throwable ex) { + LOGGER.error("Failed to shutdown disruptor", ex); + } +}
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/service/ApiService.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/service/ApiService.java index b079014..a75d016 100644 --- a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/service/ApiService.java +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/service/ApiService.java
@@ -63,7 +63,12 @@ holder.setInitOrder(1); holder.setInitParameter( "jersey.config.server.provider.packages", - "io.swagger.jaxrs.listing, io.swagger.sample.resource, org.apache.iotdb.collector.api"); + String.join( + ",", + "io.swagger.jaxrs.listing, io.swagger.sample.resource", + "org.apache.iotdb.collector.api.ping", + "org.apache.iotdb.collector.api.v1.plugin", + "org.apache.iotdb.collector.api.v1.task")); holder.setInitParameter( "jersey.config.server.provider.classnames", "org.glassfish.jersey.media.multipart.MultiPartFeature");
diff --git a/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/service/RuntimeService.java b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/service/RuntimeService.java new file mode 100644 index 0000000..5d15016 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/java/org/apache/iotdb/collector/service/RuntimeService.java
@@ -0,0 +1,54 @@ +/* + * 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.service; + +import org.apache.iotdb.collector.runtime.plugin.PluginRuntime; +import org.apache.iotdb.collector.runtime.task.TaskRuntime; + +public class RuntimeService implements IService { + + private static TaskRuntime task; + private static PluginRuntime plugin; + + @Override + public void start() { + plugin = new PluginRuntime(); + task = new TaskRuntime(); + } + + public static TaskRuntime task() { + return task; + } + + public static PluginRuntime plugin() { + return plugin; + } + + @Override + public void stop() { + task = null; + plugin = null; + } + + @Override + public String name() { + return "RuntimeService"; + } +}
diff --git a/iotdb-collector/collector-core/src/main/resources/application.properties b/iotdb-collector/collector-core/src/main/resources/application.properties index 7bcf3f0..3c69299 100644 --- a/iotdb-collector/collector-core/src/main/resources/application.properties +++ b/iotdb-collector/collector-core/src/main/resources/application.properties
@@ -17,4 +17,40 @@ # under the License. # -api_service_port=17070 \ No newline at end of file +#################### +### API Service Configuration +#################### + +# jetty service running port +# effectiveMode: first_start +# Datatype: int +api_service_port=17070 + +#################### +### Task Runtime Configuration +#################### + +# the number of concurrent threads that the source task runs +# effectiveMode: first_start +# Datatype: int +task_source_parallelism_num=4 + +# the number of concurrent threads that the process task runs +# effectiveMode: first_start +# Datatype: int +task_process_parallelism_num=4 + +# the number of concurrent threads that the sink task runs +# effectiveMode: first_start +# Datatype: int +task_sink_parallelism_num=4 + +# the disruptor ring buffer size of the processor task +# effectiveMode: first_start +# Datatype: int +task_processor_ring_buffer_size=1024 + +# the disruptor ring buffer size of the sink task +# effectiveMode: first_start +# Datatype: int +task_sink_ring_buffer_size=1024 \ No newline at end of file
diff --git a/iotdb-collector/collector-core/src/main/resources/logback.xml b/iotdb-collector/collector-core/src/main/resources/logback.xml new file mode 100644 index 0000000..60ee2c1 --- /dev/null +++ b/iotdb-collector/collector-core/src/main/resources/logback.xml
@@ -0,0 +1,44 @@ +<?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. + +--> +<configuration> + <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"> + <encoder> + <pattern>%d{yyyy-MM-dd HH:mm:ss} %-5level %logger - %msg%n</pattern> + </encoder> + </appender> + <appender name="FILE" class="ch.qos.logback.core.FileAppender"> + <file>logs/application.log</file> + <encoder> + <pattern>%date %level [%thread] %logger{100} [%file:%line] %msg%n</pattern> + </encoder> + </appender> + <root level="info"> + <appender-ref ref="CONSOLE"/> + <appender-ref ref="FILE"/> + </root> + <logger name="org.apache.iotdb.collector" level="debug" additivity="false"> + <appender-ref ref="CONSOLE"/> + <appender-ref ref="FILE"/> + </logger> + <logger name="org.eclipse.jetty" level="warn"/> + <logger name="javax.websocket.server" level="warn"/> +</configuration>
diff --git a/iotdb-collector/collector-openapi/pom.xml b/iotdb-collector/collector-openapi/pom.xml index 1c1fcc8..4563d4c 100644 --- a/iotdb-collector/collector-openapi/pom.xml +++ b/iotdb-collector/collector-openapi/pom.xml
@@ -88,13 +88,13 @@ </goals> <phase>generate-sources</phase> <configuration> - <inputSpec>${project.basedir}/src/main/openapi3/ping.yaml</inputSpec> + <inputSpec>${project.basedir}/src/main/openapi3/ping/ping.yaml</inputSpec> <output>${project.build.directory}/generated-sources/java</output> - <apiPackage>org.apache.iotdb.collector.api</apiPackage> - <modelPackage>org.apache.iotdb.collector.api.model</modelPackage> - <invokerPackage>org.apache.iotdb.collector.api.invoker</invokerPackage> + <apiPackage>org.apache.iotdb.collector.api.ping</apiPackage> + <modelPackage>org.apache.iotdb.collector.api.ping.model</modelPackage> + <invokerPackage>org.apache.iotdb.collector.api.ping.invoker</invokerPackage> <generatorName>jaxrs-jersey</generatorName> - <groupId>org.apache.iotdb.</groupId> + <groupId>org.apache.iotdb</groupId> <artifactId>iotdb-collector-rest-service</artifactId> <artifactVersion>${project.version}</artifactVersion> <addCompileSourceRoot>true</addCompileSourceRoot> @@ -109,17 +109,44 @@ </configuration> </execution> <execution> - <id>generate-java-rest-codes-v1</id> + <id>generate-java-rest-codes-v1-task</id> <goals> <goal>generate</goal> </goals> <phase>generate-sources</phase> <configuration> - <inputSpec>${project.basedir}/src/main/openapi3/iotdb_collector_rest_v1.yaml</inputSpec> + <inputSpec>${project.basedir}/src/main/openapi3/v1/task.yaml</inputSpec> <output>${project.build.directory}/generated-sources/java</output> - <apiPackage>org.apache.iotdb.collector.api.v1</apiPackage> - <modelPackage>org.apache.iotdb.collector.api.v1.model</modelPackage> - <invokerPackage>org.apache.iotdb.collector.api.v1.invoker</invokerPackage> + <apiPackage>org.apache.iotdb.collector.api.v1.task</apiPackage> + <modelPackage>org.apache.iotdb.collector.api.v1.task.model</modelPackage> + <invokerPackage>org.apache.iotdb.collector.api.v1.task.invoker</invokerPackage> + <generatorName>jaxrs-jersey</generatorName> + <groupId>org.apache.iotdb</groupId> + <artifactId>iotdb-collector-rest-service</artifactId> + <artifactVersion>${project.version}</artifactVersion> + <addCompileSourceRoot>true</addCompileSourceRoot> + <configOptions> + <licenseName>Apache License 2.0</licenseName> + <groupId>org.apache.iotdb</groupId> + <artifactId>iotdb-collector-rest-service</artifactId> + <artifactVersion>${project.version}</artifactVersion> + <dateLibrary>java8</dateLibrary> + <useGzipFeature>true</useGzipFeature> + </configOptions> + </configuration> + </execution> + <execution> + <id>generate-java-rest-codes-v1-plugin</id> + <goals> + <goal>generate</goal> + </goals> + <phase>generate-sources</phase> + <configuration> + <inputSpec>${project.basedir}/src/main/openapi3/v1/plugin.yaml</inputSpec> + <output>${project.build.directory}/generated-sources/java</output> + <apiPackage>org.apache.iotdb.collector.api.v1.plugin</apiPackage> + <modelPackage>org.apache.iotdb.collector.api.v1.plugin.model</modelPackage> + <invokerPackage>org.apache.iotdb.collector.api.v1.plugin.invoker</invokerPackage> <generatorName>jaxrs-jersey</generatorName> <groupId>org.apache.iotdb</groupId> <artifactId>iotdb-collector-rest-service</artifactId> @@ -158,11 +185,11 @@ <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-jar-plugin</artifactId> - <configuration> + <!--<configuration> <excludes> <exclude>**/impl/**</exclude> </excludes> - </configuration> + </configuration>--> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId>
diff --git a/iotdb-collector/collector-openapi/src/main/openapi3/iotdb_collector_rest_v1.yaml b/iotdb-collector/collector-openapi/src/main/openapi3/iotdb_collector_rest_v1.yaml deleted file mode 100644 index acb16b2..0000000 --- a/iotdb-collector/collector-openapi/src/main/openapi3/iotdb_collector_rest_v1.yaml +++ /dev/null
@@ -1,253 +0,0 @@ -# -# 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. -# - -openapi: 3.0.0 -info: - title: iotdb-collector-rest - description: IoTDB Rest API for Collector - license: - name: Apache 2.0 - url: https://www.apache.org/licenses/LICENSE-2.0.html - version: 1.0.0 -servers: - - url: http://127.0.0.1:17070/ - description: api -security: - - basic: [] -paths: - /admin/v1/createPipe: - post: - summary: createPipe - description: createPipe - operationId: createPipe - requestBody: - content: - application/json: - schema: - $ref: '#/components/schemas/CreatePipeRequest' - responses: - "200": - description: ExecutionStatus - content: - application/json: - schema: - $ref: '#/components/schemas/ExecutionStatus' - - /admin/v1/alterPipe: - post: - summary: alterPipe - description: alterPipe - operationId: alterPipe - requestBody: - content: - application/json: - schema: - $ref: '#/components/schemas/AlterPipeRequest' - responses: - "200": - description: ExecutionStatus - content: - application/json: - schema: - $ref: '#/components/schemas/ExecutionStatus' - - /admin/v1/startPipe: - post: - summary: startPipe - description: startPipe - operationId: startPipe - requestBody: - content: - application/json: - schema: - $ref: '#/components/schemas/StartPipeRequest' - responses: - "200": - description: ExecutionStatus - content: - application/json: - schema: - $ref: '#/components/schemas/ExecutionStatus' - - /admin/v1/stopPipe: - post: - summary: stopPipe - description: stopPipe - operationId: stopPipe - requestBody: - content: - application/json: - schema: - $ref: '#/components/schemas/StopPipeRequest' - responses: - "200": - description: ExecutionStatus - content: - application/json: - schema: - $ref: '#/components/schemas/ExecutionStatus' - - /admin/v1/dropPipe: - post: - summary: dropPipe - description: dropPipe - operationId: dropPipe - requestBody: - content: - application/json: - schema: - $ref: '#/components/schemas/DropPipeRequest' - responses: - "200": - description: ExecutionStatus - content: - application/json: - schema: - $ref: '#/components/schemas/ExecutionStatus' - -components: - schemas: - CreatePipeRequest: - title: CreatePipeRequest - type: object - properties: - sourceAttribute: - type: object - additionalProperties: - type: string - processorAttribute: - type: object - additionalProperties: - type: string - sinkAttribute: - type: object - additionalProperties: - type: string - taskId: - type: string - - - AlterPipeRequest: - title: AlterPipeRequest - type: object - properties: - timestamps: - type: array - items: - type: integer - format: int64 - measurements: - type: array - items: - type: string - dataTypes: - type: array - items: - type: string - values: - type: array - items: - type: array - items: - type: object - isAligned: - type: boolean - deviceId: - type: string - - StartPipeRequest: - title: StartPipeRequest - type: object - properties: - timestamps: - type: array - items: - type: integer - format: int64 - measurements: - type: array - items: - type: string - dataTypes: - type: array - items: - type: string - values: - type: array - items: - type: array - items: - type: object - isAligned: - type: boolean - deviceId: - type: string - - StopPipeRequest: - title: StopPipeRequest - type: object - properties: - taskId: - type: string - - DropPipeRequest: - title: DropPipeRequest - type: object - properties: - timestamps: - type: array - items: - type: integer - format: int64 - measurements: - type: array - items: - type: string - dataTypes: - type: array - items: - type: string - values: - type: array - items: - type: array - items: - type: object - isAligned: - type: boolean - deviceId: - type: string - - ExecutionStatus: - type: object - properties: - code: - type: integer - format: int32 - message: - type: string - - securitySchemes: - basic: - type: http - scheme: basic - APIKey: - type: apiKey - name: API Key - in: header
diff --git a/iotdb-collector/collector-openapi/src/main/openapi3/ping.yaml b/iotdb-collector/collector-openapi/src/main/openapi3/ping/ping.yaml similarity index 95% rename from iotdb-collector/collector-openapi/src/main/openapi3/ping.yaml rename to iotdb-collector/collector-openapi/src/main/openapi3/ping/ping.yaml index 36adab4..0a20ff9 100644 --- a/iotdb-collector/collector-openapi/src/main/openapi3/ping.yaml +++ b/iotdb-collector/collector-openapi/src/main/openapi3/ping/ping.yaml
@@ -20,19 +20,18 @@ openapi: 3.0.0 info: title: ping - description: IoTDB Rest API for Collector license: name: Apache 2.0 url: https://www.apache.org/licenses/LICENSE-2.0.html version: 1.0.0 servers: - url: http://127.0.0.1:17070/ - description: api security: - basic: [] paths: /ping: get: + operationId: tryPing responses: "200": description: ExecutionStatus @@ -40,7 +39,6 @@ application/json: schema: $ref: '#/components/schemas/ExecutionStatus' - operationId: tryPing components: schemas:
diff --git a/iotdb-collector/collector-openapi/src/main/openapi3/v1/plugin.yaml b/iotdb-collector/collector-openapi/src/main/openapi3/v1/plugin.yaml new file mode 100644 index 0000000..25ea5b5 --- /dev/null +++ b/iotdb-collector/collector-openapi/src/main/openapi3/v1/plugin.yaml
@@ -0,0 +1,156 @@ +# +# 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. +# + +openapi: 3.0.0 +info: + title: iotdb-collector-rest + description: IoTDB Rest API for Collector + license: + name: Apache 2.0 + url: https://www.apache.org/licenses/LICENSE-2.0.html + version: 1.0.0 +servers: + - url: http://127.0.0.1:17070/ + description: api +paths: + /plugin/v1/create: + post: + operationId: createPlugin + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/CreatePluginRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /plugin/v1/alter: + post: + operationId: alterPlugin + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/AlterPluginRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /plugin/v1/start: + post: + operationId: startPlugin + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/StartPluginRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /plugin/v1/stop: + post: + operationId: stopPlugin + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/StopPluginRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /plugin/v1/drop: + post: + operationId: dropPlugin + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/DropPluginRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + +components: + schemas: + CreatePluginRequest: + properties: + sourceAttribute: + type: object + additionalProperties: + type: string + processorAttribute: + type: object + additionalProperties: + type: string + sinkAttribute: + type: object + additionalProperties: + type: string + taskId: + type: string + + AlterPluginRequest: + properties: + sourceAttribute: + type: object + additionalProperties: + type: string + processorAttribute: + type: object + additionalProperties: + type: string + sinkAttribute: + type: object + additionalProperties: + type: string + taskId: + type: string + + StartPluginRequest: + properties: + taskId: + type: string + + StopPluginRequest: + properties: + taskId: + type: string + + DropPluginRequest: + properties: + taskId: + type: string + + ExecutionStatus: + properties: + code: + type: integer + format: int32 + message: + type: string + responses: + SuccessExecutionStatus: + description: ExecutionStatus + content: + application/json: + schema: + $ref: '#/components/schemas/ExecutionStatus' \ No newline at end of file
diff --git a/iotdb-collector/collector-openapi/src/main/openapi3/v1/task.yaml b/iotdb-collector/collector-openapi/src/main/openapi3/v1/task.yaml new file mode 100644 index 0000000..f2ba0f5 --- /dev/null +++ b/iotdb-collector/collector-openapi/src/main/openapi3/v1/task.yaml
@@ -0,0 +1,154 @@ +# +# 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. +# + +openapi: 3.0.0 +info: + title: iotdb-collector-rest + license: + name: Apache 2.0 + url: https://www.apache.org/licenses/LICENSE-2.0.html + version: 1.0.0 +servers: + - url: http://127.0.0.1:17070/ +paths: + /task/v1/create: + post: + operationId: createTask + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/CreateTaskRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /task/v1/alter: + post: + operationId: alterTask + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/AlterTaskRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /task/v1/start: + post: + operationId: startTask + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/StartTaskRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /task/v1/stop: + post: + operationId: stopTask + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/StopTaskRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + + /task/v1/drop: + post: + operationId: dropTask + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/DropTaskRequest' + responses: + "200": + $ref: '#/components/responses/SuccessExecutionStatus' + +components: + schemas: + CreateTaskRequest: + properties: + sourceAttribute: + type: object + additionalProperties: + type: string + processorAttribute: + type: object + additionalProperties: + type: string + sinkAttribute: + type: object + additionalProperties: + type: string + taskId: + type: string + + AlterTaskRequest: + properties: + sourceAttribute: + type: object + additionalProperties: + type: string + processorAttribute: + type: object + additionalProperties: + type: string + sinkAttribute: + type: object + additionalProperties: + type: string + taskId: + type: string + + StartTaskRequest: + properties: + taskId: + type: string + + StopTaskRequest: + properties: + taskId: + type: string + + DropTaskRequest: + properties: + taskId: + type: string + + ExecutionStatus: + properties: + code: + type: integer + format: int32 + message: + type: string + responses: + SuccessExecutionStatus: + description: ExecutionStatus + content: + application/json: + schema: + $ref: '#/components/schemas/ExecutionStatus' \ No newline at end of file
diff --git a/pom.xml b/pom.xml index f46f2fb..fca72e4 100644 --- a/pom.xml +++ b/pom.xml
@@ -845,6 +845,11 @@ <artifactId>jersey-hk2</artifactId> <version>${jersey.version}</version> </dependency> + <dependency> + <groupId>com.lmax</groupId> + <artifactId>disruptor</artifactId> + <version>${disruptor.version}</version> + </dependency> <!-- Conflict: json-smart (pulls in 9.3), cglib (pulls in 7.1)