C++에서 Elasticsearch 8.x 쓰기: libcurl REST API와 Elasticlient
들어가며: “C++에서 Elasticsearch 연동이 막막해요”
이 글이 답하는 질문
"Elasticsearch 공식 C++ 클라이언트가 없는데 어떻게 연동하나요?"
"로그 100만 건을 인덱싱·검색하려면 어떤 API를 써야 하나요?"
"벌크 인덱싱 시 타임아웃·메모리 부족은 어떻게 해결하나요?"
"오래된 인덱스를 자동으로 정리하려면?"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 7.x/8.x, Elasticlient 또는 libcurl + nlohmann/json
검색·집계·ILM이 필요해지는 상황
로그 검색이 10초 넘게 걸림
상황: 수백만 건 로그를 MySQL LIKE '%error%'로 검색
문제: 풀 테이블 스캔, 인덱스 무력, 10~30초 지연
결과: Elasticsearch 역인덱스 전문 검색으로 밀리초 단위 검색전문 검색(Full-Text Search) 필요
상황: 제품 설명에서 "무선 이어폰 블루투스 노이즈캔슬링" 검색
문제: 단어 분리, 유사어 매칭, 점수 기반 정렬이 관계형 DB에서 어려움
결과: Elasticsearch match·match_phrase·bool 쿼리로 정확한 검색실시간 집계·대시보드
상황: API 호출 수, 에러율, 평균 응답 시간을 1분 단위로 집계
문제: 매분마다 COUNT·AVG 쿼리로 DB 부하 급증
결과: date_histogram·terms 집계로 실시간 집계Connection refused 에러
상황: Elasticsearch 서버 주소 설정했는데 연결 실패
문제: localhost vs 127.0.0.1, Docker 네트워크, 방화벽 혼동
결과: curl로 연결 확인, 여러 노드 지정, 타임아웃 설정벌크 인덱싱 시 타임아웃
상황: 수십만 건을 한 번에 인덱싱하려다 RequestTimeout·EsRejectedExecutionException
문제: 배치 크기 과다, 스레드 풀 포화
결과: 1000~5000건 배치, 재시도, refresh_interval 조정수십만 건 검색 시 OOM
상황: 전체 결과를 vector에 담다 메모리 폭증
문제: size=100000으로 한 번에 조회
결과: 스크롤 API 또는 Search After로 스트리밍 조회오래된 로그 인덱스 디스크 폭증
상황: 일별 인덱스가 수백 개 쌓여 디스크 부족
문제: 수동 삭제·압축 관리 부담
결과: ILM(Index Lifecycle Management)으로 hot→warm→delete 자동 전환시나리오별 기술 선택
| 시나리오 | Elasticsearch 기능 | C++ 구현 |
|---|---|---|
| 느린 로그 검색 | 역인덱스 전문 검색 | match·match_phrase 쿼리 |
| 복합 키워드 검색 | bool·multi_match | Query DSL JSON |
| 실시간 집계 | terms·date_histogram | aggs 필드 |
| 대량 인덱싱 | _bulk API | 배치 1000~5000 |
| 대용량 검색 | Scroll·Search After | scroll_id·search_after |
| 인덱스 수명 관리 | ILM | PUT _ilm/policy |
flowchart TB
subgraph 문제[실무 문제]
P1[느린 검색] --> S1[전문 검색]
P2[복합 검색] --> S2[Query DSL]
P3[실시간 집계] --> S3[집계 API]
P4[Connection refused] --> S4[연결·타임아웃]
P5[벌크 타임아웃] --> S5[배치·재시도]
P6[대용량 OOM] --> S6[스크롤]
P7[인덱스 폭증] --> S7[ILM]
end
Elasticsearch 서버와 클라이언트 라이브러리 설치
필수 의존성
| 항목 | 버전 | 비고 |
|---|---|---|
| C++ | C++14 이상 | C++17 권장 |
| Elasticsearch | 7.x/8.x | Docker 권장 |
| Elasticlient | - | 선택, cpr 의존 |
| 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 의존성 (Ubuntu/Debian)
sudo apt-get install libcurl4-openssl-dev
# Elasticlient 소스 빌드
git clone https://github.com/seznam/elasticlient.git
cd elasticlient
mkdir build && cd build
cmake .. -DCMAKE_BUILD_TYPE=Release
cmake --build .
sudo cmake --build . --target installlibcurl + 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-jsonCMakeLists.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_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: Elasticsearch 클러스터 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"]
})";
cpr::Response resp = client.search("logs", 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 <string>
int main() {
elasticlient::Client client({"http://localhost:9200/"});
elasticlient::Bulk bulk(client);
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"("})";
bulk.index("logs", "_doc", std::to_string(i), doc);
}
cpr::Response resp = bulk.perform();
if (resp.status_code == 200) {
std::cout << "벌크 완료: " << resp.text << std::endl;
} else {
std::cerr << "벌크 실패: " << resp.status_code << " " << resp.text << std::endl;
}
return 0;
}스크롤 API (Elasticlient Scroll)
// elasticlient_scroll.cpp
#include <elasticlient/client.h>
#include <elasticlient/scroll.h>
#include <iostream>
#include <string>
int main() {
elasticlient::Client client({"http://localhost:9200/"});
std::string query = R"({
"query": {"match_all": {}},
"size": 100
})";
elasticlient::Scroll scroll(client, "logs", query, "1m");
size_t total = 0;
while (scroll.hasNext()) {
auto hits = scroll.next();
for (const auto& hit : hits) {
total++;
// 문서 처리
}
}
std::cout << "총 처리: " << total << "건\n";
return 0;
}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 시 runtime_error throw.
이 헬퍼에는 설계상 짚어 둘 점이 몇 가지 있습니다.
CURL*핸들 재사용: 핸들을 멤버로 두고 재사용하면 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"이면 이후 필드 타입이 예상과 달라질 수 있습니다. 로그처럼 필드가 정해진 데이터는 이 예제처럼 매핑을 먼저 만드는 것이 안전합니다.
전문 검색 (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 쿼리를 두면 결과가 캐시되어 더 빠릅니다.
집계 (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;
}재시도를 포함한 벌크 인덱싱
기본 벌크 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", {}}}},
{"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", {}}}},
{"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;
}ILM (Index Lifecycle Management)
search_after는 이전 페이지 마지막 문서의 정렬 값을 넘겨 그다음부터 가져오는 방식입니다. 그래서 정렬 키가 문서를 유일하게 구분해야 합니다. timestamp만으로 정렬하면 같은 밀리초에 찍힌 로그들이 페이지 경계에 걸릴 때 일부가 건너뛰어지거나 중복됩니다. 이 때문에 두 번째 정렬 키로 고유 필드가 필요하며, 적당한 필드가 없다면 PIT를 열고 PIT가 자동으로 제공하는 _shard_doc 동점 해소 키를 쓰는 것이 공식 권장 방식입니다.
원래 코드는 정렬을 {{"timestamp", "asc"}, {"_id", "asc"}}로 썼는데, nlohmann/json은 이 모양을 배열이 아니라 객체 {"_id":"asc","timestamp":"asc"}로 해석합니다. 그리고 nlohmann의 기본 객체는 키를 이름순으로 정렬해 저장하므로 _id가 앞에 옵니다. 정렬 우선순위가 조용히 뒤바뀌는 버그라, 순서가 의미 있는 값은 항상 nlohmann::json::array()로 명시하는 습관이 필요합니다. 제가 이 라이브러리로 쿼리를 만들 때 가장 자주 확인하는 것이 전송 직전의 query.dump(2) 출력인데, 대부분의 “왜 쿼리가 이상하게 동작하지?”는 이 출력 한 번으로 풀립니다.
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_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이 되므로, 실무 코드에서는 {"match_all", nlohmann::json::object()}로 쓰는 것이 안전합니다.
ILM의 min_age는 인덱스 생성 시점이 아니라 롤오버 시점부터 계산됩니다. 위 정책에서 “30일 후 삭제”는 롤오버되어 쓰기가 끝난 뒤 30일이라는 뜻이라, 롤오버 조건이 충족되지 않아 인덱스가 계속 hot에 머무르면 삭제 단계에도 영영 도달하지 않습니다. 8.x에서는 max_size보다 샤드 단위 기준인 max_primary_shard_size(예: 50gb)를 권장합니다. shrink는 모든 샤드 사본을 한 노드로 모아야 하므로 노드 디스크 여유가 부족하면 단계가 멈춘 채 남을 수 있어, GET logs-*/_ilm/explain으로 각 인덱스가 어느 단계·어느 단계에서 실패했는지 주기적으로 확인하는 것이 좋습니다. 새로 설계한다면 별칭 기반 롤오버 대신 Data Stream을 쓰는 것이 8.x의 기본 방식이며, 이 경우 rollover_alias 설정과 첫 인덱스 생성 과정이 필요 없어집니다.
롤오버 인덱스 생성 및 ILM 확인
롤오버용 별칭이 있는 첫 인덱스: PUT logs-000001 {"aliases":{"logs":{"is_write_index":true}}}
ILM 상태: GET _ilm/status, 인덱스별: GET logs-*/_ilm/explainConnection refused, mapper_parsing_exception, version conflict: 에러 해결
”Connection refused” / “Could not resolve host”
증상: Elasticsearch 연결 시도 시 실패. 원인: 서버 미실행, 잘못된 주소/포트, Docker 네트워크, 방화벽. 해결법:
# Elasticsearch 실행 확인
curl -X GET "localhost:9200/"
# Docker 컨테이너에서 호스트 접속 시
# URL: http://host.docker.internal:9200// ✅ 여러 노드 지정 (고가용성)
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
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);“RequestTimeout” / “EsRejectedExecutionException”
증상: 벌크 인덱싱 또는 대용량 검색 시 타임아웃. 해결법:
// ✅ 벌크 배치 크기 제한 (1000~5000 권장)
const size_t BATCH_SIZE = 1000;
// ✅ 타임아웃 증가
curl_easy_setopt(curl_, CURLOPT_TIMEOUT, 120L);“version_conflict_engine_exception”
증상: 동일 ID로 동시 인덱싱 시 충돌. 해결법:
// ✅ 자동 ID 사용
std::string resp = client.post("logs/_doc", doc); // ID 없이 POSTJSON 파싱 에러
증상: parse_error 예외. 해결: try-catch로 파싱, 실패 시 원본 응답 로깅.
”security_exception” / 401 Unauthorized
증상: Elasticsearch 8.x 보안 활성화 시 인증 실패. 해결: 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 경고가 서버 로그에 남습니다. 운영에서는 슈퍼유저 elastic 대신 필요한 권한만 가진 API 키를 발급해 Authorization: ApiKey <base64> 헤더로 보내는 편이 안전합니다. 키는 권한 범위와 만료를 지정할 수 있고, 유출 시 해당 키만 폐기하면 됩니다.
배치 크기·연결 재사용·_source 제한
벌크 배치 크기
// 배치 크기: 1000~5000 권장
// 너무 크면: 메모리·타임아웃
// 너무 작으면: HTTP 오버헤드
const size_t BATCH_SIZE = 2000;연결 재사용
// ❌ 나쁜 예: 매 요청마다 새 연결
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", {}}}},
{"_source", {"message", "level"}},
{"size", 100}
};스크롤 vs Search After
| 방식 | 용도 | 메모리 |
|---|---|---|
| Scroll | 대용량 일괄 처리 | 서버에 스크롤 컨텍스트 유지 |
| Search After | 페이지네이션, 실시간에 가까움 | stateless |
재시도·헬스 체크·인덱스 별칭
재시도 로직
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("재시도 실패");
}헬스 체크
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;
}
}환경 변수 기반 설정
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"}}]})");TLS/SSL (프로덕션)
EsClient client("https://elasticsearch.example.com:9200");
curl_easy_setopt(curl_, CURLOPT_CAINFO, "/etc/ssl/certs/ca-certificates.crt");Elasticsearch 연동 점검 항목
환경 설정
- Elasticsearch 서버 실행 확인 (
curl localhost:9200) - libcurl, nlohmann/json 설치 (또는 Elasticlient)
- CMake/빌드 연동
REST API 클라이언트
- Content-Type: application/json 헤더 설정
- 타임아웃 설정 (connect, read)
- HTTP 에러 코드 검사 (4xx, 5xx)
- JSON 파싱 예외 처리
인덱싱
- 인덱스 매핑 사전 정의 (필드 타입)
- 벌크 배치 크기 1000~5000
- 벌크 에러 시 개별 항목 실패 처리
검색
- _source 필드 제한 (필요 시)
- size 제한 (기본 10, 최대 10000)
- 대용량 결과 시 스크롤 또는 Search After
ILM
- 로그 인덱스에 ILM 정책 연결
- 롤오버 조건 (max_size, max_age) 설정
- delete 단계 min_age 설정
프로덕션
- 연결 재사용 (EsClient 풀/싱글톤)
- 헬스 체크 (/_cluster/health)
- 환경 변수로 URL·타임아웃 외부화
- TLS/SSL (HTTPS)
기능별 요약
| 항목 | 요약 |
|---|---|
| 공식 C++ 클라이언트 | 없음. Elasticlient 또는 REST API 사용 |
| REST API | libcurl + nlohmann/json으로 모든 API 호출 가능 |
| 인덱싱 | PUT/POST, 벌크는 _bulk API |
| 전문 검색 | match·match_phrase·bool 쿼리 |
| 집계 | aggs 필드로 terms·date_histogram 등 |
| 벌크 | 배치 1000~5000, 재시도 |
| 스크롤 | scroll_id로 대용량 순차 조회 |
| ILM | hot→warm→delete 자동 전환 |
| 에러 | Connection refused, mapper_parsing, RequestTimeout |
| 프로덕션 | 재시도, 헬스 체크, 환경 변수, TLS |
핵심 원칙:
- 벌크 사용: 단건 인덱싱 대신 _bulk API로 배치 처리
- 연결 재사용: 매 요청마다 새 클라이언트 생성 금지
- 에러 파싱: HTTP 상태·JSON error.reason 확인
- 대용량 검색: 스크롤 또는 Search After
- ILM 활용: 오래된 인덱스 자동 정리
자주 묻는 질문 (FAQ)
Q. Elasticlient와 REST API 중 어떤 것을 써야 하나요?
A. Elasticlient는 API가 단순하지만 유지보수가 느리고 cpr 의존이 있습니다. 새 프로젝트에서는 libcurl + nlohmann/json으로 REST API를 직접 호출하는 방식을 권장합니다.
Q. 대용량 로그 인덱싱 시 주의할 점은?
A. 벌크 배치 크기 1000~5000, 인덱스별 refresh_interval 조정, ILM으로 오래된 인덱스 정리 등을 고려하세요.
Q. 스크롤과 Search After의 차이는?
A. 스크롤은 대용량 일괄 처리에 적합하며 서버에 컨텍스트를 유지합니다. Search After는 stateless로 실시간 페이지네이션에 적합합니다. Elasticsearch는 공식 C++ 클라이언트가 없지만, libcurl과 nlohmann/json으로 REST API를 호출하면 전문 검색·집계·벌크 인덱싱·스크롤·ILM을 C++에서 완전히 활용할 수 있습니다. 다음 글: C++ 시리즈 목차 이전 글: Elasticsearch 개요(#52-5)