MongoDB Change Streams: watch(), Olay Türleri, Resume Token ve Pre-Image

Ne işe yarar?

Belgeler change streams için ETL servisleri, platformlar arası senkronizasyon, ortak çalışma özellikleri ve bildirim servisleri gibi örnekler verir. Ortak fikir, verinin kalıcı hale geldiği anda başka bir sistemi haberdar etmektir. Örneğin yeni bir sipariş eklendiğinde e-posta kuyruğuna iş eklemek, ürün fiyatı değiştiğinde önbelleği temizlemek ya da silinen kayıtları bir arşiv sistemine aktarmak. Bunu uygulamanın her yazma noktasına kod ekleyerek yapmak yerine tek bir dinleyiciyle merkezi olarak yaparsınız.

Gereksinimler

• Dağıtım bir replica set ya da sharded cluster olmalıdır. Kurulum için MongoDB replica set kurulumu yazısına bakın.
• Depolama motoru WiredTiger olmalıdır; şifreli depolama (encryption at rest) kullanan dağıtımlar da desteklenir.
• Replica set protokol sürümü 1 (pv1) kullanılmalıdır.
• "majority" read concern desteğinin açık ya da kapalı olması change streams kullanımını etkilemez.
• Time series koleksiyonları change streams desteklemez, çünkü değişiklikleri belge düzeyinde izlemeyen optimize bir depolama biçimi kullanırlar.

Replica set'te change stream'i veri taşıyan herhangi bir üyeden açabilirsiniz; sharded cluster'da ise mongos üzerinden açmanız gerekir.

Üç kapsam: koleksiyon, veritabanı, dağıtım

İlk change stream'i açmak

Aşağıdaki örnek magaza.siparisler koleksiyonunu izler. Belgelerin önerdiği döngü kalıbı cursor.isClosed() ile cursor.tryNext() yöntemini birlikte kullanır; böylece döngü yalnızca cursor kapandığında ve son partide işlenecek olay kalmadığında biter. tryNext() beklemeden döner, yeni olay yoksa null verir; hasNext() ise bir sonraki olay gelene kadar bekler.

const izleyici = db.getSiblingDB("magaza").siparisler.watch()

while (!izleyici.isClosed()) {
   let olay = izleyici.tryNext()
   while (olay !== null) {
      printjson(olay)
      olay = izleyici.tryNext()
   }
}

Bu döngü çalışırken başka bir mongosh oturumundan siparisler koleksiyonuna belge eklerseniz ilk oturumda o eklemeye ait olay belgesi yazdırılır. mongosh dışındaki uygulamalarda aynı işi sürücünüzün watch yöntemiyle yaparsınız; cursor'ı kapatmazsanız ne kadar süre açık kalacağı dile göre değişir.

Change event belgesinin yapısı

Her olay belgesi işlemi ve yaptığı değişikliği anlatan alanlar içerir. En sık kullanılanlar şunlardır:

Olay türlerinden insert, update, replace, delete, drop, rename, dropDatabase ve invalidate varsayılan olarak gelir. MongoDB 6.0'dan beri create, createIndexes, dropIndexes, modify ve shardCollection gibi DDL olayları da vardır, ancak bunları görmek için cursor'ı showExpandedEvents: true seçeneğiyle açmanız gerekir. Belgelerin önemli bir notu: sunucu bir güncellemeyi, daha kısa ifade edilebiliyorsa replace olayı olarak döndürebilir. Güncellemeleri dinliyorsanız replace olaylarını da dinleyin.

Olayları süzmek ve biçimlendirmek

watch() ilk parametre olarak bir aggregation pipeline alır. Bu pipeline'da yalnızca belirli aşamalar kullanılabilir: $addFields, $match, $project, $replaceRoot, $replaceWith, $redact, $set ve $unset.

// Yalnızca yeni siparişler ve silmeler
const pipeline = [
   { $match: {
      $or: [
         { operationType: "insert", "fullDocument.tutar": { $gte: 1000 } },
         { operationType: "delete" }
      ]
   } },
   { $project: { operationType: 1, documentKey: 1, "fullDocument.tutar": 1 } }
]
const izleyici = db.siparisler.watch( pipeline )

