Netty 메트릭 & 관찰성 — Micrometer로 EventLoop 모니터링하기

import io.micrometer.core.instrument.*;
import io.netty.channel.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Netty 파이프라인에 추가하는 메트릭 수집 핸들러.
 * 연결 수, 메시지 처리량, 에러율, 응답 시간을 측정한다.
 */
@ChannelHandler.Sharable
public class MetricsHandler extends ChannelDuplexHandler {

    private final AtomicInteger activeConnections = new AtomicInteger(0);
    private final Counter messageCounter;
    private final Counter errorCounter;
    private final Timer responseTimer;

    // channelRead 시작 시간을 Channel에 저장하기 위한 키
    private static final AttributeKey<Long> READ_START =
        AttributeKey.valueOf("metrics.readStart");

    public MetricsHandler(MeterRegistry registry) {
        // 활성 연결 수 — Gauge
        Gauge.builder("netty.connections.active", activeConnections, AtomicInteger::get)
            .description("현재 활성 연결 수")
            .register(registry);

        // 메시지 처리 카운터
        messageCounter = Counter.builder("netty.messages.received")
            .description("수신한 메시지 수")
            .register(registry);

        // 에러 카운터
        errorCounter = Counter.builder("netty.errors.total")
            .description("발생한 예외 수")
            .register(registry);

        // 응답 시간 타이머
        responseTimer = Timer.builder("netty.response.time")
            .description("메시지 처리 소요 시간")
            .publishPercentiles(0.5, 0.95, 0.99)  // p50, p95, p99
            .register(registry);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        activeConnections.incrementAndGet();
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        activeConnections.decrementAndGet();
        ctx.fireChannelInactive();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        messageCounter.increment();
        // 처리 시작 시간 기록
        ctx.channel().attr(READ_START).set(System.nanoTime());
        ctx.fireChannelRead(msg);
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg,
                      ChannelPromise promise) throws Exception {
        Long startTime = ctx.channel().attr(READ_START).getAndSet(null);
        if (startTime != null) {
            // 처리 시간 기록
            responseTimer.record(
                System.nanoTime() - startTime,
                java.util.concurrent.TimeUnit.NANOSECONDS
            );
        }
        ctx.write(msg, promise);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        errorCounter.increment();
        ctx.fireExceptionCaught(cause);
    }
}

// 서버 부트스트랩 설정
MetricsHandler metricsHandler = new MetricsHandler(meterRegistry);

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) {
            ch.pipeline()
                .addLast("metrics", metricsHandler)     // 가장 앞에 추가
                .addLast("decoder", new MyDecoder())
                .addLast("encoder", new MyEncoder())
                .addLast("handler", new BusinessHandler());
        }
    });

import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocatorMetric;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;

/**
 * PooledByteBufAllocator의 메모리 사용량을 Micrometer에 등록한다.
 */
public class ByteBufMetrics {

    public static void register(MeterRegistry registry) {
        PooledByteBufAllocatorMetric metric =
            PooledByteBufAllocator.DEFAULT.metric();

        // Direct Memory 사용량
        Gauge.builder("netty.bytebuf.memory.direct",
                metric, PooledByteBufAllocatorMetric::usedDirectMemory)
            .description("PooledByteBufAllocator가 사용 중인 Direct Memory (bytes)")
            .baseUnit("bytes")
            .register(registry);

        // Heap Memory 사용량
        Gauge.builder("netty.bytebuf.memory.heap",
                metric, PooledByteBufAllocatorMetric::usedHeapMemory)
            .description("PooledByteBufAllocator가 사용 중인 Heap Memory (bytes)")
            .baseUnit("bytes")
            .register(registry);

        // Arena 수 (스레드별 메모리 풀)
        Gauge.builder("netty.bytebuf.arenas.direct",
                metric, PooledByteBufAllocatorMetric::numDirectArenas)
            .description("Direct Arena 수")
            .register(registry);

        Gauge.builder("netty.bytebuf.arenas.heap",
                metric, PooledByteBufAllocatorMetric::numHeapArenas)
            .description("Heap Arena 수")
            .register(registry);
    }
}

import io.netty.channel.EventLoopGroup;
import io.netty.channel.SingleThreadEventLoop;
import io.netty.util.concurrent.EventExecutor;
import io.micrometer.core.instrument.*;

import java.util.concurrent.*;

/**
 * EventLoop별 pendingTasks를 주기적으로 수집하여 Micrometer에 보고한다.
 */
public class EventLoopMetrics {

    public static void register(MeterRegistry registry,
                                EventLoopGroup group,
                                String groupName) {
        int index = 0;
        for (EventExecutor executor : group) {
            if (executor instanceof SingleThreadEventLoop eventLoop) {
                String idx = String.valueOf(index++);

                // 각 EventLoop의 대기 중인 태스크 수
                Gauge.builder("netty.eventloop.pending.tasks",
                        eventLoop, SingleThreadEventLoop::pendingTasks)
                    .tag("group", groupName)
                    .tag("index", idx)
                    .description("EventLoop 태스크 큐에 대기 중인 작업 수")
                    .register(registry);
            }
        }
    }
}

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

EventLoopMetrics.register(meterRegistry, bossGroup, "boss");
EventLoopMetrics.register(meterRegistry, workerGroup, "worker");

# application.yml
management:
  metrics:
    enable:
      reactor.netty: true

# Reactor Netty 전용 메트릭 활성화
spring:
  reactor:
    netty:
      metrics: true

# 활성 연결 수
netty_connections_active

# Reactor Netty 사용 시
reactor_netty_http_server_connections_active

# EventLoop별 대기 태스크 수
netty_eventloop_pending_tasks{group="worker"}

# 전체 worker EventLoop의 평균
avg(netty_eventloop_pending_tasks{group="worker"})

# 최대값 — 가장 바쁜 EventLoop
max(netty_eventloop_pending_tasks{group="worker"})

# Direct Memory 사용량 (MB 단위)
netty_bytebuf_memory_direct / 1024 / 1024

# Heap Memory 사용량 (MB 단위)
netty_bytebuf_memory_heap / 1024 / 1024

# Direct Memory 사용률 (MaxDirectMemorySize가 512MB일 때)
netty_bytebuf_memory_direct / (512 * 1024 * 1024) * 100

# 초당 메시지 처리량
rate(netty_messages_received_total[1m])

# 초당 에러 수
rate(netty_errors_total[1m])

# 에러율 (%)
rate(netty_errors_total[1m])
  / rate(netty_messages_received_total[1m]) * 100

# Prometheus Alert Rule
- alert: NettyEventLoopQueueHigh
  expr: max(netty_eventloop_pending_tasks{group="worker"}) > 1000
  for: 2m
  labels:
    severity: warning
  annotations:
    summary: "EventLoop 태스크 큐가 1000을 초과했습니다"
    description: "worker EventLoop 큐가 2분 이상 1000 이상입니다. 핸들러 블로킹 여부를 확인하세요."

- alert: NettyDirectMemoryHigh
  expr: netty_bytebuf_memory_direct / (512 * 1024 * 1024) * 100 > 80
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "Direct Memory 사용률이 80%를 초과했습니다"
    description: "ByteBuf 릭이 발생했거나, 트래픽 급증으로 메모리가 부족합니다."

- alert: NettyErrorRateSpike
  expr: rate(netty_errors_total[5m]) > 10
  for: 1m
  labels:
    severity: warning
  annotations:
    summary: "Netty 에러율이 급증했습니다"
    description: "최근 5분간 초당 10회 이상의 예외가 발생하고 있습니다."