본문 바로가기
WIKI 기술 지식 베이스

Grok processor

원문 보기 위키 갱신

출처: 문서

본문

Grok 프로세서

Grok 프로세서는 패턴 매칭을 사용해 비정형 데이터를 구조화하고 중요한 키를 추출해요.

설정

다음 표는 데이터를 구조화하고 쿼리하기 쉽게 만들기 위해 Grok 프로세서에서 사용할 수 있는 옵션을 설명해요.

옵션 필수 여부 타입 설명
break_on_match 아니요 Boolean 모든 패턴을 매칭할지(true), 아니면 첫 번째 성공한 매칭을 찾는 즉시 중단할지(false) 지정해요. 기본값은 true예요.
grok_when 아니요 String grok 프로세서가 어떤 조건에서 매칭을 수행해야 하는지 지정해요. 이 표현식에 대한 정보는 표현식 문법을 참조하세요. 기본값은 조건 없음이에요.
keep_empty_captures 아니요 Boolean 처리된 출력에서 null 캡처를 유지할 수 있게 해요. 기본값은 false예요.
keys_to_overwrite 아니요 List 같은 키 값을 가진 캡처가 있을 때 어떤 기존 키를 덮어쓸지 지정해요. 기본값은 []예요.
match 아니요 Map 어떤 키가 특정 패턴과 매칭되어야 하는지 지정해요. 기본값은 빈 응답 본문이에요.
named_captures_only 아니요 Boolean 이름이 지정된 캡처만 유지할지 지정해요. 기본값은 true예요.
pattern_definitions 아니요 Map 응답 본문 안에서 인라인으로 사용할 수 있는 사용자 지정 패턴을 정의할 수 있게 해요. 기본값은 빈 응답 본문이에요.
patterns_directories 아니요 List 사용자 지정 패턴 파일이 들어 있는 디렉터리 경로를 지정해요. 기본값은 빈 리스트예요.
pattern_files_glob 아니요 String pattern_directories에 지정된 디렉터리에서 어떤 패턴 파일을 사용할지 지정해요. 기본값은 *예요.
target_key 아니요 String 모든 캡처를 저장하는 데 사용되는 상위 수준 키를 지정해요. 기본값은 null이에요.
timeout_millis 아니요 Integer 매칭이 수행될 수 있는 최대 시간이에요. 0으로 설정하면 어떤 매칭도 일어나지 않아요. 기본값은 30,000이에요.
performance_metadata 아니요 Boolean 이벤트에 성능 메타데이터를 추가할지 여부예요. 기본값은 false예요. 자세한 내용은 Grok 성능 메타데이터를 참조하세요.

Grok 성능 메타데이터

performance_metadata 옵션을 true로 설정하면 grok 프로세서가 각 이벤트에 다음 메타데이터 키를 추가해요:

  • _total_grok_processing_time: grok 프로세서가 이벤트를 매칭하는 데 걸린 총 시간으로, 단위는 밀리초예요. 이 값은 해당 이벤트에서 실행되었고 performance_metadata 옵션이 활성화된 모든 grok 프로세서의 처리 시간을 합한 값이에요.
  • _total_grok_patterns_attempted: 해당 이벤트에서 실행된 모든 grok 프로세서에 걸쳐 시도된 grok 패턴 매칭의 총 횟수예요.

파이프라인 안에서 이벤트가 싱크로 전송될 때 Grok 성능 메타데이터를 포함하려면, 다음 예시와 같이 add_entries 프로세서를 사용해 포함하려는 메타데이터를 지정하세요:

  processor:
    - grok:
        performance_metadata: true
        match:
          log: ["%{COMMONAPACHELOG}"]
        break_on_match: true
        named_captures_only: true
        target_key: "parsed"

    - add_entries:
        entries:
          - add_when: 'getMetadata("_total_grok_patterns_attempted") != null'
            key: "grok_patterns_attempted"
            value_expression: 'getMetadata("_total_grok_patterns_attempted")'
          - add_when: 'getMetadata("_total_grok_processing_time") != null'
            key: "grok_time_spent"
            value_expression: 'getMetadata("_total_grok_processing_time")'

예시

다음 예시는 grok 프로세서를 구성할 수 있는 다양한 방법을 보여줘요.

이 예시는 보안을 사용하지 않으며 데모 목적으로만 제공돼요. 프로덕션에서 이 예시를 사용하기 전에 SSL을 구성하는 것을 강력히 권장해요. {: .warning}

Apache 액세스 로그 파싱

이 예시는 표준 Apache HTTP 액세스 로그를 파싱해 클라이언트 IP, 타임스탬프, HTTP 메서드, URL, 상태 코드, 응답 크기를 추출하는 방법을 보여줘요:

