Commit 7f5248c5 authored by 吴晟's avatar 吴晟 Committed by GitHub
Browse files

Merge pull request #369 from wu-sheng/feature/365 

Feature/365
parents 2fe5362a 73d309da

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/grpc/handler/JVMMetricsServiceHandler.java

+7 −6
Original line number Diff line number Diff line
@@ -3,23 +3,23 @@ package org.skywalking.apm.collector.agentjvm.grpc.handler; 
import io.grpc.stub.StreamObserver;

import java.util.List;

import org.skywalking.apm.collector.agentjvm.worker.cpu.CpuMetricPersistenceWorker;

import org.skywalking.apm.collector.storage.define.jvm.CpuMetricDataDefine;

import org.skywalking.apm.collector.agentjvm.worker.gc.GCMetricPersistenceWorker;

import org.skywalking.apm.collector.storage.define.jvm.GCMetricDataDefine;

import org.skywalking.apm.collector.agentjvm.worker.heartbeat.InstHeartBeatPersistenceWorker;

import org.skywalking.apm.collector.agentjvm.worker.heartbeat.define.InstanceHeartBeatDataDefine;

import org.skywalking.apm.collector.agentjvm.worker.memory.MemoryMetricPersistenceWorker;

import org.skywalking.apm.collector.storage.define.jvm.MemoryMetricDataDefine;

import org.skywalking.apm.collector.agentjvm.worker.memorypool.MemoryPoolMetricPersistenceWorker;

import org.skywalking.apm.collector.storage.define.jvm.MemoryPoolMetricDataDefine;

import org.skywalking.apm.collector.core.framework.CollectorContextHelper;

import org.skywalking.apm.collector.core.util.Const;

import org.skywalking.apm.collector.core.util.TimeBucketUtils;

import org.skywalking.apm.collector.server.grpc.GRPCHandler;

import org.skywalking.apm.collector.storage.define.jvm.CpuMetricDataDefine;

import org.skywalking.apm.collector.storage.define.jvm.GCMetricDataDefine;

import org.skywalking.apm.collector.storage.define.jvm.MemoryMetricDataDefine;

import org.skywalking.apm.collector.storage.define.jvm.MemoryPoolMetricDataDefine;

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.CPU;

import org.skywalking.apm.network.proto.Downstream;

