Ingest Pipeline

학습 목표

ingest pipeline이 Elasticsearch 내부에서 document를 변환하는 방식을 설명할 수 있다.

processor chain을 설계하고 grok · date · set · remove를 조합할 수 있다.

enrich processor로 lookup index에서 필드 보강 패턴을 설명할 수 있다.

on_failure·ignore_failure로 파싱 실패를 격리할 수 있다.

Beat 직결 vs Logstash vs ingest 역할 분담을 말할 수 있다.

문제 상황

• Filebeat가 ES에 JSON 그대로 넣는데 client.ip geo·service owner가 없음 — Beat마다 enrich 중복
• Logstash 없이 공통 파싱만 필요 — JVM·5044 운영 부담 없이 가고 싶음
• grok 한 줄 실패 시 전체 bulk가 reject — 실패 document 처리 전략 없음
• 팀마다 pipeline 이름·processor 순서가 달라 재현 불가 — Git·API 관리 없음
• 6편 module이 등록하는 pipeline과 커스텀 pipeline이 충돌
• 7편 Logstash filter와 같은 grok을 두 군데에 둠 — 필드 타입·디버깅 혼란

앞 편에서 Logstash로 중앙 transform을 봤다. 이번 편은 Elasticsearch ingest pipeline — Beat→ES 직결일 때의 공통 가공. data stream은 9편이다.

1. Ingest Pipeline이란?

ingest pipeline — index 요청 직전에 document를 순서대로 변환하는 processor chain.

개념	설명
pipeline	이름 있는 processor 목록 (logs-parse)
ingest node	pipeline을 실행하는 ES 노드 역할
processor	document 한 단계 변환 (grok, set, …)
simulate	_ingest/pipeline/_simulate — 저장 없이 테스트

Filebeat bulk index request
        │
        ▼
  ingest node (pipeline logs-parse)
        │
   grok → date → geoip → set
        │
        ▼
  primary shard indexing

• 실행 시점 — primary shard에 쓰기 전 (2편 NRT 흐름의 ingest 단계)
• 클러스터 — dedicated ingest 노드 또는 data 노드에 ingest 역할
• Elastic Cloud — deployment에 ingest 용량 포함 — 16편

2. Pipeline 지정 방법

방법	설명
요청마다	POST /logs/_doc?pipeline=logs-parse
index default	index·template에 "index.default_pipeline"
final pipeline	"index.final_pipeline" — 항상 마지막 실행
Beat output	output.elasticsearch.pipeline

# code/filebeat-pipeline.example.yml (발췌)
output.elasticsearch:
  hosts: ["https://es.example.com:9243"]
  pipeline: "logs-order-parse"
  api_key: "${ES_API_KEY}"

PUT _ingest/pipeline/logs-order-parse
{
  "description": "order-api json logs",
  "processors": [ ... ]
}

• default + final — default로 소스별 파싱, final로 공통 drop·rename
• overwrite — 요청의 pipeline이 default를 대체 (final은 유지)
• 14편 index template — default_pipeline을 template에 고정 권장

3. Processor Chain

processor는 배열 순서대로 실행 — 앞 단계 필드가 뒤 단계 입력.

// code/logs-parse.pipeline.json (발췌)
{
  "description": "nginx access common parse",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": ["%{COMBINEDAPACHELOG}"],
        "on_failure": [
          { "set": { "field": "error.parse", "value": "grok_failed" } }
        ]
      }
    },
    {
      "date": {
        "field": "timestamp",
        "formats": ["dd/MMM/yyyy:HH:mm:ss Z"],
        "target_field": "@timestamp"
      }
    },
    {
      "geoip": {
        "field": "clientip",
        "target_field": "source.geo"
      }
    },
    {
      "remove": { "field": "message" }
    }
  ]
}

processor	용도
set	상수·메타 필드 추가
rename	필드 이름 변경
grok	비구조 message 분리 (7편 Logstash grok과 동일 패턴)
json	문자열 JSON 파싱
date	@timestamp 정규화
geoip	IP → geo (GeoIP DB 내장)
remove	불필요 필드 삭제
drop	조건 만족 시 document 폐기
script	Painless로 커스텀 (과용 주의)

• 3편 mapping — response는 integer, source.ip는 ip — processor 출력과 일치
• 조건부 — if (processor별) — ctx.service == 'order-api' 일 때만 grok
• simulate — 샘플 document 넣어 단계별 결과 확인

POST _ingest/pipeline/logs-order-parse/_simulate
{
  "docs": [
    { "_source": { "message": "203.0.113.1 - - [10/Oct/2025:13:55:36 +0000] \"GET / HTTP/1.1\" 200 1234" } }
  ]
}

4. Enrich

enrich processor — 별도 enrich index에서 키 매칭으로 필드 병합.