apache-access-logs-pipeline:
  source:
    http:
      path: /logs
      ssl: false

  processor:
    - grok:
        match:
          message: ['%{COMBINEDAPACHELOG}']
        break_on_match: true
        named_captures_only: true
        keep_empty_captures: false
        target_key: "parsed"

    - date:
        match:
          - key: "/parsed/timestamp"     # JSON Pointer ✔
            patterns: ["dd/MMM/yyyy:HH:mm:ss Z"]
        destination: "@timestamp"
        source_timezone: "UTC"

  sink:
    - opensearch:
        hosts: ["https://opensearch:9200"]
        insecure: true
        username: admin
        password: "admin_pass"
        index_type: custom
        index: "apache-logs-%{yyyy.MM.dd}"

{% include copy.html %}

다음 명령어로 이 파이프라인을 테스트할 수 있어요:

curl -sS -X POST "http://localhost:2021/logs" \
  -H "Content-Type: application/json" \
  -d '[
    {"message":"127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] \"GET /apache.gif HTTP/1.0\" 200 2326 \"http://www.example.com/start.html\" \"Mozilla/4.08 [en] (Win98; I ;Nav)\""},
    {"message":"192.168.1.5 - - [13/Oct/2025:17:42:10 +0000] \"POST /login HTTP/1.1\" 302 512 \"-\" \"curl/8.5.0\""}
  ]'

{% include copy.html %}

OpenSearch에 저장된 문서에는 다음 정보가 들어 있어요:

{
  ...
  "hits": {
    "total": {
      "value": 2,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "apache-logs-2025.10.13",
        "_id": "gLO73pkBpMIC6s6zUMMX",
        "_score": 1,
        "_source": {
          "message": "127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] \"GET /apache.gif HTTP/1.0\" 200 2326 \"http://www.example.com/start.html\" \"Mozilla/4.08 [en] (Win98; I ;Nav)\"",
          "parsed": {
            "request": "/apache.gif",
            "referrer": "http://www.example.com/start.html",
            "agent": "Mozilla/4.08 [en] (Win98; I ;Nav)",
            "auth": "frank",
            "ident": "-",
            "response": "200",
            "bytes": "2326",
            "clientip": "127.0.0.1",
            "verb": "GET",
            "httpversion": "1.0",
            "timestamp": "10/Oct/2000:13:55:36 -0700"
          },
          "@timestamp": "2000-10-10T20:55:36.000Z"
        }
      },
      {
        "_index": "apache-logs-2025.10.13",
        "_id": "gbO73pkBpMIC6s6zUMMX",
        "_score": 1,
        "_source": {
          "message": "192.168.1.5 - - [13/Oct/2025:17:42:10 +0000] \"POST /login HTTP/1.1\" 302 512 \"-\" \"curl/8.5.0\"",
          "parsed": {
            "request": "/login",
            "referrer": "-",
            "agent": "curl/8.5.0",
            "auth": "-",
            "ident": "-",
            "response": "302",
            "bytes": "512",
            "clientip": "192.168.1.5",
            "verb": "POST",
            "httpversion": "1.1",
            "timestamp": "13/Oct/2025:17:42:10 +0000"
          },
          "@timestamp": "2025-10-13T17:42:10.000Z"
        }
      }
    ]
  }
}

사용자 지정 패턴으로 애플리케이션 로그 파싱

이 예시는 사용자 정의 패턴으로 사용자 지정 애플리케이션 로그를 파싱해 독자적인 로그 형식에서 구조화된 데이터를 추출하는 방법을 보여줘요:

application-logs-pipeline:
  source:
    http:
      path: /logs
      ssl: false

  processor:
    - grok:
        match:
          message: ['%{TIMESTAMP_ISO8601:timestamp} \\[%{LOGLEVEL:level}\\] %{DATA:component} - %{GREEDYDATA:details}']
        pattern_definitions:
          LOGLEVEL: (?:INFO|WARN|ERROR|DEBUG|TRACE)
        break_on_match: true
        target_key: "parsed"
        keep_empty_captures: false

    - date:
        match:
          - key: "/parsed/timestamp"
            patterns:
              - "yyyy-MM-dd HH:mm:ss"
              - "yyyy-MM-dd HH:mm:ss.SSS"
        destination: "@timestamp"     # you could also use "/@timestamp"
        source_timezone: "UTC"

  sink:
    - opensearch:
        hosts: ["https://opensearch:9200"]
        insecure: true
        username: admin
        password: "admin_pass"
        index_type: custom
        index: "application-logs-%{yyyy.MM.dd}"

