Pinpoint 분석 #2

Pinpoint 3.0.1 Agent -> Collector 연결 및 Trace 수집/저장 분석

기준 코드: v3.0.1 태그

이 문서는 Pinpoint Java agent가 대상 JVM에 붙어서 bytecode instrumentation으로 trace 정보를 만들고, gRPC로 collector에 연결/전송하며, collector가 받은 데이터를 HBase 계열 저장소와 통계 인덱스에 저장하는 과정을 코드 기준으로 정리한다.

전체 흐름 요약

1. Agent 부트스트랩과 instrumentation 등록

Agent 진입점은 agent-module/bootstraps/bootstrap/src/main/java/com/navercorp/pinpoint/bootstrap/PinpointBootStrap.java의 premain()이다. JVM이 -javaagent로 agent를 로딩하면 premain(String agentArgs, Instrumentation instrumentation)이 실행되고, disable 플래그와 중복 시작을 검사한 뒤 PinpointBootStrap.start()로 넘어간다.

PinpointStarter는 agent 디렉터리, profile config, agentId/applicationName 등을 해석하고 DefaultAgentOption을 만든다. 이후 profiler 쪽 DefaultAgent가 생성된다.

핵심 초기화는 agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java:

• 생성자에서 ProfilerConfig와 logging을 준비한다.
• newApplicationContext()가 DefaultApplicationContext를 만든다.
• start()에서 applicationContext.start()를 호출한다.

DefaultApplicationContext는 실제 runtime wiring 지점이다.

• Guice injector를 만들고 ClassFileTransformer, TraceContext, AgentInfoSender, AgentStatMonitor 등을 얻는다.
• instrumentation.addTransformer(classFileTransformer, true)로 JVM instrumentation hook을 등록한다.
• RPC module lifecycle을 시작해 gRPC sender들을 초기화한다.
• start()에서 interceptor registry를 bind하고 agent info/stat monitor를 시작한다.

즉 “agent가 애플리케이션에 붙는다”는 것은 JVM Instrumentation API에 transformer를 등록하고, plugin들이 지정한 클래스가 로딩/재변환될 때 bytecode에 interceptor 호출이 삽입되는 구조다.

2. Plugin transformer와 interceptor가 trace를 가로채는 방식

Plugin은 InstrumentContext.addClassFileTransformer(...) 계열 API를 통해 대상 클래스/메서드에 transformer를 등록한다. 예를 들어 Tomcat, Spring, Kafka, DB client 등 각 plugin은 특정 class matcher와 transform callback을 등록한다.

런타임에서 instrumented method가 호출되면 bootstrap interceptor가 실행된다. 일반적인 패턴은 다음과 같다.

• 서버 진입점 interceptor는 request header를 읽어 기존 trace를 이어가거나 새 trace를 만든다.
• 내부/외부 호출 interceptor는 현재 trace의 span event를 시작/종료한다.
• outgoing request interceptor는 다음 hop으로 trace header를 전파한다.

trace context API는 agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/DefaultTraceContext.java와 factory 계열에 있다.

• newTraceObject()는 root trace를 생성한다.
• continueTraceObject(traceId)는 upstream에서 받은 trace id로 이어 붙인다.
• currentTraceObject()는 현재 thread-local trace를 가져온다.
• removeTraceObject()는 request 종료 시 trace를 thread-local에서 제거한다.

실제 span event 기록은 DefaultTrace가 담당한다.

• traceBlockBegin()이 SpanEvent를 만들고 call stack에 push한다.
• traceBlockEnd()가 call stack에서 pop하고 elapsed time을 찍은 뒤 storage.store(spanEvent)로 넘긴다.
• close()가 root span 종료 시각을 찍고 storage.store(span)으로 완성된 span을 전송한다.

3. Agent 내부에서 Span/PSpan/PSpanChunk가 만들어지는 경로

StorageFactoryProvider는 trace 저장소로 BufferedStorageFactory를 만든다. 이 저장소는 이름과 달리 agent-local 영구 저장소가 아니라, span event를 잠시 buffer하다가 gRPC sender로 넘기는 전송 버퍼다.

BufferedStorage 동작:

• store(SpanEvent)는 span event를 buffer에 넣는다.
• buffer가 넘치면 현재 span event 묶음을 SpanChunk로 만들어 전송한다.
• store(Span)은 남아 있는 span event를 root Span에 붙이고 span.finish() 후 DataSender<SpanType>.send(span)을 호출한다.
• flush()는 남은 span event만 SpanChunk로 보낸다.