Burada iki kurala dikkat edin. Birincisi, pipeline olayın _id alanını (resume token) değiştirirse change stream hata fırlatır; $project ile _id'yi çıkarmayın. İkincisi, süzme sunucu tarafında yapıldığı için uygulamanıza yalnızca ilgilendiğiniz olaylar gelir; dinleyici kodunuz her olayda tür kontrolü yapmak zorunda kalmaz.

Güncellenen belgenin tamamını almak

Varsayılan olarak bir update olayı yalnızca değişen alanların farkını (updateDescription) içerir. Belgenin tamamını istiyorsanız fullDocument: "updateLookup" seçeneğini verirsiniz; MongoDB bu durumda belgenin çoğunluk tarafından onaylanmış en güncel halini okur ve fullDocument alanına ekler.

const izleyici = db.siparisler.watch( [], { fullDocument: "updateLookup" } )

Bu okuma, değişikliğin yapıldığı anı değil, olayın işlendiği anı yansıtır. Arada başka yazmalar olduysa dönen belge güncelleme anındaki halinden çok farklı olabilir. Koleksiyon yeniden adlandırıldıysa belge dönmez; adlandırmadan sonra eski adla yeni bir koleksiyon oluşturulduysa arama yeni koleksiyonda yapılır. Ayrıca güncelleme farkı ile tam belge birlikte 16 MB BSON sınırını aşarsa sunucu cursor'ı kapatıp hata döndürür.

Pre ve post image: değişiklik öncesi ve sonrası

MongoDB 6.0'dan beri bir olayda belgenin değişiklik öncesi (pre-image) ve sonrası (post-image) hallerini alabilirsiniz. Pre-image, belgenin değiştirilmeden, güncellenmeden ya da silinmeden önceki halidir; eklenen belgenin pre-image'ı yoktur. Post-image ise ekleme, değiştirme veya güncellemeden sonraki haldir; silinen belgenin post-image'ı yoktur. Önce özelliği koleksiyon için açmanız gerekir:

db.runCommand( {
   collMod: "siparisler",
   changeStreamPreAndPostImages: { enabled: true }
} )

const izleyici = db.siparisler.watch( [], {
   fullDocumentBeforeChange: "whenAvailable",
   fullDocument: "whenAvailable"
} )

"whenAvailable" image varsa ekler, yoksa olayı imagesız verir; "required" image yoksa hata fırlatır. fullDocumentBeforeChange için varsayılan değer "off" değeridir. Pre-image'lar config.system.preimages koleksiyonuna yazılır ve bu koleksiyon büyüyebilir. Saklama süresini küme genelinde setClusterParameter komutuyla changeStreamOptions.preAndPostImages.expireAfterSeconds değerini vererek sınırlayabilirsiniz; "off" değeri, image'ların ilgili olay oplog'dan düşene kadar tutulması anlamına gelir. Oplog'dan düşen olayın image'ları da silinir.

use admin
db.runCommand( {
   setClusterParameter: { changeStreamOptions: { preAndPostImages: { expireAfterSeconds: 3600 } } }
} )

Bu özellik depolama alanı tüketir ve işlem süresi ekler; belgeler yalnızca gerçekten ihtiyaç varsa açmanızı önerir. Olay boyutunu 16 MiB altında tutmak için belgeler, pre ve post image'ı birlikte isteyecekseniz belge boyutunu 8 MB ile sınırlamayı önerir. Koleksiyon ayarlarını değiştirmeden önce yedek almayı unutmayın; collMod işlemi için MongoDB schema validation yazısında da örnekler var. MongoDB 9.0 ile pre-image temizleme işinin docsDeleted ve bytesDeleted sunucu metrikleri, bazı durumlarda kesin değer yerine tahmin raporlayabilir.

Kaldığınız yerden devam etmek: resume token