{% include copy.html %}

다음 명령어로 이 파이프라인을 테스트할 수 있어요:

curl -sS -X POST "http://localhost:2021/logs" \
  -H "Content-Type: application/json" \
  -d '[
    {"message": "2025-10-13 14:30:45 [INFO] UserService - User login successful"},
    {"message": "2025-10-13 14:31:15 [ERROR] DatabaseConnection - Connection timeout"},
    {"message": "2025-10-13 14:32:30 [DEBUG] CacheManager - Cache hit"},
    {"message": "2025-10-13 14:33:05 [WARN] MetricsCollector - High memory usage detected"}
  ]'

{% include copy.html %}

OpenSearch에 저장된 문서에는 다음 정보가 들어 있어요:

{
  ...
  "hits": {
    "total": {
      "value": 4,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "application-logs-2025.10.13",
        "_id": "i7O83pkBpMIC6s6zhsNK",
        "_score": 1,
        "_source": {
          "message": "2025-10-13 14:30:45 [INFO] UserService - User login successful",
          "parsed": {
            "component": "UserService",
            "level": "INFO",
            "details": "User login successful",
            "timestamp": "2025-10-13 14:30:45"
          },
          "@timestamp": "2025-10-13T14:30:45.000Z"
        }
      },
      {
        "_index": "application-logs-2025.10.13",
        "_id": "jLO83pkBpMIC6s6zhsNK",
        "_score": 1,
        "_source": {
          "message": "2025-10-13 14:31:15 [ERROR] DatabaseConnection - Connection timeout",
          "parsed": {
            "component": "DatabaseConnection",
            "level": "ERROR",
            "details": "Connection timeout",
            "timestamp": "2025-10-13 14:31:15"
          },
          "@timestamp": "2025-10-13T14:31:15.000Z"
        }
      },
      {
        "_index": "application-logs-2025.10.13",
        "_id": "jbO83pkBpMIC6s6zhsNK",
        "_score": 1,
        "_source": {
          "message": "2025-10-13 14:32:30 [DEBUG] CacheManager - Cache hit",
          "parsed": {
            "component": "CacheManager",
            "level": "DEBUG",
            "details": "Cache hit",
            "timestamp": "2025-10-13 14:32:30"
          },
          "@timestamp": "2025-10-13T14:32:30.000Z"
        }
      },
      {
        "_index": "application-logs-2025.10.13",
        "_id": "jrO83pkBpMIC6s6zhsNK",
        "_score": 1,
        "_source": {
          "message": "2025-10-13 14:33:05 [WARN] MetricsCollector - High memory usage detected",
          "parsed": {
            "component": "MetricsCollector",
            "level": "WARN",
            "details": "High memory usage detected",
            "timestamp": "2025-10-13 14:33:05"
          },
          "@timestamp": "2025-10-13T14:33:05.000Z"
        }
      }
    ]
  }
}

여러 패턴으로 네트워크 장치 로그 파싱

이 예시는 여러 grok 패턴을 사용해 네트워크 장치에서 오는 다양한 로그 형식을 처리하고, 로그 유형에 따라 조건부 처리를 수행하는 방법을 보여줘요:

network-device-logs-pipeline:
  source:
    http:
      path: /logs
      ssl: false

  processor:
    - grok:
        match:
          message: [
            # syslog-like
            '%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:host} %{DATA:program}(?:\\[%{POSINT:pid}\\])?: %{GREEDYDATA:message}',
            # ISO8601 + IP
            '%{TIMESTAMP_ISO8601:timestamp} %{IP:host} %{DATA:program}: %{GREEDYDATA:message}',
            # Cisco style
            '%{CISCO_TIMESTAMP:timestamp}: %{DATA:facility}-%{INT:severity}-%{DATA:mnemonic}: %{GREEDYDATA:message}'
          ]
        break_on_match: true
        named_captures_only: true
        pattern_definitions:
          CISCO_TIMESTAMP: '%{MONTH} %{MONTHDAY} %{TIME}'
        target_key: "parsed"
        timeout_millis: 5000

    # Extract login info from the parsed message text
    - grok:
        match:
          /parsed/message: ['User %{USERNAME:user} logged in from %{IP:source_ip}']
        break_on_match: true
        target_key: "login_info"

    - date:
        match:
          - key: "/parsed/timestamp"
            patterns:
              # syslog-like (no year)
              - "MMM d HH:mm:ss"
              # ISO8601 without/with millis & TZ
              - "yyyy-MM-dd'T'HH:mm:ssXXX"
              - "yyyy-MM-dd'T'HH:mm:ss.SSSXXX"
        destination: "@timestamp"
        source_timezone: "UTC"

  sink:
    - opensearch:
        hosts: ["https://opensearch:9200"]
        insecure: true
        username: admin
        password: "admin_pass"
        index_type: custom
        index: "network-logs-%{yyyy.MM.dd}"

