diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java index 14135dc73..34e643670 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsPersistentWorker.java @@ -28,6 +28,7 @@ import org.apache.skywalking.oap.server.core.exporter.ExportEvent; import org.apache.skywalking.oap.server.core.storage.IMetricsDAO; 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.client.request.PrepareRequest; import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder; import org.slf4j.*; @@ -45,10 +46,9 @@ public class MetricsPersistentWorker extends PersistenceWorker nextExportWorker; private final DataCarrier dataCarrier; - MetricsPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, int batchSize, - IMetricsDAO metricsDAO, AbstractWorker nextAlarmWorker, + MetricsPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IMetricsDAO metricsDAO, AbstractWorker nextAlarmWorker, AbstractWorker nextExportWorker) { - super(moduleDefineHolder, batchSize); + super(moduleDefineHolder); this.model = model; this.mergeDataCache = new MergeDataCache<>(); this.metricsDAO = metricsDAO; @@ -76,7 +76,6 @@ public class MetricsPersistentWorker extends PersistenceWorker prepareBatch(MergeDataCache cache) { + @Override public void prepareBatch(MergeDataCache cache, List prepareRequests) { long start = System.currentTimeMillis(); - List batchCollection = new LinkedList<>(); - Collection collection = cache.getLast().collection(); int i = 0; @@ -131,9 +116,9 @@ public class MetricsPersistentWorker extends PersistenceWorker 0) { - logger.debug("prepareBatch model {}, took time: {}", model.getName(), System.currentTimeMillis() - start); + if (prepareRequests.size() > 0) { + logger.debug("prepare batch requests for model {}, took time: {}", model.getName(), System.currentTimeMillis() - start); } - - return batchCollection; } @Override public void cacheData(Metrics input) { @@ -186,17 +169,7 @@ public class MetricsPersistentWorker extends PersistenceWorker data) { - Iterator inputIterator = data.iterator(); - - int i = 0; - while (inputIterator.hasNext()) { - Metrics metrics = inputIterator.next(); - i++; - if (i == data.size()) { - metrics.asEndOfBatch(); - } - persistent.onWork(metrics); - } + data.forEach(persistent::onWork); } @Override public void onError(List data, Throwable t) { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java index 78d03986a..e81906e4b 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/MetricsStreamProcessor.java @@ -18,25 +18,15 @@ package org.apache.skywalking.oap.server.core.analysis.worker; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import lombok.Getter; -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.*; +import org.apache.skywalking.oap.server.core.analysis.*; import org.apache.skywalking.oap.server.core.analysis.metrics.Metrics; import org.apache.skywalking.oap.server.core.config.DownsamplingConfigService; -import org.apache.skywalking.oap.server.core.storage.IMetricsDAO; -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.*; 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.core.storage.model.*; import org.apache.skywalking.oap.server.core.worker.IWorkerInstanceSetter; import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder; @@ -61,6 +51,7 @@ public class MetricsStreamProcessor implements StreamProcessor { } } + @SuppressWarnings("unchecked") public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class metricsClass) { if (DisableRegister.INSTANCE.include(stream.name())) { return; @@ -114,16 +105,14 @@ public class MetricsStreamProcessor implements StreamProcessor { AlarmNotifyWorker alarmNotifyWorker = new AlarmNotifyWorker(moduleDefineHolder); ExportWorker exportWorker = new ExportWorker(moduleDefineHolder); - MetricsPersistentWorker minutePersistentWorker = new MetricsPersistentWorker(moduleDefineHolder, model, - 1000, metricsDAO, alarmNotifyWorker, exportWorker); + MetricsPersistentWorker minutePersistentWorker = new MetricsPersistentWorker(moduleDefineHolder, model, metricsDAO, alarmNotifyWorker, exportWorker); persistentWorkers.add(minutePersistentWorker); return minutePersistentWorker; } private MetricsPersistentWorker worker(ModuleDefineHolder moduleDefineHolder, IMetricsDAO metricsDAO, Model model) { - MetricsPersistentWorker persistentWorker = new MetricsPersistentWorker(moduleDefineHolder, model, - 1000, metricsDAO, null, null); + MetricsPersistentWorker persistentWorker = new MetricsPersistentWorker(moduleDefineHolder, model, metricsDAO, null, null); persistentWorkers.add(persistentWorker); return persistentWorker; diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java index e8ac6083c..166eb9c2a 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/PersistenceWorker.java @@ -18,10 +18,11 @@ package org.apache.skywalking.oap.server.core.analysis.worker; -import java.util.*; +import java.util.List; import org.apache.skywalking.oap.server.core.analysis.data.Window; -import org.apache.skywalking.oap.server.core.storage.*; +import org.apache.skywalking.oap.server.core.storage.StorageData; import org.apache.skywalking.oap.server.core.worker.AbstractWorker; +import org.apache.skywalking.oap.server.library.client.request.PrepareRequest; import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder; import org.slf4j.*; @@ -32,28 +33,11 @@ public abstract class PersistenceWorker= batchSize) { - try { - if (getCache().trySwitchPointer()) { - getCache().switchPointer(); - - List collection = buildBatchCollection(); - batchDAO.asynchronous(collection); - } - } finally { - getCache().trySwitchPointerFinally(); - } - } cacheData(input); } @@ -73,10 +57,9 @@ public abstract class PersistenceWorker prepareBatch(CACHE cache); + public abstract void prepareBatch(CACHE cache, List prepareRequests); - public final List buildBatchCollection() { - List batchCollection = new LinkedList<>(); + public final void buildBatchRequests(List prepareRequests) { try { while (getCache().getLast().isWriting()) { try { @@ -87,11 +70,10 @@ public abstract class PersistenceWorker> { +public class RecordPersistentWorker extends AbstractWorker { private static final Logger logger = LoggerFactory.getLogger(RecordPersistentWorker.class); private final Model model; - private final NonMergeDataCache nonMergeDataCache; private final IRecordDAO recordDAO; - private final DataCarrier dataCarrier; + private final IBatchDAO batchDAO; - RecordPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, int batchSize, - IRecordDAO recordDAO) { - super(moduleDefineHolder, batchSize); + RecordPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IRecordDAO recordDAO) { + super(moduleDefineHolder); this.model = model; - this.nonMergeDataCache = new NonMergeDataCache<>(); this.recordDAO = recordDAO; - - String name = "RECORD_PERSISTENT"; - BulkConsumePool.Creator creator = new BulkConsumePool.Creator(name, 1, 20); - try { - ConsumerPoolFactory.INSTANCE.createIfAbsent(name, creator); - } catch (Exception e) { - throw new UnexpectedException(e.getMessage(), e); - } - - this.dataCarrier = new DataCarrier<>(1, 10000); - this.dataCarrier.consume(ConsumerPoolFactory.INSTANCE.get(name), new RecordPersistentWorker.PersistentConsumer(this)); + this.batchDAO = moduleDefineHolder.find(StorageModule.NAME).provider().getService(IBatchDAO.class); } @Override public void in(Record record) { - dataCarrier.produce(record); - } - - @Override public NonMergeDataCache getCache() { - return nonMergeDataCache; - } - - @Override public List prepareBatch(NonMergeDataCache cache) { - List batchCollection = new LinkedList<>(); - cache.getLast().collection().forEach(record -> { - try { - batchCollection.add(recordDAO.prepareBatchInsert(model, record)); - } catch (Throwable t) { - logger.error(t.getMessage(), t); - } - }); - return batchCollection; - } - - @Override public void cacheData(Record input) { - nonMergeDataCache.writing(); - nonMergeDataCache.add(input); - nonMergeDataCache.finishWriting(); - } - - private class PersistentConsumer implements IConsumer { - - private final RecordPersistentWorker persistent; - - private PersistentConsumer(RecordPersistentWorker persistent) { - this.persistent = persistent; - } - - @Override public void init() { - - } - - @Override public void consume(List data) { - for (Record record : data) { - persistent.onWork(record); - } - } - - @Override public void onError(List data, Throwable t) { - logger.error(t.getMessage(), t); - } - - @Override public void onExit() { + try { + InsertRequest insertRequest = recordDAO.prepareBatchInsert(model, record); + batchDAO.asynchronous(insertRequest); + } 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/RecordStreamProcessor.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java index b84d99b63..3397f7519 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java @@ -19,7 +19,6 @@ package org.apache.skywalking.oap.server.core.analysis.worker; import java.util.*; -import lombok.Getter; import org.apache.skywalking.oap.server.core.*; import org.apache.skywalking.oap.server.core.analysis.*; import org.apache.skywalking.oap.server.core.analysis.record.Record; @@ -48,8 +47,6 @@ public class RecordStreamProcessor implements StreamProcessor { } } - @Getter private List persistentWorkers = new ArrayList<>(); - @SuppressWarnings("unchecked") public void create(ModuleDefineHolder moduleDefineHolder, Stream stream, Class recordClass) { if (DisableRegister.INSTANCE.include(stream.name())) { @@ -66,9 +63,8 @@ public class RecordStreamProcessor implements StreamProcessor { IModelSetter modelSetter = moduleDefineHolder.find(CoreModule.NAME).provider().getService(IModelSetter.class); Model model = modelSetter.putIfAbsent(recordClass, stream.scopeId(), new Storage(stream.name(), true, true, Downsampling.Second), true); - RecordPersistentWorker persistentWorker = new RecordPersistentWorker(moduleDefineHolder, model, 4000, recordDAO); + RecordPersistentWorker persistentWorker = new RecordPersistentWorker(moduleDefineHolder, model, recordDAO); - persistentWorkers.add(persistentWorker); workers.put(recordClass, persistentWorker); } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java index 37a826542..0287f07dc 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/TopNWorker.java @@ -18,20 +18,21 @@ package org.apache.skywalking.oap.server.core.analysis.worker; -import java.util.*; +import java.util.List; import org.apache.skywalking.apm.commons.datacarrier.DataCarrier; import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer; import org.apache.skywalking.oap.server.core.analysis.data.LimitedSizeDataCache; import org.apache.skywalking.oap.server.core.analysis.topn.TopN; import org.apache.skywalking.oap.server.core.storage.IRecordDAO; import org.apache.skywalking.oap.server.core.storage.model.Model; +import org.apache.skywalking.oap.server.library.client.request.PrepareRequest; import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder; import org.slf4j.*; /** * Top N worker is a persistence worker, but no * - * @author wusheng + * @author wusheng, peng-yongsheng */ public class TopNWorker extends PersistenceWorker> { @@ -44,9 +45,9 @@ public class TopNWorker extends PersistenceWorker(topNSize); this.recordDAO = recordDAO; this.model = model; @@ -57,7 +58,7 @@ public class TopNWorker extends PersistenceWorker getCache() { return limitedSizeDataCache; } @@ -84,8 +76,6 @@ public class TopNWorker extends PersistenceWorker prepareBatch(LimitedSizeDataCache cache) { - List batchCollection = new LinkedList<>(); + @Override public void prepareBatch(LimitedSizeDataCache cache, List prepareRequests) { cache.getLast().collection().forEach(record -> { try { - batchCollection.add(recordDAO.prepareBatchInsert(model, record)); + prepareRequests.add(recordDAO.prepareBatchInsert(model, record)); } catch (Throwable t) { logger.error(t.getMessage(), t); } }); - return batchCollection; } @Override public void in(TopN n) { @@ -113,16 +101,17 @@ public class TopNWorker extends PersistenceWorker { + @Override public void init() { } @Override public void consume(List data) { - /** + /* * TopN is not following the batch size trigger mode. * No need to implement this method, the memory size is limited always. */ - data.forEach(row -> onWork(row)); + data.forEach(TopNWorker.this::onWork); } @Override public void onError(List data, Throwable t) { diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java index 2bd3a5655..3c1ed0c4d 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IBatchDAO.java @@ -19,13 +19,14 @@ package org.apache.skywalking.oap.server.core.storage; import java.util.List; +import org.apache.skywalking.oap.server.library.client.request.*; /** * @author peng-yongsheng */ public interface IBatchDAO extends DAO { - void asynchronous(List collection); + void asynchronous(InsertRequest insertRequest); - void synchronous(List collection); + void synchronous(List prepareRequests); } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java index 312b89f2f..e8a0724e9 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/IRecordDAO.java @@ -21,11 +21,12 @@ package org.apache.skywalking.oap.server.core.storage; import java.io.IOException; import org.apache.skywalking.oap.server.core.analysis.record.Record; import org.apache.skywalking.oap.server.core.storage.model.Model; +import org.apache.skywalking.oap.server.library.client.request.InsertRequest; /** * @author peng-yongsheng */ -public interface IRecordDAO extends DAO { +public interface IRecordDAO extends DAO { - INSERT prepareBatchInsert(Model model, Record record) throws IOException; + InsertRequest prepareBatchInsert(Model model, Record record) throws IOException; } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java index ffe3480ca..cedd0740c 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/PersistenceTimer.java @@ -23,6 +23,7 @@ import java.util.concurrent.*; import org.apache.skywalking.apm.util.RunnableWithExceptionProtection; import org.apache.skywalking.oap.server.core.CoreModuleConfig; import org.apache.skywalking.oap.server.core.analysis.worker.*; +import org.apache.skywalking.oap.server.library.client.request.PrepareRequest; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.telemetry.TelemetryModule; @@ -61,33 +62,37 @@ public enum PersistenceTimer { if (!isStarted) { Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( - new RunnableWithExceptionProtection(() -> extractDataAndSaveRecord(batchDAO), - t -> logger.error("Extract data and save record failure.", t)), 5, moduleConfig.getPersistentPeriod(), TimeUnit.SECONDS); - - Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate( - new RunnableWithExceptionProtection(() -> extractDataAndSaveMetrics(batchDAO), - t -> logger.error("Extract data and save metrics failure.", t)), 5, moduleConfig.getPersistentPeriod(), TimeUnit.SECONDS); + new RunnableWithExceptionProtection(() -> extractDataAndSave(batchDAO), + t -> logger.error("Extract data and save failure.", t)), 5, moduleConfig.getPersistentPeriod(), TimeUnit.SECONDS); this.isStarted = true; } } - private void extractDataAndSaveRecord(IBatchDAO batchDAO) { + private void extractDataAndSave(IBatchDAO batchDAO) { if (logger.isDebugEnabled()) { - logger.debug("Extract data and save record"); + logger.debug("Extract data and save"); } long startTime = System.currentTimeMillis(); try { HistogramMetrics.Timer timer = prepareLatency.createTimer(); - List records = new LinkedList(); + List prepareRequests = new LinkedList<>(); try { List persistenceWorkers = new ArrayList<>(); - persistenceWorkers.addAll(RecordStreamProcessor.getInstance().getPersistentWorkers()); persistenceWorkers.addAll(TopNStreamProcessor.getInstance().getPersistentWorkers()); + persistenceWorkers.addAll(MetricsStreamProcessor.getInstance().getPersistentWorkers()); - buildBatchCollection(persistenceWorkers, records); + persistenceWorkers.forEach(worker -> { + if (logger.isDebugEnabled()) { + logger.debug("extract {} worker data and save", worker.getClass().getName()); + } + + if (worker.flushAndSwitch()) { + worker.buildBatchRequests(prepareRequests); + } + }); if (debug) { logger.info("build batch persistence duration: {} ms", System.currentTimeMillis() - startTime); @@ -98,8 +103,8 @@ public enum PersistenceTimer { HistogramMetrics.Timer executeLatencyTimer = executeLatency.createTimer(); try { - if (CollectionUtils.isNotEmpty(records)) { - batchDAO.asynchronous(records); + if (CollectionUtils.isNotEmpty(prepareRequests)) { + batchDAO.synchronous(prepareRequests); } } finally { executeLatencyTimer.finish(); @@ -117,66 +122,4 @@ public enum PersistenceTimer { logger.info("Batch persistence duration: {} ms", System.currentTimeMillis() - startTime); } } - - private void extractDataAndSaveMetrics(IBatchDAO batchDAO) { - if (logger.isDebugEnabled()) { - logger.debug("Extract data and save metrics"); - } - - long startTime = System.currentTimeMillis(); - try { - HistogramMetrics.Timer timer = prepareLatency.createTimer(); - - List metrics = new LinkedList(); - try { - List persistenceWorkers = new ArrayList<>(MetricsStreamProcessor.getInstance().getPersistentWorkers()); - buildBatchCollection(persistenceWorkers, metrics); - - if (debug) { - logger.info("build metrics batch persistence duration: {} ms", System.currentTimeMillis() - startTime); - } - } finally { - timer.finish(); - } - - HistogramMetrics.Timer executeLatencyTimer = executeLatency.createTimer(); - try { - if (CollectionUtils.isNotEmpty(metrics)) { - batchDAO.synchronous(metrics); - } - } finally { - executeLatencyTimer.finish(); - } - } catch (Throwable e) { - errorCounter.inc(); - logger.error(e.getMessage(), e); - } finally { - if (logger.isDebugEnabled()) { - logger.debug("Persistence data save finish"); - } - } - - if (debug) { - logger.info("Batch persistence duration: {} ms", System.currentTimeMillis() - startTime); - } - } - - @SuppressWarnings("unchecked") - private void buildBatchCollection(List persistenceWorkers, List collection) { - persistenceWorkers.forEach(worker -> { - if (logger.isDebugEnabled()) { - logger.debug("extract {} worker data and save", worker.getClass().getName()); - } - - if (worker.flushAndSwitch()) { - List batchCollection = worker.buildBatchCollection(); - - if (logger.isDebugEnabled()) { - logger.debug("extract {} worker data size: {}", worker.getClass().getName(), batchCollection.size()); - } - - collection.addAll(batchCollection); - } - }); - } } diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java index b513d38c4..35cc46607 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/worker/AbstractWorker.java @@ -25,6 +25,7 @@ import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder; * @author peng-yongsheng */ public abstract class AbstractWorker { + @Getter private final ModuleDefineHolder moduleDefineHolder; public AbstractWorker(ModuleDefineHolder moduleDefineHolder) { diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java index d4d118d47..da8f4e4f8 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/BatchProcessEsDAO.java @@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base; import java.util.List; import org.apache.skywalking.oap.server.core.storage.IBatchDAO; import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; +import org.apache.skywalking.oap.server.library.client.request.*; import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.elasticsearch.action.bulk.*; import org.elasticsearch.action.index.IndexRequest; @@ -47,38 +48,23 @@ public class BatchProcessEsDAO extends EsDAO implements IBatchDAO { this.concurrentRequests = concurrentRequests; } - @Override public void asynchronous(List collection) { + @Override public void asynchronous(InsertRequest insertRequest) { if (bulkProcessor == null) { this.bulkProcessor = getClient().createBulkProcessor(bulkActions, flushInterval, concurrentRequests); } - if (logger.isDebugEnabled()) { - logger.debug("Asynchronous batch persistent data collection size: {}", collection.size()); - } - - if (CollectionUtils.isNotEmpty(collection)) { - collection.forEach(builder -> { - if (builder instanceof IndexRequest) { - this.bulkProcessor.add((IndexRequest)builder); - } - if (builder instanceof UpdateRequest) { - this.bulkProcessor.add((UpdateRequest)builder); - } - }); - this.bulkProcessor.flush(); - } + this.bulkProcessor.add((IndexRequest)insertRequest); } - @Override public void synchronous(List collection) { - if (CollectionUtils.isNotEmpty(collection)) { + @Override public void synchronous(List prepareRequests) { + if (CollectionUtils.isNotEmpty(prepareRequests)) { BulkRequest request = new BulkRequest(); - for (Object builder : collection) { - if (builder instanceof IndexRequest) { - request.add((IndexRequest)builder); - } - if (builder instanceof UpdateRequest) { - request.add((UpdateRequest)builder); + for (PrepareRequest prepareRequest : prepareRequests) { + if (prepareRequest instanceof InsertRequest) { + request.add((IndexRequest)prepareRequest); + } else { + request.add((UpdateRequest)prepareRequest); } } getClient().synchronousBulk(request); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java index 05bb6d295..2e7a86442 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/RecordEsDAO.java @@ -23,13 +23,13 @@ import org.apache.skywalking.oap.server.core.analysis.record.Record; import org.apache.skywalking.oap.server.core.storage.*; import org.apache.skywalking.oap.server.core.storage.model.Model; import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient; -import org.elasticsearch.action.index.IndexRequest; +import org.apache.skywalking.oap.server.library.client.request.InsertRequest; import org.elasticsearch.common.xcontent.XContentBuilder; /** * @author peng-yongsheng */ -public class RecordEsDAO extends EsDAO implements IRecordDAO { +public class RecordEsDAO extends EsDAO implements IRecordDAO { private final StorageBuilder storageBuilder; @@ -38,7 +38,7 @@ public class RecordEsDAO extends EsDAO implements IRecordDAO { this.storageBuilder = storageBuilder; } - @Override public IndexRequest prepareBatchInsert(Model model, Record record) throws IOException { + @Override public InsertRequest prepareBatchInsert(Model model, Record record) throws IOException { XContentBuilder builder = map2builder(storageBuilder.data2Map(record)); String modelName = TimeSeriesUtils.timeSeries(model, record.getTimeBucket()); return getClient().prepareInsert(modelName, record.id(), builder); 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/H2BatchDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2BatchDAO.java index bfc8148e3..279a6a86a 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2BatchDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2BatchDAO.java @@ -18,16 +18,18 @@ package org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao; -import java.sql.Connection; -import java.sql.SQLException; +import java.sql.*; import java.util.List; +import org.apache.skywalking.apm.commons.datacarrier.DataCarrier; +import org.apache.skywalking.apm.commons.datacarrier.consumer.*; +import org.apache.skywalking.oap.server.core.UnexpectedException; import org.apache.skywalking.oap.server.core.storage.IBatchDAO; import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; import org.apache.skywalking.oap.server.library.client.jdbc.hikaricp.JDBCHikariCPClient; +import org.apache.skywalking.oap.server.library.client.request.*; import org.apache.skywalking.oap.server.library.util.CollectionUtils; import org.apache.skywalking.oap.server.storage.plugin.jdbc.SQLExecutor; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.slf4j.*; /** * @author wusheng, peng-yongsheng @@ -37,24 +39,36 @@ public class H2BatchDAO implements IBatchDAO { private static final Logger logger = LoggerFactory.getLogger(H2BatchDAO.class); private JDBCHikariCPClient h2Client; + private final DataCarrier dataCarrier; public H2BatchDAO(JDBCHikariCPClient h2Client) { this.h2Client = h2Client; + + String name = "H2_ASYNCHRONOUS_BATCH_PERSISTENT"; + BulkConsumePool.Creator creator = new BulkConsumePool.Creator(name, 1, 20); + try { + ConsumerPoolFactory.INSTANCE.createIfAbsent(name, creator); + } catch (Exception e) { + throw new UnexpectedException(e.getMessage(), e); + } + + this.dataCarrier = new DataCarrier<>(1, 10000); + this.dataCarrier.consume(ConsumerPoolFactory.INSTANCE.get(name), new H2BatchDAO.H2BatchConsumer(this)); } - @Override public void synchronous(List collection) { - if (CollectionUtils.isEmpty(collection)) { + @Override public void synchronous(List prepareRequests) { + if (CollectionUtils.isEmpty(prepareRequests)) { return; } if (logger.isDebugEnabled()) { - logger.debug("batch sql statements execute, data size: {}", collection.size()); + logger.debug("batch sql statements execute, data size: {}", prepareRequests.size()); } try (Connection connection = h2Client.getConnection()) { - for (Object exe : collection) { + for (PrepareRequest prepareRequest : prepareRequests) { try { - SQLExecutor sqlExecutor = (SQLExecutor)exe; + SQLExecutor sqlExecutor = (SQLExecutor)prepareRequest; sqlExecutor.invoke(connection); } catch (SQLException e) { // Just avoid one execution failure makes the rest of batch failure. @@ -66,7 +80,31 @@ public class H2BatchDAO implements IBatchDAO { } } - @Override public void asynchronous(List collection) { - synchronous(collection); + @Override public void asynchronous(InsertRequest insertRequest) { + this.dataCarrier.produce(insertRequest); + } + + private class H2BatchConsumer implements IConsumer { + + private final H2BatchDAO h2BatchDAO; + + private H2BatchConsumer(H2BatchDAO h2BatchDAO) { + this.h2BatchDAO = h2BatchDAO; + } + + @Override public void init() { + + } + + @Override public void consume(List prepareRequests) { + h2BatchDAO.synchronous(prepareRequests); + } + + @Override public void onError(List prepareRequests, Throwable t) { + logger.error(t.getMessage(), t); + } + + @Override public void onExit() { + } } } 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/H2RecordDAO.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2RecordDAO.java index aa4c3751e..39158b4ca 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2RecordDAO.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/dao/H2RecordDAO.java @@ -23,12 +23,12 @@ import org.apache.skywalking.oap.server.core.analysis.record.Record; import org.apache.skywalking.oap.server.core.storage.*; 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 org.apache.skywalking.oap.server.library.client.request.InsertRequest; /** * @author wusheng */ -public class H2RecordDAO extends H2SQLExecutor implements IRecordDAO { +public class H2RecordDAO extends H2SQLExecutor implements IRecordDAO { private JDBCHikariCPClient h2Client; private StorageBuilder storageBuilder; @@ -38,7 +38,7 @@ public class H2RecordDAO extends H2SQLExecutor implements IRecordDAO getByIDs(JDBCHikariCPClient h2Client, String modelName, String[] ids, @@ -67,6 +58,7 @@ public class H2SQLExecutor { } } while (storageData != null); + return storageDataList; } } catch (SQLException | JDBCClientException e) { @@ -96,8 +88,7 @@ public class H2SQLExecutor { } } - protected StorageData toStorageData(ResultSet rs, String modelName, - StorageBuilder storageBuilder) throws SQLException { + protected StorageData toStorageData(ResultSet rs, String modelName, StorageBuilder storageBuilder) throws SQLException { if (rs.next()) { Map data = new HashMap(); List columns = TableMetaInfo.get(modelName).getColumns(); @@ -122,8 +113,7 @@ public class H2SQLExecutor { return Const.NONE; } - protected SQLExecutor getInsertExecutor(String modelName, StorageData metrics, - StorageBuilder storageBuilder) throws IOException { + protected SQLExecutor getInsertExecutor(String modelName, StorageData metrics, StorageBuilder storageBuilder) throws IOException { Map objectMap = storageBuilder.data2Map(metrics); SQLBuilder sqlBuilder = new SQLBuilder("INSERT INTO " + modelName + " VALUES"); @@ -150,8 +140,7 @@ public class H2SQLExecutor { return new SQLExecutor(sqlBuilder.toString(), param); } - protected SQLExecutor getUpdateExecutor(String modelName, StorageData metrics, - StorageBuilder storageBuilder) throws IOException { + protected SQLExecutor getUpdateExecutor(String modelName, StorageData metrics, StorageBuilder storageBuilder) throws IOException { Map objectMap = storageBuilder.data2Map(metrics); SQLBuilder sqlBuilder = new SQLBuilder("UPDATE " + modelName + " SET ");