SpanGrpcDataSender는 SpanType을 protobuf message로 변환한다.

• Span -> PSpan
• SpanChunk -> PSpanChunk
• 둘 다 PSpanMessage에 감싸서 collector의 SpanGrpc.sendSpan() client streaming RPC로 보낸다.

SpanGrpcDataSender는 생성 시 spanStub.sendSpan(response)로 stream을 열고, reconnect/renew 로직을 가진 DefaultStreamTask가 queue에서 데이터를 꺼내 stream.onNext(PSpanMessage)로 계속 밀어 넣는다.

4. Agent와 Collector의 연결 과정

Pinpoint 3.0.1 gRPC 전송은 목적별 포트를 분리한다.

• Agent channel: agent info, metadata, command, ping/lifecycle. 기본 9991.
• Stat channel: agent stat metric. 기본 9992.
• Span channel: trace span/span chunk. 기본 9993.

agent 설정은 agent-module/agent/src/main/resources/pinpoint-root.config 기준:

• profiler.transport.grpc.collector.ip=127.0.0.1
• profiler.transport.grpc.agent.collector.port=9991
• profiler.transport.grpc.stat.collector.port=9992
• profiler.transport.grpc.span.collector.port=9993

4.1 AgentInfo 전송

AgentInfoSender는 시작 시 agent 정보를 만들고 collector로 전송한다. 실패하면 3초 간격으로 재시도하고, 성공 후에는 기본 24시간마다 refresh한다.

전송 구현은 AgentGrpcDataSender.request():

• MetaDataType을 PAgentInfo로 변환한다.
• agentInfoStub.requestAgentInfo(pAgentInfo, observer) unary RPC를 호출한다.
• collector의 AgentService.requestAgentInfo()가 받는다.

collector 쪽 GrpcAgentInfoHandler는:

• PAgentInfo와 gRPC header의 agent identity를 AgentInfoBo로 매핑한다.
• AgentInfoService.insert(agentInfoBo)로 저장한다.
• 성공 여부를 PResult로 응답한다.

4.2 Ping/lifecycle stream

AgentGrpcDataSender는 agent info stub 외에 ping stub도 만든다. newAgentPingStub()는 SocketIdClientInterceptor를 붙이고, PingStreamContext가 AgentGrpc.pingSession() bi-directional stream을 유지한다.

collector의 AgentService.pingSession()은:

• 첫 PPing에서 pingEventHandler.connect()를 호출한다.
• 이후 ping마다 pingEventHandler.ping()을 호출한다.
• stream error/completed 시 pingEventHandler.close()를 호출한다.
• 응답 stream이 준비되어 있으면 빈 PPing을 echo한다.

DefaultPingEventHandler는 gRPC ServerContext의 transport metadata와 agent header로 PingSession을 만들고 registry에 보관한다.

• connect()는 PingSession 생성 후 LifecycleListener.connect() 호출.
• ping()은 session을 찾아 마지막 ping 시간을 갱신하고 handshake() 호출.
• close()는 session 제거 후 LifecycleListener.close() 호출.
• update(serviceType)는 agent info 수신 이후 ping session의 service type을 보정한다.

AgentLifecycleListener와 KeepAliveService는 lifecycle 상태를 저장한다.

• connect: ManagedAgentLifeCycle.RUNNING
• close: collector shutdown 여부에 따라 CLOSED_BY_SERVER 또는 CLOSED_BY_CLIENT
• ping: agent state 통계 갱신

최종 저장은 AgentLifeCycleAsyncTaskService -> AgentLifeCycleService -> AgentLifeCycleDao 경로다. 동시에 agent event와 agent state 통계도 갱신된다.

5. Collector가 Span stream을 받는 경로

collector span receiver 구성은 GrpcSpanReceiverConfiguration이다.

• SpanService를 만들고 spanService.bindService()를 gRPC server에 등록한다.
• SpanDispatchHandler에 GrpcSpanHandler와 GrpcSpanChunkHandler를 연결한다.
• GrpcReceiver가 bind address, interceptor, executor, server option을 받아 span gRPC server를 띄운다.

SpanService.sendSpan()은 client streaming RPC endpoint다. agent가 보낸 PSpanMessage를 받을 때마다:

• hasSpan()이면 DefaultTBaseLocator.SPAN header로 Message<PSpan>을 만든다.
• hasSpanChunk()이면 DefaultTBaseLocator.SPANCHUNK header로 Message<PSpanChunk>를 만든다.
• ServerRequestFactory로 ServerRequest를 만들고 dispatchHandler.dispatchSendMessage(request)를 호출한다.

