Aggregate processor
출처: 문서
본문
Aggregate 프로세서
aggregate 프로세서는 identification_keys 값을 기준으로 이벤트를 그룹화해요. 그런 다음 각 그룹에 액션을 수행해서 불필요한 로그 볼륨을 줄이고, 시간이 지나면서 집계된 로그를 만들 수 있도록 도와줘요. 기존 액션을 사용하거나 Java 코드로 직접 커스텀 집계를 만들 수도 있어요.
설정
다음 표에서는 aggregate 프로세서를 설정할 때 사용할 수 있는 옵션을 설명해요.
| 옵션 | 필수 여부 | 타입 | 설명 |
|---|---|---|---|
identification_keys |
예 | List | 이벤트를 그룹화할 기준이 되는, 순서가 없는 목록이에요. 이 키들과 같은 값을 가진 이벤트는 같은 그룹에 속하게 돼요. 이벤트에 identification_keys 중 하나가 없으면 그 키의 값은 null로 간주돼요. identification_key는 최소 한 개 이상 필요해요(예: ["sourceIp", "destinationIp", "port"]). |
action |
예 | AggregateAction | 각 그룹에 대해 수행할 액션이에요. 사용 가능한 aggregate 액션 중 하나를 반드시 지정하거나, 직접 커스텀 aggregate 액션을 만들 수도 있어요. 현재 사용할 수 있는 액션은 remove_duplicates와 put_all이에요. 자세한 내용은 새로운 aggregate 액션 만들기를 참고하세요. |
group_duration |
아니요 | String | 그룹이 자동으로 종료되기까지 유지되어야 하는 시간이에요. ISO_8601 표기 문자열(예: "PT20.345S" 또는 "PT15M")과 초("60s"), 밀리초("1500ms")의 간단한 표기를 지원해요. 기본값은 180s예요. |
local_mode |
아니요 | Boolean | local_mode를 true로 설정하면 해시 함수를 사용해 identification_keys를 기준으로 특정 노드에 이벤트를 전달하는 대신, 각 OpenSearch Data Prepper 노드에서 로컬로 집계가 수행돼요. 기본값은 false예요. |
output_unaggregated_events |
아니요 | Boolean | true로 설정하면 집계되지 않은 이벤트가 파이프라인의 다음 프로세서나 싱크로 전달돼요. 기본값은 false예요. |
aggregated_events_tag |
아니요 | String | 집계되지 않은 이벤트와 구별할 수 있도록 집계된 이벤트에 추가하는 태그예요. output_unaggregated_events가 true일 때는 필수예요. |
aggregate_when |
아니요 | String | aggregate 프로세서가 이벤트를 처리할지 여부를 결정하는 조건식이에요. 조건이 false로 평가되면 이벤트는 집계 없이 전달돼요. |
acknowledge_on_conclude |
아니요 | Boolean | true로 설정하면 그룹이 종료될 때(group_duration 타임아웃에 도달하거나 액션에서 정의한 조건에 의해) 그룹의 이벤트 핸들을 해제해요. 이 경우 엔드 투 엔드 승인은 포기하게 되지만 재처리는 방지할 수 있어요. 기본값은 false예요. |
disable_group_acknowledgments |
아니요 | Boolean | true로 설정하면 그룹 승인을 비활성화해요. 기본값은 false예요. |
사용 가능한 애그리게이트 액션
다음 애그리게이트 액션을 사용해서 aggregate 프로세서가 각 그룹에서 이벤트를 처리하는 방식을 결정할 수 있어요.
remove_duplicates
remove_duplicates 액션은 그룹의 첫 번째 이벤트를 즉시 처리하고, 소스에서 첫 번째 이벤트와 중복되는 이벤트는 모두 버려요. 예를 들어 identification_keys: ["sourceIp", "destinationIp"]를 사용하는 경우:
remove_duplicates액션은 소스의 첫 번째 이벤트인{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 200 }을 처리해요.- OpenSearch Data Prepper는
sourceIp와destinationIp가 소스의 첫 번째 이벤트와 일치하기 때문에{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 1000 }이벤트를 버려요. remove_duplicates액션은 다음 이벤트인{ "sourceIp": "127.0.0.2", "destinationIp": "192.168.0.1", "bytes": 1000 }을 처리해요.sourceIp가 그룹의 첫 번째 이벤트와 다르기 때문에, Data Prepper는 이 이벤트를 기반으로 새 그룹을 만들어요.
put_all
put_all 액션은 Java의 Map.putAll과 비슷하게, 기존 키를 덮어쓰고 새 키를 추가하는 방식으로 같은 그룹에 속한 이벤트들을 결합해요. 이 액션은 결합된 이벤트를 구성하는 모든 이벤트를 버려요. 예를 들어 identification_keys: ["sourceIp", "destinationIp"]를 사용하면 put_all 액션이 다음 세 개의 이벤트를 처리해요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 200 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 1000 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "http_verb": "GET" }
그런 다음 액션이 이 이벤트들을 하나로 결합해요. 이후 파이프라인은 다음과 같은 결합된 이벤트를 사용해요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 200, "bytes": 1000, "http_verb": "GET" }
count
count 액션은 같은 그룹에 속한 이벤트의 개수를 세고, identification_keys 값과 그룹 내 이벤트 수를 나타내는 count를 담은 새 이벤트를 생성해요. 이 액션은 결합된 이벤트를 구성하는 모든 이벤트를 버려요.
다음 설정 옵션으로 프로세서를 커스터마이즈할 수 있어요:
count_key: count를 저장하는 데 사용되는 키예요. 기본 이름은aggr._count예요.start_time_key: 시작 시간을 저장하는 데 사용되는 키예요. 기본 이름은aggr._start_time이에요.end_time_key: 종료 시간을 저장하는 데 사용되는 키예요. 기본 이름은aggr._end_time이에요.metric_name:otel_metrics출력 형식을 사용할 때의 메트릭 이름이에요. 기본값은count예요.unique_keys: 고유한 값을 셀 키들의 목록이에요. 지정하면 count가 전체 이벤트 수가 아니라 이 키들의 고유한 조합 개수를 반영해요.output_format: 애그리게이트된 이벤트 형식이에요. 유효한 값은 다음과 같아요:otel_metrics(기본값):value필드에 그룹에 속한 이벤트 수가 담긴SUM타입의 OpenTelemetry 메트릭을 출력해요.raw:count_key필드를 count로,start_time_key필드를 애그리게이션 시작 시간으로 담은 JSON 객체를 생성해요.
예를 들어 identification_keys: ["sourceIp", "destinationIp"]를 사용하면 count 액션이 다음 이벤트를 세고 처리해요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 200 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 503 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 400 }
프로세서는 다음 이벤트를 생성해요:
{"isMonotonic":true,"unit":"1","aggregationTemporality":"AGGREGATION_TEMPORALITY_DELTA","kind":"SUM","name":"count","description":"Number of events","startTime":"2022-12-02T19:29:51.245358486Z","time":"2022-12-02T19:30:15.247799684Z","value":3.0,"sourceIp":"127.0.0.1","destinationIp":"192.168.0.1"}
histogram
histogram 액션은 같은 그룹에 속한 이벤트들을 애그리게이트하고, 설정된 key를 기준으로 identification_keys 값과 애그리게이트된 이벤트들의 히스토그램을 담은 새 이벤트를 생성해요. 히스토그램에는 이벤트 수, 합계, 버킷, 버킷별 개수가 들어가고, 선택적으로 key에 해당하는 값들의 최솟값과 최댓값도 포함돼요. 이 액션은 결합된 이벤트를 구성하는 모든 이벤트를 버려요.
다음 설정 옵션으로 프로세서를 커스터마이즈할 수 있어요:
key: 히스토그램이 생성되는 필드의 이름이에요.generated_key_prefix: 애그리게이트된 이벤트에서 생성되는 모든 필드가 사용하는key_prefix예요. 접두사가 있으면 히스토그램 이벤트의 이름이 이벤트의 필드 이름과 충돌하지 않아요.units:key에 있는 값들의 단위예요.record_minmax: 히스토그램이 애그리게이션에 있는 값들의 최솟값과 최댓값을 포함할지 여부를 나타내는 불리언 값이에요.buckets: 히스토그램의 버킷을 나타내는 버킷 목록(double타입의 값)이에요.metric_name:otel_metrics출력 형식을 사용할 때의 메트릭 이름이에요. 기본값은histogram이에요.output_format: 애그리게이트된 이벤트 형식이에요. 유효한 값은 다음과 같아요:otel_metrics(기본값): 버킷 경계와 각 버킷에 속한 값의 개수를 담은HISTOGRAM타입의 OpenTelemetry 메트릭을 출력해요.raw: 합계, count, 버킷, 버킷별 개수, 애그리게이션 시작 시간, duration, 그리고record_minmax가 활성화된 경우 최솟값과 최댓값을 담은 JSON 객체를 생성해요. 각 필드 이름은generated_key_prefix를 사용하기 때문에, 합계는 기본적으로aggr._sum이 돼요.
예를 들어 identification_keys: ["sourceIp", "destinationIp", "request"], key: latency, buckets: [0.0, 0.25, 0.5]를 사용하면 histogram 액션이 다음 이벤트들을 처리해요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "request" : "/index.html", "latency": 0.2 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "request" : "/index.html", "latency": 0.55}
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "request" : "/index.html", "latency": 0.25 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "request" : "/index.html", "latency": 0.15 }
그러면 프로세서는 다음 이벤트를 생성해요:
{"max":0.55,"kind":"HISTOGRAM","buckets":[{"min":-3.4028234663852886E38,"max":0.0,"count":0},{"min":0.0,"max":0.25,"count":2},{"min":0.25,"max":0.50,"count":1},{"min":0.50,"max":3.4028234663852886E38,"count":1}],"count":4,"bucketCountsList":[0,2,1,1],"description":"Histogram of latency in the events","sum":1.15,"unit":"seconds","aggregationTemporality":"AGGREGATION_TEMPORALITY_DELTA","min":0.15,"bucketCounts":4,"name":"histogram","startTime":"2022-12-14T06:43:40.848762215Z","explicitBoundsCount":3,"time":"2022-12-14T06:44:04.852564623Z","explicitBounds":[0.0,0.25,0.5],"request":"/index.html","sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "key": "latency"}
sum
sum 액션은 같은 그룹에 속한 모든 이벤트에 대해 설정된 key의 숫자 값을 합산하고, identification_keys 값과 총합을 담은 새 이벤트를 생성해요. 이 액션은 결합된 이벤트를 구성하는 모든 이벤트를 버려요.
다음 설정 옵션으로 프로세서를 커스터마이즈할 수 있어요:
key: 합산할 이벤트 내 필드의 이름이에요. 이 필드의 값은 숫자여야 해요. 필수예요.metric_name:otel_metrics출력 형식을 사용할 때의 메트릭 이름이에요. 기본값은sum이에요.count_key:raw출력 형식을 사용할 때 합산에 기여한 이벤트 수를 저장하는 데 사용되는 키예요. 기본 이름은aggr._count예요.output_format: 애그리게이트된 이벤트 형식이에요. 유효한 값은 다음과 같아요:otel_metrics(기본값):value필드에 총합이 담긴SUM타입의 OpenTelemetry 메트릭을 출력해요. 합산되는 값들이 음수가 아니라고 보장할 수 없기 때문에 이 메트릭은 non-monotonic이에요.raw:aggr._sum필드를 총합으로,count_key필드를 총합에 기여한 이벤트 수로,aggr._start_time필드를 애그리게이션 시작 시간으로 담은 JSON 객체를 생성해요.
예를 들어 identification_keys: ["sourceIp", "destinationIp"]와 key: bytes_out을 사용하면 sum 액션이 다음 이벤트들을 처리해요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes_out": 1234 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes_out": 4321 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes_out": 100 }
프로세서는 다음 이벤트를 생성해요:
{"isMonotonic":false,"unit":"1","aggregationTemporality":"AGGREGATION_TEMPORALITY_DELTA","kind":"SUM","name":"sum","description":"Sum of the events","startTime":"2022-12-02T19:29:51.245358486Z","time":"2022-12-02T19:30:15.247799684Z","value":5655.0,"sourceIp":"127.0.0.1","destinationIp":"192.168.0.1"}
rate_limiter
rate_limiter 액션은 초당 애그리게이트되는 이벤트 수를 제어해요. 기본적으로 rate_limiter는 설정된 허용 수보다 많은 이벤트를 받으면 aggregate 프로세서가 실행되지 못하게 막아요. when_exceeds 설정 옵션을 사용하면 rate_limiter를 트리거하는 이벤트 수를 다르게 설정할 수 있어요.
다음 설정 옵션으로 프로세서를 커스터마이즈할 수 있어요:
events_per_second: 초당 허용되는 이벤트 수예요.when_exceeds: 받은 이벤트 수가 초당 허용되는 이벤트 수보다 클 때rate_limiter가 어떤 동작을 할지 나타내요. 기본값은block이에요. 이 값은 초당 허용되는 최대 이벤트 수에 도달한 후 다음 초가 될 때까지 프로세서의 실행을 막아요. 또는drop옵션을 사용하면 그 초에 받은 초과 이벤트들을 버려요.
예를 들어 events_per_second가 1로, when_exceeds가 drop으로 설정되어 있으면, 1초 시간 구간 동안 다음 이벤트들을 받았을 때 액션이 이 이벤트들을 처리하려고 해요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 200 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 1000 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "http_verb": "GET" }
첫 번째 이벤트는 처리되지만, when_exceeds가 drop으로 설정되어 있어서 나머지 이벤트들은 버려져요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "status": 200 }
when_exceeds가 block으로 설정되어 있으면, 프로세서는 나머지 이벤트를 처리하기 전에 다음 초까지 잠시 멈춰요.
append
append 액션은 그룹 내 모든 이벤트에 걸쳐 지정된 키들의 값을 이어 붙여서 여러 이벤트를 하나의 이벤트로 결합해요. 값을 덮어쓰는 put_all과 달리, append는 지정된 키들의 모든 값을 리스트로 모아요.
다음 설정 옵션으로 프로세서를 커스터마이즈할 수 있어요:
keys_to_append: 그룹 내 이벤트에 걸쳐 값을 이어 붙일 키들의 목록이에요. 애그리게이트된 이벤트에는 각 키에 대해 만난 모든 값들이 리스트로 들어가요.
예를 들어 identification_keys: ["sourceIp"]와 keys_to_append: ["status"]를 사용하면 append 액션이 다음 이벤트들을 처리해요:
{ "sourceIp": "127.0.0.1", "status": 200 }
{ "sourceIp": "127.0.0.1", "status": 503 }
{ "sourceIp": "127.0.0.1", "status": 400 }
프로세서는 다음 이벤트를 생성해요:
{ "sourceIp": "127.0.0.1", "status": [200, 503, 400] }
tail_sampler
tail_sampler 액션은 그룹 지속 시간 내에서 한 트레이스의 모든 스팬을 수집한 후 OpenTelemetry 트레이스를 샘플링해요. 성공한 트레이스 중 일부 퍼센트를 샘플링하는 동시에 모든 에러 트레이스는 유지할 수 있어서, 에러 트레이스를 모두 보존하면서 스토리지 사용량을 줄일 수 있어요.
다음 설정 옵션으로 프로세서를 커스터마이즈할 수 있어요:
wait_period: 트레이스가 완료됐다고 간주하기 전에 기다리는 시간이에요. 0보다 크고 60초를 넘지 않아야 해요. 필수예요.percent: 샘플링할 에러가 아닌 트레이스의 퍼센트예요(0--100, 경계값 제외). 모든 에러 트레이스는 항상 유지돼요. 필수예요.condition: 이벤트가 에러 이벤트인지 판단하는 조건식이에요. 이 조건과 일치하는 이벤트는percent설정과 관계없이 항상 유지돼요.
예를 들어 identification_keys: ["traceId"], wait_period: "10s", percent: 20, condition: '/status_code == 2'를 사용하면, tail_sampler 액션은 status_code == 2(에러)인 스팬을 하나 이상 포함하는 트레이스는 모두 유지하고, 나머지 성공 트레이스 중 20%를 샘플링해요.
percent_sampler
percent_sampler 액션은 이벤트 퍼센트를 기준으로 애그리게이트되는 이벤트 수를 제어해요. 이 액션은 퍼센트에 포함되지 않는 이벤트는 모두 버려요.
percent 설정을 사용해서 이벤트 퍼센트를 지정할 수 있는데, 이 값은 1초 구간 동안 처리되는 이벤트의 퍼센트를 나타내요(0%--100%).
예를 들어 percent가 50으로 설정되어 있으면, 액션은 1초 구간에서 다음 이벤트들을 처리하려고 해요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 2500 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 500 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 1000 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 3100 }
파이프라인은 이벤트의 50%를 처리하고 나머지 이벤트들은 버리며, 새 이벤트는 생성하지 않아요:
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 500 }
{ "sourceIp": "127.0.0.1", "destinationIp": "192.168.0.1", "bytes": 3100 }
메트릭
다음 표는 공통 Abstract processor 메트릭을 설명해요.
| 메트릭 이름 | 타입 | 설명 |
|---|---|---|
recordsIn |
Counter | 파이프라인 컴포넌트로 레코드가 유입되는 것을 나타내는 메트릭이에요. |
recordsOut |
Counter | 파이프라인 컴포넌트에서 레코드가 나가는 것을 나타내는 메트릭이에요. |
timeElapsed |
Timer | 파이프라인 컴포넌트 실행 중 경과된 시간을 나타내는 메트릭이에요. |
aggregate 프로세서에는 다음 커스텀 메트릭이 포함되어 있어요.
Counter
actionHandleEventsOut: 설정된 action으로의handleEvent호출에서 반환된 이벤트 수예요.actionHandleEventsDropped: 설정된 action으로의handleEvent호출에서 반환되지 않은 이벤트 수예요.actionHandleEventsProcessingErrors: 설정된 action의handleEvent호출 중 에러가 발생한 호출 수예요.actionConcludeGroupEventsOut: 설정된 action으로의concludeGroup호출에서 반환된 이벤트 수예요.actionConcludeGroupEventsDropped: 설정된 action으로의concludeGroup호출에서 반환되지 않은 이벤트 수예요.actionConcludeGroupEventsProcessingErrors: 설정된 action의concludeGroup호출 중 에러가 발생한 호출 수예요.
Gauge
currentAggregateGroups: 이 게이지는 현재 활성 상태인 애그리게이트 그룹의 수를 나타내요. 애그리게이트 그룹이 완료되어 결과가 출력되면 값이 감소하고, 새 이벤트가 새 애그리게이트 그룹 생성을 시작하면 값이 증가해요.