diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/dao/CpuMetricEsDAO.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/dao/CpuMetricEsDAO.java index 221f1c012..39db58b45 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/dao/CpuMetricEsDAO.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/dao/CpuMetricEsDAO.java @@ -4,7 +4,7 @@ import java.util.HashMap; import java.util.Map; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.skywalking.apm.collector.agentjvm.worker.cpu.define.CpuMetricTable; +import org.skywalking.apm.collector.storage.table.jvm.CpuMetricTable; import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; import org.skywalking.apm.collector.stream.worker.impl.dao.IPersistenceDAO; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricDataDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricDataDefine.java index 2622e2e95..e843a3420 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricDataDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricDataDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.cpu.define; import org.skywalking.apm.collector.core.framework.UnexpectedException; import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; +import org.skywalking.apm.collector.storage.table.jvm.CpuMetricTable; import org.skywalking.apm.collector.stream.worker.impl.data.Attribute; import org.skywalking.apm.collector.stream.worker.impl.data.AttributeType; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricEsTableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricEsTableDefine.java index 12a22447b..f1f88c84a 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricEsTableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricEsTableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.cpu.define; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine; +import org.skywalking.apm.collector.storage.table.jvm.CpuMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricH2TableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricH2TableDefine.java index 1ae29cbee..23e45f50d 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricH2TableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricH2TableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.cpu.define; import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine; import org.skywalking.apm.collector.storage.h2.define.H2TableDefine; +import org.skywalking.apm.collector.storage.table.jvm.CpuMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/dao/GCMetricEsDAO.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/dao/GCMetricEsDAO.java index 4ae4f63cb..a06ff0d58 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/dao/GCMetricEsDAO.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/dao/GCMetricEsDAO.java @@ -4,7 +4,7 @@ import java.util.HashMap; import java.util.Map; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.skywalking.apm.collector.agentjvm.worker.gc.define.GCMetricTable; +import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable; import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; import org.skywalking.apm.collector.stream.worker.impl.dao.IPersistenceDAO; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricDataDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricDataDefine.java index a579cda1f..d4d9c609a 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricDataDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricDataDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.gc.define; import org.skywalking.apm.collector.core.framework.UnexpectedException; import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; +import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable; import org.skywalking.apm.collector.stream.worker.impl.data.Attribute; import org.skywalking.apm.collector.stream.worker.impl.data.AttributeType; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricEsTableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricEsTableDefine.java index 976e501bc..b37adc1c7 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricEsTableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricEsTableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.gc.define; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine; +import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricH2TableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricH2TableDefine.java index 39dbbdcaa..00a50e42b 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricH2TableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricH2TableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.gc.define; import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine; import org.skywalking.apm.collector.storage.h2.define.H2TableDefine; +import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/dao/MemoryMetricEsDAO.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/dao/MemoryMetricEsDAO.java index 193f4885a..09af9112e 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/dao/MemoryMetricEsDAO.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/dao/MemoryMetricEsDAO.java @@ -4,7 +4,7 @@ import java.util.HashMap; import java.util.Map; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.skywalking.apm.collector.agentjvm.worker.memory.define.MemoryMetricTable; +import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable; import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; import org.skywalking.apm.collector.stream.worker.impl.dao.IPersistenceDAO; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricDataDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricDataDefine.java index 927458e83..035bacfa2 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricDataDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricDataDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.memory.define; import org.skywalking.apm.collector.core.framework.UnexpectedException; import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; +import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable; import org.skywalking.apm.collector.stream.worker.impl.data.Attribute; import org.skywalking.apm.collector.stream.worker.impl.data.AttributeType; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricEsTableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricEsTableDefine.java index a4861382d..6e70cc175 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricEsTableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricEsTableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.memory.define; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine; +import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricH2TableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricH2TableDefine.java index 8cd4093b5..4e61ce68c 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricH2TableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricH2TableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.memory.define; import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine; import org.skywalking.apm.collector.storage.h2.define.H2TableDefine; +import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/dao/MemoryPoolMetricEsDAO.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/dao/MemoryPoolMetricEsDAO.java index 3dd44559d..023e92578 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/dao/MemoryPoolMetricEsDAO.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/dao/MemoryPoolMetricEsDAO.java @@ -4,7 +4,7 @@ import java.util.HashMap; import java.util.Map; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.skywalking.apm.collector.agentjvm.worker.memorypool.define.MemoryPoolMetricTable; +import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable; import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; import org.skywalking.apm.collector.stream.worker.impl.dao.IPersistenceDAO; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricDataDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricDataDefine.java index dd2daaad3..b2d9ce0a8 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricDataDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricDataDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.memorypool.define; import org.skywalking.apm.collector.core.framework.UnexpectedException; import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; +import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable; import org.skywalking.apm.collector.stream.worker.impl.data.Attribute; import org.skywalking.apm.collector.stream.worker.impl.data.AttributeType; import org.skywalking.apm.collector.stream.worker.impl.data.Data; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricEsTableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricEsTableDefine.java index bec83bcc8..87974f004 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricEsTableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricEsTableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.memorypool.define; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine; import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine; +import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricH2TableDefine.java b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricH2TableDefine.java index 7275f7939..98278e08d 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricH2TableDefine.java +++ b/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricH2TableDefine.java @@ -2,6 +2,7 @@ package org.skywalking.apm.collector.agentjvm.worker.memorypool.define; import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine; import org.skywalking.apm.collector.storage.h2.define.H2TableDefine; +import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/InstPerformancePersistenceWorker.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/InstPerformancePersistenceWorker.java new file mode 100644 index 000000000..9a7d68275 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/InstPerformancePersistenceWorker.java @@ -0,0 +1,71 @@ +package org.skywalking.apm.collector.agentstream.worker.instance.performance; + +import org.skywalking.apm.collector.agentstream.worker.instance.performance.dao.IInstPerformanceDAO; +import org.skywalking.apm.collector.agentstream.worker.instance.performance.define.InstPerformanceDataDefine; +import org.skywalking.apm.collector.storage.dao.DAOContainer; +import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider; +import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext; +import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException; +import org.skywalking.apm.collector.stream.worker.Role; +import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker; +import org.skywalking.apm.collector.stream.worker.impl.dao.IPersistenceDAO; +import org.skywalking.apm.collector.stream.worker.impl.data.DataDefine; +import org.skywalking.apm.collector.stream.worker.selector.HashCodeSelector; +import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector; + +/** + * @author pengys5 + */ +public class InstPerformancePersistenceWorker extends PersistenceWorker { + + public InstPerformancePersistenceWorker(Role role, ClusterWorkerContext clusterContext) { + super(role, clusterContext); + } + + @Override public void preStart() throws ProviderNotFoundException { + super.preStart(); + } + + @Override protected boolean needMergeDBData() { + return true; + } + + @Override protected IPersistenceDAO persistenceDAO() { + return (IPersistenceDAO)DAOContainer.INSTANCE.get(IInstPerformanceDAO.class.getName()); + } + + public static class Factory extends AbstractLocalAsyncWorkerProvider { + @Override + public Role role() { + return WorkerRole.INSTANCE; + } + + @Override + public InstPerformancePersistenceWorker workerInstance(ClusterWorkerContext clusterContext) { + return new InstPerformancePersistenceWorker(role(), clusterContext); + } + + @Override + public int queueSize() { + return 1024; + } + } + + public enum WorkerRole implements Role { + INSTANCE; + + @Override + public String roleName() { + return InstPerformancePersistenceWorker.class.getSimpleName(); + } + + @Override + public WorkerSelector workerSelector() { + return new HashCodeSelector(); + } + + @Override public DataDefine dataDefine() { + return new InstPerformanceDataDefine(); + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/InstPerformanceSpanListener.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/InstPerformanceSpanListener.java new file mode 100644 index 000000000..d82c2d9c8 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/InstPerformanceSpanListener.java @@ -0,0 +1,59 @@ +package org.skywalking.apm.collector.agentstream.worker.instance.performance; + +import org.skywalking.apm.collector.agentstream.worker.instance.performance.define.InstPerformanceDataDefine; +import org.skywalking.apm.collector.agentstream.worker.segment.EntrySpanListener; +import org.skywalking.apm.collector.agentstream.worker.segment.FirstSpanListener; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; +import org.skywalking.apm.collector.core.util.Const; +import org.skywalking.apm.collector.stream.StreamModuleContext; +import org.skywalking.apm.collector.stream.StreamModuleGroupDefine; +import org.skywalking.apm.collector.stream.worker.WorkerInvokeException; +import org.skywalking.apm.collector.stream.worker.WorkerNotFoundException; +import org.skywalking.apm.collector.stream.worker.util.TimeBucketUtils; +import org.skywalking.apm.network.proto.SpanObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class InstPerformanceSpanListener implements EntrySpanListener, FirstSpanListener { + + private final Logger logger = LoggerFactory.getLogger(InstPerformanceSpanListener.class); + + private int applicationId; + private int applicationInstanceId; + private long cost; + private long timeBucket; + + @Override + public void parseEntry(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { + } + + @Override + public void parseFirst(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { + this.applicationId = applicationId; + this.applicationInstanceId = applicationInstanceId; + this.cost = spanObject.getEndTime() - spanObject.getStartTime(); + timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(spanObject.getStartTime()); + } + + @Override public void build() { + StreamModuleContext context = (StreamModuleContext)CollectorContextHelper.INSTANCE.getContext(StreamModuleGroupDefine.GROUP_NAME); + + InstPerformanceDataDefine.InstPerformance instPerformance = new InstPerformanceDataDefine.InstPerformance(); + instPerformance.setId(timeBucket + Const.ID_SPLIT + applicationInstanceId); + instPerformance.setApplicationId(applicationId); + instPerformance.setInstanceId(applicationInstanceId); + instPerformance.setCallTimes(1); + instPerformance.setCostTotal(cost); + instPerformance.setTimeBucket(timeBucket); + + try { + logger.debug("send to instance performance persistence worker, id: {}", instPerformance.getId()); + context.getClusterWorkerContext().lookup(InstPerformancePersistenceWorker.WorkerRole.INSTANCE).tell(instPerformance.toData()); + } catch (WorkerInvokeException | WorkerNotFoundException e) { + logger.error(e.getMessage(), e); + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/IInstPerformanceDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/IInstPerformanceDAO.java new file mode 100644 index 000000000..96c810a33 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/IInstPerformanceDAO.java @@ -0,0 +1,7 @@ +package org.skywalking.apm.collector.agentstream.worker.instance.performance.dao; + +/** + * @author pengys5 + */ +public interface IInstPerformanceDAO { +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/InstPerformanceEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/InstPerformanceEsDAO.java new file mode 100644 index 000000000..52ae19ca3 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/InstPerformanceEsDAO.java @@ -0,0 +1,56 @@ +package org.skywalking.apm.collector.agentstream.worker.instance.performance.dao; + +import java.util.HashMap; +import java.util.Map; +import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; +import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; +import org.skywalking.apm.collector.storage.table.instance.InstPerformanceTable; +import org.skywalking.apm.collector.stream.worker.impl.dao.IPersistenceDAO; +import org.skywalking.apm.collector.stream.worker.impl.data.Data; +import org.skywalking.apm.collector.stream.worker.impl.data.DataDefine; + +/** + * @author pengys5 + */ +public class InstPerformanceEsDAO extends EsDAO implements IInstPerformanceDAO, IPersistenceDAO { + + @Override public Data get(String id, DataDefine dataDefine) { + GetResponse getResponse = getClient().prepareGet(InstPerformanceTable.TABLE, id).get(); + if (getResponse.isExists()) { + Data data = dataDefine.build(id); + Map source = getResponse.getSource(); + data.setDataInteger(0, (Integer)source.get(InstPerformanceTable.COLUMN_APPLICATION_ID)); + data.setDataInteger(1, (Integer)source.get(InstPerformanceTable.COLUMN_INSTANCE_ID)); + data.setDataInteger(2, (Integer)source.get(InstPerformanceTable.COLUMN_CALL_TIMES)); + data.setDataLong(0, (Long)source.get(InstPerformanceTable.COLUMN_COST_TOTAL)); + data.setDataLong(1, (Long)source.get(InstPerformanceTable.COLUMN_TIME_BUCKET)); + return data; + } else { + return null; + } + } + + @Override public IndexRequestBuilder prepareBatchInsert(Data data) { + Map source = new HashMap<>(); + source.put(InstPerformanceTable.COLUMN_APPLICATION_ID, data.getDataInteger(0)); + source.put(InstPerformanceTable.COLUMN_INSTANCE_ID, data.getDataInteger(1)); + source.put(InstPerformanceTable.COLUMN_CALL_TIMES, data.getDataInteger(2)); + source.put(InstPerformanceTable.COLUMN_COST_TOTAL, data.getDataLong(0)); + source.put(InstPerformanceTable.COLUMN_TIME_BUCKET, data.getDataLong(1)); + + return getClient().prepareIndex(InstPerformanceTable.TABLE, data.getDataString(0)).setSource(source); + } + + @Override public UpdateRequestBuilder prepareBatchUpdate(Data data) { + Map source = new HashMap<>(); + source.put(InstPerformanceTable.COLUMN_APPLICATION_ID, data.getDataInteger(0)); + source.put(InstPerformanceTable.COLUMN_INSTANCE_ID, data.getDataInteger(1)); + source.put(InstPerformanceTable.COLUMN_CALL_TIMES, data.getDataInteger(2)); + source.put(InstPerformanceTable.COLUMN_COST_TOTAL, data.getDataLong(0)); + source.put(InstPerformanceTable.COLUMN_TIME_BUCKET, data.getDataLong(1)); + + return getClient().prepareUpdate(InstPerformanceTable.TABLE, data.getDataString(0)).setDoc(source); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/InstPerformanceH2DAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/InstPerformanceH2DAO.java new file mode 100644 index 000000000..9dbe1800c --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/dao/InstPerformanceH2DAO.java @@ -0,0 +1,10 @@ +package org.skywalking.apm.collector.agentstream.worker.instance.performance.dao; + +import org.skywalking.apm.collector.storage.h2.dao.H2DAO; + +/** + * @author pengys5 + */ +public class InstPerformanceH2DAO extends H2DAO implements IInstPerformanceDAO { + +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/define/InstPerformanceDataDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/define/InstPerformanceDataDefine.java new file mode 100644 index 000000000..831357d74 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/instance/performance/define/InstPerformanceDataDefine.java @@ -0,0 +1,131 @@ +package org.skywalking.apm.collector.agentstream.worker.instance.performance.define; + +import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; +import org.skywalking.apm.collector.storage.table.instance.InstPerformanceTable; +import org.skywalking.apm.collector.stream.worker.impl.data.Attribute; +import org.skywalking.apm.collector.stream.worker.impl.data.AttributeType; +import org.skywalking.apm.collector.stream.worker.impl.data.Data; +import org.skywalking.apm.collector.stream.worker.impl.data.DataDefine; +import org.skywalking.apm.collector.stream.worker.impl.data.Transform; +import org.skywalking.apm.collector.stream.worker.impl.data.operate.AddOperation; +import org.skywalking.apm.collector.stream.worker.impl.data.operate.CoverOperation; +import org.skywalking.apm.collector.stream.worker.impl.data.operate.NonOperation; + +/** + * @author pengys5 + */ +public class InstPerformanceDataDefine extends DataDefine { + + @Override protected int initialCapacity() { + return 6; + } + + @Override protected void attributeDefine() { + addAttribute(0, new Attribute(InstPerformanceTable.COLUMN_ID, AttributeType.STRING, new NonOperation())); + addAttribute(1, new Attribute(InstPerformanceTable.COLUMN_APPLICATION_ID, AttributeType.INTEGER, new CoverOperation())); + addAttribute(2, new Attribute(InstPerformanceTable.COLUMN_INSTANCE_ID, AttributeType.INTEGER, new CoverOperation())); + addAttribute(3, new Attribute(InstPerformanceTable.COLUMN_CALL_TIMES, AttributeType.INTEGER, new AddOperation())); + addAttribute(4, new Attribute(InstPerformanceTable.COLUMN_COST_TOTAL, AttributeType.LONG, new AddOperation())); + addAttribute(5, new Attribute(InstPerformanceTable.COLUMN_TIME_BUCKET, AttributeType.LONG, new CoverOperation())); + } + + @Override public Object deserialize(RemoteData remoteData) { + return null; + } + + @Override public RemoteData serialize(Object object) { + return null; + } + + public static class InstPerformance implements Transform { + private String id; + private int applicationId; + private int instanceId; + private int callTimes; + private long costTotal; + private long timeBucket; + + public InstPerformance(String id, int applicationId, int instanceId, int callTimes, long costTotal, + long timeBucket) { + this.id = id; + this.applicationId = applicationId; + this.instanceId = instanceId; + this.callTimes = callTimes; + this.costTotal = costTotal; + this.timeBucket = timeBucket; + } + + public InstPerformance() { + } + + @Override public Data toData() { + InstPerformanceDataDefine define = new InstPerformanceDataDefine(); + Data data = define.build(id); + data.setDataString(0, this.id); + data.setDataInteger(0, this.applicationId); + data.setDataInteger(1, this.instanceId); + data.setDataInteger(2, this.callTimes); + data.setDataLong(0, this.costTotal); + data.setDataLong(1, this.timeBucket); + return data; + } + + @Override public InstPerformance toSelf(Data data) { + this.id = data.getDataString(0); + this.applicationId = data.getDataInteger(0); + this.instanceId = data.getDataInteger(1); + this.callTimes = data.getDataInteger(2); + this.costTotal = data.getDataLong(0); + this.timeBucket = data.getDataLong(1); + return this; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public int getInstanceId() { + return instanceId; + } + + public void setInstanceId(int instanceId) { + this.instanceId = instanceId; + } + + public int getCallTimes() { + return callTimes; + } + + public void setCallTimes(int callTimes) { + this.callTimes = callTimes; + } + + public long getCostTotal() { + return costTotal; + } + + public void setCostTotal(long costTotal) { + this.costTotal = costTotal; + } + + public long getTimeBucket() { + return timeBucket; + } + + public void setTimeBucket(long timeBucket) { + this.timeBucket = timeBucket; + } + + public int getApplicationId() { + return applicationId; + } + + public void setApplicationId(int applicationId) { + this.applicationId = applicationId; + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java index 4485db639..8f58c2ecf 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java @@ -3,6 +3,7 @@ package org.skywalking.apm.collector.agentstream.worker.segment; import java.util.ArrayList; import java.util.List; import org.skywalking.apm.collector.agentstream.worker.global.GlobalTraceSpanListener; +import org.skywalking.apm.collector.agentstream.worker.instance.performance.InstPerformanceSpanListener; import org.skywalking.apm.collector.agentstream.worker.node.component.NodeComponentSpanListener; import org.skywalking.apm.collector.agentstream.worker.node.mapping.NodeMappingSpanListener; import org.skywalking.apm.collector.agentstream.worker.noderef.reference.NodeRefSpanListener; @@ -45,6 +46,7 @@ public class SegmentParse { spanListeners.add(new GlobalTraceSpanListener()); spanListeners.add(new ServiceEntrySpanListener()); spanListeners.add(new ServiceRefSpanListener()); + spanListeners.add(new InstPerformanceSpanListener()); } public void parse(List traceIds, TraceSegmentObject segmentObject) { diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define index 63656c106..98ec500d2 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define @@ -9,4 +9,5 @@ org.skywalking.apm.collector.agentstream.worker.noderef.summary.dao.NodeRefSumEs org.skywalking.apm.collector.agentstream.worker.segment.cost.dao.SegmentCostEsDAO org.skywalking.apm.collector.agentstream.worker.global.dao.GlobalTraceEsDAO org.skywalking.apm.collector.agentstream.worker.service.entry.dao.ServiceEntryEsDAO -org.skywalking.apm.collector.agentstream.worker.serviceref.reference.dao.ServiceRefEsDAO \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.serviceref.reference.dao.ServiceRefEsDAO +org.skywalking.apm.collector.agentstream.worker.instance.performance.dao.InstPerformanceEsDAO \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define index f0dd190a7..d8170fa44 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define @@ -10,7 +10,4 @@ org.skywalking.apm.collector.agentstream.worker.segment.cost.dao.SegmentCostH2DA org.skywalking.apm.collector.agentstream.worker.global.dao.GlobalTraceH2DAO org.skywalking.apm.collector.agentstream.worker.service.entry.dao.ServiceEntryH2DAO org.skywalking.apm.collector.agentstream.worker.serviceref.reference.dao.ServiceRefH2DAO -org.skywalking.apm.collector.agentjvm.worker.cpu.dao.CpuMetricH2DAO -org.skywalking.apm.collector.agentjvm.worker.memory.dao.MemoryMetricH2DAO -org.skywalking.apm.collector.agentjvm.worker.memorypool.dao.MemoryPoolMetricH2DAO -org.skywalking.apm.collector.agentjvm.worker.gc.dao.GCMetricH2DAO \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.instance.performance.dao.InstPerformanceH2DAO \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_worker_provider.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_worker_provider.define index 39434ad03..45ea31146 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_worker_provider.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_worker_provider.define @@ -20,11 +20,8 @@ org.skywalking.apm.collector.agentstream.worker.segment.origin.SegmentPersistenc org.skywalking.apm.collector.agentstream.worker.segment.cost.SegmentCostPersistenceWorker$Factory org.skywalking.apm.collector.agentstream.worker.global.GlobalTracePersistenceWorker$Factory -org.skywalking.apm.collector.agentjvm.worker.cpu.CpuMetricPersistenceWorker$Factory -org.skywalking.apm.collector.agentjvm.worker.memory.MemoryMetricPersistenceWorker$Factory -org.skywalking.apm.collector.agentjvm.worker.memorypool.MemoryPoolMetricPersistenceWorker$Factory -org.skywalking.apm.collector.agentjvm.worker.gc.GCMetricPersistenceWorker$Factory - org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterSerialWorker$Factory org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceRegisterSerialWorker$Factory -org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameRegisterSerialWorker$Factory \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameRegisterSerialWorker$Factory + +org.skywalking.apm.collector.agentstream.worker.instance.performance.InstPerformancePersistenceWorker$Factory \ No newline at end of file diff --git a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/instance/InstPerformanceTable.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/instance/InstPerformanceTable.java new file mode 100644 index 000000000..9b72c6c25 --- /dev/null +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/instance/InstPerformanceTable.java @@ -0,0 +1,14 @@ +package org.skywalking.apm.collector.storage.table.instance; + +import org.skywalking.apm.collector.storage.table.CommonTable; + +/** + * @author pengys5 + */ +public class InstPerformanceTable extends CommonTable { + public static final String TABLE = "instance_performance"; + public static final String COLUMN_APPLICATION_ID = "application_id"; + public static final String COLUMN_INSTANCE_ID = "instance_id"; + public static final String COLUMN_CALL_TIMES = "call_times"; + public static final String COLUMN_COST_TOTAL = "cost_total"; +} diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricTable.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/CpuMetricTable.java similarity index 84% rename from apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricTable.java rename to apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/CpuMetricTable.java index 3c5378500..4a0b9277e 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/define/CpuMetricTable.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/CpuMetricTable.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agentjvm.worker.cpu.define; +package org.skywalking.apm.collector.storage.table.jvm; import org.skywalking.apm.collector.storage.table.CommonTable; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricTable.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/GCMetricTable.java similarity index 87% rename from apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricTable.java rename to apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/GCMetricTable.java index 6e828c082..9b70277c7 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricTable.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/GCMetricTable.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agentjvm.worker.gc.define; +package org.skywalking.apm.collector.storage.table.jvm; import org.skywalking.apm.collector.storage.table.CommonTable; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricTable.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/MemoryMetricTable.java similarity index 89% rename from apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricTable.java rename to apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/MemoryMetricTable.java index ccd355a9d..3551dabfb 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memory/define/MemoryMetricTable.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/MemoryMetricTable.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agentjvm.worker.memory.define; +package org.skywalking.apm.collector.storage.table.jvm; import org.skywalking.apm.collector.storage.table.CommonTable; diff --git a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricTable.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/MemoryPoolMetricTable.java similarity index 89% rename from apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricTable.java rename to apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/MemoryPoolMetricTable.java index 55648a036..d03c3b889 100644 --- a/apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/memorypool/define/MemoryPoolMetricTable.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/table/jvm/MemoryPoolMetricTable.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agentjvm.worker.memorypool.define; +package org.skywalking.apm.collector.storage.table.jvm; import org.skywalking.apm.collector.storage.table.CommonTable; diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/GCMetricEsDAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/GCMetricEsDAO.java new file mode 100644 index 000000000..81017dea7 --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/GCMetricEsDAO.java @@ -0,0 +1,51 @@ +package org.skywalking.apm.collector.ui.dao; + +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.search.SearchType; +import org.elasticsearch.index.query.BoolQueryBuilder; +import org.elasticsearch.index.query.MatchQueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; +import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; +import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable; +import org.skywalking.apm.network.proto.GCPhrase; + +/** + * @author pengys5 + */ +public class GCMetricEsDAO extends EsDAO implements IGCMetricDAO { + + @Override public GCCount getGCCount(long timestamp, int instanceId) { + SearchRequestBuilder searchRequestBuilder = getClient().prepareSearch(GCMetricTable.TABLE); + searchRequestBuilder.setTypes(GCMetricTable.TABLE_TYPE); + searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH); + + BoolQueryBuilder boolQuery = QueryBuilders.boolQuery(); + MatchQueryBuilder matchApplicationId = QueryBuilders.matchQuery(GCMetricTable.COLUMN_APPLICATION_INSTANCE_ID, instanceId); + MatchQueryBuilder matchTimeBucket = QueryBuilders.matchQuery(GCMetricTable.COLUMN_TIME_BUCKET, timestamp); + + boolQuery.must().add(matchApplicationId); + boolQuery.must().add(matchTimeBucket); + + searchRequestBuilder.setQuery(boolQuery); + searchRequestBuilder.setSize(100); + + SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); + SearchHit[] searchHits = searchResponse.getHits().getHits(); + + GCCount gcCount = new GCCount(); + for (SearchHit searchHit : searchHits) { + int phrase = (Integer)searchHit.getSource().get(GCMetricTable.COLUMN_PHRASE); + int count = (Integer)searchHit.getSource().get(GCMetricTable.COLUMN_COUNT); + + if (phrase == GCPhrase.NEW_VALUE) { + gcCount.setYoung(count); + } else if (phrase == GCPhrase.OLD_VALUE) { + gcCount.setOld(count); + } + } + + return gcCount; + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/GCMetricH2DAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/GCMetricH2DAO.java new file mode 100644 index 000000000..42f0d0924 --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/GCMetricH2DAO.java @@ -0,0 +1,13 @@ +package org.skywalking.apm.collector.ui.dao; + +import org.skywalking.apm.collector.storage.h2.dao.H2DAO; + +/** + * @author pengys5 + */ +public class GCMetricH2DAO extends H2DAO implements IGCMetricDAO { + + @Override public GCCount getGCCount(long timestamp, int instanceId) { + return null; + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IGCMetricDAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IGCMetricDAO.java new file mode 100644 index 000000000..aa162e499 --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IGCMetricDAO.java @@ -0,0 +1,39 @@ +package org.skywalking.apm.collector.ui.dao; + +/** + * @author pengys5 + */ +public interface IGCMetricDAO { + + GCCount getGCCount(long timestamp, int instanceId); + + class GCCount { + private int young; + private int old; + private int full; + + public int getYoung() { + return young; + } + + public int getOld() { + return old; + } + + public int getFull() { + return full; + } + + public void setYoung(int young) { + this.young = young; + } + + public void setOld(int old) { + this.old = old; + } + + public void setFull(int full) { + this.full = full; + } + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IInstPerformanceDAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IInstPerformanceDAO.java new file mode 100644 index 000000000..294d37f6f --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IInstPerformanceDAO.java @@ -0,0 +1,34 @@ +package org.skywalking.apm.collector.ui.dao; + +import java.util.List; + +/** + * @author pengys5 + */ +public interface IInstPerformanceDAO { + List getMultiple(long timestamp, int applicationId); + + class InstPerformance { + private final int instanceId; + private final int callTimes; + private final long costTotal; + + public InstPerformance(int instanceId, int callTimes, long costTotal) { + this.instanceId = instanceId; + this.callTimes = callTimes; + this.costTotal = costTotal; + } + + public int getInstanceId() { + return instanceId; + } + + public int getCallTimes() { + return callTimes; + } + + public long getCostTotal() { + return costTotal; + } + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IInstanceDAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IInstanceDAO.java index d8bbee131..3d40c136b 100644 --- a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IInstanceDAO.java +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/IInstanceDAO.java @@ -1,5 +1,7 @@ package org.skywalking.apm.collector.ui.dao; +import java.util.List; + /** * @author pengys5 */ @@ -7,4 +9,24 @@ public interface IInstanceDAO { Long lastHeartBeatTime(); Long instanceLastHeartBeatTime(long applicationInstanceId); + + List getApplications(long time); + + class Application { + private final int applicationId; + private final long count; + + public Application(int applicationId, long count) { + this.applicationId = applicationId; + this.count = count; + } + + public int getApplicationId() { + return applicationId; + } + + public long getCount() { + return count; + } + } } diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstPerformanceEsDAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstPerformanceEsDAO.java new file mode 100644 index 000000000..82ae92b7d --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstPerformanceEsDAO.java @@ -0,0 +1,48 @@ +package org.skywalking.apm.collector.ui.dao; + +import java.util.LinkedList; +import java.util.List; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.search.SearchType; +import org.elasticsearch.index.query.BoolQueryBuilder; +import org.elasticsearch.index.query.MatchQueryBuilder; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; +import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; +import org.skywalking.apm.collector.storage.table.instance.InstPerformanceTable; + +/** + * @author pengys5 + */ +public class InstPerformanceEsDAO extends EsDAO implements IInstPerformanceDAO { + + @Override public List getMultiple(long timestamp, int applicationId) { + SearchRequestBuilder searchRequestBuilder = getClient().prepareSearch(InstPerformanceTable.TABLE); + searchRequestBuilder.setTypes(InstPerformanceTable.TABLE_TYPE); + searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH); + + BoolQueryBuilder boolQuery = QueryBuilders.boolQuery(); + MatchQueryBuilder matchApplicationId = QueryBuilders.matchQuery(InstPerformanceTable.COLUMN_APPLICATION_ID, applicationId); + MatchQueryBuilder matchTimeBucket = QueryBuilders.matchQuery(InstPerformanceTable.COLUMN_TIME_BUCKET, timestamp); + + boolQuery.must().add(matchApplicationId); + boolQuery.must().add(matchTimeBucket); + + searchRequestBuilder.setQuery(boolQuery); + searchRequestBuilder.setSize(100); + + SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); + List instPerformances = new LinkedList<>(); + SearchHit[] searchHits = searchResponse.getHits().getHits(); + for (SearchHit searchHit : searchHits) { + int instanceId = (Integer)searchHit.getSource().get(InstPerformanceTable.COLUMN_INSTANCE_ID); + int callTimes = (Integer)searchHit.getSource().get(InstPerformanceTable.COLUMN_CALL_TIMES); + long costTotal = (Long)searchHit.getSource().get(InstPerformanceTable.COLUMN_COST_TOTAL); + + instPerformances.add(new InstPerformance(instanceId, callTimes, costTotal)); + } + + return instPerformances; + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstPerformanceH2DAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstPerformanceH2DAO.java new file mode 100644 index 000000000..311f4d5fc --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstPerformanceH2DAO.java @@ -0,0 +1,14 @@ +package org.skywalking.apm.collector.ui.dao; + +import java.util.List; +import org.skywalking.apm.collector.storage.h2.dao.H2DAO; + +/** + * @author pengys5 + */ +public class InstPerformanceH2DAO extends H2DAO implements IInstPerformanceDAO { + + @Override public List getMultiple(long timestamp, int applicationId) { + return null; + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstanceEsDAO.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstanceEsDAO.java index 85e542eba..88f4103f3 100644 --- a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstanceEsDAO.java +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/dao/InstanceEsDAO.java @@ -1,5 +1,7 @@ package org.skywalking.apm.collector.ui.dao; +import java.util.LinkedList; +import java.util.List; import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchType; @@ -9,10 +11,11 @@ import org.elasticsearch.index.query.MatchQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.RangeQueryBuilder; import org.elasticsearch.search.SearchHit; +import org.elasticsearch.search.aggregations.AggregationBuilders; +import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.sort.SortBuilders; import org.elasticsearch.search.sort.SortMode; import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO; -import org.skywalking.apm.collector.storage.table.node.NodeComponentTable; import org.skywalking.apm.collector.storage.table.register.InstanceTable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,7 +45,7 @@ public class InstanceEsDAO extends EsDAO implements IInstanceDAO { private Long heartBeatTime(AbstractQueryBuilder queryBuilder) { SearchRequestBuilder searchRequestBuilder = getClient().prepareSearch(InstanceTable.TABLE); - searchRequestBuilder.setTypes(NodeComponentTable.TABLE_TYPE); + searchRequestBuilder.setTypes(InstanceTable.TABLE_TYPE); searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH); searchRequestBuilder.setQuery(queryBuilder); searchRequestBuilder.setSize(1); @@ -60,4 +63,27 @@ public class InstanceEsDAO extends EsDAO implements IInstanceDAO { } return heartBeatTime; } + + @Override public List getApplications(long time) { + SearchRequestBuilder searchRequestBuilder = getClient().prepareSearch(InstanceTable.TABLE); + searchRequestBuilder.setTypes(InstanceTable.TABLE_TYPE); + searchRequestBuilder.setSearchType(SearchType.DFS_QUERY_THEN_FETCH); + + RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(InstanceTable.COLUMN_HEARTBEAT_TIME).gt(time); + + searchRequestBuilder.setQuery(rangeQueryBuilder); + searchRequestBuilder.setSize(0); + searchRequestBuilder.addAggregation(AggregationBuilders.terms(InstanceTable.COLUMN_APPLICATION_ID).field(InstanceTable.COLUMN_APPLICATION_ID).size(100)); + + SearchResponse searchResponse = searchRequestBuilder.execute().actionGet(); + Terms genders = searchResponse.getAggregations().get(InstanceTable.COLUMN_APPLICATION_ID); + + List applications = new LinkedList<>(); + for (Terms.Bucket entry : genders.getBuckets()) { + Integer applicationId = entry.getKeyAsNumber().intValue(); + long instanceCount = entry.getDocCount(); + applications.add(new Application(applicationId, instanceCount)); + } + return applications; + } } diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/jetty/handler/instancehealth/ApplicationsGetHandler.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/jetty/handler/instancehealth/ApplicationsGetHandler.java new file mode 100644 index 000000000..be2de6f68 --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/jetty/handler/instancehealth/ApplicationsGetHandler.java @@ -0,0 +1,41 @@ +package org.skywalking.apm.collector.ui.jetty.handler.instancehealth; + +import com.google.gson.JsonElement; +import javax.servlet.http.HttpServletRequest; +import org.skywalking.apm.collector.server.jetty.ArgumentsParseException; +import org.skywalking.apm.collector.server.jetty.JettyHandler; +import org.skywalking.apm.collector.ui.service.InstanceHealthService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class ApplicationsGetHandler extends JettyHandler { + + private final Logger logger = LoggerFactory.getLogger(ApplicationsGetHandler.class); + + @Override public String pathSpec() { + return "/instance/health/applications"; + } + + private InstanceHealthService service = new InstanceHealthService(); + + @Override protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException { + String timestamp = req.getParameter("timestamp"); + logger.debug("instance health applications get time: {}", timestamp); + + long time; + try { + time = Long.parseLong(timestamp); + } catch (NumberFormatException e) { + throw new ArgumentsParseException("time must be long"); + } + + return service.getApplications(time); + } + + @Override protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException { + throw new UnsupportedOperationException(); + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/jetty/handler/instancehealth/InstanceHealthGetHandler.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/jetty/handler/instancehealth/InstanceHealthGetHandler.java new file mode 100644 index 000000000..107730e13 --- /dev/null +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/jetty/handler/instancehealth/InstanceHealthGetHandler.java @@ -0,0 +1,49 @@ +package org.skywalking.apm.collector.ui.jetty.handler.instancehealth; + +import com.google.gson.JsonElement; +import javax.servlet.http.HttpServletRequest; +import org.skywalking.apm.collector.server.jetty.ArgumentsParseException; +import org.skywalking.apm.collector.server.jetty.JettyHandler; +import org.skywalking.apm.collector.ui.service.InstanceHealthService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class InstanceHealthGetHandler extends JettyHandler { + + private final Logger logger = LoggerFactory.getLogger(InstanceHealthGetHandler.class); + + @Override public String pathSpec() { + return "/instance/health/applicationId"; + } + + private InstanceHealthService service = new InstanceHealthService(); + + @Override protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException { + String timestampStr = req.getParameter("timestamp"); + String applicationIdStr = req.getParameter("applicationId"); + logger.debug("instance health get timestamp: {}", timestampStr); + + long timestamp; + try { + timestamp = Long.parseLong(timestampStr); + } catch (NumberFormatException e) { + throw new ArgumentsParseException("timestamp must be long"); + } + + int applicationId; + try { + applicationId = Integer.parseInt(applicationIdStr); + } catch (NumberFormatException e) { + throw new ArgumentsParseException("application id must be integer"); + } + + return service.getInstances(timestamp, applicationId); + } + + @Override protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException { + throw new UnsupportedOperationException(); + } +} diff --git a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/service/InstanceHealthService.java b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/service/InstanceHealthService.java index 0367c4366..72ea89207 100644 --- a/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/service/InstanceHealthService.java +++ b/apm-collector/apm-collector-ui/src/main/java/org/skywalking/apm/collector/ui/service/InstanceHealthService.java @@ -1,8 +1,85 @@ package org.skywalking.apm.collector.ui.service; +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; +import java.util.List; +import org.skywalking.apm.collector.storage.dao.DAOContainer; +import org.skywalking.apm.collector.ui.cache.ApplicationCache; +import org.skywalking.apm.collector.ui.dao.GCMetricEsDAO; +import org.skywalking.apm.collector.ui.dao.IGCMetricDAO; +import org.skywalking.apm.collector.ui.dao.IInstPerformanceDAO; +import org.skywalking.apm.collector.ui.dao.IInstanceDAO; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + /** * @author pengys5 */ public class InstanceHealthService { + private final Logger logger = LoggerFactory.getLogger(InstanceHealthService.class); + + public JsonObject getApplications(long timestamp) { + IInstanceDAO instanceDAO = (IInstanceDAO)DAOContainer.INSTANCE.get(IInstanceDAO.class.getName()); + List applications = instanceDAO.getApplications(timestamp); + + JsonObject response = new JsonObject(); + JsonArray applicationArray = new JsonArray(); + + response.addProperty("timestamp", timestamp); + response.add("applicationList", applicationArray); + + applications.forEach(application -> { + JsonObject applicationJson = new JsonObject(); + String applicationCode = ApplicationCache.get(application.getApplicationId()); + applicationJson.addProperty("applicationId", application.getApplicationId()); + applicationJson.addProperty("applicationCode", applicationCode); + applicationJson.addProperty("instanceCount", application.getCount()); + }); + + return response; + } + + public JsonObject getInstances(long timestamp, int applicationId) { + JsonObject response = new JsonObject(); + + IInstPerformanceDAO instPerformanceDAO = (IInstPerformanceDAO)DAOContainer.INSTANCE.get(IInstPerformanceDAO.class.getName()); + List performances = instPerformanceDAO.getMultiple(timestamp, applicationId); + + response.addProperty("timestamp", timestamp); + + JsonArray instances = new JsonArray(); + response.addProperty("applicationCode", ApplicationCache.get(applicationId)); + response.addProperty("applicationId", applicationId); + response.add("appInstances", instances); + + GCMetricEsDAO gcMetricEsDAO = (GCMetricEsDAO)DAOContainer.INSTANCE.get(GCMetricEsDAO.class.getName()); + performances.forEach(instance -> { + JsonObject instanceJson = new JsonObject(); + instanceJson.addProperty("id", instance.getInstanceId()); + instanceJson.addProperty("tps", instance.getCallTimes()); + + int avg = (int)((instance.getCostTotal() / instance.getCallTimes()) / 1000); + instanceJson.addProperty("avg", avg); + + if (avg > 5) { + instanceJson.addProperty("healthLevel", 0); + } else if (avg > 3 && avg <= 5) { + instanceJson.addProperty("healthLevel", 1); + } else if (avg > 1 && avg <= 3) { + instanceJson.addProperty("healthLevel", 2); + } else { + instanceJson.addProperty("healthLevel", 3); + } + + instanceJson.addProperty("status", 0); + + IGCMetricDAO.GCCount gcCount = gcMetricEsDAO.getGCCount(timestamp, instance.getInstanceId()); + instanceJson.addProperty("ygc", gcCount.getYoung()); + instanceJson.addProperty("ogc", gcCount.getOld()); + + instances.add(instanceJson); + }); + return response; + } } diff --git a/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/es_dao.define b/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/es_dao.define index 1ea705134..8cbfd212a 100644 --- a/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/es_dao.define +++ b/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/es_dao.define @@ -7,4 +7,6 @@ org.skywalking.apm.collector.ui.dao.GlobalTraceEsDAO org.skywalking.apm.collector.ui.dao.SegmentEsDAO org.skywalking.apm.collector.ui.dao.ApplicationEsDAO org.skywalking.apm.collector.ui.dao.ServiceNameEsDAO -org.skywalking.apm.collector.ui.dao.InstanceEsDAO \ No newline at end of file +org.skywalking.apm.collector.ui.dao.InstanceEsDAO +org.skywalking.apm.collector.ui.dao.InstPerformanceEsDAO +org.skywalking.apm.collector.ui.dao.GCMetricEsDAO \ No newline at end of file diff --git a/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/h2_dao.define b/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/h2_dao.define index a501a251e..f45bcb397 100644 --- a/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/h2_dao.define +++ b/apm-collector/apm-collector-ui/src/main/resources/META-INF/defines/h2_dao.define @@ -7,4 +7,6 @@ org.skywalking.apm.collector.ui.dao.GlobalTraceH2DAO org.skywalking.apm.collector.ui.dao.SegmentH2DAO org.skywalking.apm.collector.ui.dao.ApplicationH2DAO org.skywalking.apm.collector.ui.dao.ServiceNameH2DAO -org.skywalking.apm.collector.ui.dao.InstanceH2DAO \ No newline at end of file +org.skywalking.apm.collector.ui.dao.InstanceH2DAO +org.skywalking.apm.collector.ui.dao.InstPerformanceH2DAO +org.skywalking.apm.collector.ui.dao.GCMetricH2DAO \ No newline at end of file