Monitoramento e alertas de Atlas Stream Processing

O Atlas Stream Processing fornece monitoramento e alertas para que os usuários possam aproveitar as percepções de desempenho e status para refinar seus fluxos de trabalho.

Monitore espaços de trabalho do processador de fluxo na IU do Atlas

Para cada um dos seus espaços de trabalho de processamento de fluxos, você pode monitorar seus processadores de fluxo na IU do Atlas:

No Atlas, vá para a página Stream Processing do seu projeto.

Se ainda não tiver sido exibido, selecione a organização que contém seu projeto no menu Organizations na barra de navegação.
Se ainda não estiver exibido, selecione seu projeto no menu Projects na barra de navegação.
Na barra lateral, clique em Stream Processing sob o título Streaming Data.

A página Processamento de fluxo é exibida.

Selecione o espaço de trabalho do processamento de fluxos que você deseja monitorar.

No painel do espaço de trabalho do processamento de fluxos que você deseja monitorar, clique em Manage.

Clique na Monitoring aba.

A aba Monitoring exibe uma variedade de estatísticas de tempo de execução sobre um processador de fluxos de sua escolha, incluindo, mas não se limitando a:

Número de mensagens ingeridas
Número de mensagens processadas com êxito
Número de mensagens enviadas para sua fila de Dead Letter

Se a sua conexão de fonte é Apache Kafka, você pode monitorar o atraso entre o deslocamento atual e o deslocamento mais recente no broker para a partição de um tópico e a soma de todos os atrasos das partições.

Filtre os gráficos de monitoramento conforme necessário.

Você pode filtrar os gráficos por nome do processador de fluxos, intervalo de tempo e granularidade.

Métodos de monitoramento do processador de fluxo

O Atlas Stream Processing fornece os seguintes métodos para geração de relatórios sob demanda sobre seus processadores de fluxo:

O método sp.processor.sample() permite que você veja uma pequena amostra da saída de documentos por um processador de fluxo em execução no momento de sua escolha. Os usuários podem comparar os resultados amostrados com os resultados esperados para diagnosticar erros no design do pipeline de agregação.

O método sp.processor.stats() retorna estatísticas de tempo de execução sobre um processador de stream de sua escolha.

O documento de saída tem os seguintes campos:

