메모하는 습관

Datadog에서 Armeria 모니터링을 위한 Custom

오늘은 Datadog APM에서 Armeria gRPC server 모니터링을 위해 Datadog 수동 계측을 적용했던 개발기를 포스팅 합니다.

개발 과정까지는 Datadog에서 HTTP/2 기반의 armeria-grpc 통신을 자동 계측 하지 못한다는 사실은 전혀 몰랐습니다. AWS Cloud 환경이 구축되어 가며, 플랫폼 엔지니어를 통해 Webflux netty와는 다르게 자동 계측이 안된다는 안타까운 소식을 접하게 되었습니다. 한 가지 희망은 자동 계측은 불가하지만 수동 계측으로 Datadog 모니터링을 할 수 있다며.. Link를 전달받았는데..

내용을 보면 OpenTelemetry library로 armeria-grpc 통신을 추적하고 수집하여 Dagadog Agent를 통해 Datadog Backend로 가져와 데쉬 보드를 구성하는 것으로 이해했는데, 자세한 내용은 아래 링크에서 참고하시기 바랍니다.

Datadog API Docs
https://docs.datadoghq.com/ko/tracing/trace_collection/custom_instrumentation/java/

https://docs.datadoghq.com/ko/tracing/trace_collection/opentracing/java/

Datadog에서 수동 계측을 위한 Sample 가이드를 참고하여, Opentracing library를 사용하여 수동 계측을 구현하였습니다. 사실 Opentracing은 @Deprecate 된 library 란 것을 Production 환경에 적용한 후에 알게 되었네요.. Datadog... API 가이드를 잘 봤어야 하는데, OpenTelemetry로 통합됐다고 하네요. 

그리고, Datadog v1.22 version에서 armeria-grpc 자동 계측을 지원하기위해 dd-trace-java library PR 중이고, 최근 Release 되어, 플랫폼 엔지니어분과 개발기에 적용 후 테스트를 하였지만, 그럼에도 자동 계측이 안되네요.. 언젠가 되길 바라며..

https://github.com/DataDog/dd-trace-java/releases/tag/v1.22.0 

https://github.com/DataDog/dd-trace-java/pull/5819

Datadog 수동 계측 구현은 다음과 같이 진행하였습니다.

Package 구성

build.gradle.kts

dependencies {
    implementation("com.datadoghq:dd-trace-api:1.21.0")
    implementation("io.opentracing:opentracing-api:0.33.0")
    implementation("io.opentracing:opentracing-util:0.33.0")
}

수동 계측 구현

GlobalTracerProvider.kt

OpenTracing interface의 GlobalTracer()를 통해 Noop span이 생성되고 스레드 추적을 시작할 수 있습니다. extracParentSpan()는 현재 실행되는 Active span 정보를 수동으로 생성할 때 end-point를 추가하여 분산 추적을 위해 Root span 정보를 받아오는 메서드입니다.
Kotlin object class로 정의하여 Singleton pattern을 적용하여 GlobalTracer() 인스턴스가 중복 생성되지 않도록 구현하였습니다.

@Component
object GlobalTracerProvider {

    fun getTracer(): Tracer {
        return GlobalTracer.get()
    }

    fun determineSpanType(req: HttpRequest): String {
        return if (req.headers()?.contentType().toString().endsWith("grpc")) {
            "grpc.server"
        } else {
            "http.server"
        }
    }

    fun extractParentSpan(tracer: Tracer, ctx: RequestContext): SpanContext? {
        val headers: Map<String, String>? = ctx.request()?.headers()?.associate { it.key.toString() to it.value }
        return tracer.extract(Format.Builtin.HTTP_HEADERS, TextMapAdapter(headers))
    }
}

https://kotlinlang.org/docs/object-declarations.html#object-declarations-overview

GlobalSpanBuilder.kt

@Component
object GlobalSpanBuilder {

    fun buildSpan(tracer: Tracer, spanType: String, parentSpan: SpanContext?, ctx: RequestContext): Span {
        return tracer
            .buildSpan(spanType)
            .asChildOf(parentSpan)  // Root span add
            .withTag(Tags.SPAN_KIND, Tags.SPAN_KIND_SERVER)
            .withTag(DDTags.SERVICE_NAME, "armeria-service")
            .withTag(DDTags.RESOURCE_NAME, ctx.method().name + ' ' + ctx.path())
            .withTag(DDTags.HTTP_QUERY, ctx.query())
            .withTag(Tags.HTTP_METHOD, ctx.method().name)
            .withTag(Tags.HTTP_URL, ctx.uri().toString())
            .start()  // 현재 스레드를 span으로 생성
    }
}