// code/service-enrich.policy.json (발췌)
PUT _enrich/policy/service-owner
{
  "match": {
    "indices": "enrich-service-owner",
    "match_field": "service.name",
    "enrich_fields": ["owner", "tier", "slack_channel"]
  }
}

{
  "enrich": {
    "policy_name": "service-owner",
    "field": "service",
    "target_field": "service.meta"
  }
}

단계	설명
소스 index	service, owner, tier CSV·DB 스냅샷
enrich policy	match_field·enrich_fields 정의
execute policy	enrich index 생성 (주기적 refresh)
enrich processor	document의 service로 lookup

• 정적 메타 — 서비스 카탈로그·DC·팀 매핑에 적합
• 실시간 API — Logstash http filter·외부 캐시가 나을 수 있음
• policy 업데이트 후 _execute — stale enrich 주의
• enrich index는 시스템 관리 — 직접 document 넣지 않음

5. Failure Handling

processor 실패 시 전체 chain 중단이 기본 — on_failure로 분기.

옵션	동작
(기본)	실패 시 document reject — bulk item error
on_failure	실패 시 대체 processor 실행
ignore_failure: true	실패 무시하고 다음 processor
tag	on_failure에서 _tag 필드로 표시

{
  "grok": {
    "field": "message",
    "patterns": ["%{COMBINEDAPACHELOG}"],
    "on_failure": [
      {
        "set": {
          "field": "error.ingest",
          "value": "grok_parse_failure"
        }
      },
      {
        "set": {
          "field": "_index",
          "value": "logs-deadletter"
        }
      }
    ]
  }
}

• dead letter index — 파싱 실패 document 격리 — 7편 Logstash output 분기와 동일 목적
• 모니터링 — error.ingest·event.original 보존 여부 팀 규칙
• bulk 응답 — 실패 item만 재시도 vs DLQ 재적재
• event.original — 실패 분석용 원문 보존 (processor로 copy)

ignore_failure vs on_failure

	ignore_failure	on_failure
의도	해당 단계 선택	실패 처리 경로 명시
위험	조용히 필드 누락	DLQ·tag로 가시성
권장	geoip 등 부가 필드	grok·json 핵심 파싱

6. Beat · Logstash와 역할 분담

단계	Beat processor	Logstash filter	Ingest pipeline
위치	호스트	LS JVM	ES ingest
적합	host·cloud 메타	소스별 heavy grok	공통 parse·enrich
배포	beat yml	logstash.conf	_ingest/pipeline API
버전	Beat 배치	LS 배치	ES 클러스터 동기

• 권장 — JSON 로그: Beat 최소 + ingest 공통 / 비구조 Nginx: module 또는 ingest grok 한 곳만
• 6편 module — filebeat setup이 pipeline 등록 — 이름 충돌·덮어쓰기 정책 필요
• 7편 output.elasticsearch.pipeline — LS filter 후 ingest 추가 가능 — 중복 grok 금지

7. 배포·운영

# 등록·갱신
PUT _ingest/pipeline/logs-order-parse
# ...
GET _ingest/pipeline/logs-order-parse

# 삭제 (주의)
DELETE _ingest/pipeline/logs-order-parse

practice	설명
Git	pipeline JSON을 소스로 — CI가 PUT
simulate	배포 전 샘플 검증
버전	pipeline 이름에 suffix (v2) 또는 alias 패턴
권한	manage_ingest_pipelines — prod는 role 분리 (20편)

• Elastic Cloud — Kibana Stack Management → Ingest Pipelines UI
• breaking change — 필드 rename·remove는 대시보드·alert 영향 — 10~12편
• ingest CPU — 무거운 grok이 많으면 ingest 노드 스케일 또는 Logstash로 이전

8. 실무 체크리스트

#	질문
1	grok·json 실패 시 DLQ 또는 tag가 있는가
2	pipeline이 Git·simulate로 관리되는가
3	Logstash와 같은 grok이 중복되지 않는가
4	enrich policy refresh 주기가 메타 변경에 맞는가
5	processor 출력 필드가 3편 mapping과 타입이 맞는가
6	default_pipeline이 14편 template에 연결됐는가

정리

• ingest pipeline — ES 내부 processor chain, index 직전 변환
• processor — grok·date·geoip·set·remove … 순서가 결과를 결정
• enrich — lookup index로 정적 메타 병합
• on_failure — 파싱 실패 격리·DLQ
• Beat 직결 + ingest — Logstash 없이 공통 파싱 가능

다음에 다룰 것

• Data Stream
• data stream vs classic index, @timestamp, backing index

---

해당 내용은 Elasticsearch Guide (Ingest pipelines, Enrich, Simulate pipeline API) 를 기반으로 합니다.