Campo	Tipo	Descrição
`ns`	string	O namespace no qual o processador de stream está definido.
`stats`	objeto	Um documento que descreve o estado operacional do processador de fluxo.
`stats.name`	string	O nome do processador de fluxo.
`stats.status`	string	O status do processador de fluxo. Este campo pode ter os seguintes valores: `created` `running` `error` `stopping`
`stats.scaleFactor`	inteiro	A escala na qual o campo de tamanho é exibido. Se definido como `1`, os tamanhos são exibidos em bytes. Se definido como `1024` , os tamanhos são exibidos em kilobytes.
`stats.inputMessageCount`	inteiro	O número de documentos publicados no stream. Um documento é considerado "publicado" no fluxo quando passa pelo estágio `$source` , não quando passa por todo o pipeline.
`stats.inputMessageSize`	inteiro	O número de bytes ou kilobytes publicados no stream. Os bytes são considerados "publicados" no fluxo quando passam pelo estágio `$source` , não quando ele passa por todo o pipeline.
`stats.outputMessageCount`	inteiro	O número de documentos processados pelo fluxo. Um documento é considerado "processado" pelo fluxo quando passa por todo o pipeline.
`stats.outputMessageSize`	inteiro	O número de bytes ou kilobytes processados pelo stream. Os bytes são considerados "processados" pelo fluxo quando passam por todo o pipeline.
`stats.dlqMessageCount`	inteiro	O número de documento enviados para a fila de mensagens não entregues (DLQ).
`stats.dlqMessageSize`	inteiro	O número de bytes ou kilobytes enviados para a fila de mensagens não entregues (DLQ).
`stats.changeStreamTimeDifferenceSecs`	inteiro	A diferença, em segundos, entre o tempo do evento representado pelo token de retomada do fluxo de alterações mais recente e o evento mais recente no oplog.
`stats.changeStreamState`	token	O token de retomada do fluxo de alterações mais recente. Só se aplica a processadores de fluxo com uma fonte do fluxo de alterações.
`stats.latency`	documento	Estatísticas de latência para o processador de fluxo como um todo. O Atlas Stream Processing retornará este campo somente se você fornecer a opção `verbose`.
`stats.latency.p50`	inteiro	A latência estimada do 50º percentil de todos os documentos processados nos últimos 30 segundos. Se o seu pipeline incluir um estágio de janela, as medições de latência incluirão o intervalo da janela. Por exemplo, se sua etapa `$tumblingWindow` tiver um intervalo de 5 minutos, as medições de latência incluirão esses 5 minutos.
`stats.latency.p99`	inteiro	A latência estimada do 99º percentil de todos os documentos processados nos últimos 30 segundos. Se o seu pipeline incluir um estágio de janela, as medições de latência incluirão o intervalo da janela. Por exemplo, se sua etapa `$tumblingWindow` tiver um intervalo de 5 minutos, as medições de latência incluirão esses 5 minutos.
`stats.latency.start`	datetime	Tempo decorrido real em que o período de medição de 30 segundos mais recente começou.
`stats.latency.end`	datetime	Tempo decorrido real em que o período de medição mais recente de 30 segundos terminou.
`stats.latency.unit`	string	Unidade de tempo usada para medir a latência. Este valor é sempre expresso em `microseconds`.
`stats.latency.count`	inteiro	Número de documentos que o processador de fluxo processou no período de medição mais recente de 30 segundos.
`stats.latency.sum`	inteiro	Soma de todas as medições de latência individuais, em microssegundos, realizadas na janela de medição mais recente de 30 segundos.
`stats.stateSize`	inteiro	O número de bytes usados pelo Windows para armazenar o estado do processador.
`stats.watermark`	data	O carimbo de data/hora da marca d'Água atual.
`stats.operatorStats`	array	As estatísticas de cada operador no pipeline do processador. O Atlas Stream Processing retorna este campo somente se você passar a opção `verbose` . `stats.operatorStats` fornece versões por operador de muitos campos principais do `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` inclui os seguintes campos exclusivos: `stats.operatorStats.maxMemoryUsage` `stats.operatorStats.executionTimeMillis` `stats.operatorStats` também inclui os seguintes campos quando você passa a opção `verbose` e seu processador inclui um estágio de janela: `stats.minOpenWindowStartTime` `stats.maxOpenWindowStartTime` `stats.operatorStats` Também pode incluir o seguinte campo para determinados operadores de origem e destino: `stats.operatorStats.targetStats`
`stats.operatorStats.maxMemoryUsage`	inteiro	O uso máximo de memória do operador em bytes ou kilobytes.
`stats.operatorStats.executionTimeMillis`	inteiro	O tempo total de execução do operador em milissegundos.
`stats.minOpenWindowStartTime`	data	O horário de início da janela mínima aberta. Este valor é opcional.
`stats.maxOpenWindowStartTime`	data	O horário de início da janela de tempo máxima aberta. Este valor é opcional.
`stats.operatorStats.targetStats`	array	As estatísticas por destino para determinados operadores de origem e coletor. Cada elemento dessa array é um documento que representa um único destino de entrada ou saída, como uma coleção de entrada ou saída ou um tópico do Apache Kafka. Dependendo do operador, cada documento contém um subconjunto dos seguintes campos: Para operadores de origem, como Apache Kafka `$source` ou fontes de fluxo de alterações: Para destinos MongoDB, pode-se usar `db` ou `coll`, ou `topic` para destinos do Apache Kafka. `inputMessageCount` `inputMessageSize` Para operadores de coleta, como o MongoDB `$merge` ou `$emit` coletores: Para destinos MongoDB, pode-se usar `db` ou `coll`, ou `topic` para destinos do Apache Kafka. `outputMessageCount` `outputMessageSize` O Atlas Stream Processing coleta estatísticas por alvo apenas para algumas etapas de agregação de fluxos, como Apache Kafka `$source` e MongoDB `$merge`. Ele faz o registro de estatísticas por destino para no máximo 100 destinos distintos; depois disso, o processador de fluxo para de adicionar novas entradas a `targetStats`, mas continua atualizando as estatísticas agregadas por operador.
`stats.kafkaPartitions`	array	Informações de deslocamento para um Apache Kafka partição do banking. `kafkaPartitions` aplica-se apenas a conexões que usam um Apache Kafka fonte.
`stats.kafkaPartitions.partition`	inteiro	O número da partição do tópico do Apache Kafka.
`stats.kafkaPartitions.currentOffset`	inteiro	O deslocamento em que o processador de fluxo está ativado para a partição especificada. Esse valor é igual ao offset anterior que o processador de stream processou mais `1`.
`stats.kafkaPartitions.checkpointOffset`	inteiro	O deslocamento que o processador de fluxo confirmou pela última vez ao agente Apache Kafka e o checkpoint da partição especificada. Todas as mensagens através deste deslocamento são registradas no último checkpoint.
`stats.kafkaPartitions.isIdle`	booleano	O sinalizador que indica se a partição está inativa. O valor padrão é `false`.

