From 2fcf7c4649118237f6e2ac6de617e3ee394eef24 Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Fri, 27 Dec 2019 14:58:15 +0800 Subject: [PATCH] provide config stream processor (#4133) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * provide config stream processor * 1. rename config to none stream 2. add comments on none stream relate classes Co-authored-by: 吴晟 Wu Sheng --- .../analysis/StreamAnnotationListener.java | 2 + .../core/analysis/config/NoneStream.java | 30 +++++++ .../worker/NoneStreamPersistentWorker.java | 57 +++++++++++++ .../worker/NoneStreamingProcessor.java | 82 +++++++++++++++++++ .../server/core/storage/INoneStreamDAO.java | 35 ++++++++ .../oap/server/core/storage/StorageDAO.java | 3 + .../elasticsearch/base/NoneStreamEsDAO.java | 50 +++++++++++ .../elasticsearch/base/StorageEsDAO.java | 6 ++ .../elasticsearch7/dao/StorageEs7DAO.java | 13 +-- .../plugin/jdbc/h2/dao/H2NoneStreamDAO.java | 56 +++++++++++++ .../plugin/jdbc/h2/dao/H2StorageDAO.java | 6 ++ 11 files changed, 335 insertions(+), 5 deletions(-) create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/config/NoneStream.java create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamPersistentWorker.java create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamingProcessor.java create mode 100644 oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/INoneStreamDAO.java create mode 100644 oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java create mode 100644 oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2NoneStreamDAO.java diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java index df522612a2..6f61739dc8 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/StreamAnnotationListener.java +++ b/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."); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/config/NoneStream.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/config/NoneStream.java new file mode 100644 index 0000000000..2967c44081 --- /dev/null +++ b/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 { +} + diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamPersistentWorker.java new file mode 100644 index 0000000000..ea3a671452 --- /dev/null +++ b/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 { + + 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); + } + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamingProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/NoneStreamingProcessor.java new file mode 100644 index 0000000000..164b158618 --- /dev/null +++ b/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 { + + private static final NoneStreamingProcessor PROCESSOR = new NoneStreamingProcessor(); + + private Map, 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 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); + } +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/INoneStreamDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/INoneStreamDAO.java new file mode 100644 index 0000000000..9dbc033c46 --- /dev/null +++ b/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; + +} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java index 4d3f26d60d..6b7da06ac5 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/StorageDAO.java +++ b/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 storageBuilder); IRecordDAO newRecordDao(StorageBuilder storageBuilder); + + INoneStreamDAO newNoneStreamDao(StorageBuilder storageBuilder); } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/NoneStreamEsDAO.java new file mode 100644 index 0000000000..1608d2583c --- /dev/null +++ b/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 storageBuilder; + + public NoneStreamEsDAO(ElasticSearchClient client, StorageBuilder 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); + } +} diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java index 52fe3fb97a..851f984817 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/StorageEsDAO.java +++ b/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 storageBuilder) { return new RecordEsDAO(getClient(), storageBuilder); } + + @Override + public INoneStreamDAO newNoneStreamDao(StorageBuilder storageBuilder) { + return new NoneStreamEsDAO(getClient(), storageBuilder); + } } diff --git a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/dao/StorageEs7DAO.java b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/dao/StorageEs7DAO.java index 6ba0cf1478..5b75493644 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/dao/StorageEs7DAO.java +++ b/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.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.core.storage.*; 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 storageBuilder) { return new RecordEsDAO(getClient(), storageBuilder); } + + @Override + public INoneStreamDAO newNoneStreamDao(StorageBuilder storageBuilder) { + return new NoneStreamEsDAO(getClient(), storageBuilder); + } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2NoneStreamDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2NoneStreamDAO.java new file mode 100644 index 0000000000..e527005557 --- /dev/null +++ b/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 storageBuilder; + + public H2NoneStreamDAO(JDBCHikariCPClient h2Client, StorageBuilder 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); + } + } +} diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2StorageDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2StorageDAO.java index 1d95881348..8cca092707 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2StorageDAO.java +++ b/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 storageBuilder) { return new H2RecordDAO(h2Client, storageBuilder); } + + @Override + public INoneStreamDAO newNoneStreamDao(StorageBuilder storageBuilder) { + return new H2NoneStreamDAO(h2Client, storageBuilder); + } } -- GitLab