provide config stream processor (#4133)

* provide config stream processor

* 1. rename config to none stream
2. add comments on none stream relate classes

Co-authored-by: 吴晟 Wu Sheng <wu.sheng@foxmail.com>

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java

@@ -53,6 +53,8 @@ public class StreamAnnotationListener implements AnnotationListener {
                 MetricsStreamProcessor.getInstance().create(moduleDefineHolder, stream, aClass);
             } else if (stream.processor().equals(TopNStreamProcessor.class)) {
                 TopNStreamProcessor.getInstance().create(moduleDefineHolder, stream, aClass);
+            } else if (stream.processor().equals(NoneStreamingProcessor.class)) {
+                NoneStreamingProcessor.getInstance().create(moduleDefineHolder, stream, aClass);
             } else {
                 throw new UnexpectedException("Unknown stream processor.");
             }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/config/NoneStream.java

@@ -0,0 +1,30 @@
+/*
+ * 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.skywalking.oap.server.core.analysis.config;
+
+import org.apache.skywalking.oap.server.core.analysis.record.Record;
+
+/**
+ * None stream data base on record, support time bucket field to TTL.
+ *
+ * @author MrPro
+ */
+public abstract class NoneStream extends Record {
+}
+

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamPersistentWorker.java

@@ -0,0 +1,57 @@
+/*
+ * 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.skywalking.oap.server.core.analysis.worker;
+
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.storage.INoneStreamDAO;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
+import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * None persistent use {@link INoneStreamDAO#insert(Model, NoneStream)} on get new data
+ *
+ * @author MrPro
+ */
+public class NoneStreamPersistentWorker extends AbstractWorker<NoneStream> {
+
+    private static final Logger logger = LoggerFactory.getLogger(NoneStreamPersistentWorker.class);
+
+    private final Model model;
+    private final INoneStreamDAO configDAO;
+
+    public NoneStreamPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, INoneStreamDAO configDAO) {
+        super(moduleDefineHolder);
+        this.model = model;
+        this.configDAO = configDAO;
+    }
+
+    @Override
+    public void in(NoneStream noneStream) {
+        try {
+            configDAO.insert(model, noneStream);
+        } catch (IOException e) {
+            logger.error(e.getMessage(), e);
+        }
+    }
+}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamingProcessor.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.analysis.worker;
+
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.UnexpectedException;
+import org.apache.skywalking.oap.server.core.analysis.DisableRegister;
+import org.apache.skywalking.oap.server.core.analysis.Downsampling;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.StreamProcessor;
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.storage.INoneStreamDAO;
+import org.apache.skywalking.oap.server.core.storage.StorageDAO;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
+import org.apache.skywalking.oap.server.core.storage.model.IModelSetter;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * none streaming is designed for user operation configuration in UI interface. It uses storage (synchronization) similar to Inventory and supports TTL deletion mode similar to the record.
+ *
+ * @author MrPro
+ */
+public class NoneStreamingProcessor implements StreamProcessor<NoneStream> {
+
+    private static final NoneStreamingProcessor PROCESSOR = new NoneStreamingProcessor();
+
+    private Map<Class<? extends NoneStream>, NoneStreamPersistentWorker> workers = new HashMap<>();
+
+    public static NoneStreamingProcessor getInstance() {
+        return PROCESSOR;
+    }
+
+    @Override
+    public void in(NoneStream noneStream) {
+        final NoneStreamPersistentWorker worker = workers.get(noneStream.getClass());
+        if (worker != null) {
+            worker.in(noneStream);
+        }
+    }
+
+    @Override
+    public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class<? extends NoneStream> streamClass) {
+        if (DisableRegister.INSTANCE.include(stream.name())) {
+            return;
+        }
+
+        StorageDAO storageDAO = moduleDefineHolder.find(StorageModule.NAME).provider().getService(StorageDAO.class);
+        INoneStreamDAO noneStream;
+        try {
+            noneStream = storageDAO.newNoneStreamDao(stream.builder().newInstance());
+        } catch (InstantiationException | IllegalAccessException e) {
+            throw new UnexpectedException("Create " + stream.builder().getSimpleName() + " none stream record DAO failure.", e);
+        }
+
+        IModelSetter modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(IModelSetter.class);
+        Model model = modelSetter.putIfAbsent(streamClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Second), true);
+
+        final NoneStreamPersistentWorker persistentWorker = new NoneStreamPersistentWorker(moduleDefineHolder, model, noneStream);
+        workers.put(streamClass, persistentWorker);
+    }
+}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/INoneStreamDAO.java

@@ -0,0 +1,35 @@
+/*
+ * 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.skywalking.oap.server.core.storage;
+
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+
+import java.io.IOException;
+
+/**
+ * Use synchronize storage to insert none stream data
+ *
+ * @author MrPro
+ */
+public interface INoneStreamDAO extends DAO {
+
+    void insert(Model model, NoneStream noneStream) throws IOException;
+
+}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.core.storage;
 
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.record.Record;
 import org.apache.skywalking.oap.server.core.register.RegisterSource;
