C++에서 Elasticsearch 8.x 쓰기: libcurl REST API와 Elasticlient

들어가며

Elasticsearch는 Java, Python, Go, JavaScript, .NET 등에는 공식 클라이언트를 제공하지만 C++용 공식 클라이언트는 없습니다. 대신 모든 기능이 HTTP + JSON REST API로 노출되어 있으므로, C++에서는 HTTP 라이브러리와 JSON 라이브러리만 있으면 모든 API를 쓸 수 있습니다. 이 글은 서드파티 래퍼인 Elasticlient와, libcurl + nlohmann/json으로 REST API를 직접 호출하는 방식을 비교하고, 인덱싱·전문 검색·집계·벌크 인덱싱·스크롤·ILM(Index Lifecycle Management)을 예제로 다룹니다. 예제마다 실제로 자주 틀리는 부분(부분 실패하는 bulk 응답, nlohmann 초기화 규칙, _id 정렬 금지 등)을 함께 짚습니다.

예제는 C++17, Elasticsearch 8.x(대부분 7.x에서도 동작), libcurl + nlohmann/json을 기준으로 합니다.

관계형 DB의 LIKE '%error%'는 앞에 와일드카드가 붙어 B-tree 인덱스를 쓰지 못하고 전체를 스캔합니다. Elasticsearch는 텍스트를 토큰으로 쪼개 역인덱스(inverted index)에 저장하므로 단어 단위 검색과 점수 기반 정렬을 인덱스 조회로 처리하고, 같은 데이터에 대해 terms·date_histogram 같은 집계도 제공합니다. 로그처럼 시간이 지나면 가치가 떨어지는 데이터는 ILM으로 오래된 인덱스를 자동으로 줄이거나 지울 수 있습니다. 아래 표는 이 글에서 다루는 기능과 해당 API를 정리한 것입니다.

필요	Elasticsearch 기능	REST API
단어 단위 검색	역인덱스 전문 검색	match·match_phrase·bool 쿼리
통계·대시보드	집계	aggs의 terms·date_histogram
대량 적재	벌크 인덱싱	_bulk
많은 결과를 끝까지 읽기	스크롤, search_after	_search?scroll=, search_after
오래된 인덱스 정리	ILM	PUT _ilm/policy/...

---

Elasticsearch 서버와 클라이언트 라이브러리 설치

필수 의존성

항목	버전	비고
C++	C++17	예제 기준
Elasticsearch	7.x/8.x	Docker 권장
Elasticlient	-	선택, cpr·jsoncpp 의존
libcurl	7.x+	REST API용
nlohmann/json	3.x	JSON 파싱

Elasticsearch 서버 실행 (Docker)

# 단일 노드 (개발용)
docker run -d --name elasticsearch -p 9200:9200 -p 9300:9300 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
  docker.elastic.co/elasticsearch/elasticsearch:8.11.0
# 연결 확인
curl -X GET "localhost:9200/?pretty"

Elasticlient 설치 (선택)

# cpr가 사용하는 libcurl (Ubuntu/Debian)
sudo apt-get install libcurl4-openssl-dev
# Elasticlient 소스 빌드 (cpr·jsoncpp 등은 서브모듈로 포함됨)
git clone --recursive https://github.com/seznam/elasticlient.git
cd elasticlient
mkdir build && cd build
cmake .. -DCMAKE_BUILD_TYPE=Release
cmake --build .
sudo cmake --build . --target install

libcurl + nlohmann/json 설치 (REST API용, 권장)

# Ubuntu/Debian
sudo apt-get install libcurl4-openssl-dev
# macOS (Homebrew)
brew install curl nlohmann-json
# vcpkg
vcpkg install curl nlohmann-json

CMakeLists.txt 기본 설정

REST API를 직접 호출할 때의 설정입니다.

cmake_minimum_required(VERSION 3.16)
project(elasticsearch_demo LANGUAGES CXX)
set(CMAKE_CXX_STANDARD 17)
find_package(CURL REQUIRED)
find_package(nlohmann_json REQUIRED)
add_executable(es_demo main.cpp)
target_link_libraries(es_demo PRIVATE CURL::libcurl nlohmann_json::nlohmann_json)

Elasticlient를 쓸 때의 설정입니다. 설치 방식에 따라 CMake 패키지·타깃 이름이 다를 수 있으므로, 설치된 config 파일이 없다면 소스 트리를 add_subdirectory로 포함하는 편이 간단합니다.

cmake_minimum_required(VERSION 3.16)
project(elasticsearch_demo LANGUAGES CXX)
set(CMAKE_CXX_STANDARD 17)
find_package(elasticlient REQUIRED)
find_package(cpr REQUIRED)
add_executable(es_demo main.cpp)
target_link_libraries(es_demo PRIVATE elasticlient::elasticlient cpr::cpr)

---

Elasticlient로 인덱싱·검색·벌크·스크롤

최소 동작: 인덱싱·조회·삭제