SpanDispatchHandler는 header type으로 분기한다.

• SPAN -> GrpcSpanHandler
• SPANCHUNK -> GrpcSpanChunkHandler

GrpcSpanHandler는 PSpan을 GrpcSpanFactory.buildSpanBo()로 SpanBo로 변환한다. 이때 ServerContext.getAgentInfo()의 agent header와 collector accepted time이 BindAttribute로 들어간다. sampling filter를 통과하면 모든 TraceService에 insertSpan(spanBo)를 호출한다.

GrpcSpanChunkHandler도 동일하게 PSpanChunk를 SpanChunkBo로 변환하고 TraceService.insertSpanChunk(spanChunkBo)를 호출한다.

6. Collector 저장 경로

3.0.1의 기본 trace service는 HbaseTraceService다.

6.1 Span 저장

HbaseTraceService.insertSpan(SpanBo) 순서:

1. publisher.captureContext(spanBo)로 downstream event context를 잡는다.
2. traceDao.asyncInsert(spanBo)로 원본 trace를 비동기 저장한다.
3. applicationTraceIndexDao.insert(spanBo)로 application/time 기반 trace 검색 인덱스를 저장한다.
4. insertAcceptorHost(spanBo)로 host-application map을 저장한다.
5. insertSpanStat(spanBo)로 server map caller/callee/response time 통계를 갱신한다.
6. insertSpanEventStat(spanBo)로 span event 기반 외부 호출 통계를 갱신한다.
7. trace 원본 저장 future 완료 시 span store event를 publish한다.

원본 trace 저장은 HbaseTraceDaoV2.asyncInsert():

• TransactionId를 traceRowKeyEncoderV2로 row key로 인코딩한다.
• Put(rowKey, acceptedTime, true)를 만든다.
• SpanSerializerV2.serialize(spanBo, put, null)로 SpanBo를 HBase Put column에 직렬화한다.
• HbaseColumnFamily.TRACE_V2_SPAN의 table, 즉 TraceV2 계열 테이블에 HbasePutWriter.put()으로 기록한다.

6.2 SpanChunk 저장

HbaseTraceService.insertSpanChunk(SpanChunkBo) 순서:

1. span chunk event context capture.
2. traceDao.insertSpanChunk(spanChunkBo)로 원본 chunk 저장.
3. chunk 내부 SpanEventBo 목록을 돌며 external call/statistics를 갱신한다.
4. span chunk store event publish.

HbaseTraceDaoV2.insertSpanChunk():

• TransactionId를 같은 trace row key로 인코딩한다.
• accepted time timestamp로 Put 생성.
• SpanChunkSerializerV2.serialize(spanChunkBo, put, null) 실행.
• Put이 비어 있지 않으면 TraceV2 table에 기록한다.

즉 root span과 chunk는 같은 transactionId row에 저장되고, column qualifier/serializer 구조로 span과 span event 조각들이 구분된다.

6.3 ApplicationTraceIndex 저장

HbaseApplicationTraceIndexDao.insert(SpanBo)는 trace list 조회용 인덱스를 만든다.

• row key: applicationIndexRowKeyEncoder.encodeRowKey(span)
  일반적으로 application id와 시간 slot/distribution을 포함하는 application trace index row key다.
• qualifier: SpanUtils.getVarTransactionId(span)
  trace를 찾아갈 수 있는 transaction id 계열 값이다.
• APPLICATION_TRACE_INDEX_TRACE column family value: elapsed, error code, agent id.
• APPLICATION_TRACE_INDEX_META column family value: span id, start time, rpc, remote address, endpoint, agent name.

Web/API가 특정 application의 시간 범위 trace 목록을 찾을 때 TraceV2 전체를 scan하지 않고 이 인덱스를 먼저 조회할 수 있다.

6.4 Server map/statistics 저장

HbaseTraceService는 span 원본 저장 외에도 application map에 필요한 통계를 갱신한다.

• root span이면 virtual user 또는 queue node 기준 caller/callee 통계를 만든다.
• parent application 정보가 있으면 parent -> current callee 통계를 만든다.
• response time 통계를 현재 application 기준으로 저장한다.
• span event가 외부 호출이고 ServiceType.isRecordStatistics()이면 caller 통계를 갱신한다.
• alias 타입이면 HostApplicationMapDao에 endpoint/destination/application 관계를 기록한다.

