Commit 34aad0f0 authored by peng-yongsheng's avatar peng-yongsheng
Browse files

jvm stream finish.

parent 714b4ccb

apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/AgentModuleGRPCProvider.java

+3 −3
Original line number Diff line number Diff line
@@ -42,7 +42,7 @@ import org.skywalking.apm.collector.naming.service.NamingHandlerRegisterService; 
import org.skywalking.apm.collector.remote.RemoteModule;

import org.skywalking.apm.collector.server.Server;

import org.skywalking.apm.collector.storage.StorageModule;

import org.skywalking.apm.collector.stream.worker.base.WorkerCreateListener;

import org.skywalking.apm.collector.stream.StreamModule;



/**

 * @author peng-yongsheng
@@ -82,7 +82,7 @@ public class AgentModuleGRPCProvider extends ModuleProvider { 
        GRPCManagerService managerService = getManager().find(GRPCManagerModule.NAME).getService(GRPCManagerService.class);

        Server gRPCServer = managerService.createIfAbsent(host, port);



        AgentStreamSingleton.getInstance(getManager(), new WorkerCreateListener());

        AgentStreamSingleton.createInstanceIfAbsent(getManager());

        addHandlers(gRPCServer);

    }
@@ -91,7 +91,7 @@ public class AgentModuleGRPCProvider extends ModuleProvider { 
    }



    @Override public String[] requiredModules() {

        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, GRPCManagerModule.NAME, CacheModule.NAME, RemoteModule.NAME};

        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, GRPCManagerModule.NAME, CacheModule.NAME, RemoteModule.NAME, StreamModule.NAME};

    }



    private void addHandlers(Server gRPCServer) {

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

+2 −2
Original line number Diff line number Diff line
@@ -68,7 +68,7 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe 


        request.getMetricsList().forEach(metric -> {

            long time = TimeBucketUtils.INSTANCE.getSecondTimeBucket(metric.getTime());

            senToInstanceHeartBeatPersistenceWorker(instanceId, metric.getTime());

            sendToInstanceHeartBeatPersistenceWorker(instanceId, metric.getTime());

            sendToCpuMetricPersistenceWorker(instanceId, time, metric.getCpu());

            sendToMemoryMetricPersistenceWorker(instanceId, time, metric.getMemoryList());

            sendToMemoryPoolMetricPersistenceWorker(instanceId, time, metric.getMemoryPoolList());
@@ -79,7 +79,7 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe 
        responseObserver.onCompleted();

    }



    private void senToInstanceHeartBeatPersistenceWorker(int instanceId, long heartBeatTime) {

    private void sendToInstanceHeartBeatPersistenceWorker(int instanceId, long heartBeatTime) {

        Instance instance = new Instance(String.valueOf(instanceId));

        instance.setHeartBeatTime(TimeBucketUtils.INSTANCE.getSecondTimeBucket(heartBeatTime));

        instance.setInstanceId(instanceId);

apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/AgentModuleJettyProvider.java

+4 −6
Original line number Diff line number Diff line
@@ -36,7 +36,7 @@ import org.skywalking.apm.collector.naming.NamingModule; 
import org.skywalking.apm.collector.naming.service.NamingHandlerRegisterService;

import org.skywalking.apm.collector.server.Server;

import org.skywalking.apm.collector.storage.StorageModule;

import org.skywalking.apm.collector.storage.service.DAOService;

import org.skywalking.apm.collector.stream.StreamModule;



/**

 * @author peng-yongsheng
@@ -75,11 +75,9 @@ public class AgentModuleJettyProvider extends ModuleProvider { 
        NamingHandlerRegisterService namingHandlerRegisterService = getManager().find(NamingModule.NAME).getService(NamingHandlerRegisterService.class);

        namingHandlerRegisterService.register(new AgentJettyNamingHandler(namingListener));



        DAOService daoService = getManager().find(StorageModule.NAME).getService(DAOService.class);



        JettyManagerService managerService = getManager().find(JettyManagerModule.NAME).getService(JettyManagerService.class);

        Server jettyServer = managerService.createIfAbsent(host, port, contextPath);

        addHandlers(daoService, jettyServer);

        addHandlers(jettyServer);

    }



    @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
@@ -87,10 +85,10 @@ public class AgentModuleJettyProvider extends ModuleProvider { 
    }



    @Override public String[] requiredModules() {

        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, JettyManagerModule.NAME, CacheModule.NAME};

        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, JettyManagerModule.NAME, CacheModule.NAME, StreamModule.NAME};

    }



    private void addHandlers(DAOService daoService, Server jettyServer) {

    private void addHandlers(Server jettyServer) {

        jettyServer.addHandler(new TraceSegmentServletHandler());

    }

}

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/AgentStreamSingleton.java

+22 −9
Original line number Diff line number Diff line
@@ -18,9 +18,11 @@ 


package org.skywalking.apm.collector.agent.stream;



import org.skywalking.apm.collector.agent.stream.graph.JvmMetricStreamGraph;

import org.skywalking.apm.collector.agent.stream.graph.RegisterStreamGraph;

import org.skywalking.apm.collector.core.module.ModuleManager;

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

import org.skywalking.apm.collector.stream.timer.PersistenceTimer;

import org.skywalking.apm.collector.stream.worker.base.WorkerCreateListener;



/**
@@ -33,24 +35,35 @@ public class AgentStreamSingleton { 
    private final ModuleManager moduleManager;

    private final WorkerCreateListener workerCreateListener;



    public AgentStreamSingleton(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {

    private AgentStreamSingleton(ModuleManager moduleManager) {

        this.moduleManager = moduleManager;

        this.workerCreateListener = workerCreateListener;

        createJVMGraph();

        createRegisterGraph();

        createTraceGraph();

        this.workerCreateListener = new WorkerCreateListener();

        this.create();

    }



    public static synchronized AgentStreamSingleton getInstance(ModuleManager moduleManager,

        WorkerCreateListener workerCreateListener) {

    public static synchronized AgentStreamSingleton createInstanceIfAbsent(ModuleManager moduleManager) {

        if (ObjectUtils.isEmpty(INSTANCE)) {

            INSTANCE = new AgentStreamSingleton(moduleManager, workerCreateListener);

            INSTANCE = new AgentStreamSingleton(moduleManager);

        }

        return INSTANCE;

    }



    private void createJVMGraph() {

    private void create() {

        createJVMGraph();

        createRegisterGraph();

        createTraceGraph();



        PersistenceTimer timer = new PersistenceTimer();

        timer.start(moduleManager, workerCreateListener.getPersistenceWorkers());

    }



    private void createJVMGraph() {

        JvmMetricStreamGraph jvmMetricStreamGraph = new JvmMetricStreamGraph(moduleManager, workerCreateListener);

        jvmMetricStreamGraph.createCpuMetricGraph();

        jvmMetricStreamGraph.createGcMetricGraph();

        jvmMetricStreamGraph.createMemoryMetricGraph();

        jvmMetricStreamGraph.createMemoryPoolMetricGraph();

        jvmMetricStreamGraph.createHeartBeatGraph();

    }



    private void createRegisterGraph() {

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/graph/JvmMetricStreamGraph.java

+38 −2
Original line number Diff line number Diff line
@@ -18,14 +18,22 @@ 


package org.skywalking.apm.collector.agent.stream.graph;



import org.skywalking.apm.collector.agent.stream.worker.jvm.CpuMetricPersistenceWorker;

import org.skywalking.apm.collector.agent.stream.worker.jvm.GCMetricPersistenceWorker;

import org.skywalking.apm.collector.agent.stream.worker.jvm.InstHeartBeatPersistenceWorker;

import org.skywalking.apm.collector.agent.stream.worker.jvm.MemoryMetricPersistenceWorker;

import org.skywalking.apm.collector.agent.stream.worker.jvm.MemoryPoolMetricPersistenceWorker;

import org.skywalking.apm.collector.core.graph.Graph;

import org.skywalking.apm.collector.core.graph.GraphManager;

import org.skywalking.apm.collector.core.module.ModuleManager;

import org.skywalking.apm.collector.queue.QueueModule;

import org.skywalking.apm.collector.queue.service.QueueCreatorService;

import org.skywalking.apm.collector.storage.table.jvm.CpuMetric;

import org.skywalking.apm.collector.storage.table.jvm.GCMetric;

import org.skywalking.apm.collector.storage.table.jvm.MemoryMetric;

import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric;

import org.skywalking.apm.collector.storage.table.register.Instance;

import org.skywalking.apm.collector.stream.worker.base.ProviderNotFoundException;

import org.skywalking.apm.collector.stream.worker.base.WorkerCreateListener;



/**

 * @author peng-yongsheng
@@ -38,28 +46,56 @@ public class JvmMetricStreamGraph { 
    public static final int CPU_METRIC_GRAPH_ID = 103;

    public static final int INST_HEART_BEAT_GRAPH_ID = 104;



    private final ModuleManager moduleManager;

    private final WorkerCreateListener workerCreateListener;



    public JvmMetricStreamGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {

        this.moduleManager = moduleManager;

        this.workerCreateListener = workerCreateListener;

    }



    @SuppressWarnings("unchecked")

    public Graph<GCMetric> createGcMetricGraph() {

        QueueCreatorService<GCMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);



        Graph<GCMetric> graph = GraphManager.INSTANCE.createIfAbsent(GC_METRIC_GRAPH_ID, GCMetric.class);

        graph.addNode(new GCMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));

        return graph;

    }



    public Graph<CpuMetric> createCpuMetricGraph() throws ProviderNotFoundException {

    @SuppressWarnings("unchecked")

    public Graph<CpuMetric> createCpuMetricGraph() {

        QueueCreatorService<CpuMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);



        Graph<CpuMetric> graph = GraphManager.INSTANCE.createIfAbsent(CPU_METRIC_GRAPH_ID, CpuMetric.class);

        graph.addNode(new CpuMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));

        return graph;

    }



    @SuppressWarnings("unchecked")

    public Graph<MemoryMetric> createMemoryMetricGraph() {

        QueueCreatorService<MemoryMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);



        Graph<MemoryMetric> graph = GraphManager.INSTANCE.createIfAbsent(MEMORY_METRIC_GRAPH_ID, MemoryMetric.class);

        graph.addNode(new MemoryMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));

        return graph;

    }



    @SuppressWarnings("unchecked")

    public Graph<MemoryPoolMetric> createMemoryPoolMetricGraph() {

        QueueCreatorService<MemoryPoolMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);



        Graph<MemoryPoolMetric> graph = GraphManager.INSTANCE.createIfAbsent(MEMORY_POOL_METRIC_GRAPH_ID, MemoryPoolMetric.class);

        graph.addNode(new MemoryPoolMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));

        return graph;

    }



    @SuppressWarnings("unchecked")

    public Graph<Instance> createHeartBeatGraph() {

        QueueCreatorService<Instance> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);



        Graph<Instance> graph = GraphManager.INSTANCE.createIfAbsent(INST_HEART_BEAT_GRAPH_ID, Instance.class);

        graph.addNode(new InstHeartBeatPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));

        return graph;

    }

}