// elasticlient_basic.cpp
#include <elasticlient/client.h>
#include <iostream>
#include <string>
int main() {
    try {
        elasticlient::Client client({"http://localhost:9200/"});
        // 1. 문서 인덱싱
        std::string doc = R"({"message": "Hello Elasticsearch!", "timestamp": "2024-01-15T10:00:00"})";
        cpr::Response indexResp = client.index("logs", "_doc", "1", doc);
        if (indexResp.status_code != 200 && indexResp.status_code != 201) {
            std::cerr << "인덱싱 실패: " << indexResp.status_code << " " << indexResp.text << std::endl;
            return 1;
        }
        std::cout << "인덱싱 완료: " << indexResp.text << std::endl;
        // 2. 문서 조회
        cpr::Response getResp = client.get("logs", "_doc", "1");
        if (getResp.status_code == 200) {
            std::cout << "조회 결과: " << getResp.text << std::endl;
        }
        // 3. 문서 삭제
        cpr::Response delResp = client.remove("logs", "_doc", "1");
        std::cout << "삭제 완료: " << delResp.status_code << std::endl;
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

Client 생성자는 클러스터 노드 URL 목록을 받으며, 노드를 여러 개 주면 요청이 실패할 때 다음 노드로 넘어갑니다. index(index, type, id, doc)의 문서 타입은 7.x에서 폐기 예고되고 8.x에서 제거되었으므로 항상 _doc을 넘깁니다. 그러면 URL이 /logs/_doc/1이 되어 8.x에서도 동작합니다. get과 remove도 같은 방식으로 문서를 조회·삭제합니다.

Elasticlient의 인터페이스가 type 인자를 요구하는 것 자체가 이 라이브러리가 6.x 시절에 설계되었다는 흔적입니다. 단건 인덱싱, 검색, bulk, scroll 정도만 쓴다면 충분하지만, 8.x에서 추가된 PIT(point in time), Data Streams, 인덱스 템플릿 v2 같은 API는 결국 raw 요청을 직접 만들어 보내야 합니다. 그럴 바에는 처음부터 얇은 REST 헬퍼를 두는 편이 의존성도 적고 동작도 투명하다는 것이 이 글이 뒤쪽에서 libcurl 방식을 권하는 이유입니다.

전문 검색 (Search)

// elasticlient_search.cpp
#include <elasticlient/client.h>
#include <iostream>
#include <string>
int main() {
    elasticlient::Client client({"http://localhost:9200/"});
    // match 쿼리: message 필드에서 "error" 검색
    std::string query = R"({
        "query": {
            "match": {
                "message": "error"
            }
        },
        "size": 10,
        "_source": ["message", "level", "timestamp"]
    })";
    // client.search(index, docType, body)는 /index/docType/_search로 요청하는데,
    // 8.x는 타입이 들어간 검색 경로를 받지 않으므로 경로를 직접 지정합니다.
    cpr::Response resp = client.performRequest(
        elasticlient::Client::HTTPMethod::POST, "logs/_search", query);
    if (resp.status_code == 200) {
        std::cout << "검색 결과:\n" << resp.text << std::endl;
    } else {
        std::cerr << "검색 실패: " << resp.status_code << " " << resp.text << std::endl;
    }
    return 0;
}

벌크 인덱싱 (Bulk)

// elasticlient_bulk.cpp
#include <elasticlient/client.h>
#include <elasticlient/bulk.h>
#include <iostream>
#include <memory>
#include <string>
int main() {
    // Bulk·Scroll은 Client를 shared_ptr로 받습니다.
    auto client = std::make_shared<elasticlient::Client>(
        std::vector<std::string>{"http://localhost:9200/"});
    elasticlient::SameIndexBulkData data("logs", 100);  // 한 번에 보낼 최대 문서 수
    for (int i = 1; i <= 100; ++i) {
        std::string doc = R"({"msg":"log)" + std::to_string(i) +
            R"(","level":")" + (i % 5 == 0 ? "error" : "info") + R"("})";
        data.indexDocument("_doc", std::to_string(i), doc);
    }
    elasticlient::Bulk indexer(client);
    std::size_t errors = indexer.perform(data);  // 실패한 문서 수를 반환
    std::cout << data.size() << "건 중 실패 " << errors << "건\n";
    return 0;
}

perform()은 HTTP 응답이 아니라 실패한 문서 수를 돌려줍니다. 뒤에서 설명하듯 _bulk는 일부 문서가 실패해도 HTTP 200을 반환하므로, 이 값이 0인지 확인해야 합니다.

스크롤 API (Elasticlient Scroll)

// elasticlient_scroll.cpp
#include <elasticlient/client.h>
#include <elasticlient/scroll.h>
#include <json/json.h>  // Elasticlient는 jsoncpp로 결과를 돌려줍니다
#include <iostream>
#include <memory>
#include <string>
int main() {
    auto client = std::make_shared<elasticlient::Client>(
        std::vector<std::string>{"http://localhost:9200/"});
    // 한 번에 100건, 컨텍스트 유지 시간 1분
    elasticlient::Scroll scroll(client, 100, "1m");
    std::string query = R"({"sort": ["_doc"], "query": {"match_all": {}}})";
    scroll.init("logs", "_doc", query);
    Json::Value res;
    std::size_t total = 0;
    while (scroll.next(res)) {
        if (res["hits"].empty()) break;  // 더 이상 결과 없음
        total += res["hits"].size();
    }
    std::cout << "총 처리: " << total << "건\n";
    return 0;
}

Elasticlient의 스크롤도 내부적으로 문서 타입을 경로에 넣기 때문에, 8.x 클러스터에서는 버전에 따라 경로 오류가 날 수 있습니다. 이 경우 아래 REST 방식의 스크롤을 쓰면 됩니다.

---

libcurl REST API 헬퍼

Elasticlient 없이 libcurl과 nlohmann/json으로 REST API를 직접 호출합니다. 의존성이 적고 모든 Elasticsearch API를 사용할 수 있습니다.

HTTP 헬퍼 클래스

// es_client.hpp
#pragma once
#include <curl/curl.h>
#include <string>
class EsClient {
public:
    EsClient(const std::string& base_url = "http://localhost:9200");
    ~EsClient();
    std::string get(const std::string& path);
    std::string put(const std::string& path, const std::string& body = "{}");
    std::string post(const std::string& path, const std::string& body = "{}");
    std::string del(const std::string& path);
private:
    static size_t write_cb(void* c, size_t s, size_t n, void* u) {
        size_t t = s * n;
        static_cast<std::string*>(u)->append(static_cast<char*>(c), t);
        return t;
    }
    std::string request(const std::string& method, const std::string& path, const std::string& body);
    std::string base_url_;
    CURL* curl_;
};