Uygulamanız yeniden başladığında ya da bağlantı koptuğunda olay kaçırmamak için son işlediğiniz olayın _id alanını (resume token) kalıcı bir yerde saklarsınız. Sonra change stream'i bu token ile açarsınız:

let izleyici = db.siparisler.watch()
let olay = izleyici.next()
const sonToken = olay._id   // bunu kalici bir yere kaydedin

// Daha sonra: o olaydan sonrasini almaya devam et
let devam = db.siparisler.watch( [], { resumeAfter: sonToken } )

Bu üç seçenek birbirini dışlar; aynı anda yalnızca birini verebilirsiniz. Her durumda oplog'da token'a ait işlemi bulacak kadar geçmiş olmalıdır; işlem oplog'dan düştüyse change stream devam edemez. Bu yüzden uygulamanızın en uzun kesinti süresini oplog boyutunu planlarken hesaba katın. Bir de uyarı: token ile devam ederken, token'ı ürettiğiniz pipeline ve seçeneklerin aynısını kullanın; farklı pipeline öngörülemeyen davranışa ve veri tutarlılığı sorunlarına yol açabilir.

Önemli bir fark: MongoDB sürücüleri belirli hatalardan sonra change stream'i otomatik olarak bir kez devam ettirmeyi dener, mongosh ise hata sonrası otomatik devam etmez. Resume token'lar iki türdür: belirli bir olayı gösteren olay token'ı ve olay olmasa da küme zamanının ilerlediğini gösteren highwatermark token. aggregate ve getMore komutlarının yanıtlarında bu token cursor.postBatchResumeToken alanında bulunur.

MongoDB 9.0: seçimlerden sonra read preference zorlaması

MongoDB 9.0 ile, readPreference değeri primary ya da secondary olan bir change stream, bir replica set seçimi cursor'ı sunan üyenin rolünü değiştirip read preference artık karşılanamaz hale gelirse getMore çağrısında devam ettirilebilir InterruptedDueToReplStateChange hatasını (kod 11602) döndürür. Önceki sürümlerde cursor aynı üyeden sonuç vermeye devam ediyordu. Uyumlu sürücüler ve mongos bu hatada son token ile otomatik devam eder; elle getMore döngüsü yazan uygulamalar ve mongosh ise hatayı yakalayıp resumeAfter ile devam etmelidir. primaryPreferred, secondaryPreferred ve nearest her zaman karşılanabildiği için etkilenmez. Read preference kavramı için MongoDB read concern ve read preference yazısına bakın.

Cursor ne zaman kapanır?

• Cursor açıkça kapatıldığında.
• Bir invalidate olayı olduğunda; örneğin izlenen koleksiyon silindiğinde ya da yeniden adlandırıldığında.
• MongoDB bağlantısı kapandığında ya da zaman aşımına uğradığında.
• Sharded cluster'da bir shard kaldırıldığında; bu durumda kapanan cursor tam olarak devam ettirilemeyebilir.

Garantiler, performans ve yetkiler

Change streams yalnızca replica set'teki veri taşıyan üyelerin çoğunluğuna kalıcı olarak yazılmış değişiklikler için bildirim üretir. Üç üyeli bir replica set'te primary'ye eklenen bir belge, çoğunluğa ulaşana kadar olay olarak görünmez. Bu sayede failover sonrası geri alınacak bir değişikliği dinleyicinize iletmezsiniz.

Performans tarafında her change stream, sıradaki olayı beklerken bir getMore işlemiyle bir bağlantıyı açık tutar. Bir veritabanına açık change stream sayısı bağlantı havuzu boyutunu aşarsa bildirimlerde gecikme yaşanabilir; havuz boyutunun açık change stream sayısından büyük olmasını sağlayın. maxAwaitTimeMS sunucunun boş bir parti döndürmeden önce ne kadar bekleyeceğini belirler (varsayılan 1000 ms); batchSize parti başına en fazla olay sayısını sınırlar. Sharded cluster'da mongos her shard üzerinde ayrı bir change stream açar, sonuçları sıralar ve süzer; belgeler change stream içinde $lookup kullanımını sınırlamanızı önerir.

