From f32d3d072081943e29a367ccb714b362f5acddc7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E6=99=9F=20Wu=20Sheng?= Date: Thu, 1 Jul 2021 23:25:52 +0800 Subject: [PATCH] Enhance the storage session mechanism (#7221) --- CHANGES.md | 3 ++ .../setup/backend/configuration-vocabulary.md | 1 + .../src/main/resources/application.yml | 2 ++ .../oap/server/core/CoreModuleConfig.java | 4 +++ .../oap/server/core/CoreModuleProvider.java | 6 ++-- .../server/core/analysis/metrics/Metrics.java | 22 +++++++++---- .../worker/MetricsPersistentWorker.java | 33 +++++++++++-------- .../worker/MetricsStreamProcessor.java | 9 +++-- .../analysis/worker/PersistenceWorker.java | 4 +-- .../core/analysis/worker/TopNWorker.java | 2 +- .../server/core/storage/PersistenceTimer.java | 10 +++--- 11 files changed, 62 insertions(+), 34 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index ea53ebb79..053f51aeb 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -63,6 +63,9 @@ Release Notes. metrics. The timeout of the cache for minute and hour level metrics has been prolonged to ~5 min. * Performance: Add L1 aggregation flush period, which reduce the CPU load and help young GC. * Support connectTimeout and socketTimeout settings for ElasticSearch6 and ElasticSearch7 storages. +* Re-implement storage session mechanism, cached metrics are removed only according to their last access timestamp, + rather than first time. This makes sure hot data never gets removed unexpectedly. +* Support session expired threshold configurable. #### UI diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md index d5be1dacb..6969c8035 100644 --- a/docs/en/setup/backend/configuration-vocabulary.md +++ b/docs/en/setup/backend/configuration-vocabulary.md @@ -24,6 +24,7 @@ core|default|role|Option values, `Mixed/Receiver/Aggregator`. **Receiver** mode | - | - | recordDataTTL|The lifecycle of record data. Record data includes traces, top n sampled records, and logs. Unit is day. Minimal value is 2.|SW_CORE_RECORD_DATA_TTL|3| | - | - | metricsDataTTL|The lifecycle of metrics data, including the metadata. Unit is day. Recommend metricsDataTTL >= recordDataTTL. Minimal value is 2.| SW_CORE_METRICS_DATA_TTL|7| | - | - | l1FlushPeriod| The period of L1 aggregation flush to L2 aggregation. Unit is ms. | SW_CORE_L1_AGGREGATION_FLUSH_PERIOD | 500 | +| - | - | storageSessionTimeout| The threshold of session time. Unit is ms. Default value is 70s. | SW_CORE_STORAGE_SESSION_TIMEOUT | 70000 | | - | - | enableDatabaseSession|Cache metrics data for 1 minute to reduce database queries, and if the OAP cluster changes within that minute.|SW_CORE_ENABLE_DATABASE_SESSION|true| | - | - | topNReportPeriod|The execution period of top N sampler, which saves sampled data into the storage. Unit is minute|SW_CORE_TOPN_REPORT_PERIOD|10| | - | - | activeExtraModelColumns|Append the names of entity, such as service name, into the metrics storage entities.|SW_CORE_ACTIVE_EXTRA_MODEL_COLUMNS|false| diff --git a/oap-server/server-bootstrap/src/main/resources/application.yml b/oap-server/server-bootstrap/src/main/resources/application.yml index 71769c0bd..d27121b94 100755 --- a/oap-server/server-bootstrap/src/main/resources/application.yml +++ b/oap-server/server-bootstrap/src/main/resources/application.yml @@ -87,6 +87,8 @@ core: metricsDataTTL: ${SW_CORE_METRICS_DATA_TTL:7} # Unit is day # The period of L1 aggregation flush to L2 aggregation. Unit is ms. l1FlushPeriod: ${SW_CORE_L1_AGGREGATION_FLUSH_PERIOD:500} + # The threshold of session time. Unit is ms. Default value is 70s. + storageSessionTimeout: ${SW_CORE_STORAGE_SESSION_TIMEOUT:70000} # Cache metrics data for 1 minute to reduce database queries, and if the OAP cluster changes within that minute, # the metrics may not be accurate within that minute. enableDatabaseSession: ${SW_CORE_ENABLE_DATABASE_SESSION:true} diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java index af5f735b3..5d3a02204 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleConfig.java @@ -55,6 +55,10 @@ public class CoreModuleConfig extends ModuleConfig { * Enable database flush session. */ private boolean enableDatabaseSession; + /** + * The threshold of session time. Unit is ms. Default value is 70s. + */ + private long storageSessionTimeout = 70_000; private final List downsampling; /** * The period of doing data persistence. Unit is second. diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java index a8bbcbe8e..af6c2ecb3 100755 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java @@ -288,8 +288,10 @@ public class CoreModuleProvider extends ModuleProvider { this.registerServiceImplementation( UITemplateManagementService.class, new UITemplateManagementService(getManager())); - MetricsStreamProcessor.getInstance().setEnableDatabaseSession(moduleConfig.isEnableDatabaseSession()); - MetricsStreamProcessor.getInstance().setL1FlushPeriod(moduleConfig.getL1FlushPeriod()); + final MetricsStreamProcessor metricsStreamProcessor = MetricsStreamProcessor.getInstance(); + metricsStreamProcessor.setEnableDatabaseSession(moduleConfig.isEnableDatabaseSession()); + metricsStreamProcessor.setL1FlushPeriod(moduleConfig.getL1FlushPeriod()); + metricsStreamProcessor.setStorageSessionTimeout(moduleConfig.getStorageSessionTimeout()); TopNStreamProcessor.getInstance().setTopNWorkerReportCycle(moduleConfig.getTopNReportPeriod()); apdexThresholdConfig = new ApdexThresholdConfig(this); ApdexMetrics.setDICT(apdexThresholdConfig); diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/Metrics.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/Metrics.java index 904929e4a..1bbcd063c 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/Metrics.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/Metrics.java @@ -50,13 +50,14 @@ public abstract class Metrics extends StreamData implements StorageData { * Time in the cache, only work when MetricsPersistentWorker#enableDatabaseSession == true. */ @Getter - private long survivalTime = 0L; + private long lastUpdateTimestamp = 0L; /** * Merge the given metrics instance, these two must be the same metrics type. * * @param metrics to be merged - * @return {@code true} if the combined metrics should be continuously processed. {@code false} means it should be abandoned, and the implementation needs to keep the data unaltered in this case. + * @return {@code true} if the combined metrics should be continuously processed. {@code false} means it should be + * abandoned, and the implementation needs to keep the data unaltered in this case. */ public abstract boolean combine(Metrics metrics); @@ -80,12 +81,21 @@ public abstract class Metrics extends StreamData implements StorageData { public abstract Metrics toDay(); /** - * Extend the {@link #survivalTime} + * Set the last update timestamp * - * @param value to extend + * @param timestamp last update timestamp */ - public void extendSurvivalTime(long value) { - survivalTime += value; + public void setLastUpdateTimestamp(long timestamp) { + lastUpdateTimestamp = timestamp; + } + + /** + * @param timestamp of current time + * @param expiredThreshold represents the duration between last update time and the time point removing from cache. + * @return true means this metrics should be removed from cache. + */ + public boolean isExpired(long timestamp, long expiredThreshold) { + return timestamp - lastUpdateTimestamp > expiredThreshold; } public long toTimeBucketInHour() { 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 b7c073570..5dd29db2a 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 @@ -65,12 +65,13 @@ public class MetricsPersistentWorker extends PersistenceWorker { private final Optional transWorker; private final boolean enableDatabaseSession; private final boolean supportUpdate; + private long sessionTimeout; private CounterMetrics aggregationCounter; - private long sessionTimeout = 70_000; // Unit, ms. 70,000ms means more than one minute. MetricsPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IMetricsDAO metricsDAO, AbstractWorker nextAlarmWorker, AbstractWorker nextExportWorker, - MetricsTransWorker transWorker, boolean enableDatabaseSession, boolean supportUpdate) { + MetricsTransWorker transWorker, boolean enableDatabaseSession, boolean supportUpdate, + long storageSessionTimeout) { super(moduleDefineHolder, new ReadWriteSafeCache<>(new MergableBufferedData(), new MergableBufferedData())); this.model = model; this.context = new HashMap<>(100); @@ -80,6 +81,7 @@ public class MetricsPersistentWorker extends PersistenceWorker { this.nextExportWorker = Optional.ofNullable(nextExportWorker); this.transWorker = Optional.ofNullable(transWorker); this.supportUpdate = supportUpdate; + this.sessionTimeout = storageSessionTimeout; String name = "METRICS_L2_AGGREGATION"; int size = BulkConsumePool.Creator.recommendMaxSize() / 8; @@ -111,15 +113,15 @@ public class MetricsPersistentWorker extends PersistenceWorker { * Create the leaf and down-sampling MetricsPersistentWorker, no next step. */ MetricsPersistentWorker(ModuleDefineHolder moduleDefineHolder, Model model, IMetricsDAO metricsDAO, - boolean enableDatabaseSession, boolean supportUpdate) { + boolean enableDatabaseSession, boolean supportUpdate, long storageSessionTimeout) { this(moduleDefineHolder, model, metricsDAO, null, null, null, - enableDatabaseSession, supportUpdate + enableDatabaseSession, supportUpdate, storageSessionTimeout ); // For a down-sampling metrics, we prolong the session timeout for 4 times, nearly 5 minutes. // And add offset according to worker creation sequence, to avoid context clear overlap, // eventually optimize load of IDs reading. - this.sessionTimeout = sessionTimeout * 4 + SESSION_TIMEOUT_OFFSITE_COUNTER * 200; + this.sessionTimeout = this.sessionTimeout * 4 + SESSION_TIMEOUT_OFFSITE_COUNTER * 200; } /** @@ -171,6 +173,7 @@ public class MetricsPersistentWorker extends PersistenceWorker { try { loadFromStorage(metricsList); + long timestamp = System.currentTimeMillis(); for (Metrics metrics : metricsList) { Metrics cachedMetrics = context.get(metrics); if (cachedMetrics != null) { @@ -191,10 +194,12 @@ public class MetricsPersistentWorker extends PersistenceWorker { cachedMetrics.calculate(); prepareRequests.add(metricsDAO.prepareBatchUpdate(model, cachedMetrics)); nextWorker(cachedMetrics); + cachedMetrics.setLastUpdateTimestamp(timestamp); } else { metrics.calculate(); prepareRequests.add(metricsDAO.prepareBatchInsert(model, metrics)); nextWorker(metrics); + metrics.setLastUpdateTimestamp(timestamp); } /* @@ -221,14 +226,14 @@ public class MetricsPersistentWorker extends PersistenceWorker { */ private void loadFromStorage(List metrics) { try { - List noInCacheMetrics = metrics.stream() - .filter(m -> !context.containsKey(m) || !enableDatabaseSession) - .collect(Collectors.toList()); - if (noInCacheMetrics.isEmpty()) { + List notInCacheMetrics = metrics.stream() + .filter(m -> !context.containsKey(m) || !enableDatabaseSession) + .collect(Collectors.toList()); + if (notInCacheMetrics.isEmpty()) { return; } - - final List dbMetrics = metricsDAO.multiGet(model, noInCacheMetrics); + + final List dbMetrics = metricsDAO.multiGet(model, notInCacheMetrics); if (!enableDatabaseSession) { // Clear the cache only after results from DB are returned successfully. context.clear(); @@ -240,14 +245,14 @@ public class MetricsPersistentWorker extends PersistenceWorker { } @Override - public void endOfRound(long tookTime) { + public void endOfRound() { if (enableDatabaseSession) { Iterator iterator = context.values().iterator(); + long timestamp = System.currentTimeMillis(); while (iterator.hasNext()) { Metrics metrics = iterator.next(); - metrics.extendSurvivalTime(tookTime); - if (metrics.getSurvivalTime() > sessionTimeout) { + if (metrics.isExpired(timestamp, sessionTimeout)) { iterator.remove(); } } 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 3a407771a..367d7296f 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 @@ -82,6 +82,11 @@ public class MetricsStreamProcessor implements StreamProcessor { @Setter @Getter private boolean enableDatabaseSession; + /** + * The threshold of session time. Unit is ms. Default value is 70s. + */ + @Setter + private long storageSessionTimeout = 70_000; public static MetricsStreamProcessor getInstance() { return PROCESSOR; @@ -191,7 +196,7 @@ public class MetricsStreamProcessor implements StreamProcessor { MetricsPersistentWorker minutePersistentWorker = new MetricsPersistentWorker( moduleDefineHolder, model, metricsDAO, alarmNotifyWorker, exportWorker, transWorker, enableDatabaseSession, - supportUpdate + supportUpdate, storageSessionTimeout ); persistentWorkers.add(minutePersistentWorker); @@ -203,7 +208,7 @@ public class MetricsStreamProcessor implements StreamProcessor { Model model, boolean supportUpdate) { MetricsPersistentWorker persistentWorker = new MetricsPersistentWorker( - moduleDefineHolder, model, metricsDAO, enableDatabaseSession, supportUpdate); + moduleDefineHolder, model, metricsDAO, enableDatabaseSession, supportUpdate, storageSessionTimeout); 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 415371bdc..e38089338 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 @@ -55,10 +55,8 @@ public abstract class PersistenceWorker extends Abstr /** * The persistence process is driven by the {@link org.apache.skywalking.oap.server.core.storage.PersistenceTimer}. * This is a notification method for the worker when every round finished. - * - * @param tookTime The time costs in this round. */ - public abstract void endOfRound(long tookTime); + public abstract void endOfRound(); /** * Prepare the batch persistence, transfer all prepared data to the executable data format based on the storage 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 9093a2fb5..1465bf09d 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 @@ -86,7 +86,7 @@ public class TopNWorker extends PersistenceWorker { * This method used to clear the expired cache, but TopN is not following it. */ @Override - public void endOfRound(long tookTime) { + public void endOfRound() { } @Override 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 aaafc0674..a7e1e89ce 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 @@ -54,9 +54,8 @@ public enum PersistenceTimer { private HistogramMetrics prepareLatency; private HistogramMetrics executeLatency; private HistogramMetrics allLatency; - private long lastTime = System.currentTimeMillis(); private int syncOperationThreadsNum; - private int maxSyncoperationNum; + private int maxSyncOperationNum; private ExecutorService executorService; private ExecutorService prepareExecutorService; @@ -89,7 +88,7 @@ public enum PersistenceTimer { ); syncOperationThreadsNum = moduleConfig.getSyncThreads(); - maxSyncoperationNum = moduleConfig.getMaxSyncOperationNum(); + maxSyncOperationNum = moduleConfig.getMaxSyncOperationNum(); executorService = Executors.newFixedThreadPool(syncOperationThreadsNum); prepareExecutorService = Executors.newFixedThreadPool(moduleConfig.getPrepareThreads()); if (!isStarted) { @@ -116,7 +115,7 @@ public enum PersistenceTimer { AtomicBoolean stop = new AtomicBoolean(false); DefaultBlockingBatchQueue prepareQueue = new DefaultBlockingBatchQueue( - this.maxSyncoperationNum); + this.maxSyncOperationNum); try { List> persistenceWorkers = new ArrayList<>(); persistenceWorkers.addAll(TopNStreamProcessor.getInstance().getPersistentWorkers()); @@ -142,7 +141,7 @@ public enum PersistenceTimer { // Push the prepared requests into DefaultBlockingBatchQueue, // the executorService consumes from it when it reaches the size of batch. prepareQueue.offer(innerPrepareRequests); - worker.endOfRound(System.currentTimeMillis() - lastTime); + worker.endOfRound(); } finally { timer.finish(); prepareStageCountDownLatch.countDown(); @@ -194,7 +193,6 @@ public enum PersistenceTimer { stop.set(true); allTimer.finish(); - lastTime = System.currentTimeMillis(); } if (debug) {