[Grafana Loki] 스케줄러 Peer 연결 미종료로 인한 메모리 누수 수정

PR 링크: grafana/loki#20268 상태: Merged | 변경: +9 / -6

들어가며

Grafana Loki의 쿼리 엔진에서 threadJob은 하나 이상의 streamSink를 가지며, 각 streamSink는 다른 워커와의 피어 연결입니다. streamSink가 종료될 때 피어의 Stop() 메서드는 호출했지만, 실제 네트워크 연결은 닫지 않았습니다. 이로 인해 반대편 워커의 Serve() 메서드가 영원히 반환되지 않아 고루틴과 메모리가 계속 누적되었습니다.

핵심 코드 분석

Before: Serve()에서 연결 종료 없음

func (p *Peer) Serve(ctx context.Context) error {
    p.lazyInit()
    // 연결 종료 로직 없음 - Peer에 Close 메서드도 없음

    g, ctx := errgroup.WithContext(ctx)
    g.Go(func() error { return p.recvMessages(ctx) })
    // ...
}

func (p *Peer) recvMessages(ctx context.Context) error {
    for {
        frame, err := p.Conn.Recv(ctx)
        if err != nil && ctx.Err() != nil {
            return nil
        } else if err != nil {
            return fmt.Errorf("recv: %w", err)  // 연결이 안 닫혀서 여기 도달 불가
        }
    }
}

After: defer로 연결 종료 보장

func (p *Peer) Serve(ctx context.Context) error {
    p.lazyInit()

    // Peer에 명시적 Close 메서드가 없으므로 Serve에서 연결 종료
    defer p.Conn.Close()

    g, ctx := errgroup.WithContext(ctx)
    g.Go(func() error { return p.recvMessages(ctx) })
    // ...
}

func (p *Peer) recvMessages(ctx context.Context) error {
    for {
        frame, err := p.Conn.Recv(ctx)
        if err != nil {
            if ctx.Err() != nil {
                return nil
            }
            return err  // 에러 래핑 제거로 간결화
        }
    }
}

왜 이게 좋은가

1. 근본 원인 해결: 연결이 닫히지 않으면 Recv()가 블로킹 상태로 유지되어 고루틴이 해제되지 않는다. defer p.Conn.Close()로 Serve 종료 시 연결을 반드시 닫는다.
2. 최소한의 변경: defer 한 줄과 에러 처리 정리만으로 메모리 누수를 해결했다.
3. 에러 처리 개선: ctx.Err() 체크를 에러 존재 여부 확인 내부로 이동하여, 에러 없이 컨텍스트가 취소된 경우의 불필요한 체크를 제거했다.

참고 자료

• Go defer 패턴과 리소스 정리
• Grafana Loki 쿼리 엔진 아키텍처