Erişim denetimi açık bir dağıtımda bir koleksiyonu izlemek için kullanıcının o koleksiyonda find ve changeStream yetki eylemlerine sahip olması gerekir; yerleşik read rolü bu yetkileri verir. Veritabanı veya dağıtım düzeyinde izlemek için aynı eylemler ilgili tüm system dışı koleksiyonlarda gerekir. Ayrıntılı rol tanımı için MongoDB kullanıcı ve rol yönetimi yazısına bakın.

MongoDB 9.0 ile serverStatus her mongod ve mongos sürecinde change stream cursor metriklerini raporlar: açık cursor sayısı (metrics.changeStreams.cursor.open.total), aktif kullanılanlar (open.pinned), toplam açılan sayısı (totalOpened) ve kapanan cursor'ların ortalama ömrü (lifespan). currentOp ayrıca cursor'ın oplog'daki konumunu currentOp.cursor.changeStreams.optime alanında gösterir.

Olaydaki belgeyi güvenle sorgulamak

Bir olaydan yola çıkıp ilgili belgeyi başka bir koleksiyonda aramak istediğinizde documentKey alanını filtre olarak kullanabilirsiniz. Belgeler her değeri $eq içine sarmayı önerir: aksi halde düzenli ifade olan bir değer desen gibi, $ ile başlayan alan adı içeren gömülü bir belge de sorgu operatörü gibi yorumlanabilir.

const izleyici = db.siparisler.watch()
while (izleyici.hasNext()) {
   const olay = izleyici.next()
   if (!olay.documentKey) { continue }   // documentKey olmayan DDL olaylari
   const filtre = {}
   for (const [alan, deger] of Object.entries(olay.documentKey)) {
      filtre[alan] = { $eq: deger }
   }
   printjson( db.siparisArsivi.find( filtre ).toArray() )
}

Sık yapılan hatalar

• Standalone sunucuda denemek: Change streams replica set ister.
• Resume token saklamamak: Yeniden başlatmada aradaki olaylar kaybolur.
• Pipeline'da _id'yi değiştirmek: Change stream hata fırlatır.
• Yalnızca update dinlemek: Sunucu bazı güncellemeleri replace olarak verebilir.
• Oplog'u küçük tutmak: Uzun kesintiden sonra token oplog'dan düşmüş olabilir.
• mongosh'ta otomatik devam beklemek: mongosh hata sonrası kendiliğinden devam etmez.
• Çok sayıda change stream açmak: Her biri bir bağlantı tutar; havuz boyutunu buna göre ayarlayın.

Sık Sorulan Sorular

Change streams ile tetikleyici (trigger) aynı şey mi?

Change streams sunucu içinde kod çalıştırmaz; olayları uygulamanıza akıtır ve tepkiyi siz yazarsınız. Bu yüzden iş mantığı uygulamanızda kalır ve dinleyiciyi istediğiniz dille yazabilirsiniz.

Transaction içindeki değişiklikler nasıl görünür?

Transaction içindeki işlemler de olay üretir ve olay belgesi txnNumber ile lsid alanlarını içerir; böylece hangi olayların aynı transaction'a ait olduğunu ayırt edebilirsiniz. Transaction kullanımı için MongoDB transaction kullanımı yazısına bakın.

Belirli bir zamandan itibaren olayları alabilir miyim?

Evet, startAtOperationTime seçeneğiyle bir zaman damgası verirsiniz. Zaman geçmişteyse oplog'un kapsadığı aralıkta olmalıdır; oplog aralığını rs.printReplicationInfo() ile görebilirsiniz.

Change stream Node.js uygulamasından nasıl açılır?

Sürücüler aynı kavramları kendi watch yöntemleriyle sunar ve belirli hatalarda bir kez otomatik devam eder. Node.js ile ilgili genel bilgi için Node.js nedir yazısına bakabilirsiniz; sürücü API ayrıntıları için sürücünün kendi belgelerine bakın.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: MongoDB öğrenme rehberi · Replica set kurulumu · Aggregation pipeline · Read concern ve read preference · MongoDB 9.0 yenilikleri