import org.skywalking.apm.network.proto.GC;
@@ -111,7 +111,7 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe 


        memoryPools.forEach(memoryPool -> {

            MemoryPoolMetricDataDefine.MemoryPoolMetric memoryPoolMetric = new MemoryPoolMetricDataDefine.MemoryPoolMetric();

            memoryPoolMetric.setId(timeBucket + Const.ID_SPLIT + applicationInstanceId + Const.ID_SPLIT + String.valueOf(memoryPool.getType().getNumber()));

            memoryPoolMetric.setId(timeBucket + Const.ID_SPLIT + applicationInstanceId + Const.ID_SPLIT + memoryPool.getIsHeap() + Const.ID_SPLIT + String.valueOf(memoryPool.getType().getNumber()));

            memoryPoolMetric.setApplicationInstanceId(applicationInstanceId);

            memoryPoolMetric.setPoolType(memoryPool.getType().getNumber());

            memoryPoolMetric.setHeap(memoryPool.getIsHeap());
@@ -139,6 +139,7 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe 
            gcMetric.setCount(gc.getCount());

            gcMetric.setTime(gc.getTime());

            gcMetric.setTimeBucket(timeBucket);

            gcMetric.setS5TimeBucket(TimeBucketUtils.INSTANCE.getFiveSecondTimeBucket(timeBucket));

            try {

                logger.debug("send to gc metric persistence worker, id: {}", gcMetric.getId());

                context.getClusterWorkerContext().lookup(GCMetricPersistenceWorker.WorkerRole.INSTANCE).tell(gcMetric.toData());

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/cpu/dao/CpuMetricEsDAO.java

+7 −2
Original line number Diff line number Diff line
@@ -4,17 +4,21 @@ 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.core.stream.Data;

import org.skywalking.apm.collector.storage.define.DataDefine;

import org.skywalking.apm.collector.storage.define.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.core.stream.Data;

import org.skywalking.apm.collector.storage.define.DataDefine;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



/**

 * @author pengys5

 */

public class CpuMetricEsDAO extends EsDAO implements ICpuMetricDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder> {



    private final Logger logger = LoggerFactory.getLogger(CpuMetricEsDAO.class);



    @Override public Data get(String id, DataDefine dataDefine) {

        return null;

    }
@@ -25,6 +29,7 @@ public class CpuMetricEsDAO extends EsDAO implements ICpuMetricDAO, IPersistence 
        source.put(CpuMetricTable.COLUMN_USAGE_PERCENT, data.getDataDouble(0));

        source.put(CpuMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(0));



        logger.debug("prepare cpu metric batch insert, id: {}", data.getDataString(0));

        return getClient().prepareIndex(CpuMetricTable.TABLE, data.getDataString(0)).setSource(source);

    }

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/dao/GCMetricEsDAO.java

+4 −3
Original line number Diff line number Diff line
@@ -4,11 +4,11 @@ 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.core.stream.Data;

import org.skywalking.apm.collector.storage.define.DataDefine;

import org.skywalking.apm.collector.storage.define.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.core.stream.Data;

import org.skywalking.apm.collector.storage.define.DataDefine;



/**

 * @author pengys5
@@ -22,10 +22,11 @@ public class GCMetricEsDAO extends EsDAO implements IGCMetricDAO, IPersistenceDA 
    @Override public IndexRequestBuilder prepareBatchInsert(Data data) {

        Map<String, Object> source = new HashMap<>();

        source.put(GCMetricTable.COLUMN_APPLICATION_INSTANCE_ID, data.getDataInteger(0));

        source.put(GCMetricTable.COLUMN_PHRASE, data.getDataInteger(0));

        source.put(GCMetricTable.COLUMN_PHRASE, data.getDataInteger(1));

        source.put(GCMetricTable.COLUMN_COUNT, data.getDataLong(0));

        source.put(GCMetricTable.COLUMN_TIME, data.getDataLong(1));

        source.put(GCMetricTable.COLUMN_TIME_BUCKET, data.getDataLong(2));

        source.put(GCMetricTable.COLUMN_5S_TIME_BUCKET, data.getDataLong(3));



        return getClient().prepareIndex(GCMetricTable.TABLE, data.getDataString(0)).setSource(source);

    }

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricEsTableDefine.java

+2 −1
Original line number Diff line number Diff line 
package org.skywalking.apm.collector.agentjvm.worker.gc.define;



import org.skywalking.apm.collector.storage.define.jvm.GCMetricTable;

import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine;

import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine;

import org.skywalking.apm.collector.storage.define.jvm.GCMetricTable;



/**

 * @author pengys5
@@ -31,5 +31,6 @@ public class GCMetricEsTableDefine extends ElasticSearchTableDefine { 
        addColumn(new ElasticSearchColumnDefine(GCMetricTable.COLUMN_COUNT, ElasticSearchColumnDefine.Type.Long.name()));

        addColumn(new ElasticSearchColumnDefine(GCMetricTable.COLUMN_TIME, ElasticSearchColumnDefine.Type.Long.name()));

        addColumn(new ElasticSearchColumnDefine(GCMetricTable.COLUMN_TIME_BUCKET, ElasticSearchColumnDefine.Type.Long.name()));

        addColumn(new ElasticSearchColumnDefine(GCMetricTable.COLUMN_5S_TIME_BUCKET, ElasticSearchColumnDefine.Type.Long.name()));

    }

}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/worker/gc/define/GCMetricH2TableDefine.java

+1 −0
Original line number Diff line number Diff line
@@ -20,5 +20,6 @@ public class GCMetricH2TableDefine extends H2TableDefine { 
        addColumn(new H2ColumnDefine(GCMetricTable.COLUMN_COUNT, H2ColumnDefine.Type.Bigint.name()));

        addColumn(new H2ColumnDefine(GCMetricTable.COLUMN_TIME, H2ColumnDefine.Type.Bigint.name()));

        addColumn(new H2ColumnDefine(GCMetricTable.COLUMN_TIME_BUCKET, H2ColumnDefine.Type.Bigint.name()));

        addColumn(new H2ColumnDefine(GCMetricTable.COLUMN_5S_TIME_BUCKET, H2ColumnDefine.Type.Bigint.name()));

    }

}