DatadogTracerService.kt
Armeria DecoratingHttpServiceFunction class를 상속받고 serve() 메서드를 override하여 해당 스레드의 Span을 생성하고 Root span 정보를 추가하여 분산 추적이 가능한 수동 계측 로직을 구현하였습니다.

@Component
class DatadogTracerService(
    private val globalSpanBuilder: GlobalSpanBuilder,
    private val globalTracerProvider: GlobalTracerProvider
): DecoratingHttpServiceFunction {
    constructor(): this(GlobalSpanBuilder, GlobalTracerProvider)

    override fun serve(delegate: HttpService, ctx: ServiceRequestContext, req: HttpRequest): HttpResponse {
        val tracer = globalTracerProvider.getTracer()
        val spanType = globalTracerProvider.determineSpanType(req)
        val parentSpan = globalTracerProvider.extractParentSpan(tracer, ctx)

        val span = globalSpanBuilder.buildSpan(tracer, spanType, parentSpan, ctx)

        try {
            tracer.activateSpan(span).use {
                return delegate.serve(ctx, req)
            }
        } catch (e: Exception) {
            span.setTag(Tags.ERROR, true)
            span.setTag(DDTags.ERROR_MSG, e.message)
            span.setTag(DDTags.ERROR_TYPE, e.javaClass.name)
            span.setTag(DDTags.ERROR_STACK, e.stackTraceToString())
            throw e
        } finally {
            span.finish()
        }
    }
}

다행히 Production 환경에서 수동 계측한 Span 정보와 Root span 정보가 맵핑되어 armeria-grpc 분산 추적을 할 수 있게 되었습니다. Request 성공은 현재 로직을 Armeria @Decorator pattern으로 서비스 레이어에 추가하여 분산 계측이 가능하였지만, gRPC Exception case는 Datadog dash board에서 추적 되지 않는 문제점을 발견하게 되었습니다...🥶🥶

@Service
@Decorator(DatadogTracerService::class)
class PurchaseGrpcService(private val purchaseComponent: PurchaseComponent): PurchaseServiceGrpcKt.PurchaseServiceCoroutineImplBase() {
}

gRPC Exception Handler Function 구현

DatadogTracerService에서 생성한 Tracer 객체를 ActiveSpan()으로 상태 체크 후 gRPC Exception 메시지를 add 하여 Datadog으로 수집되도록 로직을 구현하였습니다.
Armeria 1.26.0 이상 version에서는 GrpcExceptionHandlerFunction class를 상속받아 apply() 메서드를 override 하도록 업데이트 되었으니 참고하시기 바랍니다. (@Deprecated GrpcStatusFunction)

@Component
class GlobalGrpcExceptionHandler(
    private val globalTracerProvider: GlobalTracerProvider,
): GrpcExceptionHandlerFunction {
    constructor(): this(GlobalTracerProvider)

    override fun apply(ctx: RequestContext, throwable: Throwable, metadata: Metadata): Status {
        val activeSpan = globalTracerProvider.getTracer().activeSpan()

        if (activeSpan != null) {
            with(activeSpan) {
                setTag(Tags.ERROR, true)
                setTag(DDTags.ERROR_MSG, throwable.message)
                setTag(DDTags.ERROR_TYPE, throwable.javaClass.name)
                setTag(DDTags.ERROR_STACK, throwable.stackTraceToString())
            }
            activeSpan.finish()
        }
        return GrpcStatus.fromThrowable(throwable)
    }
}

gRPCService 적용
v1.26 

@Service
@Decorator(DatadogTracerService::class)
@GrpcExceptionHandler(GlobalGrpcExceptionHandler::class)
class PurchaseGrpcService(private val purchaseComponent: PurchaseComponent): PurchaseServiceGrpcKt.PurchaseServiceCoroutineImplBase() {
}

v.1.26 이하
ArmeriaConfiguration에서 addExceptionMapping() 으로 ExceptionHandler() 처리

private fun configureGrpcService(sb: ServerBuilder) {
    val grpcServiceBuilder = GrpcService.builder()
        .addExceptionMapping(Throwable::class.java, GlobalGrpcExceptionHandler())
        .apply {
            grpcServices.forEach { addService(it) }
        }
}

하루빨리.. Armeria에 대한 자동 계측이 되기를 바라는 마음으로.. 🙏🙏.. 개발기 마무리 합니다.

OpenTelemetry와 OpenTracing 비교
https://signoz.io/blog/opentelemetry-vs-opentracing/

https://uptrace.dev/blog/opentelemetry-vs-opentracing.html

OpenTelemetry API
https://opentelemetry.io/docs/instrumentation/java/