로드 밸런서 세션 모니터링 시스템

수행기간: 2023.06 ~ 2023.09 (3개월)

프로젝트 개요

개선 전

기존 로드 밸런서는 HAProxy 유닉스 소켓을 통해 세션 데이터 정보를 가져와 이를 사용자에게 제공합니다. 이 데이터는 소켓 명령어를 통해 가져오므로 휘발성이며, 요청 시점 기준으로 최근 5분간의 세션만 모니터링할 수 있습니다.

이를 개선하기 위해 로드 밸런서 세션 수집 에이전트를 개발 하고, 수집된 로드 밸런서 세션 데이터를 사용자에게 제공하는 시스템을 구축했습니다.

개선 후

개발 환경

Language: Java
Framework: Spring Boot
Load Balancer Agent: 자체 스택
Load Balancer: HAProxy
Infra: Metricbeat, Filebeat, Logstash, Kibana

아키텍처

1안

Agent를 직접 개발하여 Host 머신에 직접적으로 socket 명령어를 주기적으로 실행하고 이를 수집하며 또한 기존의 존재하는 RDB에 수집된 세션 데이터를 저장하는 구조입니다.

선택하지 않은 이유는 다음과 같습니다.

Agent 개발: Agent를 직접 개발해야 하며, 이는 추가적인 개발 리소스를 소모합니다.
RDB 사용: 시계열 데이터를 RDB에 저장하는 것은 비효율적이며, 데이터의 양이 많아질 경우 성능 저하가 발생할 수 있습니다.

2안

Metricbeat이 직접 HAProxy의 유닉스 소켓에서 세션 데이터를 수집하고, Logstash를 통해 이를 가공하여 Elasticsearch에 저장하는 구조입니다.

선택하지 않은 이유는 다음과 같습니다.

Metric 유실: Metricbeat가 HAProxy의 유닉스 소켓에서 세션 데이터를 수집 후 중앙화된 서버에 전송할 때 내부망 통신이 불안정할 경우 Metric이 유실될 수 있습니다.
로그 중앙화된 서버 관리: 관리 포인트가 늘어나는 것은 운영 측면에서 복잡성을 증가시킵니다.

최종안

Metricbeat로 수집된 로드 밸런서 세션 데이터를 호스트 머신의 파일로 저장하고 이를 Filebeat가 수집하여 Logstash로 전송하는 구조입니다. Logstash는 데이터를 가공하여 Elasticsearch에 저장하고, Kibana를 통해 시각화합니다.

또한 네트워크 단절로 인해 Metricbeat가 데이터를 수집하지 못하더라도, Filebeat가 로컬 파일을 모니터링하여 데이터가 쌓이면 이를 Logstash로 전송합니다. Filebeat의 registry 파일을 통해 전송을 보장 할 수 있습니다.

이에 따라 기존 로드 밸런서 RPM 에이전트 배포시 Metricbeat와 Filebeat 를 함께 배포되로록 service 구성하였습니다.

로드 밸런서 설정이 변경되면 그에 따른 메트릭 수집 설정 동기화 구현
세션 데이터를 중앙화된 Elasticsearch에 전송하도록 구현 (네트워크 단절 시에도 복구 가능)
로드 밸런서 RPM 에이전트 배포시 Metricbeat, Filebeat 설치되도록 패키징

API 서버 ES 쿼리 예시


@Repository
@RequiredArgsConstructor
public class ESLoadbalancerSearchRepository implements LoadbalancerSearchRepository {
 
    private final ElasticClient client;
    private final ObjectMapper mapper;
    private static final String INDEX_PATTERN = "haproxy-stat-*";
    private static final ZoneOffset KST_OFFSET = ZoneOffset.ofHours(9);
 