request()는 curl_easy_setopt로 URL, WRITEFUNCTION, CONNECTTIMEOUT(5초), TIMEOUT(30초), Content-Type: application/json 헤더를 설정하고, HTTP 상태가 4xx/5xx이면 std::runtime_error를 던지도록 구현합니다. 이 헬퍼에는 설계상 짚어 둘 점이 몇 가지 있습니다.

핸들을 멤버로 두고 재사용하면 libcurl이 keep-alive 연결을 유지해 요청마다 TCP·TLS 핸드셰이크를 반복하지 않습니다. 대신 하나의 CURL* 핸들을 여러 스레드가 동시에 쓰면 안 되므로, 한 EsClient를 스레드 간에 공유할 수 없습니다. 스레드마다 EsClient를 두거나 여러 핸들을 담은 풀을 만들어야 합니다. 프로그램 시작 시 curl_global_init(CURL_GLOBAL_DEFAULT)를 한 번 호출하는 것도 잊지 마세요.

4xx/5xx에서 예외를 던질 때는 응답 본문을 메시지에 포함해야 합니다. Elasticsearch는 {"error":{"type":"mapper_parsing_exception","reason":"..."}}처럼 원인을 본문에 담아 주므로, 상태 코드만 남기면 디버깅할 정보가 사라집니다.

CUSTOMREQUEST, POSTFIELDS 같은 옵션은 핸들에 남아 있습니다. 이전 POST 요청의 본문이 다음 GET에 딸려 가지 않도록 요청마다 명시적으로 다시 설정하거나 curl_easy_reset을 호출해야 합니다.

---

인덱싱·전문 검색·집계

인덱스 매핑 생성 및 문서 인덱싱

// rest_indexing.cpp
#include "es_client.hpp"
#include <iostream>
#include <nlohmann/json.hpp>
int main() {
    try {
        EsClient client("http://localhost:9200");
        // 1. 인덱스 매핑 생성
        std::string mapping = R"({
            "mappings": {
                "properties": {
                    "message": { "type": "text" },
                    "level": { "type": "keyword" },
                    "timestamp": { "type": "date" },
                    "userId": { "type": "keyword" }
                }
            }
        })";
        client.put("logs", mapping);
        // 2. 문서 인덱싱 (ID 지정)
        nlohmann::json doc = {
            {"message", "Application started"},
            {"level", "info"},
            {"timestamp", "2024-01-15T10:00:00Z"},
            {"userId", "user123"}
        };
        std::string indexResp = client.put("logs/_doc/1", doc.dump());
        std::cout << "인덱싱: " << indexResp << std::endl;
        // 3. 자동 ID 생성 (POST)
        std::string autoResp = client.post("logs/_doc", doc.dump());
        std::cout << "자동 ID: " << autoResp << std::endl;
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

한 가지 주의할 점은 client.put("logs", mapping)이 처음 한 번만 성공한다는 것입니다. 인덱스가 이미 있으면 resource_already_exists_exception(400)이 나므로, 애플리케이션 시작 시마다 호출한다면 HEAD logs로 존재 여부를 먼저 확인하거나 이 오류를 무시하도록 처리합니다. 또 매핑을 정하지 않고 바로 문서를 넣으면 Elasticsearch가 동적 매핑으로 타입을 추측하는데, 문자열은 text와 keyword 하위 필드를 함께 만들고, 처음 들어온 값이 "123"이면 이후 필드 타입이 예상과 달라질 수 있습니다. 로그처럼 필드가 정해진 데이터는 이 예제처럼 매핑을 먼저 만드는 것이 안전합니다. userId를 숫자 타입이 아니라 keyword로 둔 것도 의도적입니다. ID는 범위 검색이나 합계를 낼 일이 없고 정확 일치로만 찾는데, 숫자 타입은 범위 검색에 맞춘 구조로 저장되므로 term 조회에는 keyword가 더 잘 맞습니다.

인덱싱 응답이 성공으로 왔어도 그 문서가 곧바로 검색되지는 않습니다. Elasticsearch는 refresh가 일어날 때(기본 1초 간격) 새 세그먼트를 열어 검색에 노출하므로, 문서를 넣은 직후 _search를 호출하면 결과가 비어 있는 것이 정상입니다. “넣고 바로 찾는” 테스트가 간헐적으로 실패하는 흔한 원인이라, 테스트에서는 인덱싱 경로에 ?refresh=wait_for를 붙이거나 POST logs/_refresh를 호출합니다. 운영 코드에서 매 요청마다 강제 refresh를 하면 작은 세그먼트가 계속 생겨 병합 비용이 커지므로 피해야 합니다.

전문 검색 (Full-Text Search)

