From a798bb64986d6be8704393889ecf9d609276c08c Mon Sep 17 00:00:00 2001 From: pengys5 <8082209@qq.com> Date: Wed, 9 Aug 2017 23:47:20 +0800 Subject: [PATCH] Add cpu metric record persistence, but not test. --- .../grpc/AgentStreamGRPCModuleDefine.java | 2 + .../handler/JVMMetricsServiceHandler.java | 56 ++++++++++ .../cpu/CpuMetricPersistenceWorker.java | 71 ++++++++++++ .../jvmmetric/cpu/dao/CpuMetricEsDAO.java | 34 ++++++ .../jvmmetric/cpu/dao/CpuMetricH2DAO.java | 9 ++ .../jvmmetric/cpu/dao/ICpuMetricDAO.java | 7 ++ .../cpu/define/CpuMetricDataDefine.java | 102 ++++++++++++++++++ .../cpu/define/CpuMetricEsTableDefine.java | 32 ++++++ .../cpu/define/CpuMetricH2TableDefine.java | 21 ++++ .../jvmmetric/cpu/define/CpuMetricTable.java | 12 +++ .../worker/util/TimeBucketUtils.java | 8 ++ .../resources/META-INF/defines/es_dao.define | 3 +- .../resources/META-INF/defines/h2_dao.define | 3 +- .../defines/local_worker_provider.define | 2 + .../resources/META-INF/defines/storage.define | 5 +- .../util/TimeBucketUtilsTestCase.java | 47 ++++++++ .../define/ElasticSearchColumnDefine.java | 2 +- .../storage/h2/define/H2ColumnDefine.java | 2 +- .../worker/impl/data/AttributeType.java | 2 +- .../stream/worker/impl/data/Data.java | 24 ++--- .../stream/worker/impl/data/DataDefine.java | 16 +-- .../stream/worker/impl/data/Operation.java | 2 +- .../impl/data/operate/AddOperation.java | 2 +- .../impl/data/operate/CoverOperation.java | 2 +- .../impl/data/operate/NonOperation.java | 2 +- .../src/main/proto/RemoteCommonService.proto | 4 +- 26 files changed, 440 insertions(+), 32 deletions(-) create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/handler/JVMMetricsServiceHandler.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/CpuMetricPersistenceWorker.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricEsDAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricH2DAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/ICpuMetricDAO.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricDataDefine.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricEsTableDefine.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricH2TableDefine.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricTable.java create mode 100644 apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/util/TimeBucketUtilsTestCase.java diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java index 1cc1bdfad..4e190caed 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java @@ -4,6 +4,7 @@ import java.util.LinkedList; import java.util.List; import org.skywalking.apm.collector.agentstream.AgentStreamModuleDefine; import org.skywalking.apm.collector.agentstream.AgentStreamModuleGroupDefine; +import org.skywalking.apm.collector.agentstream.grpc.handler.JVMMetricsServiceHandler; import org.skywalking.apm.collector.agentstream.grpc.handler.TraceSegmentServiceHandler; import org.skywalking.apm.collector.core.cluster.ClusterDataListener; import org.skywalking.apm.collector.core.framework.Handler; @@ -46,6 +47,7 @@ public class AgentStreamGRPCModuleDefine extends AgentStreamModuleDefine { @Override public List handlerList() { List handlers = new LinkedList<>(); handlers.add(new TraceSegmentServiceHandler()); + handlers.add(new JVMMetricsServiceHandler()); return handlers; } } diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/handler/JVMMetricsServiceHandler.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/handler/JVMMetricsServiceHandler.java new file mode 100644 index 000000000..f7f895ced --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/handler/JVMMetricsServiceHandler.java @@ -0,0 +1,56 @@ +package org.skywalking.apm.collector.agentstream.grpc.handler; + +import io.grpc.stub.StreamObserver; +import org.skywalking.apm.collector.agentstream.worker.Const; +import org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.CpuMetricPersistenceWorker; +import org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define.CpuMetricDataDefine; +import org.skywalking.apm.collector.agentstream.worker.util.TimeBucketUtils; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; +import org.skywalking.apm.collector.server.grpc.GRPCHandler; +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.network.proto.CPU; +import org.skywalking.apm.network.proto.Downstream; +import org.skywalking.apm.network.proto.JVMMetrics; +import org.skywalking.apm.network.proto.JVMMetricsServiceGrpc; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsServiceImplBase implements GRPCHandler { + + private final Logger logger = LoggerFactory.getLogger(JVMMetricsServiceHandler.class); + + @Override public void collect(JVMMetrics request, StreamObserver responseObserver) { + int applicationInstanceId = request.getApplicationInstanceId(); + logger.debug("receive the jvm metric from application instance, id: {}", applicationInstanceId); + + StreamModuleContext context = (StreamModuleContext)CollectorContextHelper.INSTANCE.getContext(StreamModuleGroupDefine.GROUP_NAME); + request.getMetricsList().forEach(metric -> { + long time = TimeBucketUtils.INSTANCE.getSecondTimeBucket(metric.getTime()); + sendToCpuMetricPersistenceWorker(context, applicationInstanceId, time, metric.getCpu()); + }); + + responseObserver.onNext(Downstream.newBuilder().build()); + responseObserver.onCompleted(); + } + + private void sendToCpuMetricPersistenceWorker(StreamModuleContext context, int applicationInstanceId, + long timeBucket, CPU cpu) { + CpuMetricDataDefine.CpuMetric cpuMetric = new CpuMetricDataDefine.CpuMetric(); + cpuMetric.setId(timeBucket + Const.ID_SPLIT + applicationInstanceId); + cpuMetric.setApplicationInstanceId(applicationInstanceId); + cpuMetric.setUsagePercent(cpu.getUsagePercent()); + cpuMetric.setTimeBucket(timeBucket); + try { + logger.debug("send to cpu metric persistence worker, id: {}", cpuMetric.getId()); + context.getClusterWorkerContext().lookup(CpuMetricPersistenceWorker.WorkerRole.INSTANCE).tell(cpuMetric.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/jvmmetric/cpu/CpuMetricPersistenceWorker.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/CpuMetricPersistenceWorker.java new file mode 100644 index 000000000..02b8eeb2e --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/CpuMetricPersistenceWorker.java @@ -0,0 +1,71 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu; + +import org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.dao.ICpuMetricDAO; +import org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define.CpuMetricDataDefine; +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 CpuMetricPersistenceWorker extends PersistenceWorker { + + public CpuMetricPersistenceWorker(Role role, ClusterWorkerContext clusterContext) { + super(role, clusterContext); + } + + @Override public void preStart() throws ProviderNotFoundException { + super.preStart(); + } + + @Override protected boolean needMergeDBData() { + return false; + } + + @Override protected IPersistenceDAO persistenceDAO() { + return (IPersistenceDAO)DAOContainer.INSTANCE.get(ICpuMetricDAO.class.getName()); + } + + public static class Factory extends AbstractLocalAsyncWorkerProvider { + @Override + public Role role() { + return WorkerRole.INSTANCE; + } + + @Override + public CpuMetricPersistenceWorker workerInstance(ClusterWorkerContext clusterContext) { + return new CpuMetricPersistenceWorker(role(), clusterContext); + } + + @Override + public int queueSize() { + return 1024; + } + } + + public enum WorkerRole implements Role { + INSTANCE; + + @Override + public String roleName() { + return CpuMetricPersistenceWorker.class.getSimpleName(); + } + + @Override + public WorkerSelector workerSelector() { + return new HashCodeSelector(); + } + + @Override public DataDefine dataDefine() { + return new CpuMetricDataDefine(); + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricEsDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricEsDAO.java new file mode 100644 index 000000000..e1c7982f3 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricEsDAO.java @@ -0,0 +1,34 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.dao; + +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.agentstream.worker.jvmmetric.cpu.define.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; +import org.skywalking.apm.collector.stream.worker.impl.data.DataDefine; + +/** + * @author pengys5 + */ +public class CpuMetricEsDAO extends EsDAO implements ICpuMetricDAO, IPersistenceDAO { + + @Override public Data get(String id, DataDefine dataDefine) { + return null; + } + + @Override public IndexRequestBuilder prepareBatchInsert(Data data) { + Map source = new HashMap<>(); + source.put(CpuMetricTable.COLUMN_APPLICATION_INSTANCE_ID, data.getDataInteger(0)); + source.put(CpuMetricTable.COLUMN_USAGE_PERCENT, data.getDataDouble(0)); + source.put(CpuMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(0)); + + return getClient().prepareIndex(CpuMetricTable.TABLE, data.getDataString(0)).setSource(source); + } + + @Override public UpdateRequestBuilder prepareBatchUpdate(Data data) { + return null; + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricH2DAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricH2DAO.java new file mode 100644 index 000000000..a4cec0a62 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/CpuMetricH2DAO.java @@ -0,0 +1,9 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.dao; + +import org.skywalking.apm.collector.storage.h2.dao.H2DAO; + +/** + * @author pengys5 + */ +public class CpuMetricH2DAO extends H2DAO implements ICpuMetricDAO { +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/ICpuMetricDAO.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/ICpuMetricDAO.java new file mode 100644 index 000000000..8569314ad --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/dao/ICpuMetricDAO.java @@ -0,0 +1,7 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.dao; + +/** + * @author pengys5 + */ +public interface ICpuMetricDAO { +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricDataDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricDataDefine.java new file mode 100644 index 000000000..b859b5382 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricDataDefine.java @@ -0,0 +1,102 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define; + +import org.skywalking.apm.collector.remote.grpc.proto.RemoteData; +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.CoverOperation; +import org.skywalking.apm.collector.stream.worker.impl.data.operate.NonOperation; + +/** + * @author pengys5 + */ +public class CpuMetricDataDefine extends DataDefine { + + @Override protected int initialCapacity() { + return 4; + } + + @Override protected void attributeDefine() { + addAttribute(0, new Attribute(CpuMetricTable.COLUMN_ID, AttributeType.STRING, new NonOperation())); + addAttribute(1, new Attribute(CpuMetricTable.COLUMN_APPLICATION_INSTANCE_ID, AttributeType.INTEGER, new CoverOperation())); + addAttribute(2, new Attribute(CpuMetricTable.COLUMN_USAGE_PERCENT, AttributeType.DOUBLE, new CoverOperation())); + addAttribute(2, new Attribute(CpuMetricTable.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 CpuMetric implements Transform { + private String id; + private int applicationInstanceId; + private double usagePercent; + private long timeBucket; + + public CpuMetric(String id, int applicationInstanceId, double usagePercent, long timeBucket) { + this.id = id; + this.applicationInstanceId = applicationInstanceId; + this.usagePercent = usagePercent; + this.timeBucket = timeBucket; + } + + public CpuMetric() { + } + + @Override public Data toData() { + CpuMetricDataDefine define = new CpuMetricDataDefine(); + Data data = define.build(id); + data.setDataString(0, this.id); + data.setDataInteger(0, this.applicationInstanceId); + data.setDataDouble(0, this.usagePercent); + data.setDataLong(0, this.timeBucket); + return data; + } + + @Override public CpuMetric toSelf(Data data) { + this.id = data.getDataString(0); + this.applicationInstanceId = data.getDataInteger(0); + this.usagePercent = data.getDataDouble(0); + this.timeBucket = data.getDataLong(0); + return this; + } + + public void setId(String id) { + this.id = id; + } + + public void setApplicationInstanceId(int applicationInstanceId) { + this.applicationInstanceId = applicationInstanceId; + } + + public void setUsagePercent(double usagePercent) { + this.usagePercent = usagePercent; + } + + public void setTimeBucket(long timeBucket) { + this.timeBucket = timeBucket; + } + + public String getId() { + return id; + } + + public int getApplicationInstanceId() { + return applicationInstanceId; + } + + public double getUsagePercent() { + return usagePercent; + } + + public long getTimeBucket() { + return timeBucket; + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricEsTableDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricEsTableDefine.java new file mode 100644 index 000000000..ee3bb0bbb --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricEsTableDefine.java @@ -0,0 +1,32 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define; + +import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine; +import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine; + +/** + * @author pengys5 + */ +public class CpuMetricEsTableDefine extends ElasticSearchTableDefine { + + public CpuMetricEsTableDefine() { + super(CpuMetricTable.TABLE); + } + + @Override public int refreshInterval() { + return 1; + } + + @Override public int numberOfShards() { + return 2; + } + + @Override public int numberOfReplicas() { + return 0; + } + + @Override public void initialize() { + addColumn(new ElasticSearchColumnDefine(CpuMetricTable.COLUMN_APPLICATION_INSTANCE_ID, ElasticSearchColumnDefine.Type.Integer.name())); + addColumn(new ElasticSearchColumnDefine(CpuMetricTable.COLUMN_USAGE_PERCENT, ElasticSearchColumnDefine.Type.Double.name())); + addColumn(new ElasticSearchColumnDefine(CpuMetricTable.COLUMN_TIME_BUCKET, ElasticSearchColumnDefine.Type.Long.name())); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricH2TableDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricH2TableDefine.java new file mode 100644 index 000000000..60e714b8e --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricH2TableDefine.java @@ -0,0 +1,21 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define; + +import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine; +import org.skywalking.apm.collector.storage.h2.define.H2TableDefine; + +/** + * @author pengys5 + */ +public class CpuMetricH2TableDefine extends H2TableDefine { + + public CpuMetricH2TableDefine() { + super(CpuMetricTable.TABLE); + } + + @Override public void initialize() { + addColumn(new H2ColumnDefine(CpuMetricTable.COLUMN_ID, H2ColumnDefine.Type.Varchar.name())); + addColumn(new H2ColumnDefine(CpuMetricTable.COLUMN_APPLICATION_INSTANCE_ID, H2ColumnDefine.Type.Int.name())); + addColumn(new H2ColumnDefine(CpuMetricTable.COLUMN_USAGE_PERCENT, H2ColumnDefine.Type.Double.name())); + addColumn(new H2ColumnDefine(CpuMetricTable.COLUMN_TIME_BUCKET, H2ColumnDefine.Type.Bigint.name())); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricTable.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricTable.java new file mode 100644 index 000000000..05f71dbff --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/jvmmetric/cpu/define/CpuMetricTable.java @@ -0,0 +1,12 @@ +package org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define; + +import org.skywalking.apm.collector.agentstream.worker.CommonTable; + +/** + * @author pengys5 + */ +public class CpuMetricTable extends CommonTable { + public static final String TABLE = "cpu_metric"; + public static final String COLUMN_APPLICATION_INSTANCE_ID = "application_instance_id"; + public static final String COLUMN_USAGE_PERCENT = "application_instance_id"; +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/util/TimeBucketUtils.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/util/TimeBucketUtils.java index de8390297..92a306fe9 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/util/TimeBucketUtils.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/util/TimeBucketUtils.java @@ -13,6 +13,7 @@ public enum TimeBucketUtils { private final SimpleDateFormat dayDateFormat = new SimpleDateFormat("yyyyMMdd"); private final SimpleDateFormat hourDateFormat = new SimpleDateFormat("yyyyMMddHH"); private final SimpleDateFormat minuteDateFormat = new SimpleDateFormat("yyyyMMddHHmm"); + private final SimpleDateFormat secondDateFormat = new SimpleDateFormat("yyyyMMddHHmmss"); public long getMinuteTimeBucket(long time) { Calendar calendar = Calendar.getInstance(); @@ -21,6 +22,13 @@ public enum TimeBucketUtils { return Long.valueOf(timeStr); } + public long getSecondTimeBucket(long time) { + Calendar calendar = Calendar.getInstance(); + calendar.setTimeInMillis(time); + String timeStr = secondDateFormat.format(calendar.getTime()); + return Long.valueOf(timeStr); + } + public long getHourTimeBucket(long time) { Calendar calendar = Calendar.getInstance(); calendar.setTimeInMillis(time); 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..9fc387d93 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.jvmmetric.cpu.dao.CpuMetricEsDAO \ 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 3f18e0daa..6499e03ef 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 @@ -9,4 +9,5 @@ org.skywalking.apm.collector.agentstream.worker.noderef.summary.dao.NodeRefSumH2 org.skywalking.apm.collector.agentstream.worker.segment.cost.dao.SegmentCostH2DAO 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 \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.serviceref.reference.dao.ServiceRefH2DAO +org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.dao.CpuMetricH2DAO \ 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 5ce5803b7..3e688d9e6 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,6 +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.agentstream.worker.jvmmetric.cpu.CpuMetricPersistenceWorker$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 diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define index c462b86ec..bd82f294f 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define @@ -32,4 +32,7 @@ org.skywalking.apm.collector.agentstream.worker.service.entry.define.ServiceEntr org.skywalking.apm.collector.agentstream.worker.service.entry.define.ServiceEntryH2TableDefine org.skywalking.apm.collector.agentstream.worker.serviceref.reference.define.ServiceRefEsTableDefine -org.skywalking.apm.collector.agentstream.worker.serviceref.reference.define.ServiceRefH2TableDefine \ No newline at end of file +org.skywalking.apm.collector.agentstream.worker.serviceref.reference.define.ServiceRefH2TableDefine + +org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define.CpuMetricEsTableDefine +org.skywalking.apm.collector.agentstream.worker.jvmmetric.cpu.define.CpuMetricH2TableDefine \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/util/TimeBucketUtilsTestCase.java b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/util/TimeBucketUtilsTestCase.java new file mode 100644 index 000000000..1ac55cc1e --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/util/TimeBucketUtilsTestCase.java @@ -0,0 +1,47 @@ +package org.skywalking.apm.collector.agentstream.util; + +import java.util.Calendar; +import java.util.TimeZone; +import org.junit.Assert; +import org.junit.Test; +import org.skywalking.apm.collector.agentstream.worker.util.TimeBucketUtils; + +/** + * @author pengys5 + */ +public class TimeBucketUtilsTestCase { + + @Test + public void testUTCLocation() { + TimeZone.setDefault(TimeZone.getTimeZone("UTC")); + long timeBucket = 201703310915L; + long changedTimeBucket = TimeBucketUtils.INSTANCE.changeToUTCTimeBucket(timeBucket); + Assert.assertEquals(201703310115L, changedTimeBucket); + } + + @Test + public void testUTC8Location() { + TimeZone.setDefault(TimeZone.getTimeZone("GMT+08:00")); + long timeBucket = 201703310915L; + long changedTimeBucket = TimeBucketUtils.INSTANCE.changeToUTCTimeBucket(timeBucket); + Assert.assertEquals(201703310915L, changedTimeBucket); + } + + @Test + public void testGetSecondTimeBucket() { + long timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(1490922929258L); + Assert.assertEquals(20170331091529L, timeBucket); + } + + @Test + public void test() { + Calendar calendar = Calendar.getInstance(); + calendar.setTimeInMillis(1490922929258L); + calendar.set(Calendar.SECOND, calendar.get(Calendar.SECOND) - 3); +// System.out.println(calendar.getTimeInMillis()); + calendar.set(Calendar.SECOND, calendar.get(Calendar.SECOND) - 2); +// System.out.println(calendar.getTimeInMillis()); + calendar.set(Calendar.SECOND, calendar.get(Calendar.SECOND) - 2); +// System.out.println(calendar.getTimeInMillis()); + } +} diff --git a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchColumnDefine.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchColumnDefine.java index 883480a44..2d7d38595 100644 --- a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchColumnDefine.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/elasticsearch/define/ElasticSearchColumnDefine.java @@ -11,6 +11,6 @@ public class ElasticSearchColumnDefine extends ColumnDefine { } public enum Type { - Binary, Boolean, Date, Keyword, Long, Integer + Binary, Boolean, Keyword, Long, Integer, Double } } diff --git a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java index ca7cf49f1..4122b20a2 100644 --- a/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java +++ b/apm-collector/apm-collector-storage/src/main/java/org/skywalking/apm/collector/storage/h2/define/H2ColumnDefine.java @@ -12,6 +12,6 @@ public class H2ColumnDefine extends ColumnDefine { } public enum Type { - Boolean, Varchar, Int, Bigint, BINARY + Boolean, Varchar, Int, Bigint, BINARY, Double } } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java index fa53ad9da..480ba1a93 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/AttributeType.java @@ -4,5 +4,5 @@ package org.skywalking.apm.collector.stream.worker.impl.data; * @author pengys5 */ public enum AttributeType { - STRING, LONG, FLOAT, INTEGER, BYTE, BOOLEAN + STRING, LONG, DOUBLE, INTEGER, BYTE, BOOLEAN } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java index 5a8832ba3..c4b2dd99b 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Data.java @@ -10,29 +10,29 @@ import org.skywalking.apm.collector.stream.worker.selector.AbstractHashMessage; public class Data extends AbstractHashMessage { private final int stringCapacity; private final int longCapacity; - private final int floatCapacity; + private final int doubleCapacity; private final int integerCapacity; private final int booleanCapacity; private final int byteCapacity; private String[] dataStrings; private Long[] dataLongs; - private Float[] dataFloats; + private Double[] dataDoubles; private Integer[] dataIntegers; private Boolean[] dataBooleans; private byte[][] dataBytes; - public Data(String id, int stringCapacity, int longCapacity, int floatCapacity, int integerCapacity, + public Data(String id, int stringCapacity, int longCapacity, int doubleCapacity, int integerCapacity, int booleanCapacity, int byteCapacity) { super(id); this.dataStrings = new String[stringCapacity]; this.dataLongs = new Long[longCapacity]; - this.dataFloats = new Float[floatCapacity]; + this.dataDoubles = new Double[doubleCapacity]; this.dataIntegers = new Integer[integerCapacity]; this.dataBooleans = new Boolean[booleanCapacity]; this.dataBytes = new byte[byteCapacity][]; this.stringCapacity = stringCapacity; this.longCapacity = longCapacity; - this.floatCapacity = floatCapacity; + this.doubleCapacity = doubleCapacity; this.integerCapacity = integerCapacity; this.booleanCapacity = booleanCapacity; this.byteCapacity = byteCapacity; @@ -46,8 +46,8 @@ public class Data extends AbstractHashMessage { dataLongs[position] = value; } - public void setDataFloat(int position, Float value) { - dataFloats[position] = value; + public void setDataDouble(int position, Double value) { + dataDoubles[position] = value; } public void setDataInteger(int position, Integer value) { @@ -70,8 +70,8 @@ public class Data extends AbstractHashMessage { return dataLongs[position]; } - public Float getDataFloat(int position) { - return dataFloats[position]; + public Double getDataDouble(int position) { + return dataDoubles[position]; } public Integer getDataInteger(int position) { @@ -93,7 +93,7 @@ public class Data extends AbstractHashMessage { public RemoteData serialize() { RemoteData.Builder builder = RemoteData.newBuilder(); builder.setIntegerCapacity(integerCapacity); - builder.setFloatCapacity(floatCapacity); + builder.setDoubleCapacity(doubleCapacity); builder.setStringCapacity(stringCapacity); builder.setLongCapacity(longCapacity); builder.setByteCapacity(byteCapacity); @@ -105,8 +105,8 @@ public class Data extends AbstractHashMessage { for (int i = 0; i < dataIntegers.length; i++) { builder.setDataIntegers(i, dataIntegers[i]); } - for (int i = 0; i < dataFloats.length; i++) { - builder.setDataFloats(i, dataFloats[i]); + for (int i = 0; i < dataDoubles.length; i++) { + builder.setDataDoubles(i, dataDoubles[i]); } for (int i = 0; i < dataLongs.length; i++) { builder.setDataLongs(i, dataLongs[i]); diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java index ddbd9c2a3..8e98a736d 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/DataDefine.java @@ -9,7 +9,7 @@ public abstract class DataDefine { private Attribute[] attributes; private int stringCapacity; private int longCapacity; - private int floatCapacity; + private int doubleCapacity; private int integerCapacity; private int booleanCapacity; private int byteCapacity; @@ -26,8 +26,8 @@ public abstract class DataDefine { stringCapacity++; } else if (AttributeType.LONG.equals(attribute.getType())) { longCapacity++; - } else if (AttributeType.FLOAT.equals(attribute.getType())) { - floatCapacity++; + } else if (AttributeType.DOUBLE.equals(attribute.getType())) { + doubleCapacity++; } else if (AttributeType.INTEGER.equals(attribute.getType())) { integerCapacity++; } else if (AttributeType.BOOLEAN.equals(attribute.getType())) { @@ -47,13 +47,13 @@ public abstract class DataDefine { protected abstract void attributeDefine(); public final Data build(String id) { - return new Data(id, stringCapacity, longCapacity, floatCapacity, integerCapacity, booleanCapacity, byteCapacity); + return new Data(id, stringCapacity, longCapacity, doubleCapacity, integerCapacity, booleanCapacity, byteCapacity); } public void mergeData(Data newData, Data oldData) { int stringPosition = 0; int longPosition = 0; - int floatPosition = 0; + int doublePosition = 0; int integerPosition = 0; int booleanPosition = 0; int bytePosition = 0; @@ -65,9 +65,9 @@ public abstract class DataDefine { } else if (AttributeType.LONG.equals(attribute.getType())) { attribute.getOperation().operate(newData.getDataLong(longPosition), oldData.getDataLong(longPosition)); longPosition++; - } else if (AttributeType.FLOAT.equals(attribute.getType())) { - attribute.getOperation().operate(newData.getDataFloat(floatPosition), oldData.getDataFloat(floatPosition)); - floatPosition++; + } else if (AttributeType.DOUBLE.equals(attribute.getType())) { + attribute.getOperation().operate(newData.getDataDouble(doublePosition), oldData.getDataDouble(doublePosition)); + doublePosition++; } else if (AttributeType.INTEGER.equals(attribute.getType())) { attribute.getOperation().operate(newData.getDataInteger(integerPosition), oldData.getDataInteger(integerPosition)); integerPosition++; diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java index 667051ab0..19bb7c583 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/Operation.java @@ -8,7 +8,7 @@ public interface Operation { Long operate(Long newValue, Long oldValue); - Float operate(Float newValue, Float oldValue); + Double operate(Double newValue, Double oldValue); Integer operate(Integer newValue, Integer oldValue); diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/AddOperation.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/AddOperation.java index 0772806e6..847b54928 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/AddOperation.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/AddOperation.java @@ -15,7 +15,7 @@ public class AddOperation implements Operation { return newValue + oldValue; } - @Override public Float operate(Float newValue, Float oldValue) { + @Override public Double operate(Double newValue, Double oldValue) { return newValue + oldValue; } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java index 06b93fdb0..95432820c 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/CoverOperation.java @@ -14,7 +14,7 @@ public class CoverOperation implements Operation { return newValue; } - @Override public Float operate(Float newValue, Float oldValue) { + @Override public Double operate(Double newValue, Double oldValue) { return newValue; } diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java index 62ebb2f95..ee7400902 100644 --- a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/worker/impl/data/operate/NonOperation.java @@ -14,7 +14,7 @@ public class NonOperation implements Operation { return oldValue; } - @Override public Float operate(Float newValue, Float oldValue) { + @Override public Double operate(Double newValue, Double oldValue) { return oldValue; } diff --git a/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto b/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto index 5f933c024..159bfe9bc 100644 --- a/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto +++ b/apm-collector/apm-collector-stream/src/main/proto/RemoteCommonService.proto @@ -16,13 +16,13 @@ message RemoteMessage { message RemoteData { int32 stringCapacity = 1; int32 longCapacity = 2; - int32 floatCapacity = 3; + int32 doubleCapacity = 3; int32 integerCapacity = 4; int32 byteCapacity = 5; int32 booleanCapacity = 6; repeated string dataStrings = 7; repeated int64 dataLongs = 8; - repeated float dataFloats = 9; + repeated double dataDoubles = 9; repeated int32 dataIntegers = 10; repeated bytes dataBytes = 11; repeated bool dataBooleans = 12;