@@ -33,4 +34,6 @@ public interface StorageDAO extends Service {
     IRegisterDAO newRegisterDao(StorageBuilder<RegisterSource> storageBuilder);
 
     IRecordDAO newRecordDao(StorageBuilder<Record> storageBuilder);
+
+    INoneStreamDAO newNoneStreamDao(StorageBuilder<NoneStream> storageBuilder);
 }

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
+
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.storage.INoneStreamDAO;
+import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
+/**
+ * Synchronize storage Elasticsearch implements
+ *
+ * @author MrPro
+ */
+public class NoneStreamEsDAO extends EsDAO implements INoneStreamDAO {
+
+    private final StorageBuilder<NoneStream> storageBuilder;
+
+    public NoneStreamEsDAO(ElasticSearchClient client, StorageBuilder<NoneStream> storageBuilder) {
+        super(client);
+        this.storageBuilder = storageBuilder;
+    }
+
+    @Override
+    public void insert(Model model, NoneStream noneStream) throws IOException {
+        XContentBuilder builder = map2builder(storageBuilder.data2Map(noneStream));
+        String modelName = model.getName();
+        getClient().forceInsert(modelName, noneStream.id(), builder);
+    }
+}

oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base;
 
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.record.Record;
 import org.apache.skywalking.oap.server.core.register.RegisterSource;
@@ -44,4 +45,9 @@ public class StorageEsDAO extends EsDAO implements StorageDAO {
     @Override public IRecordDAO newRecordDao(StorageBuilder<Record> storageBuilder) {
         return new RecordEsDAO(getClient(), storageBuilder);
     }
+
+    @Override
+    public INoneStreamDAO newNoneStreamDao(StorageBuilder<NoneStream> storageBuilder) {
+        return new NoneStreamEsDAO(getClient(), storageBuilder);
+    }
 }

oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/dao/StorageEs7DAO.java

@@ -18,15 +18,13 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.dao;
 
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.record.Record;
 import org.apache.skywalking.oap.server.core.register.RegisterSource;
-import org.apache.skywalking.oap.server.core.storage.IMetricsDAO;
+import org.apache.skywalking.oap.server.core.storage.*;
-import org.apache.skywalking.oap.server.core.storage.IRecordDAO;
-import org.apache.skywalking.oap.server.core.storage.IRegisterDAO;
-import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
-import org.apache.skywalking.oap.server.core.storage.StorageDAO;
 import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.NoneStreamEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.RecordEsDAO;
 import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.RegisterEsDAO;
@@ -51,4 +49,9 @@ public class StorageEs7DAO extends EsDAO implements StorageDAO {
     @Override public IRecordDAO newRecordDao(StorageBuilder<Record> storageBuilder) {
         return new RecordEsDAO(getClient(), storageBuilder);
     }
+
+    @Override
+    public INoneStreamDAO newNoneStreamDao(StorageBuilder<NoneStream> storageBuilder) {
+        return new NoneStreamEsDAO(getClient(), storageBuilder);
+    }
 }

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2NoneStreamDAO.java

@@ -0,0 +1,56 @@
+/*
+ * 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.skywalking.oap.server.storage.plugin.jdbc.h2.dao;
+
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
+import org.apache.skywalking.oap.server.core.storage.INoneStreamDAO;
+import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient;
+import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLExecutor;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/**
+ * Synchronize storage H2 implements
+ *
+ * @author MrPro
+ */
+public class H2NoneStreamDAO extends H2SQLExecutor implements INoneStreamDAO {
+
+    private JDBCHikariCPClient h2Client;
+    private StorageBuilder<NoneStream> storageBuilder;
+
+    public H2NoneStreamDAO(JDBCHikariCPClient h2Client, StorageBuilder<NoneStream> storageBuilder) {
+        this.h2Client = h2Client;
+        this.storageBuilder = storageBuilder;
+    }
+
+    @Override
+    public void insert(Model model, NoneStream noneStream) throws IOException {
+        try (Connection connection = h2Client.getConnection()) {
+            SQLExecutor insertExecutor = getInsertExecutor(model.getName(), noneStream, storageBuilder);
+            insertExecutor.invoke(connection);
+        } catch (IOException | SQLException e) {
+            throw new IOException(e.getMessage(), e);
+        }
+    }
+}

oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2StorageDAO.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao;
 
+import org.apache.skywalking.oap.server.core.analysis.config.NoneStream;
 import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics;
 import org.apache.skywalking.oap.server.core.analysis.record.Record;
 import org.apache.skywalking.oap.server.core.register.RegisterSource;
@@ -46,4 +47,9 @@ public class H2StorageDAO implements StorageDAO {
     @Override public IRecordDAO newRecordDao(StorageBuilder<Record> storageBuilder) {
         return new H2RecordDAO(h2Client, storageBuilder);
     }
+
+    @Override
+    public INoneStreamDAO newNoneStreamDao(StorageBuilder<NoneStream> storageBuilder) {
+        return new H2NoneStreamDAO(h2Client, storageBuilder);
+    }
 }