// rest_fulltext_search.cpp
#include "es_client.hpp"
#include <iostream>
#include <nlohmann/json.hpp>
int main() {
    try {
        EsClient client("http://localhost:9200");
        // match: 단어 분리 후 검색 (OR 기본)
        // match_phrase: 구문 검색 (순서 유지)
        // bool: must/should/must_not/filter로 복합 조건
        nlohmann::json match_query = {
            {"query", {
                {"match", {{"message", "error timeout"}}}
            }},
            {"size", 10},
            {"_source", {"message", "level", "timestamp"}}
        };
        std::string resp = client.post("logs/_search", match_query.dump());
        auto j = nlohmann::json::parse(resp);
        auto hits = j["hits"]["hits"];
        std::cout << "총 " << j["hits"]["total"]["value"] << "건\n";
        for (const auto& hit : hits) {
            std::cout << "  - " << hit["_source"]["message"] << " ["
                      << hit["_source"]["level"] << "]\n";
        }
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

match 쿼리는 검색어를 분석기(analyzer)로 토큰화한 뒤 토큰 중 하나라도 맞는 문서를 찾습니다(operator의 기본값이 or). 그래서 "error timeout"은 둘 중 하나만 있는 문서도 반환하고, 둘 다 있는 문서가 점수(_score)가 높아 위에 옵니다. 모두 포함해야 한다면 {"match": {"message": {"query": "error timeout", "operator": "and"}}}로 씁니다. level처럼 keyword로 매핑한 필드에 정확히 일치하는 조건을 걸 때는 점수 계산이 필요 없으므로 bool.filter 안에 term 쿼리를 두면 결과가 캐시되어 더 빠릅니다.

반대 방향의 실수, 즉 text 필드에 term 쿼리를 쓰는 경우가 더 찾기 어렵습니다. message는 색인 시 분석기를 거쳐 소문자 토큰으로 쪼개져 저장되므로 {"term": {"message": "Connection timeout"}}은 대문자와 공백이 섞인 원문 전체를 하나의 토큰으로 찾다가 아무것도 찾지 못합니다. 에러 없이 빈 결과만 돌아오기 때문에 C++ 쪽 파싱 코드를 의심하게 되기 쉽습니다. text 필드에는 match 계열, keyword 필드에는 term 계열이라는 짝을 지키고, 검색어가 어떻게 토큰화되는지 궁금하면 POST logs/_analyze에 {"field": "message", "text": "Connection timeout"}을 보내 직접 확인합니다.

집계 (Aggregation)

// rest_aggregation.cpp
#include "es_client.hpp"
#include <iostream>
#include <nlohmann/json.hpp>
int main() {
    try {
        EsClient client("http://localhost:9200");
        // terms: level별 문서 수. date_histogram: 1시간 단위. 중첩 aggs로 복합 집계 가능
        nlohmann::json terms_agg = {
            {"size", 0},
            {"aggs", {
                {"levels", {
                    {"terms", {{"field", "level"}}}
                }}
            }}
        };
        std::string resp = client.post("logs/_search", terms_agg.dump());
        auto j = nlohmann::json::parse(resp);
        std::cout << "level별 건수:\n";
        for (const auto& bucket : j["aggregations"]["levels"]["buckets"]) {
            std::cout << "  " << bucket["key"] << ": " << bucket["doc_count"] << "\n";
        }
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

집계 대상 필드는 keyword(또는 숫자·날짜)여야 합니다. text 필드에 terms 집계를 걸면 Fielddata is disabled on text fields by default(illegal_argument_exception)로 거부되는데, 전문 검색과 집계를 모두 해야 하는 필드라면 매핑에 "fields": {"keyword": {"type": "keyword"}} 서브필드를 두고 집계에서는 message.keyword를 지정합니다.

집계 결과를 대시보드에 그대로 쓰기 전에 알아 둘 점도 있습니다. 샤드가 여러 개일 때 terms 집계는 각 샤드가 자기 상위 항목만 조정 노드로 보내 합치는 방식이라 근사값입니다. 응답의 doc_count_error_upper_bound가 0이 아니면 오차가 있다는 뜻이고, shard_size를 키우면 정확도가 올라가는 대신 비용이 늘어납니다. percentiles 집계도 TDigest 기반 근사치라서 데이터가 적거나 극단값(P99.9 등) 근처에서는 실제 값과 차이가 날 수 있습니다. 모니터링에는 충분하지만 과금이나 SLA 판정처럼 정확한 값이 필요하면 원본 데이터로 따로 계산하는 편이 안전합니다.

---

재시도를 포함한 벌크 인덱싱

기본 벌크 API

// rest_bulk.cpp
#include "es_client.hpp"
#include <iostream>
#include <sstream>
#include <nlohmann/json.hpp>
int main() {
    try {
        EsClient client("http://localhost:9200");
        // 벌크 형식: 액션\n문서\n액션\n문서...
        std::ostringstream bulk;
        for (int i = 1; i <= 1000; ++i) {
            nlohmann::json action = {
                {"index", {
                    {"_index", "logs"},
                    {"_id", std::to_string(i)}
                }}
            };
            nlohmann::json doc = {
                {"message", "Log entry " + std::to_string(i)},
                {"level", (i % 5 == 0 ? "error" : "info")},
                {"timestamp", "2024-01-15T10:00:00Z"}
            };
            bulk << action.dump() << "\n" << doc.dump() << "\n";
        }
        std::string resp = client.post("_bulk", bulk.str());
        auto j = nlohmann::json::parse(resp);
        if (j.contains("errors") && j["errors"]) {
            std::cerr << "벌크 중 일부 실패\n";
            for (const auto& item : j["items"]) {
                if (item.contains("index") && item["index"].contains("error")) {
                    std::cerr << "  " << item["index"]["error"]["reason"] << "\n";
                }
            }
        } else {
            std::cout << "벌크 인덱싱 완료: 1000건\n";
        }
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

벌크 요청에서 가장 흔한 오해는 HTTP 200이면 성공이라고 생각하는 것입니다. _bulk는 요청 자체를 처리하기만 하면 개별 문서가 실패해도 200을 반환하고, 응답 본문의 "errors": true와 items[i].index.status로 실패를 알립니다. 위 코드처럼 errors를 반드시 확인해야 하며, 실패 항목의 status에 따라 대응을 나눠야 합니다. 429(es_rejected_execution_exception, 쓰기 스레드 풀 큐 포화)는 잠시 뒤 그 항목만 다시 보내면 되지만, 400(mapper_parsing_exception 등)은 문서 자체가 잘못된 것이라 몇 번을 재시도해도 실패합니다.

벌크 본문은 NDJSON 형식이라 마지막 줄도 줄바꿈으로 끝나야 합니다. 마지막 \n이 빠지면 The bulk request must be terminated by a newline [\n] 오류가 납니다. 공식 문서는 Content-Type: application/x-ndjson을 권하지만 application/json도 받아 줍니다.

대용량 배치 벌크 (재시도 포함)

// batch_size 2000, max_retries 3으로 배치별 _bulk 호출
// 실패 시 100*(retry+1)ms 선형 백오프 후 재시도
void bulk_index_batch(EsClient& client, const std::vector<nlohmann::json>& docs,
                      size_t batch_size = 2000, int max_retries = 3) {
    for (size_t offset = 0; offset < docs.size(); offset += batch_size) {
        std::ostringstream bulk;
        for (size_t i = offset; i < std::min(offset + batch_size, docs.size()); ++i) {
            bulk << R"({"index":{"_index":"logs","_id":")" << (i+1) << R"("}})" << "\n"
                 << docs[i].dump() << "\n";
        }
        for (int r = 0; r < max_retries; ++r) {
            try { client.post("_bulk", bulk.str()); break; }
            catch (const std::exception& e) {
                if (r == max_retries - 1) throw;
                std::this_thread::sleep_for(std::chrono::milliseconds(100 * (r + 1)));
            }
        }
    }
}

이 함수는 client.post가 예외를 던질 때, 즉 연결 실패나 HTTP 4xx/5xx일 때만 재시도합니다. 앞에서 본 것처럼 문서 일부가 429로 거절되어도 HTTP 상태는 200이므로 이 루프는 부분 실패를 재시도하지 못합니다. 실무용으로 쓰려면 응답의 items를 순회하며 status == 429인 문서만 모아 다음 시도의 본문으로 만들고, 재시도할 때마다 대기 시간을 두 배로 늘리는 지수 백오프를 적용해야 합니다. 또 문서 ID를 배열 인덱스(i+1)로 만들면 같은 데이터를 다시 넣을 때 기존 문서를 덮어써서 재시도가 멱등해지는 장점이 있지만, 다른 배치 작업과 ID가 겹치면 서로 덮어쓰므로 원본 데이터의 고유 키로 ID를 만드는 편이 안전합니다.

대량 초기 적재라면 적재하는 동안 인덱스의 refresh_interval을 -1로, number_of_replicas를 0으로 낮췄다가 끝난 뒤 되돌리면 색인 속도가 크게 좋아집니다. 배치 크기는 문서 수보다 요청 바이트 크기(수 MB~수십 MB)를 기준으로 잡는 것이 좋습니다. 문서 크기가 제각각이면 문서 2000개가 어떤 때는 1MB, 어떤 때는 100MB가 되기 때문입니다.

---

스크롤과 Search After

스크롤로 대용량 검색

// rest_scroll.cpp
#include "es_client.hpp"
#include <iostream>
#include <nlohmann/json.hpp>
int main() {
    try {
        EsClient client("http://localhost:9200");
        // 1. 스크롤 시작 (scroll=1m: 1분간 컨텍스트 유지)
        nlohmann::json search = {
            {"query", {{"match_all", nlohmann::json::object()}}},  // {}는 null이 됨
            {"size", 100},
            {"sort", nlohmann::json::array({"_doc"})}  // 스크롤은 _doc 순서가 가장 효율적
        };
        std::string resp = client.post("logs/_search?scroll=1m", search.dump());
        auto j = nlohmann::json::parse(resp);
        std::string scroll_id = j["_scroll_id"];
        auto hits = j["hits"]["hits"];
        size_t total = 0;
        while (!hits.empty()) {
            total += hits.size();
            for (const auto& hit : hits) {
                // 문서 처리 (예: 파일 저장, 변환)
                (void)hit;
            }
            // 2. 다음 스크롤
            nlohmann::json scroll_req = {{"scroll", "1m"}, {"scroll_id", scroll_id}};
            resp = client.post("_search/scroll", scroll_req.dump());
            j = nlohmann::json::parse(resp);
            scroll_id = j["_scroll_id"];
            hits = j["hits"]["hits"];
        }
        std::cout << "총 처리: " << total << "건\n";
        // 3. 스크롤 컨텍스트 해제 (서버 자원 반환)
        client.del("_search/scroll/" + scroll_id);
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

정렬 키로 _id를 쓰면 안 됩니다. Elasticsearch 8.x는 기본 설정(indices.id_field_data.enabled: false)에서 _id 필드의 fielddata 접근을 막기 때문에 Fielddata access on the _id field is disallowed 오류가 납니다. 스크롤은 순서가 중요하지 않은 일괄 처리용이므로 정렬 비용이 없는 _doc을 쓰는 것이 권장 방식입니다.

스크롤 컨텍스트는 서버의 힙과 파일 핸들을 잡고 있다가 scroll 시간(여기서는 1분)이 지나야 풀립니다. 처리가 끝나면 DELETE _search/scroll로 명시적으로 해제하고, 배치 하나를 처리하는 시간이 scroll 값보다 길어지지 않게 해야 합니다. 그렇지 않으면 다음 요청에서 No search context found for id 오류가 납니다. 동시에 열 수 있는 스크롤 컨텍스트 수에도 상한(search.max_open_scroll_context)이 있어, 해제를 빠뜨린 스크롤 작업을 여러 개 동시에 돌리면 Trying to create too many scroll contexts 오류를 만나게 됩니다.

Elasticsearch 7.10부터 공식 문서는 깊은 페이지 조회에 스크롤 대신 PIT(point in time)와 search_after 조합을 권합니다. PIT는 스크롤처럼 특정 시점의 스냅숏을 고정하면서도, 요청이 stateless해서 여러 작업자가 나눠 읽기 쉽습니다.

Search After (실시간 페이지네이션)

// rest_search_after.cpp
#include "es_client.hpp"
#include <iostream>
#include <nlohmann/json.hpp>
int main() {
    try {
        EsClient client("http://localhost:9200");
        nlohmann::json query = {
            {"query", {{"match_all", nlohmann::json::object()}}},
            {"size", 100},
            // 정렬 키 순서가 중요하므로 배열로 명시 (중괄호만 쓰면 키 이름순 객체가 됨)
            {"sort", nlohmann::json::array({
                {{"timestamp", "asc"}},
                {{"event_id", "asc"}}   // 동점 해소용 고유 keyword 필드 (_id 정렬은 8.x 기본 금지)
            })}
        };
        // 첫 페이지
        std::string resp = client.post("logs/_search", query.dump());
        auto j = nlohmann::json::parse(resp);
        auto hits = j["hits"]["hits"];
        while (!hits.empty()) {
            for (const auto& hit : hits) {
                std::cout << hit["_source"]["message"] << "\n";
            }
            auto last = hits.back();
            if (!last.contains("sort")) break;
            query["search_after"] = last["sort"];
            resp = client.post("logs/_search", query.dump());
            j = nlohmann::json::parse(resp);
            hits = j["hits"]["hits"];
        }
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

from/size로 페이지를 넘기는 방식은 from + size가 인덱스 설정 index.max_result_window(기본 10000)를 넘으면 Result window is too large 오류로 거부됩니다. 깊은 페이지일수록 각 샤드가 from + size개를 정렬해 올려야 해서 비용이 급격히 커지기 때문에 막아 둔 것이고, search_after는 몇 번째 페이지든 비용이 일정합니다.

search_after는 이전 페이지 마지막 문서의 정렬 값을 넘겨 그다음부터 가져오는 방식입니다. 그래서 정렬 키가 문서를 유일하게 구분해야 합니다. timestamp만으로 정렬하면 같은 밀리초에 찍힌 로그들이 페이지 경계에 걸릴 때 일부가 건너뛰어지거나 중복됩니다. 이 때문에 두 번째 정렬 키로 고유 필드가 필요하며, 적당한 필드가 없다면 PIT를 열고 PIT가 자동으로 제공하는 _shard_doc 동점 해소 키를 쓰는 것이 공식 권장 방식입니다.

정렬을 {"sort", {{"timestamp", "asc"}, {"event_id", "asc"}}}처럼 중괄호만으로 쓰면, nlohmann/json은 이 모양을 배열이 아니라 객체 {"event_id":"asc","timestamp":"asc"}로 해석합니다. 그리고 nlohmann의 기본 객체는 키를 이름순으로 정렬해 저장하므로 event_id가 앞에 옵니다. 정렬 우선순위가 조용히 뒤바뀌는 버그라, 순서가 의미 있는 값은 항상 nlohmann::json::array()로 명시하는 습관이 필요합니다. 제가 이 라이브러리로 쿼리를 만들 때 가장 자주 확인하는 것이 전송 직전의 query.dump(2) 출력인데, 대부분의 “왜 쿼리가 이상하게 동작하지?”는 이 출력 한 번으로 풀립니다.

---

ILM (Index Lifecycle Management)

ILM 정책 생성

// rest_ilm.cpp
#include "es_client.hpp"
#include <iostream>
#include <nlohmann/json.hpp>
int main() {
    try {
        EsClient client("http://localhost:9200");
        // ILM 정책: hot → 7일 후 warm → 30일 후 delete
        nlohmann::json policy = {
            {"policy", {
                {"phases", {
                    {"hot", {
                        {"min_age", "0ms"},
                        {"actions", {
                            {"rollover", {
                                {"max_primary_shard_size", "50gb"},
                                {"max_age", "1d"}
                            }}
                        }}
                    }},
                    {"warm", {
                        {"min_age", "7d"},
                        {"actions", {
                            {"shrink", {{"number_of_shards", 1}}},
                            {"forcemerge", {{"max_num_segments", 1}}}
                        }}
                    }},
                    {"delete", {
                        {"min_age", "30d"},
                        {"actions", {{"delete", nlohmann::json::object()}}}
                    }}
                }}
            }}
        };
        client.put("_ilm/policy/logs_policy", policy.dump());
        std::cout << "ILM 정책 생성 완료\n";
        // 인덱스 템플릿에 ILM 연결
        nlohmann::json template_body = {
            {"index_patterns", {"logs-*"}},
            {"template", {
                {"settings", {
                    {"number_of_shards", 3},
                    {"number_of_replicas", 1},
                    {"index.lifecycle.name", "logs_policy"},
                    {"index.lifecycle.rollover_alias", "logs"}
                }}
            }}
        };
        client.put("_index_template/logs_template", template_body.dump());
        std::cout << "인덱스 템플릿 생성 완료\n";
    } catch (const std::exception& e) {
        std::cerr << "에러: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

이 정책은 nlohmann 초기화 규칙을 잘못 쓰면 엉뚱한 JSON이 됩니다. {"shrink", {"number_of_shards", 1}}처럼 쓰면 안쪽 {"number_of_shards", 1}은 키-값 쌍이 아니라 2개짜리 배열로 해석되어 "shrink": ["number_of_shards", 1]이 되고, {"delete", {}}의 {}는 빈 객체가 아니라 null이 됩니다. Elasticsearch는 이런 정책을 x_content_parse_exception으로 거부합니다. 위처럼 한 겹 더 감싸 객체로 만들고, 빈 객체는 nlohmann::json::object()로 명시해야 합니다. 같은 이유로 {"match_all", {}}도 "match_all": null이 되므로 이 글의 쿼리 예제는 모두 nlohmann::json::object()를 씁니다.

ILM의 min_age는 인덱스가 롤오버되었다면 생성 시점이 아니라 롤오버 시점부터 계산됩니다. 위 정책에서 “30일 후 삭제”는 롤오버되어 쓰기가 끝난 뒤 30일이라는 뜻이라, 롤오버 조건이 충족되지 않아 인덱스가 계속 hot에 머무르면 삭제 단계에도 도달하지 않습니다. 롤오버 조건으로는 인덱스 전체 크기(max_size)보다 샤드 단위 기준인 max_primary_shard_size가 8.x에서 권장되므로 위 코드도 그렇게 썼습니다. shrink는 모든 샤드 사본을 한 노드로 모아야 하므로 노드 디스크 여유가 부족하면 단계가 멈춘 채 남을 수 있습니다. GET logs-*/_ilm/explain으로 각 인덱스가 어느 단계에 있고 어디서 멈췄는지 주기적으로 확인하는 것이 좋습니다.

rollover_alias로 지정한 logs는 별칭 이름이므로, 앞의 예제처럼 같은 이름의 실제 인덱스 logs가 이미 있으면 별칭을 만들 수 없습니다. 롤오버를 쓸 클러스터에서는 인덱스 이름과 별칭 이름이 겹치지 않게 정해야 합니다. 새로 설계한다면 별칭 기반 롤오버 대신 Data Stream을 쓰는 것이 8.x의 기본 방식이며, 이 경우 rollover_alias 설정과 첫 인덱스 생성 과정이 필요 없어집니다.

롤오버 인덱스 생성 및 ILM 확인

롤오버용 별칭이 있는 첫 인덱스: PUT logs-000001 {"aliases":{"logs":{"is_write_index":true}}}
ILM 상태: GET _ilm/status, 인덱스별: GET logs-*/_ilm/explain

---

Connection refused, mapper_parsing_exception, version conflict: 에러 해결

”Connection refused” / “Could not resolve host”

연결 단계에서 실패하는 경우로, 서버가 떠 있지 않거나 주소·포트가 틀렸거나, Docker 네트워크나 방화벽 때문에 닿지 않는 것이 보통의 원인입니다. 먼저 같은 위치에서 curl로 확인합니다.

# Elasticsearch 실행 확인
curl -X GET "localhost:9200/"
# Docker 컨테이너에서 호스트 접속 시
# URL: http://host.docker.internal:9200

// 이 글의 EsClient는 URL을 하나만 받으므로, 노드가 여러 개라면
// 앞단에 로드밸런서를 두거나 실패 시 다른 URL로 바꿔 재시도하는 로직이 필요합니다.
EsClient client("http://node1:9200");
// 서버가 응답하지 않을 때 무한정 기다리지 않도록 타임아웃 설정
curl_easy_setopt(curl_, CURLOPT_CONNECTTIMEOUT, 5L);
curl_easy_setopt(curl_, CURLOPT_TIMEOUT, 30L);

”mapper_parsing_exception” / “400 Bad Request”

문서 인덱싱이 400으로 거부되는 경우로, JSON 형식이 깨졌거나 값이 매핑된 필드 타입과 맞지 않을 때 납니다. 문자열을 손으로 이어 붙이지 말고 JSON 라이브러리로 만드는 것이 첫 번째 대책입니다.

// ❌ 잘못된 JSON
std::string bad = R"({"message": "test" "level": "info"})";  // 쉼표 누락
// ✅ nlohmann::json으로 안전하게 구성
nlohmann::json doc = {
    {"message", "test"},
    {"level", "info"},
    {"timestamp", "2024-01-15T10:00:00Z"}
};
std::string body = doc.dump();

”index_not_found_exception”

존재하지 않는 인덱스를 검색·조회할 때 납니다. 문서를 넣을 때는 기본 설정상 인덱스가 자동 생성되지만, 검색은 그렇지 않으므로 시작 시 인덱스를 미리 만들어 둡니다.

std::string mapping = R"({
    "mappings": {
        "properties": {
            "message": { "type": "text" },
            "level": { "type": "keyword" }
        }
    }
})";
client.put("logs", mapping);

타임아웃 / “es_rejected_execution_exception”

벌크 인덱싱이나 무거운 검색에서 클라이언트 타임아웃이 나거나, 서버의 쓰기 스레드 풀 큐가 가득 차 429가 돌아오는 경우입니다. 배치를 작게 나누고, 대량 작업용 요청에만 타임아웃을 늘립니다.

const size_t BATCH_SIZE = 1000;
curl_easy_setopt(curl_, CURLOPT_TIMEOUT, 120L);

”version_conflict_engine_exception”

같은 ID로 일반 인덱싱(PUT logs/_doc/1)을 여러 번 하면 충돌 없이 덮어쓰기됩니다. 이 오류는 op_type=create(PUT logs/_create/1)로 이미 있는 ID를 만들려 하거나, if_seq_no·if_primary_term으로 낙관적 동시성 제어를 걸었는데 그사이 다른 쓰기가 끼어들었을 때, 또는 동시에 _update가 경합할 때 납니다. 중복 생성 방지가 목적이라면 이 오류를 “이미 있음”으로 처리하고, 갱신 경합이라면 문서를 다시 읽어 재시도합니다. ID가 의미 없는 로그라면 ID 없이 POST해 자동 ID를 받는 방법도 있습니다.

std::string resp = client.post("logs/_doc", doc.dump());  // ID 없이 POST하면 서버가 ID 생성

JSON 파싱 에러

nlohmann의 parse는 응답이 JSON이 아니면 parse_error 예외를 던집니다. 프록시가 HTML 오류 페이지를 돌려주는 경우 등이 있으므로, 파싱을 try-catch로 감싸고 실패 시 원본 응답을 로그로 남깁니다.

”security_exception” / 401 Unauthorized

보안이 켜진 8.x 클러스터에 인증 없이 요청하면 401이 납니다. 기본 인증은 curl_easy_setopt(curl_, CURLOPT_USERPWD, "elastic:password");로 지정합니다.

8.x는 기본으로 보안과 HTTPS가 켜진 상태로 시작합니다. 이 글의 Docker 예제는 개발 편의를 위해 xpack.security.enabled=false로 껐지만, 운영 클러스터에 http://로 요청하면 연결이 끊기거나 received plaintext http traffic on an https channel 경고가 서버 로그에 남습니다. 클라이언트 쪽에서는 Connection refused가 아니라 curl: (52) Empty reply from server처럼 응답 없이 끊기는 증상으로 보이는데, 이때는 서버는 살아 있고 프로토콜만 맞지 않는 것입니다. https://로 바꾸고, 설치 시 자동 생성된 CA 인증서(http_ca.crt)를 CURLOPT_CAINFO로 지정합니다. 운영에서는 슈퍼유저 elastic 대신 필요한 권한만 가진 API 키를 발급해 Authorization: ApiKey <base64> 헤더로 보내는 편이 안전합니다. 키는 권한 범위와 만료를 지정할 수 있고, 유출 시 해당 키만 폐기하면 됩니다.

---

배치 크기·연결 재사용·_source 제한

벌크 배치 크기

배치가 너무 크면 클라이언트와 조정 노드의 메모리를 많이 쓰고 요청 하나가 타임아웃에 걸리기 쉬우며, 너무 작으면 요청 수가 늘어 HTTP 왕복 비용이 커집니다. 수천 건 단위에서 시작해, 앞서 말한 대로 요청 바이트 크기를 보면서 조정합니다.

연결 재사용

// ❌ 나쁜 예: 매 요청마다 새 연결
void handle_request() {
    EsClient client("http://localhost:9200");
    client.get("logs/_search");
}
// ✅ 좋은 예: EsClient 싱글톤 또는 풀
class SearchService {
    EsClient client_;
public:
    SearchService() : client_("http://localhost:9200") {}
    std::string search(const std::string& query) {
        return client_.post("logs/_search", query);
    }
};

이 SearchService를 여러 스레드가 공유한다면 앞에서 말한 CURL* 스레드 안전성 문제가 생깁니다. 웹 서버의 요청 핸들러처럼 동시에 여러 요청이 들어오는 곳에서는 thread_local EsClient를 쓰거나, 스레드 수만큼 클라이언트를 담은 풀에서 빌려 쓰는 구조가 필요합니다.

_source 필드 제한

// 필요한 필드만 요청 (네트워크·파싱 부하 감소)
nlohmann::json query = {
    {"query", {{"match_all", nlohmann::json::object()}}},
    {"_source", {"message", "level"}},
    {"size", 100}
};

스크롤 vs Search After

방식	용도	메모리
Scroll	대용량 일괄 처리	서버에 스크롤 컨텍스트 유지
Search After	페이지네이션, 실시간에 가까움	요청 자체는 stateless (PIT를 함께 쓰면 PIT 컨텍스트 유지)

---

재시도·헬스 체크·인덱스 별칭

재시도 로직

std::string request_with_retry(const std::string& method, const std::string& path,
                               const std::string& body, int max_retries = 3) {
    for (int i = 0; i < max_retries; ++i) {
        try {
            if (method == "GET") return client_.get(path);
            if (method == "POST") return client_.post(path, body);
            if (method == "PUT") return client_.put(path, body);
        } catch (const std::exception& e) {
            if (i == max_retries - 1) throw;
            std::this_thread::sleep_for(std::chrono::milliseconds(100 * (i + 1)));
        }
    }
    throw std::runtime_error("재시도 실패");
}

이 함수는 예외 종류를 가리지 않고 재시도하므로 400처럼 다시 보내도 결과가 같은 오류까지 반복합니다. 실제로는 연결 실패, 429, 503 같은 일시적 오류만 재시도하도록 헬퍼가 상태 코드를 담은 예외 타입을 던지게 하는 편이 좋습니다. 또 ID 없이 보내는 POST logs/_doc는 멱등하지 않아서, 서버가 처리한 뒤 응답만 유실된 경우 재시도하면 문서가 두 번 들어갑니다.

헬스 체크

bool is_healthy() {
    try {
        std::string resp = client_.get("_cluster/health");
        auto j = nlohmann::json::parse(resp);
        std::string status = j["status"];
        return status == "green" || status == "yellow";
    } catch (...) {
        return false;
    }
}

yellow는 레플리카 일부가 할당되지 않은 상태로, 레플리카를 둘 노드가 없는 단일 노드 개발 환경에서는 항상 yellow이므로 정상으로 취급했습니다.

환경 변수 기반 설정

struct EsConfig {
    std::string url = "http://localhost:9200";
    int timeout_sec = 30;
    size_t bulk_size = 2000;
};
EsConfig load_config() {
    EsConfig c;
    if (const char* u = std::getenv("ELASTICSEARCH_URL")) c.url = u;
    if (const char* t = std::getenv("ES_TIMEOUT")) c.timeout_sec = std::stoi(t);
    return c;
}

인덱스 별칭 (Zero-Downtime 재인덱싱)

// POST _aliases: remove logs_v1, add logs_v2 to alias "logs"
client_.post("_aliases", R"({"actions":[{"remove":{"index":"logs_v1","alias":"logs"}},{"add":{"index":"logs_v2","alias":"logs"}}]})");

한 _aliases 요청 안의 액션들은 원자적으로 적용되므로, 애플리케이션이 별칭 logs로만 접근한다면 새 인덱스로 재인덱싱을 마친 뒤 이 요청 하나로 끊김 없이 전환할 수 있습니다.

TLS/SSL (프로덕션)

EsClient client("https://elasticsearch.example.com:9200");
// 8.x가 설치 시 생성한 자체 서명 CA라면 시스템 CA 번들이 아니라 그 인증서를 지정
curl_easy_setopt(curl_, CURLOPT_CAINFO, "/etc/elasticsearch/certs/http_ca.crt");

---

다음 글: C++ 시리즈 목차

---

참고 자료

• Elasticsearch REST API 문서
• Elasticlient GitHub
• Index Lifecycle Management
• nlohmann/json
• libcurl

---

같이 보면 좋은 글

• C++ HTTP 클라이언트 직접 만들기
• C++에서 MongoDB 쓰기

---