ApsaraDB for MongoDB:Capture alterações de dados em tempo real com change streams do MongoDB

O que é um change stream?

Um change stream converte eventos de alteração do banco de dados em um fluxo em tempo real. Os clientes podem assinar esse fluxo e receber notificações imediatas sempre que dados forem inseridos, atualizados ou excluídos. Os cenários típicos incluem:

• Sincronização de dados entre clusters: replique dados incrementalmente entre clusters MongoDB.

• Auditoria de operações: rastreie operações de alto risco, como a exclusão de um banco de dados ou coleção.

• Arquitetura orientada a eventos: envie alterações para sistemas downstream para análises em tempo real, atualizações de cache ou notificações.

Limites

Tipos de instância suportados: instância de replica set ou instância de sharded cluster.

Configurar change stream

cursor = db.getSiblingDB("test").myCollection.watch([], 
  {
    fullDocument: 'updateLookup'
  }
);
cursor.next();
              

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // Add split stage
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // Unit: seconds
    }
  }
})

Melhores práticas para change streams

1. Use pre-images e post-images com cautela:

   • Ativar fullDocumentBeforeChange (pre-image) e fullDocument (post-image) aumenta a sobrecarga de armazenamento (na coleção config.system.preimages) e a latência das requisições.

   • Ative esses recursos apenas quando sua aplicação realmente precisar do conteúdo completo do documento antes e depois de uma alteração.

2. Considerações importantes para implantações de sharded cluster:

   • Sempre crie listeners de change stream no mongos para garantir a ordenação global dos eventos.

   • Sob altas cargas de escrita, os change streams podem se tornar um gargalo, pois o mongos precisa ordenar e mesclar eventos provenientes dos shards.

   • Uma distribuição desigual de escritas entre os shards (por exemplo, devido a uma chave de shard mal projetada) aumenta significativamente a latência do change stream.

3. Evite updateLookup:

   • O updateLookup executa uma consulta findOne separada para cada evento de atualização, o que é ineficiente.

   • Em sharded clusters, as operações de moveChunk agravam ainda mais a latência do updateLookup.

4. Previna interrupções no change stream:

   • ⚠️ Os cenários a seguir tornam um cursor de change stream inválido (operationType: "invalidate") ou geram erros.

     • Atraso no consumidor downstream: o consumidor processa eventos mais lentamente do que são gerados, fazendo com que o resumeToken fique fora da janela do oplog.

     • **resumeToken inválido: uso de um resumeToken desatualizado cujo timestamp não está mais presente no oplog.

     • Impacto de failover: após um failover, o oplog do novo nó primário pode não conter o resumeToken original.

     • Alterações de metadados: operações como drop, rename e dropDatabase podem disparar um evento de invalidate.

     • Expiração da pre-image: definir expireAfterSeconds com um valor muito curto enquanto o consumo é lento causa perda de pre-images.

   • Estratégias de mitigação:

     • Monitore a latência do change stream.

     • Garanta que a janela do oplog seja suficientemente grande.

     • Implemente lógica robusta de tratamento de erros e recuperação: capture eventos de invalidate, registre o último resumeToken válido e recrie o listener.

     • Defina um valor razoável para expireAfterSeconds.

