You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2018/10/26 06:40:12 UTC
[incubator-skywalking] branch mesh-receiver updated: Remove debug
log, and restore TTL timer for real scenarios.
This is an automated email from the ASF dual-hosted git repository.
wusheng pushed a commit to branch mesh-receiver
in repository https://gitbox.apache.org/repos/asf/incubator-skywalking.git
The following commit(s) were added to refs/heads/mesh-receiver by this push:
new 5872cfc Remove debug log, and restore TTL timer for real scenarios.
5872cfc is described below
commit 5872cfca973690f7ca41d2b494b15aba33fbd757
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Fri Oct 26 14:40:00 2018 +0800
Remove debug log, and restore TTL timer for real scenarios.
---
.../commons/datacarrier/consumer/ConsumerPool.java | 7 +++---
.../analysis/indicator/ThermodynamicIndicator.java | 23 +++++++-----------
.../analysis/worker/IndicatorAggregateWorker.java | 21 +++++-----------
.../analysis/worker/IndicatorPersistentWorker.java | 28 +++++++---------------
.../analysis/worker/IndicatorRemoteWorker.java | 12 ++++------
.../register/service/ServiceInventoryRegister.java | 1 -
.../register/worker/RegisterDistinctWorker.java | 7 ------
.../register/worker/RegisterPersistentWorker.java | 4 ----
.../core/storage/ttl/DataTTLKeeperTimer.java | 9 ++++---
.../receiver/mesh/TelemetryDataDispatcher.java | 7 ------
10 files changed, 36 insertions(+), 83 deletions(-)
diff --git a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java
index 7ba0620..d3ba23e 100644
--- a/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java
+++ b/apm-commons/apm-datacarrier/src/main/java/org/apache/skywalking/apm/commons/datacarrier/consumer/ConsumerPool.java
@@ -32,10 +32,11 @@ public class ConsumerPool<T> {
private Channels<T> channels;
private ReentrantLock lock;
- public ConsumerPool(String name, Channels<T> channels, Class<? extends IConsumer<T>> consumerClass, int num, long consumeCycle) {
+ public ConsumerPool(String name, Channels<T> channels, Class<? extends IConsumer<T>> consumerClass, int num,
+ long consumeCycle) {
this(channels, num);
for (int i = 0; i < num; i++) {
- consumerThreads[i] = new ConsumerThread("DataCarrier."+ name + ".Consumser." + i + ".Thread", getNewConsumerInstance(consumerClass), consumeCycle);
+ consumerThreads[i] = new ConsumerThread("DataCarrier." + name + ".Consumser." + i + ".Thread", getNewConsumerInstance(consumerClass), consumeCycle);
consumerThreads[i].setDaemon(true);
}
}
@@ -44,7 +45,7 @@ public class ConsumerPool<T> {
this(channels, num);
prototype.init();
for (int i = 0; i < num; i++) {
- consumerThreads[i] = new ConsumerThread("DataCarrier."+ name + ".Consumser." + i + ".Thread", prototype, consumeCycle);
+ consumerThreads[i] = new ConsumerThread("DataCarrier." + name + ".Consumser." + i + ".Thread", prototype, consumeCycle);
consumerThreads[i].setDaemon(true);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/ThermodynamicIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/ThermodynamicIndicator.java
index cef4342..830683c 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/ThermodynamicIndicator.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/indicator/ThermodynamicIndicator.java
@@ -18,13 +18,15 @@
package org.apache.skywalking.oap.server.core.analysis.indicator;
-import java.util.*;
-import java.util.concurrent.atomic.AtomicInteger;
-import lombok.*;
-import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.*;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.Arg;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.Entrance;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorOperator;
+import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.SourceFrom;
import org.apache.skywalking.oap.server.core.storage.annotation.Column;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Thermodynamic indicator represents the calculator for heat map.
@@ -38,9 +40,6 @@ import org.slf4j.LoggerFactory;
*/
@IndicatorOperator
public abstract class ThermodynamicIndicator extends Indicator {
- private static final Logger logger = LoggerFactory.getLogger(ThermodynamicIndicator.class);
- private static AtomicInteger INDEX = new AtomicInteger(0);
-
public static final String DETAIL_GROUP = "detail_group";
public static final String STEP = "step";
public static final String NUM_OF_STEPS = "num_of_steps";
@@ -63,7 +62,6 @@ public abstract class ThermodynamicIndicator extends Indicator {
*/
@Entrance
public final void combine(@SourceFrom int value, @Arg int step, @Arg int maxNumOfSteps) {
- logger.info("ThermodynamicIndicator get " + INDEX.addAndGet(1));
if (this.step == 0) {
this.step = step;
}
@@ -97,7 +95,6 @@ public abstract class ThermodynamicIndicator extends Indicator {
thermodynamicIndicator.detailIndex.forEach((key, element) -> {
IntKeyLongValue existingElement = self.detailIndex.get(key);
- logger.info("prepare merging id =" + key + ", value=" + element.getValue());
if (existingElement == null) {
existingElement = new IntKeyLongValue();
existingElement.setKey(key);
@@ -106,11 +103,7 @@ public abstract class ThermodynamicIndicator extends Indicator {
} else {
existingElement.addValue(element.getValue());
}
-
- logger.info("result=" + self.detailGroup.toStorageData());
});
-
- logger.info("after combine, " + self.detailGroup.toStorageData());
}
/**
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java
index 69999b8..0f02651 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorAggregateWorker.java
@@ -18,14 +18,16 @@
package org.apache.skywalking.oap.server.core.analysis.worker;
-import java.util.*;
+import java.util.Iterator;
+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.*;
-import org.apache.skywalking.oap.server.core.analysis.generated.all.AllHeatmapIndicator;
+import org.apache.skywalking.oap.server.core.analysis.data.EndOfBatchContext;
+import org.apache.skywalking.oap.server.core.analysis.data.MergeDataCache;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
-import org.slf4j.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* @author peng-yongsheng
@@ -55,10 +57,6 @@ public class IndicatorAggregateWorker extends AbstractWorker<Indicator> {
}
private void onWork(Indicator indicator) {
- if (modelName.equals("all_heatmap")) {
- AllHeatmapIndicator allHeatmapIndicator = (AllHeatmapIndicator)indicator;
- logger.info("aggregate indicator: {}", allHeatmapIndicator.getDetailGroup().toStorageData());
- }
messageNum++;
aggregate(indicator);
@@ -96,13 +94,6 @@ public class IndicatorAggregateWorker extends AbstractWorker<Indicator> {
mergeDataCache.put(indicator);
}
- mergeDataCache.getLast().collection().forEach(indicator1 -> {
- if (modelName.equals("all_heatmap")) {
- AllHeatmapIndicator allHeatmapIndicator = (AllHeatmapIndicator)indicator1;
- logger.warn("aggregate indicator aggregate method: {}", allHeatmapIndicator.getDetailGroup().toStorageData());
- }
- });
-
mergeDataCache.finishWriting();
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java
index e3cde52..8763d8b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorPersistentWorker.java
@@ -18,16 +18,20 @@
package org.apache.skywalking.oap.server.core.analysis.worker;
-import java.util.*;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Objects;
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.*;
-import org.apache.skywalking.oap.server.core.analysis.generated.all.AllHeatmapIndicator;
+import org.apache.skywalking.oap.server.core.analysis.data.EndOfBatchContext;
+import org.apache.skywalking.oap.server.core.analysis.data.MergeDataCache;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.storage.IIndicatorDAO;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.slf4j.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import static java.util.Objects.nonNull;
@@ -56,10 +60,6 @@ public class IndicatorPersistentWorker extends PersistenceWorker<Indicator, Merg
}
@Override void onWork(Indicator indicator) {
- if (modelName.equals("all_heatmap")) {
- AllHeatmapIndicator allHeatmapIndicator = (AllHeatmapIndicator)indicator;
- logger.warn("persistent indicator: {}", allHeatmapIndicator.getDetailGroup().toStorageData());
- }
super.onWork(indicator);
}
@@ -95,23 +95,11 @@ public class IndicatorPersistentWorker extends PersistenceWorker<Indicator, Merg
}
try {
if (nonNull(dbData)) {
- if (modelName.equals("all_heatmap")) {
- AllHeatmapIndicator dbBefore = (AllHeatmapIndicator)dbData;
- AllHeatmapIndicator stremBefore = (AllHeatmapIndicator)data;
- logger.warn("need to combine, db data before: {}, stream data before: {}", dbBefore.getDetailGroup().toStorageData(), stremBefore.getDetailGroup().toStorageData());
- }
data.combine(dbData);
data.calculate();
- if (modelName.equals("all_heatmap")) {
- AllHeatmapIndicator stremAfter = (AllHeatmapIndicator)data;
- logger.warn("need to combine, stream data after: {}", stremAfter.getDetailGroup().toStorageData());
- }
batchCollection.add(indicatorDAO.prepareBatchUpdate(modelName, data));
} else {
- if (modelName.equals("all_heatmap")) {
- logger.warn("insert all_heatmap id: {}", data.id());
- }
batchCollection.add(indicatorDAO.prepareBatchInsert(modelName, data));
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorRemoteWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorRemoteWorker.java
index 4c7b137..2a100c0 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorRemoteWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/IndicatorRemoteWorker.java
@@ -19,13 +19,13 @@
package org.apache.skywalking.oap.server.core.analysis.worker;
import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.analysis.generated.all.AllHeatmapIndicator;
import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
import org.apache.skywalking.oap.server.core.remote.RemoteSenderService;
import org.apache.skywalking.oap.server.core.remote.selector.Selector;
import org.apache.skywalking.oap.server.core.worker.AbstractWorker;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.slf4j.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* @author peng-yongsheng
@@ -38,7 +38,8 @@ public class IndicatorRemoteWorker extends AbstractWorker<Indicator> {
private final RemoteSenderService remoteSender;
private final String modelName;
- IndicatorRemoteWorker(int workerId, ModuleManager moduleManager, AbstractWorker<Indicator> nextWorker, String modelName) {
+ IndicatorRemoteWorker(int workerId, ModuleManager moduleManager, AbstractWorker<Indicator> nextWorker,
+ String modelName) {
super(workerId);
this.remoteSender = moduleManager.find(CoreModule.NAME).getService(RemoteSenderService.class);
this.nextWorker = nextWorker;
@@ -46,11 +47,6 @@ public class IndicatorRemoteWorker extends AbstractWorker<Indicator> {
}
@Override public final void in(Indicator indicator) {
- if (modelName.equals("all_heatmap")) {
- AllHeatmapIndicator allHeatmapIndicator = (AllHeatmapIndicator)indicator;
- logger.error("remote indicator: {}", allHeatmapIndicator.getDetailGroup().toStorageData());
- }
-
try {
remoteSender.send(nextWorker.getWorkerId(), indicator, Selector.HashCode);
} catch (Throwable e) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java
index 85a9c73..531d3f1 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java
@@ -52,7 +52,6 @@ public class ServiceInventoryRegister implements IServiceInventoryRegister {
@Override public int getOrCreate(String serviceName) {
int serviceId = getServiceInventoryCache().getServiceId(serviceName);
- logger.info("register service: {}, service id: {}", serviceName, serviceId);
if (serviceId == Const.NONE) {
ServiceInventory serviceInventory = new ServiceInventory();
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterDistinctWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterDistinctWorker.java
index d5821f6..c02f150 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterDistinctWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterDistinctWorker.java
@@ -54,10 +54,6 @@ public class RegisterDistinctWorker extends AbstractWorker<RegisterSource> {
private void onWork(RegisterSource source) {
messageNum++;
- if (source instanceof ServiceInventory) {
- logger.info("service register distinct, name {}", ((ServiceInventory)source).getName());
- }
-
if (!sources.containsKey(source)) {
sources.put(source, source);
} else {
@@ -66,9 +62,6 @@ public class RegisterDistinctWorker extends AbstractWorker<RegisterSource> {
if (messageNum >= 1000 || source.getEndOfBatchContext().isEndOfBatch()) {
sources.values().forEach(source1 -> {
- if (source instanceof ServiceInventory) {
- logger.info("register distinct send to next, name: {}", ((ServiceInventory)source).getName());
- }
nextWorker.in(source1);
});
messageNum = 0;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java
index cb8e615..0be299c 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/worker/RegisterPersistentWorker.java
@@ -61,10 +61,6 @@ public class RegisterPersistentWorker extends AbstractWorker<RegisterSource> {
}
private void onWork(RegisterSource registerSource) {
- if (registerSource instanceof ServiceInventory) {
- logger.info("service register persistent, name {}", ((ServiceInventory)registerSource).getName());
- }
-
if (!sources.containsKey(registerSource)) {
sources.put(registerSource, registerSource);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java
index 13e249a..2d920ad 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/ttl/DataTTLKeeperTimer.java
@@ -20,7 +20,10 @@ package org.apache.skywalking.oap.server.core.storage.ttl;
import java.io.IOException;
import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import lombok.Setter;
+import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.DataTTL;
@@ -57,9 +60,9 @@ public enum DataTTLKeeperTimer {
this.moduleManager = moduleManager;
this.clusterNodesQuery = moduleManager.find(ClusterModule.NAME).getService(ClusterNodesQuery.class);
-// Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
-// new RunnableWithExceptionProtection(this::delete,
-// t -> logger.error("Remove data in background failure.", t)), 1, 5, TimeUnit.MINUTES);
+ Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
+ new RunnableWithExceptionProtection(this::delete,
+ t -> logger.error("Remove data in background failure.", t)), 1, 5, TimeUnit.MINUTES);
}
private void delete() {
diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java
index 35121d0..70b755a 100644
--- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java
+++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java
@@ -18,7 +18,6 @@
package org.apache.skywalking.aop.server.receiver.mesh;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.logging.log4j.util.Strings;
import org.apache.skywalking.apm.network.servicemesh.Protocol;
import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric;
@@ -36,8 +35,6 @@ import org.apache.skywalking.oap.server.core.source.ServiceRelation;
import org.apache.skywalking.oap.server.core.source.SourceReceiver;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.util.TimeBucketUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* TelemetryDataDispatcher processes the {@link ServiceMeshMetric} format telemetry data, transfers it to source
@@ -46,9 +43,6 @@ import org.slf4j.LoggerFactory;
* @author wusheng
*/
public class TelemetryDataDispatcher {
- private static final Logger logger = LoggerFactory.getLogger(MeshGRPCHandler.class);
- private static AtomicInteger INDEX = new AtomicInteger(0);
-
private static MeshDataBufferFileCache CACHE;
private static ServiceInventoryCache SERVICE_CACHE;
private static ServiceInstanceInventoryCache SERVICE_INSTANCE_CACHE;
@@ -84,7 +78,6 @@ public class TelemetryDataDispatcher {
long minuteTimeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(metric.getStartTime());
if (org.apache.skywalking.apm.network.common.DetectPoint.server.equals(metric.getDetectPoint())) {
- logger.info("dispatch server: " + INDEX.addAndGet(1));
toAll(decorator, minuteTimeBucket);
toService(decorator, minuteTimeBucket);
toEndpoint(decorator, minuteTimeBucket);