NodeJs Üzerinde RabbitMQ Kullanarak MongoDB’de Audit Log Tutma
Selamlar,
https://youtu.be/JkhLPpJkO64
Bu makalede NodeJs üzerinde yapılan tüm Audit işlemleri, MongoDB’de RabbitMQ kullanarak loglayacağız. Örneğin, bir data güncellendiğinde bunu anlık olarak değil de, performansı arttırmak amacı ile bir RabbitMQ’ya atacağız. Daha sonra Microservices olarak çalışan başka bir NodeJs uygulaması, RabbitMQ’nun ilgili kanalını dinleyip, MongoDB’ye kaydedecek. Bu makale Angular ve NodeJs Üzerinde SocketIO Kullanılarak Real Time Data Bildirilmesi makalesinin devamıdır.
Gelin ilk önce bir kayıt güncellendiğinde, bu işlemi RabbitMQ’ya, daha sonradan işlenmek üzere atalım. RabbitMQ hakkında daha detaylı bilgiye, bu makaleden erişebilirsiniz.
Öncelikle NodeJS projesine, RabbitMQ ile işlem yapılabilmesi için “AMQP” kütüphanesi, aşağıdaki komut ile yüklenir.
|
1 |
npm install amqplib --save |
İlk önce RabbitMQ, Mac tarafında “rabbitmq-server” komutu ile ayağa kaldırılır.
Eğer herşey yolunda ise, “http://localhost:15672/” url satırından aşağıdaki gibi bir ekran ile karşılaşılması lazımdır.
Şimdi sıra geldi NodeJs üzerindeki service.js sayfasından, “/updatePeople” methodunun güncellenmesine. Amaç, güncellenen kaydın logunun MongoDB’de tutulmasıdır.
service.js: Öncelikle NodeJs üzerindeki service.js sayfasına, aşağıdaki gibi RabbitMQ’ya bağlanılıp, istenen bir paketin belirlenen bir queue’ya atabilecek function yazılır.
- “var amqp = require(‘amqplib/callback_api’)”: RabbitMq’ya ilgili paketin atılabilmesi için gerekli “amqplib” projeye eklenir.
- “const rabbitUrl = ‘amqp://localhost'” : RabbitMQ locale çalıştığı için, ilgili URL atanır.
- “function sendRabbitMQ(queueName, data) {” : RabbitMQ’ta atma işleminin yapıldığı functiondır. “queueName” : Atılacak channel, “data” : String olarak atılcak datayı belirler.
- “amqp.connect(‘amqp://localhost’, function (error0, connection) {: RabbitMQ’ya bağlanılır.
- “if (error0) { throw error0; }” : Hata var ise, fırlatılır.
- “connection.createChannel(function (error1, channel) {“: Kanal yaratılır.
- “if (error1) { throw error1; }” : Kanal yaratma işleminde, hata var ise dönülür.
- “var queue = queueName” : Datanın atılacağı queue, burada gelen string parametre ile atanır.
- “channel.assertQueue(queue, { durable: false })” : RabbitMQ’daki datanın hızdan kazanılması adına, disk’e yazılması engellenir. “durable: true” ==> yapılır ise, sunucu kapanıp açılsa dahi kuyruktaki paketler kaybolmaz.
- “channel.sendToQueue(queue, Buffer.from(data))” : Belirtilen channel’a güncellenen person data, json string olarak gönderilir.
- “setTimeout(function ()”: Yarım saniye sonra connection kapanır.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 |
. . var amqp = require('amqplib/callback_api'); const rabbitUrl = 'amqp://localhost'; function sendRabbitMQ(queueName, data) { amqp.connect('amqp://localhost', function (error0, connection) { if (error0) { throw error0; } connection.createChannel(function (error1, channel) { if (error1) { throw error1; } var queue = queueName; channel.assertQueue(queue, { durable: false }); channel.sendToQueue(queue, Buffer.from(data)); console.log(" [x] Sent %s", data); }); setTimeout(function () { connection.close(); }, 500); }); } |
service/updatePeople():
“sendRabbitMQ(“logChannel”, JSON.stringify(updatePerson))” : Tek eklenen satır burasıdır. Var olan bir person kaydı güncellendiği zaman, güncellene datanın RabitMQ’nın “logChannel”‘ına json olarak atıldığı kısım burasıdır. İstenir ise loglama işlem direk burada da yapılabilirdi. Ama amaç, sonucun client’a en hızlı şekilde dönülmesi durumlarında, ya da her kayıt işleminde data üzerinde bir takım vakit alacak işler yaplacak ise var olan sistemden bağımsız bu işleri yürütmek çok daha sağlıklıdır.
“Microservicesler, bir çeşit asenkron çalışan mekanizmalardır.”
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
app.post('/updatePeople', token, async (req, res) => { console.log("req.username : " + req.body.username); try { var updatePerson = new Human(req.body); const person = await Human.findOne({ username: updatePerson.username }); await person.updateOne(updatePerson); //Redis Update if (redisClient.connected) { var userName = 'User:' + updatePerson.username; var data = JSON.stringify(updatePerson); redisClient.set(userName, data, function (err, res) { }); } //-------------- /* return res.send("succesfully saved"); */ /* return true; */ //YENİ EKLENEN SATIR. ADD RABBITMQ sendRabbitMQ("logChannel", JSON.stringify(updatePerson)); //------------------- return res.status(200).json({ status: "succesfully update" }); } catch (error) { res.status(500).send(error); } }) |
RabbitMQ’ya atıldıktan sonra, “http://localhost:15672/#/queues” url satırına gidilince aşağıdaki gibi bir ekran ile karşılaşılır. Görüldüğü gibi “logChannel” Queue’lere eklenmiştir. Ve işlenmeyi bekleyen 1 paket bulunmaktadır. (Ready 1)
service/insertPeople():
“sendRabbitMQ(“logChannelInsert”, JSON.stringify(person))” : Ekelenen tek satır budur. RabbitMQ’nun “logChannel” kanalına yeni kaydedilen kayıt, atılır. Ve microservisler tarafından işlenmek amacı ile, sıraya konur.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
app.post('/insertPeople', token, async (req, res) => { console.log("req.body : " + req.body.username); console.log("req.body.last : " + req.body.name.last); try { var person = new Human(req.body); var result = await person.save(); //Redis Insert if (redisClient.connected) { var userName = 'User:' + req.body.username; var data = JSON.stringify(person); redisClient.set(userName, data, function (err, res) { }); } //-------------- sendRabbitMQ("logChannel", JSON.stringify(person)); res.send(result); } catch (error) { res.status(500).send(error); } }) |
Şimdi sıra geldi, NodeJS ile yeni bir RabbitMQ Consumer Microservice yazmaya.
consumer.js:
- Yeni bir consumer.js dosyası oluşturulur ve aşağıda görüldüğü gibi “amqplib” kütüphanesi RabbitMQ işlemleri için eklenir.
|
1 |
npm install amqplib --save |
- Express kütüphanesi aşağıdaki gibi eklenir.
|
1 |
npm install express --save |
- RabbitMQ ile listeden çekilen data, MongoDB’ye yazılması için eklenmesi gereken mongoose kütüphanesi aşağıdaki gibi projeye eklenir.
|
1 |
npm i mongoose |
- “amqp.connect(‘amqp://localhost’, function (error0, connection) {” : Local’de çalışan RabbitMQ’ya bağlanılır.
- “connection.createChannel(function (error1, channel) {” : Channel yaratılır.
- “var queue = ‘logChannel'”: Dinlenecek kuyruk, burada tanımlanır.
- “channel.assertQueue(queue, { durable: false })” : Queue alınan datanın, harddisk’e yazılması engellenir.
- “channel.consume(queue, function (data) {” : İlgili kanal dinlenir ve person data’sı, parametre olarak beklenir.
- “person = JSON.parse(data.content.toString())” : Çekilen Json string data, person Objesine cast edilir.
- “noAck: true” : Çekilen datanın RabbitMQ kuyruğundan, kaldırılması sağlanır.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 |
const app = require('express')(); var mongoose = require("mongoose"); var amqp = require('amqplib/callback_api'); amqp.connect('amqp://localhost', function (error0, connection) { if (error0) { throw error0; } connection.createChannel(function (error1, channel) { if (error1) { throw error1; } var queue = 'logChannel'; channel.assertQueue(queue, { durable: false }); console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue); channel.consume(queue, function (data) { person = JSON.parse(data.content.toString()) console.log(" [x] Received Person:", person.name.first + " - " + person.name.last); }, { noAck: true }); }); }); |
Şimdi sıra geldi, çekilen datanın mongoDB’ye yazılmasına.
consumer.js: Öncelikle localdeki, mongoDB’ye bağlanılır.
|
1 2 |
mongoose.connect('mongodb://localhost/people', { useNewUrlParser: true, useUnifiedTopology: true, useCreateIndex: true }); mongoose.set('useFindAndModify', false); |
Kaydedilecek “userslog” Scheması, aşağıdaki gibi oluşturulur. mongoDB’deki döküman karşılığı, “Userlog” değişkenine aktarılır.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
var userlogSchema = new mongoose.Schema({ gender: String, email: String, username: String, name: { title: String, first: String, last: String }, fullName: String }); var Userlog = mongoose.model('UserLog', userlogSchema, 'usersLog'); |
Kuyruktan gelen person kayıt, mongoDB’de “usersLog” dökümanına kaydedilir.
|
1 2 3 4 |
//Write MongoDB var personLog = new Userlog(person); var result = personLog.save(); //-------------------- |
consumer.js: Kodun son hali aşağıdaki gibidir.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
const app = require('express')(); var mongoose = require("mongoose"); var amqp = require('amqplib/callback_api'); mongoose.connect('mongodb://localhost/people', { useNewUrlParser: true, useUnifiedTopology: true, useCreateIndex: true }); mongoose.set('useFindAndModify', false); var userlogSchema = new mongoose.Schema({ gender: String, email: String, username: String, name: { title: String, first: String, last: String }, fullName: String }); var Userlog = mongoose.model('UserLog', userlogSchema, 'usersLog'); amqp.connect('amqp://localhost', function (error0, connection) { if (error0) { throw error0; } connection.createChannel(function (error1, channel) { if (error1) { throw error1; } var queue = 'logChannel'; channel.assertQueue(queue, { durable: false }); console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue); channel.consume(queue, function (data) { person = JSON.parse(data.content.toString()) //Write MongoDB var personLog = new Userlog(person); var result = personLog.save(); //-------------------- console.log(" [x] Received Person:", person.name.first + " - " + person.name.last); }, { noAck: true }); }); }); |
Bu makalede, NodeJs’üzerinde microservicesler ile işlem yapılan tüm data bir kuyruğa kondu. Amaç, üzerinde işlem yapılan herbir satırın yedeğinin alınması idi. Bunu da var olan sistemin üzerine yük bindirmeden, hatta gerektiğinde, araya girip data’nın üzerinde işlem yapılmasına olanak veren bir yöntem ile yapıldı. Bu makale serisinde, NodeJs üzerinde aynı bir puzzle’ın parçaları gibi birçok bileşeni çevresinde yapılandırıldı. Böylece test, debug ve refactoring işleri kolaylaştırıldı. Ayrıca dağıtık mimari ile, ekipte birden fazla kişinin farklı teknolojilerle çalışmasına imkan sağlandı.
Geldik bir makalenin daha sonuna. Yeni bir makalede görüşmek üzere hepinize hoşçakalın.
Source :
eline sağlık
Çok Teşekkür ederim.
mongodb de neden schema kullanmadan gelen datayi collection’a yazmadiniz, schema kullanmanin ve schema ya cast etmenin bu case de bir avantaji var mi?