Logs de Stream Processing

O Atlas Stream Processing fornece dois tipos de logs de atividades do workspace de stream processing:

Logs operacionais, que acompanham principalmente o comportamento de
processadores de stream individuais.
Logs de auditar, que acompanham principalmente autenticação e segurança
atividade no nível do espaço de trabalho de stream processing.

Para baixar logs operacionais ou logs de auditar do Atlas Stream Processing:

No Atlas, vá para a página Stream Processing do seu projeto.

Se ainda não tiver sido exibido, selecione a organização que contém seu projeto no menu Organizations na barra de navegação.
Se ainda não estiver exibido, selecione seu projeto no menu Projects na barra de navegação.
Na barra lateral, clique em Stream Processing sob o título Streaming Data.

A página Processamento de fluxo é exibida.

Navegue até o painel do espaço de trabalho de stream processing do qual deseja baixar os logs e clique nas reticências.

No menu suspenso, selecione Download Logs.

Na janela modal, selecione o tipo de log que deseja baixar.

No campo Stream processor, forneça o nome do processador de stream para o qual você deseja baixar os logs. Deixe este campo em branco para baixar logs para todos os processadores de stream.

Na lista suspensa Time Period, selecione o intervalo para o qual deseja baixar os logs.

Clique no botão Download Logs.

Observação

Os logs levam até cinco minutos após a criação para se tornarem consultáveis.

dead letter queue (DLQ)

Você pode revisar documentos em sua fila de mensagens não entregues (DLQ) para analisar modos de falha em seus processadores de fluxo. Os documentos gravados em seu DLQ assumem o seguinte formato:

{
  "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>",
      . . .
    },
    . . .
  }
}

Os documentos DLQ contêm os seguintes campos:

Campo	Tipo	Condicionalidade	Descrição
processorName	string	Sempre	Nome do processador de fluxo do qual o documento se origina.
instanceName	string	Condicional	Nome do Stream Processing workspace no qual o processador executa. Só será gravado se o seu projeto contiver vários workspaces.
dlqTime.$date	Data ISO	Sempre	Data e hora em que o Atlas Stream Processing registrou o documento na fila de mensagens não entregues (DLQ).
operatorName	string	Sempre	Atlas Stream Processing operador de agregação ou estágio em que o erro ocorreu.
errInfo.reason	string	Sempre	Motivo da falha em processamento do documento.
doc	documento	Sempre	O documento completo gravado no DLQ como resultado deste evento de falha.
_stream_meta	documento	Sempre	Documento contendo metadados de fluxo. O conteúdo deste documento é idêntico à saída do estágio $meta para uma determinada entrada. Para obter mais detalhes, consulte a referência de sintaxe $meta.

Integrações de métricas de terceiros

O Atlas Stream Processing oferece suporte a integrações de métricas de terceiros, permitindo que os usuários registrem e analisem o desempenho do processador de stream sem precisar desenvolver lógica ou exibições personalizadas.

O Datadog acompanha as métricas do Atlas Stream Processing com o prefixo mongodb.atlas.stream_processing. Para configurar a integração e visualizar as métricas disponíveis, consulte Integração com Datadog.

Para enviar métricas para um coletor OpenTelemetry (OTel), consulte Integração com OpenTelemetry (OTel).

Alertas do processador de fluxo

O Atlas Stream Processing aciona alertas quando os processadores mudam de estado ou quando um processador atinge vários limites de ingestão ou saída. Para obter uma lista de alertas disponíveis do Atlas Stream Processing, consulte Alertas do Atlas Stream Processing. Para saber mais sobre a configuração de alertas, consulte Definir configurações de alerta.

Você pode direcionar os alertas do Atlas Stream Processing das seguintes maneiras:

Todos os processadores de fluxo dentro de um projeto
Todos os processadores do processamento de fluxos dentro de um espaço de trabalho do processamento de fluxos que correspondam ao predicado configurado
Todos os processadores de fluxo cujos nomes correspondem ao predicado configurado

Para destinos que não sejam todos os processadores de fluxo, é possível configurar vários destinos para o mesmo alerta.

O Atlas Stream Processing configura um alerta Stream Processor State is failed por padrão. Como este é um alerta de nível de projeto, ele se aplica a qualquer processador de fluxo em execução em qualquer espaço de trabalho de stream processing dentro do projeto para o qual está configurado.

Voltar

Ações de privilégio de processamento de fluxo

Limitações