Atlas Stream Processing 모니터링 및 알림

Atlas Stream Processing은 모니터링 및 경고 기능을 제공하여 사용자가 성능 및 상태 인사이트를 활용하여 워크플로를 개선할 수 있도록 합니다.

Atlas UI에서 스트림 프로세서 작업 공간 모니터

각 스트림 처리 작업 공간에서 Atlas UI를 통해 스트림 프로세서를 모니터링할 수 있습니다.

Atlas 에서 프로젝트 의 Stream Processing 페이지로 이동합니다.

아직 표시되지 않은 경우 탐색 표시줄의 Organizations 메뉴에서 프로젝트가 포함된 조직을 선택합니다.
아직 표시되지 않은 경우 내비게이션 바의 Projects 메뉴에서 프로젝트를 선택합니다.
사이드바에서 Streaming Data 제목 아래의 Stream Processing를 클릭합니다.

스트림 처리 페이지가 표시됩니다.

모니터링하려는 스트림 처리 작업영역을 선택합니다.

모니터하려는 스트림 처리 작업 공간의 패널에서 Manage를 클릭하세요.

Monitoring 탭 클릭합니다.

Monitoring 탭에는 선택한 스트림 프로세서에 대한 다양한 런타임 통계가 표시됩니다. 여기에는 다음이 포함되지만 이에 국한되지는 않습니다.

수집된 메시지 수
성공적으로 처리된 메시지 수
Dead Letter Queue에 전송된 메시지 수

소스 연결이 Apache Kafka인 경우, 주제의 파티션에 대한 브로커의 현재 오프셋과 최신 오프셋 간의 지연 및 모든 파티션 지연의 합계를 모니터링할 수 있습니다.

필요에 따라 모니터링 차트를 필터링하세요.

스트림 프로세서 이름, 시간 범위, 그리고 세분성에 따라 차트를 필터링할 수 있습니다.

스트림 프로세서 모니터링 방법

Atlas Stream Processing은 스트림 프로세서에 대한 주문형 보고를 위한 다음과 같은 방법을 제공합니다.

sp.processor.sample() 메서드를 사용하면 현재 실행 중인 선택한 스트림 프로세서에서 출력한 문서의 작은 샘플을 볼 수 있습니다. 사용자는 샘플링된 결과를 예상 결과와 비교하여 집계 파이프라인 설계의 오류를 진단할 수 있습니다.

sp.processor.stats() 메서드는 선택한 스트림 프로세서에 대한 런타임 통계를 반환합니다.

출력 문서에는 다음과 같은 필드가 있습니다.