5. Estratégia de seleção de escopo:

   • Change stream único vs. múltiplos change streams no nível da coleção:

     • Stream único (nível de banco de dados/instância): menor sobrecarga de recursos (busca de oplog single-threaded), mas exige filtragem e despacho no downstream. Sob alto volume de eventos, o mongos pode se tornar um gargalo.

     • Múltiplos streams no nível da coleção: podem aproveitar a filtragem no lado do servidor para reduzir o tráfego de rede e oferecer melhor concorrência. No entanto, muitos streams aumentam a contenção por leituras do oplog e o consumo de recursos.

   • Recomendação: teste com base na sua carga de trabalho (volume de eventos, número de coleções) e escolha a abordagem ideal. Geralmente, use streams dedicados para um pequeno número de coleções altamente ativas e utilize streams no nível de banco de dados/instância com filtragem downstream para muitas coleções com baixa atividade.

   FAQ

   **1. Por que os slow logs de change stream sempre mostram COLLSCAN?**

   Esse comportamento é normal e esperado. Nenhuma otimização é necessária. Um cursor de change stream lê, em última instância, de local.oplog.rs (o único local que registra todas as modificações da instância). Essa coleção não possui índices, portanto, sempre realiza um COLLSCAN. Isso não pode ser evitado e não há margem para otimização.

   Se você usar COLLSCAN como palavra-chave para filtrar slow logs em busca de consultas a otimizar, filtre também a palavra-chave $changeStream.

   Investigue slow logs relacionados a change streams apenas se encontrar problemas de desempenho (por exemplo, aumento de latência no consumo downstream).

   2. Por que, em uma instância com shard, vejo cursores de change stream em shards diferentes do shard primário ao escutar uma coleção sem shard?

   Isso prepara o sistema para possíveis operações de shardCollection. Você pode converter uma coleção sem shard com um listener ativo em uma coleção com shard a qualquer momento, distribuindo seus dados por todos os shards (semelhante ao movePrimary). Criar cursores de change stream em outros shards antecipadamente lida com esse cenário. Normalmente, esses cursores em outros shards não retornam eventos de alteração e incorrem em sobrecarga mínima de desempenho.

   Da mesma forma, para lidar com potenciais operações de addShard/removeShard, quando você configura um listener em uma instância com shard, o mongos também cria os cursores correspondentes no Config Server.

   3. Por que o slow log de um cursor de change stream aparece no nó primário mesmo eu tendo especificado readPreference:secondary ao criar o cursor?

   Um cursor fica "fixado" em um nó específico após a criação e não migra automaticamente quando as funções dos nós mudam. Uma vez criado, um cursor de change stream permanece fixo em um nó mongod específico (primário ou secundário) e continua consumindo via getMore. Após eventos como troca de primário/secundário, redimensionamento de instância ou migração, um cursor originalmente em um secundário pode acabar no primário.

   Se você não quiser que essa carga persista no nó primário, limpe os cursores de change stream relevantes usando killCursors. Sua lógica de consumidor downstream pode então restaurar o cursor para um nó secundário usando o resumeToken e readPreference. Esta operação não causa perda ou interrupção de eventos.

   db.runCommand(
      {
        killCursors: <collection>,
        cursors: [ <cursor id1>, ... ], comment: <any>
      }
   )
   db.getSiblingDB("<testDB>").runCommand( { killCursors: "<testColl>", cursors: [NumberLong("2452840976689696187") ] } ) 

   4. Por que vejo muitos slow logs sobre change streams no mongos com duração de 1000 ms?

   2020-08-26T04:34:45.045+0000 I COMMAND [conn21283] command altconfig-b2b-perf.oplog command: getMore \{ getMore: 3513599116181216748, collection: "oplog", $db: "altconfig-b2b-perf", $clusterTime: { clusterTime: Timestamp(1598416483, 1), signature: { hash: BinData(0, EC3841EB1FB7A34F897688BB5983E32E2ADF6763), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } originatingCommand: \{ find: "oplog", filter: { timestamp: { $gte: new Date(1597895390009) } }, tailable: true, awaitData: true, $db: "altconfig-b2b-perf", $clusterTime: \{ clusterTime: Timestamp(1597895487, 1), signature: { hash: BinData(0, AAAE33D5688F935C70469CBDB8EB1F6882749A40), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } nShards:1 cursorid:3513599116181216748 numYields:0 nreturned:0 reslen:237 protocol:op_msg 1000ms

   Este é um comportamento normal em versões do MongoDB anteriores à 6.0. Indica um timeout de espera, não um gargalo de desempenho. Em arquiteturas de sharded cluster, o mongos cria cursores de change stream nos shards com tailable:true, awaitData:true e maxTimeMS:1000 para retornar o máximo possível de eventos de alteração dentro de 1000 ms.

   Nas versões principais anteriores à 6.0, o mongos registrava esses slow logs de 1000 ms, o que poderia enganar os usuários. O MongoDB otimizou esse comportamento. Para detalhes, consulte SERVER-50559.