    @Override
    public List<DateSessions> searchPortSessionsListByIdAndPortAndProtocol(
            String lbId,
            int lbPort,
            String protocol,
            DatePeriodQueryString period
    ) throws Exception {
        SearchRequest req = buildPortSessionsSearch(lbId, lbPort, protocol, period);
        SearchResponse resp = client.search(req, RequestOptions.DEFAULT);
        Terms serviceTerms = resp.getAggregations().get("sessions");
        // Convert buckets to DateSessions list
        return serviceTerms.getBuckets().stream()
            .flatMap(serviceBucket -> {
                Terms hist = serviceBucket.getAggregations().get("units");
                return hist.getBuckets().stream().map(timeBucket -> {
                    Sum rate = timeBucket.getAggregations().get("session_rate");
                    Sum session = timeBucket.getAggregations().get("session");
                    return new DateSessions(
                        timeBucket.getKeyAsString(),
                        rate.getValue(),
                        session.getValue()
                    );
                });
            })
            .collect(Collectors.toList());
    }
 
    // ... Other methods follow the same pattern: building request, executing, mapping aggs ...
 
    private SearchRequest buildPrivateIpSearch(
            String lbId,
            int lbPort,
            DatePeriodQueryString period
    ) {
        BoolQueryBuilder bool = QueryBuilders.boolQuery()
                .filter(QueryBuilders.rangeQuery("@timestamp")
                        .gte(period.getStartDate())
                        .lte(period.getEndDateForQuery())
                        .timeZone(KST_OFFSET.toString())
                )
                .filter(QueryBuilders.termQuery("haproxy.stat.lb_id", lbId))
                .filter(QueryBuilders.termQuery("haproxy.stat.lb_port", lbPort))
                .mustNot(QueryBuilders.matchQuery("haproxy.stat.service_name", "FRONTEND"))
                .mustNot(QueryBuilders.matchQuery("haproxy.stat.service_name", "BACKEND"));
 
        SearchSourceBuilder src = new SearchSourceBuilder()
                .query(bool)
                .size(0)
                .aggregation(AggregationBuilders.terms("privateIpList")
                        .field("haproxy.stat.source.address.keyword")
                );
 
        SearchRequest req = new SearchRequest(INDEX_PATTERN)
                .source(src);
        return req;
    }
 
    private SearchRequest buildPortSessionsSearch(
            String lbId,
            int lbPort,
            String protocol,
            DatePeriodQueryString period
    ) {
        int intervalMinutes = period.getTimeUnit().getTimeSetMinutes();
        BoolQueryBuilder bool = QueryBuilders.boolQuery()
                .filter(QueryBuilders.rangeQuery("@timestamp")
                        .gte(period.getStartDate())
                        .lte(period.getEndDateForQuery())
                        .timeZone(KST_OFFSET.toString())
                )
                .filter(QueryBuilders.termQuery("haproxy.stat.lb_id", lbId))
                .filter(QueryBuilders.termQuery("haproxy.stat.lb_port", lbPort))
                .filter(QueryBuilders.matchQuery("haproxy.stat.service_name", "FRONTEND"));
        if (protocol != null) {
            bool.filter(QueryBuilders.termQuery("haproxy.stat.proxy.mode", protocol));
        }
 
        DateHistogramAggregationBuilder dateHist = AggregationBuilders.dateHistogram("units")
                .field("@timestamp")
                .fixedInterval(DateHistogramInterval.minutes(period.getTimeUnit().getTimeSet()))
                .minDocCount(0)
                .format("strict_date_hour_minute_second")
                .timeZone(KST_OFFSET.toString())
                .subAggregation(AggregationBuilders.sum("session_rate")
                        .field("haproxy.stat.session.rate.value")
                        .script("_value/" + intervalMinutes)
                )
                .subAggregation(AggregationBuilders.sum("session")
                        .field("haproxy.stat.session.current")
                        .script("_value/" + intervalMinutes)
                );
 
        SearchSourceBuilder src = new SearchSourceBuilder()
                .query(bool)
                .size(0)
                .aggregation(
                    AggregationBuilders.terms("sessions")
                        .field("haproxy.stat.service_name.keyword")
                        .subAggregation(dateHist)
                );
 
        return new SearchRequest(INDEX_PATTERN).source(src);
    }
}