필드	유형	설명
`ns`	문자열	스트림 프로세서가 정의된 네임스페이스입니다.
`stats`	객체	스트림 프로세서의 작동 상태를 설명하는 문서입니다.
`stats.name`	문자열	스트림 프로세서의 이름입니다.
`stats.status`	문자열	스트림 프로세서의 상태입니다. 이 필드는 다음과 같은 값을 가질 수 있습니다: `created` `running` `error` `stopping`
`stats.scaleFactor`	integer	크기 필드가 표시되는 배율입니다. `1` 로 설정하면 크기가 바이트 단위로 표시됩니다. `1024` 로 설정하면 크기가 킬로바이트 단위로 표시됩니다.
`stats.inputMessageCount`	integer	스트림에 게시된 문서 수입니다. 문서가 전체 파이프라인을 통과할 때가 아니라 `$source` 단계를 통과하면 스트림에 '게시된' 것으로 간주됩니다.
`stats.inputMessageSize`	integer	스트림에 게시된 바이트 또는 킬로바이트의 수입니다. 바이트는 전체 파이프라인을 통과할 때가 아니라 `$source` 단계를 통과한 후에 스트림에 '게시된' 것으로 간주됩니다.
`stats.outputMessageCount`	integer	스트림에서 처리한 문서 수입니다. 문서가 전체 파이프라인을 통과하면 스트림에서 '처리'된 것으로 간주됩니다.
`stats.outputMessageSize`	integer	스트림에서 처리한 바이트 또는 킬로바이트의 수입니다. 바이트는 전체 파이프라인을 통과하면 스트림에서 '처리'된 것으로 간주됩니다.
`stats.dlqMessageCount`	integer	데드 레터 대기열로 전송된 문서 수입니다.
`stats.dlqMessageSize`	integer	데드 레터 큐로 전송된 바이트 또는 킬로바이트의 수입니다.
`stats.changeStreamTimeDifferenceSecs`	integer	가장 최근의 변경 스트림 재개 토큰이 나타내는 이벤트 시간과 oplog의 최신 이벤트 간의 차이(초)입니다.
`stats.changeStreamState`	token	가장 최근의 변경 스트림 재개 토큰입니다. 변경 스트림 소스가 있는 스트림 프로세서에만 적용됩니다.
`stats.latency`	문서	스트림 프로세서 전체에 대한 지연 시간 통계입니다. Atlas Stream Processing은 `verbose` 옵션을 전달해야만 이 필드를 반환합니다.
`stats.latency.p50`	integer	지난 30초 동안 처리된 모든 문서의 추정 50번째 백분위 지연 시간입니다. 파이프라인에 창 단계가 포함된 경우 지연 시간 측정에는 창의 간격이 포함됩니다. 예를 들어 `$tumblingWindow` 단계의 간격이 5 분이면 지연 시간 측정에 이 5 분이 포함됩니다.
`stats.latency.p99`	integer	지난 30초 동안 처리된 모든 문서의 추정 99번째 백분위 지연 시간입니다. 파이프라인에 창 단계가 포함된 경우 지연 시간 측정에는 창의 간격이 포함됩니다. 예를 들어 `$tumblingWindow` 단계의 간격이 5 분이면 지연 시간 측정에 이 5 분이 포함됩니다.
`stats.latency.start`	datetime	가장 최근의 30초 측정 창이 시작된 실제 시간입니다.
`stats.latency.end`	datetime	가장 최근의 30초 측정 창이 종료된 실제 시간입니다.
`stats.latency.unit`	문자열	지연 시간이 측정되는 시간 단위입니다. 이 값은 항상 `microseconds`입니다.
`stats.latency.count`	integer	가장 최근의 30초 측정 창에서 스트림 프로세서가 처리한 문서의 수입니다.
`stats.latency.sum`	integer	가장 최근의 30 초 측정 창에서 수행된 모든 개별 지연 시간 측정값의 합계(마이크로초)입니다.
`stats.stateSize`	integer	Windows에서 프로세서 상태를 저장하는 데 사용하는 바이트 수입니다.
`stats.watermark`	날짜	현재 워터마크의 타임스탬프입니다.
`stats.operatorStats`	배열	프로세서 파이프라인의 각 연산자에 대한 통계입니다. Atlas Stream Processing은 `verbose` 옵션을 전달한 경우에만 이 필드를 반환합니다. `stats.operatorStats` 다양한 핵심 `stats` 필드의 연산자별 버전을 제공합니다. `stats.operatorStats.name` `stats.operatorStats.inputMessageCount` `stats.operatorStats.inputMessageSize` `stats.operatorStats.outputMessageCount` `stats.operatorStats.outputMessageSize` `stats.operatorStats.dlqMessageCount` `stats.operatorStats.dlqMessageSize` `stats.operatorStats.latency` `stats.operatorStats.stateSize` `stats.operatorStats` 에는 고유 필드가 포함되어 있습니다: `stats.operatorStats.maxMemoryUsage` `stats.operatorStats.executionTimeMillis` `stats.operatorStats` `verbose` 옵션을 전달하고 프로세서에 창 단계가 포함되어 있는 경우 다음 필드도 포함됩니다. `stats.minOpenWindowStartTime` `stats.maxOpenWindowStartTime` `stats.operatorStats` 특정 소스 및 싱크 연산자에 대해 다음 필드를 포함할 수도 있습니다. `stats.operatorStats.targetStats`
`stats.operatorStats.maxMemoryUsage`	integer	연산자의 최대 메모리 사용량(바이트 또는 킬로바이트)입니다.
`stats.operatorStats.executionTimeMillis`	integer	연산자의 총 실행 시간(밀리초).
`stats.minOpenWindowStartTime`	날짜	최소 열린 창의 시작 시간입니다. 이 값은 선택 사항입니다.
`stats.maxOpenWindowStartTime`	날짜	최대 열린 창의 시작 시간입니다. 이 값은 선택 사항입니다.
`stats.operatorStats.targetStats`	배열	특정 소스 및 싱크 연산자에 대한 대상별 통계입니다. 이 배열 의 각 요소는 단일 입력 또는 출력 대상(예: 입력 또는 출력 컬렉션 또는 Apache Kafka 주제) 을 나타내는 문서 입니다. 연산자 에 따라 각 문서 다음 필드의 하위 집합이 포함됩니다. Apache Kafka `$source` 또는 변경 스트림 소스와 같은 소스 연산자의 경우: MongoDB 대상은 `db` `coll`, Apache Kafka 대상은 `topic` 중 하나입니다. `inputMessageCount` `inputMessageSize` 싱크 연산자(예: MongoDB `$merge` 또는 `$emit` 싱크와 같은 싱크 연산자의 경우: MongoDB 대상은 `db` `coll`, Apache Kafka 대상은 `topic` 중 하나입니다. `outputMessageCount` `outputMessageSize` Atlas Stream Processing은 Apache Kafka `$source` 및 MongoDB `$merge`와 같은 일부 스트림 집계 단계에 대해서만 대상별 통계를 수집합니다. 최대 100개의 서로 다른 목표물에 대한 표적별 통계를 기록합니다. 그 이후로 스트림 프로세서는 `targetStats`에 새로운 항목을 추가하는 것을 멈추지만, 연산자별 통계 집계 통계는 계속 업데이트합니다.
`stats.kafkaPartitions`	배열	Apache Kafka 브로커 의 파티션에 대한 오프셋 정보입니다. `kafkaPartitions` 는 Apache Kafka 소스를 사용하는 연결에만 적용됩니다.
`stats.kafkaPartitions.partition`	integer	Apache Kafka 주제 파티션 번호입니다.
`stats.kafkaPartitions.currentOffset`	integer	지정된 파티션에 대해 스트림 프로세서가 있는 오프셋입니다. 이 값은 스트림 프로세서가 처리한 이전 오프셋에 `1` 을 더한 것과 같습니다.
`stats.kafkaPartitions.checkpointOffset`	integer	스트림 프로세서가 Apache Kafka 브로커에 마지막으로 커밋한 오프셋과 지정된 파티션의 체크포인트. 이 오프셋을 통한 모든 메시지는 마지막 체크포인트 에 기록됩니다.
`stats.kafkaPartitions.isIdle`	부울	파티션이 유휴 상태인지 여부를 나타내는 플래그입니다. 이 값의 기본값은 `false`입니다.

Stream Processing 로그

Atlas Stream Processing 두 가지 유형의 stream processing 작업 공간 활동 로그를 제공합니다.

주로 다음의 동작을 추적 운영 로그
개별 스트림 프로세서.
주로 인증 및 보안을 추적 감사 로그
stream processing 작업 공간 수준에서 활동.

Atlas Stream Processing Stream Processing 또는 감사 로그를 다운로드 하려면 다음 단계를 따르세요.

Atlas 에서 프로젝트 의 Stream Processing 페이지로 이동합니다.

아직 표시되지 않은 경우 탐색 표시줄의 Organizations 메뉴에서 프로젝트가 포함된 조직을 선택합니다.
아직 표시되지 않은 경우 내비게이션 바의 Projects 메뉴에서 프로젝트를 선택합니다.
사이드바에서 Streaming Data 제목 아래의 Stream Processing를 클릭합니다.

스트림 처리 페이지가 표시됩니다.

로그를 다운로드 할 스트림 처리 작업 공간의 창으로 이동하여 줄임표를 클릭합니다.

드롭다운 메뉴에서 Download Logs을 선택합니다.

모달 창 에서 다운로드 하려는 로그 유형을 선택합니다.

Stream processor 필드 에 로그를 다운로드 하려는 스트림 프로세서의 이름을 입력합니다. 모든 스트림 프로세서에 대한 로그를 다운로드 하려면 이 필드 비워 둡니다.

Time Period 드롭다운에서 로그를 다운로드 할 간격을 선택합니다.

버튼을 Download Logs 클릭합니다.

참고

로그를 쿼리할 수 있게 되려면 생성 후 최대 5분이 소요됩니다.

데드 레터 대기열

데드 레터 큐 의 문서를 검토하여 스트림 프로세서의 실패 모드를 분석할 수 있습니다. DLQ에 작성된 문서는 다음 형식을 딱랍니다.

{
  "processorName": "<stream-processor-name>",
  "instanceName": "<workspace-name>",
  "dlqTime": {
    "$date": "<datetime>"
  },
  "operatorName": "<stream-operator-name>",
  "errInfo": {
    "reason": "<error-description>"
  },
  "doc": {
    "_id": {
      "$oid": "<object-id>"
    },
    . . .
  },
  "_stream_meta": {
    "<stage-metadata>": {
      "<metadata-field>": "<metadata-value>",
      . . .
    },
    . . .
  }
}

DLQ 문서에는 다음 필드가 포함되어 있습니다:

필드	유형	조건부	설명
processorName	문자열	항상	문서가 시작되는 스트림 프로세서의 이름입니다.
instanceName	문자열	조건부	프로세서가 실행되는 Stream Processing 워크스페이스의 이름입니다. 프로젝트에 여러 개의 워크스페이스가 포함된 경우에만 작성됩니다.
dlqTime.$date	ISODate	항상	Atlas Stream Processing이 문서를 데드 레터 큐에 등록한 날짜 및 시간.
operatorName	문자열	항상	오류가 발생한 Atlas Stream Processing 집계 연산자 또는 단계.
errInfo.reason	문자열	항상	문서 처리 실패 사유.
doc	문서	항상	이 실패 이벤트의 결과로 DLQ에 쓰여진 전체 문서입니다.
_stream_meta	문서	항상	스트림 메타데이터가 포함된 문서입니다. 이 문서의 내용은 주어진 입력에 대한 $meta 단계의 출력과 동일합니다. 자세한 내용은 $meta 구문 참조를 참조하십시오.

타사 지표 통합

Atlas Stream Processing은 사용자가 사용자 지정 노리 또는 보기를 개발하지 않고도 Stream Processing 성능을 기록하고 분석할 수 있도록 하는 제3자 지표 통합을 지원합니다.

DataDog는 mongodb.atlas.stream_processing 접두사를 사용하여 Atlas Stream Processing 지표를 추적합니다. 통합을 설정하고 사용 가능한 지표를 확인하려면 DataDog 통합을 참조하세요.

OpenTelemetry(OTel) 수집기에 지표를 보내려면 OpenTelemetry(OTel)와 통합을 참조하세요.

스트림 프로세서 경고

Atlas Stream Processing은 프로세서의 상태가 변경되거나 프로세서가 다양한 수집 또는 출력 임계값을 충족할 때 경고를 발생시킵니다. 사용 가능한 Atlas Stream Processing 알림 목록을 보려면 Atlas Stream Processing 경고를 참조하세요. 경고 구성에 대한 자세한 내용은 경고 설정 구성을 참조하세요.

다음과 같은 방법으로 Atlas Stream Processing 경고를 대상으로 지정할 수 있습니다.

프로젝트 내의 모든 스트림 프로세서
구성된 조건자와 일치하는 스트림 처리 작업 공간 내의 모든 스트림 프로세서
이름이 구성된 술어와 일치하는 모든 스트림 프로세서

모든 스트림 프로세서가 아닌 다른 대상의 경우, 동일한 경고에 대해 여러 대상을 구성할 수 있습니다.

Atlas Stream Processing 기본값으로 Stream Processor State is failed 경고 구성합니다. 이는 프로젝트 수준 경고이므로 구성된 프로젝트 내의 모든 stream processing 작업 공간에서 실행되는 모든 스트림 프로세서에 적용됩니다.

돌아가기

Stream Processing 권한 조치

제한 사항