{% include copy.html %}

다음 명령어로 이 파이프라인을 테스트할 수 있어요:

curl -sS -X POST "http://localhost:2021/logs" \
  -H "Content-Type: application/json" \
  -d '[
    {"message":"Oct 13 14:30:45 router1 sshd[1234]: User alice logged in from 10.0.0.5"},
    {"message":"2025-10-13T16:01:22Z 192.168.0.10 dhcpd: Lease granted to 192.168.0.55"},
    {"message":"Oct 13 16:30:45: LOCAL4-3-LINK_UPDOWN: Interface Gi0/1 changed state to up"}
  ]'

{% include copy.html %}

OpenSearch에 저장된 문서에는 다음 정보가 들어 있어요:

{
  ...
  "hits": {
    "total": {
      "value": 3,
      "relation": "eq"
    },
    "max_score": 1,
    "hits": [
      {
        "_index": "network-logs-2025.10.13",
        "_id": "-kzC3pkBl88jNjkRQ1TJ",
        "_score": 1,
        "_source": {
          "message": "Oct 13 14:30:45 router1 sshd[1234]: User alice logged in from 10.0.0.5",
          "parsed": {
            "host": "router1",
            "pid": "1234",
            "program": "sshd",
            "message": "User alice logged in from 10.0.0.5",
            "timestamp": "Oct 13 14:30:45"
          },
          "login_info": {
            "user": "alice",
            "source_ip": "10.0.0.5"
          },
          "@timestamp": "2025-10-13T14:30:45.000Z"
        }
      },
      {
        "_index": "network-logs-2025.10.13",
        "_id": "-0zC3pkBl88jNjkRQ1TJ",
        "_score": 1,
        "_source": {
          "message": "2025-10-13T16:01:22Z 192.168.0.10 dhcpd: Lease granted to 192.168.0.55",
          "parsed": {
            "host": "192.168.0.10",
            "program": "dhcpd",
            "message": "Lease granted to 192.168.0.55",
            "timestamp": "2025-10-13T16:01:22Z"
          },
          "login_info": {},
          "@timestamp": "2025-10-13T16:01:22.000Z"
        }
      },
      {
        "_index": "network-logs-2025.10.13",
        "_id": "_EzC3pkBl88jNjkRQ1TJ",
        "_score": 1,
        "_source": {
          "message": "Oct 13 16:30:45: LOCAL4-3-LINK_UPDOWN: Interface Gi0/1 changed state to up",
          "parsed": {
            "severity": "3",
            "mnemonic": "LINK_UPDOWN",
            "message": "Interface Gi0/1 changed state to up",
            "facility": "LOCAL4",
            "timestamp": "Oct 13 16:30:45"
          },
          "login_info": {},
          "@timestamp": "2025-10-13T16:30:45.000Z"
        }
      }
    ]
  }
}

메트릭

다음 표는 일반적인 Abstract processor 메트릭을 설명해요.

메트릭 이름 타입 설명
recordsIn Counter 파이프라인 컴포넌트로 레코드가 유입되는 것을 나타내는 메트릭이에요.
recordsOut Counter 파이프라인 컴포넌트에서 레코드가 나가는 것을 나타내는 메트릭이에요.
timeElapsed Timer 파이프라인 컴포넌트 실행 중 경과 시간을 나타내는 메트릭이에요.

Grok 프로세서에는 다음 사용자 지정 메트릭이 포함되어 있어요.

Counter

  • grokProcessingMismatch: match 필드에 지정된 어떤 패턴과도 일치하지 않은 레코드 수를 기록해요.
  • grokProcessingMatch: match 필드의 패턴 중 최소 하나와 일치한 레코드 수를 기록해요.
  • grokProcessingErrors: 레코드 처리 오류의 총 개수를 기록해요.
  • grokProcessingTimeouts: 매칭 중 시간이 초과된 레코드의 총 개수를 기록해요.

Timer

  • grokProcessingTime: 개별 레코드가 match 패턴과 매칭되는 데 걸리는 시간이에요. 이 타이머에서는 avg 메트릭이 가장 유용한데, 레코드 매칭에 걸린 평균 시간을 알려주기 때문이에요.

더 알아보기 (Learn more)

함께 읽는 실무 이야기

우리 조직에도 적용하고 싶다면, AI 시스템 구축 서비스를 살펴보세요.