실제 DAO는 StatisticsService 구현에 주입된 map statistics DAO들이 담당한다. 대표적으로 caller/callee/response time 계열 HBase DAO가 사용된다.

7. 데이터 모델 관점에서 본 저장 결과

데이터	Agent 생성 위치	Collector 처리	저장 위치/목적
AgentInfo	AgentInfoSender	AgentService -> GrpcAgentInfoHandler	AgentInfo table. agent identity, JVM/server metadata
Ping/Lifecycle	AgentGrpcDataSender ping stream	AgentService -> DefaultPingEventHandler -> KeepAliveService	AgentLifeCycle, AgentEvent, agent state statistics
Span	DefaultTrace.close() -> BufferedStorage.store(Span)	SpanService -> GrpcSpanHandler -> HbaseTraceService	TraceV2 원본, ApplicationTraceIndex, host map, statistics
SpanChunk	BufferedStorage.flush/overflow	SpanService -> GrpcSpanChunkHandler -> HbaseTraceService	TraceV2 원본 chunk, span event statistics
SpanEvent	interceptor의 traceBlockBegin/End	Span 또는 SpanChunk 내부로 collector 도착	TraceV2 column 및 server map 통계

8. 핵심 클래스 경로

Agent bootstrap / instrumentation:

• agent-module/bootstraps/bootstrap/src/main/java/com/navercorp/pinpoint/bootstrap/PinpointBootStrap.java
• agent-module/bootstraps/bootstrap/src/main/java/com/navercorp/pinpoint/bootstrap/PinpointStarter.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/module/DefaultApplicationContext.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/ClassFileTransformerProvider.java

Agent trace 생성 / 전송:

• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/DefaultTraceContext.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/DefaultTrace.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/storage/BufferedStorage.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/context/provider/StorageFactoryProvider.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/grpc/SpanGrpcDataSender.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/grpc/AgentGrpcDataSender.java
• agent-module/profiler/src/main/java/com/navercorp/pinpoint/profiler/AgentInfoSender.java

Collector gRPC 수신:

• collector/src/main/java/com/navercorp/pinpoint/collector/grpc/config/GrpcAgentConfiguration.java
• collector/src/main/java/com/navercorp/pinpoint/collector/grpc/config/GrpcSpanReceiverConfiguration.java
• collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/service/AgentService.java
• collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/service/SpanService.java
• collector/src/main/java/com/navercorp/pinpoint/collector/receiver/SpanDispatchHandler.java
• collector/src/main/java/com/navercorp/pinpoint/collector/handler/grpc/GrpcAgentInfoHandler.java
• collector/src/main/java/com/navercorp/pinpoint/collector/handler/grpc/GrpcSpanHandler.java
• collector/src/main/java/com/navercorp/pinpoint/collector/handler/grpc/GrpcSpanChunkHandler.java

Collector lifecycle / 저장:

• grpc/src/main/java/com/navercorp/pinpoint/grpc/server/lifecycle/DefaultPingEventHandler.java
• collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/service/AgentLifecycleListener.java
• collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/service/KeepAliveService.java
• collector/src/main/java/com/navercorp/pinpoint/collector/service/async/AgentLifeCycleAsyncTaskService.java
• collector/src/main/java/com/navercorp/pinpoint/collector/service/HbaseTraceService.java
• collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseTraceDaoV2.java
• collector/src/main/java/com/navercorp/pinpoint/collector/dao/hbase/HbaseApplicationTraceIndexDao.java

9. 정리

Pinpoint 3.0.1에서 agent와 collector의 관계는 두 단계로 나뉜다.

첫째, agent lifecycle 연결이다. Agent는 9991 agent gRPC channel로 agent info를 보내고 ping stream을 유지한다. Collector는 이 stream과 metadata를 통해 agent가 실행 중인지, 어떤 application/service type인지, 언제 연결/종료되었는지를 기록한다.

둘째, trace data path다. Agent plugin이 bytecode를 변환해 interceptor를 삽입하고, interceptor는 request 진입/종료 및 downstream call을 Trace, Span, SpanEvent로 기록한다. Agent 내부 BufferedStorage가 span/span chunk를 gRPC protobuf로 변환해 9993 span stream에 보내면, Collector는 PSpan/PSpanChunk -> SpanBo/SpanChunkBo로 매핑한 뒤 TraceV2 원본 테이블, ApplicationTraceIndex, host map, server-map/statistics 계열 저장소에 나누어 기록한다.