From 307bd4c2198a0c54f38ca1e726856172d479b64c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=BD=AD=E5=8B=87=E5=8D=87=20pengys?= <8082209@qq.com>
Date: Thu, 31 May 2018 09:35:47 +0800
Subject: [PATCH] #1291 (#1294)

Block streaming, mandatory persistent cache data to storage then clean the cache data.
---
 .../worker/model/impl/PersistenceWorker.java  | 10 ++----
 .../src/main/resources/application.yml        |  4 ++-
 .../configuration/ConfigurationModule.java    | 13 ++-----
 .../service/IWorkerCacheSizeConfig.java       | 28 +++++++++++++++
 .../ConfigurationModuleConfig.java            |  9 +++++
 .../ConfigurationModuleProvider.java          |  3 ++
 .../service/WorkerCacheSizeConfigService.java | 35 +++++++++++++++++++
 7 files changed, 83 insertions(+), 19 deletions(-)
 create mode 100644 apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/service/IWorkerCacheSizeConfig.java
 create mode 100644 apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/service/WorkerCacheSizeConfigService.java

diff --git a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/impl/PersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/impl/PersistenceWorker.java
index 9331b2ae9..056f2cc03 100644
--- a/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/impl/PersistenceWorker.java
+++ b/apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/impl/PersistenceWorker.java
@@ -20,12 +20,13 @@ package org.apache.skywalking.apm.collector.analysis.worker.model.impl;
 
 import java.util.*;
 import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorker;
+import org.apache.skywalking.apm.collector.configuration.ConfigurationModule;
+import org.apache.skywalking.apm.collector.configuration.service.IWorkerCacheSizeConfig;
 import org.apache.skywalking.apm.collector.core.annotations.trace.GraphComputingMetric;
 import org.apache.skywalking.apm.collector.core.cache.Collection;
 import org.apache.skywalking.apm.collector.core.cache.*;
 import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
-import org.apache.skywalking.apm.collector.core.util.StringUtils;
 import org.apache.skywalking.apm.collector.storage.StorageModule;
 import org.apache.skywalking.apm.collector.storage.base.dao.*;
 import org.slf4j.*;
@@ -43,12 +44,7 @@ public abstract class PersistenceWorker<INPUT_AND_OUTPUT extends StreamData, COL
     PersistenceWorker(ModuleManager moduleManager) {
         super(moduleManager);
         this.batchDAO = moduleManager.find(StorageModule.NAME).getService(IBatchDAO.class);
-
-        if (StringUtils.isNotEmpty(System.getProperty("batchSize"))) {
-            this.blockBatchPersistenceSize = Integer.valueOf(System.getProperty("batchSize"));
-        } else {
-            this.blockBatchPersistenceSize = 500000;
-        }
+        this.blockBatchPersistenceSize = moduleManager.find(ConfigurationModule.NAME).getService(IWorkerCacheSizeConfig.class).cacheSize();
     }
 
     public boolean flushAndSwitch() {
diff --git a/apm-collector/apm-collector-boot/src/main/resources/application.yml b/apm-collector/apm-collector-boot/src/main/resources/application.yml
index 46133ab4f..a5f995ded 100644
--- a/apm-collector/apm-collector-boot/src/main/resources/application.yml
+++ b/apm-collector/apm-collector-boot/src/main/resources/application.yml
@@ -91,4 +91,6 @@ configuration:
     applicationAverageResponseTimeThreshold: 2000
 # thermodynamic
     thermodynamicResponseTimeStep: 50
-    thermodynamicCountOfResponseTimeSteps: 40
\ No newline at end of file
+    thermodynamicCountOfResponseTimeSteps: 40
+# max collection's size of worker cache collection, setting it smaller when collector OutOfMemory crashed.
+    workerCacheMaxSize: 10000
\ No newline at end of file
diff --git a/apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModule.java b/apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModule.java
index 9e9d0abb1..3b970daf9 100644
--- a/apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModule.java
+++ b/apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModule.java
@@ -18,16 +18,7 @@
 
 package org.apache.skywalking.apm.collector.configuration;
 
-import org.apache.skywalking.apm.collector.configuration.service.IApdexThresholdService;
-import org.apache.skywalking.apm.collector.configuration.service.IApplicationAlarmRuleConfig;
-import org.apache.skywalking.apm.collector.configuration.service.IApplicationReferenceAlarmRuleConfig;
-import org.apache.skywalking.apm.collector.configuration.service.ICollectorConfig;
-import org.apache.skywalking.apm.collector.configuration.service.IComponentLibraryCatalogService;
-import org.apache.skywalking.apm.collector.configuration.service.IInstanceAlarmRuleConfig;
-import org.apache.skywalking.apm.collector.configuration.service.IInstanceReferenceAlarmRuleConfig;
-import org.apache.skywalking.apm.collector.configuration.service.IResponseTimeDistributionConfigService;
-import org.apache.skywalking.apm.collector.configuration.service.IServiceAlarmRuleConfig;
-import org.apache.skywalking.apm.collector.configuration.service.IServiceReferenceAlarmRuleConfig;
+import org.apache.skywalking.apm.collector.configuration.service.*;
 import org.apache.skywalking.apm.collector.core.module.ModuleDefine;
 
 /**
@@ -47,6 +38,6 @@ public class ConfigurationModule extends ModuleDefine {
             IApdexThresholdService.class,
             IServiceAlarmRuleConfig.class, IInstanceAlarmRuleConfig.class, IApplicationAlarmRuleConfig.class,
             IServiceReferenceAlarmRuleConfig.class, IInstanceReferenceAlarmRuleConfig.class, IApplicationReferenceAlarmRuleConfig.class,
-            IComponentLibraryCatalogService.class, IResponseTimeDistributionConfigService.class};
+            IComponentLibraryCatalogService.class, IResponseTimeDistributionConfigService.class, IWorkerCacheSizeConfig.class};
     }
 }
diff --git a/apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/service/IWorkerCacheSizeConfig.java b/apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/service/IWorkerCacheSizeConfig.java
new file mode 100644
index 000000000..34d312f7b
--- /dev/null
+++ b/apm-collector/apm-collector-configuration/collector-configuration-define/src/main/java/org/apache/skywalking/apm/collector/configuration/service/IWorkerCacheSizeConfig.java
@@ -0,0 +1,28 @@
+/*
+ * 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.apm.collector.configuration.service;
+
+import org.apache.skywalking.apm.collector.core.module.Service;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IWorkerCacheSizeConfig extends Service {
+    int cacheSize();
+}
diff --git a/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleConfig.java b/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleConfig.java
index 1b2ac6c97..3cdf372f2 100644
--- a/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleConfig.java
+++ b/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleConfig.java
@@ -35,6 +35,7 @@ class ConfigurationModuleConfig extends ModuleConfig {
     private int applicationAverageResponseTimeThreshold;
     private int thermodynamicResponseTimeStep;
     private int thermodynamicCountOfResponseTimeSteps;
+    private int workerCacheMaxSize;
 
     public String getNamespace() {
         return namespace;
@@ -115,4 +116,12 @@ class ConfigurationModuleConfig extends ModuleConfig {
     public void setThermodynamicCountOfResponseTimeSteps(int thermodynamicCountOfResponseTimeSteps) {
         this.thermodynamicCountOfResponseTimeSteps = thermodynamicCountOfResponseTimeSteps;
     }
+
+    public int getWorkerCacheMaxSize() {
+        return workerCacheMaxSize;
+    }
+
+    public void setWorkerCacheMaxSize(int workerCacheMaxSize) {
+        this.workerCacheMaxSize = workerCacheMaxSize;
+    }
 }
diff --git a/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleProvider.java b/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleProvider.java
index 432dd14ea..9172407e9 100644
--- a/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleProvider.java
+++ b/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/ConfigurationModuleProvider.java
@@ -59,6 +59,8 @@ public class ConfigurationModuleProvider extends ModuleProvider {
         int thermodynamicResponseTimeStep = config.getThermodynamicResponseTimeStep() == 0 ? 50 : config.getThermodynamicResponseTimeStep();
         int thermodynamicCountOfResponseTimeSteps = config.getThermodynamicCountOfResponseTimeSteps() == 0 ? 40 : config.getThermodynamicCountOfResponseTimeSteps();
 
+        int workerCacheMaxSize = config.getWorkerCacheMaxSize() == 0 ? 10000 : config.getWorkerCacheMaxSize();
+
         this.registerServiceImplementation(ICollectorConfig.class, new CollectorConfigService(namespace));
         this.registerServiceImplementation(IComponentLibraryCatalogService.class, new ComponentLibraryCatalogService());
         this.registerServiceImplementation(IApdexThresholdService.class, new ApdexThresholdService(applicationApdexThreshold));
@@ -69,6 +71,7 @@ public class ConfigurationModuleProvider extends ModuleProvider {
         this.registerServiceImplementation(IInstanceReferenceAlarmRuleConfig.class, new InstanceReferenceAlarmRuleConfig(instanceErrorRateThreshold, instanceAverageResponseTimeThreshold));
         this.registerServiceImplementation(IApplicationReferenceAlarmRuleConfig.class, new ApplicationReferenceAlarmRuleConfig(applicationErrorRateThreshold, applicationAverageResponseTimeThreshold));
         this.registerServiceImplementation(IResponseTimeDistributionConfigService.class, new ResponseTimeDistributionConfigService(thermodynamicResponseTimeStep, thermodynamicCountOfResponseTimeSteps));
+        this.registerServiceImplementation(IWorkerCacheSizeConfig.class, new WorkerCacheSizeConfigService(workerCacheMaxSize));
     }
 
     @Override public void start() {
diff --git a/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/service/WorkerCacheSizeConfigService.java b/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/service/WorkerCacheSizeConfigService.java
new file mode 100644
index 000000000..a66855b81
--- /dev/null
+++ b/apm-collector/apm-collector-configuration/collector-configuration-provider/src/main/java/org/apache/skywalking/apm/collector/configuration/service/WorkerCacheSizeConfigService.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.apm.collector.configuration.service;
+
+/**
+ * @author peng-yongsheng
+ */
+public class WorkerCacheSizeConfigService implements IWorkerCacheSizeConfig {
+
+    private int workerCacheMaxSize;
+
+    public WorkerCacheSizeConfigService(int workerCacheMaxSize) {
+        this.workerCacheMaxSize = workerCacheMaxSize;
+    }
+
+    @Override public int cacheSize() {
+        return workerCacheMaxSize;
+    }
+}