✨ feat: added kafka producer (#3268)

* ✨ feat: added kafka producer Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * 🐛 fix: eslint warn Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * 🐛 fix: typings and auth problems Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * 🐛 fix: better variable name to trrack disconnection Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * 🐛 fix: grouping Kafka Producer special settings into one template Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * ✨ feat: add kafka producer translations into `en.json` Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * 🐛 fix: disable close-on-select on kafka broker picker Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * 🐛 fix: `en.json` invalid json (conflict resolve) Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> * Nostr dm notifications (#3051) * Add nostr DM notification provider * require crypto for node 18 compatibility * remove whitespace Co-authored-by: Frank Elsinga <frank@elsinga.de> * move closer to where it is used * simplify success or failure logic * don't clobber the non-alert msg * Update server/notification-providers/nostr.js Co-authored-by: Frank Elsinga <frank@elsinga.de> * polyfills required for node <= 18 * resolve linter warnings * missing comma --------- Co-authored-by: Frank Elsinga <frank@elsinga.de> * Drop nostr * Minor * Fix a bug of clone --------- Signed-off-by: Muhammed Hussein Karimi <info@karimi.dev> Co-authored-by: Frank Elsinga <frank@elsinga.de> Co-authored-by: Louis Lam <louislam@users.noreply.github.com>
2025-07-24 07:21:09 -04:00 · 2023-07-17 11:45:44 +03:30 · 2023-07-17 11:45:44 +03:30 · 278b88a9d9
commit 278b88a9d9
parent 084cf01fcd
9 changed files with 296 additions and 3 deletions
--- a/server/util-server.js
+++ b/server/util-server.js
@ -28,8 +28,11 @@ const {
 } = require("node-radius-utils");
 const dayjs = require("dayjs");

-const isWindows = process.platform === /^win/.test(process.platform);
+// SASLOptions used in JSDoc
+// eslint-disable-next-line no-unused-vars
+const { Kafka, SASLOptions } = require("kafkajs");

+const isWindows = process.platform === /^win/.test(process.platform);
 /**
 * Init or reset JWT secret
 * @returns {Promise<Bean>}
@ -196,6 +199,94 @@ exports.mqttAsync = function (hostname, topic, okMessage, options = {}) {
    });
 };

+/**
+ * Monitor Kafka using Producer
+ * @param {string} topic Topic name to produce into
+ * @param {string} message Message to produce
+ * @param {Object} [options={interval = 20, allowAutoTopicCreation = false, ssl = false, clientId = "Uptime-Kuma"}]
+ * Kafka client options. Contains ssl, clientId, allowAutoTopicCreation and
+ * interval (interval defaults to 20, allowAutoTopicCreation defaults to false, clientId defaults to "Uptime-Kuma"
+ * and ssl defaults to false)
+ * @param {string[]} brokers List of kafka brokers to connect, host and port joined by ':'
+ * @param {SASLOptions} [saslOptions={}] Options for kafka client Authentication (SASL) (defaults to
+ * {})
+ * @returns {Promise<string>}
+ */
+exports.kafkaProducerAsync = function (brokers, topic, message, options = {}, saslOptions = {}) {
+    return new Promise((resolve, reject) => {
+        const { interval = 20, allowAutoTopicCreation = false, ssl = false, clientId = "Uptime-Kuma" } = options;
+
+        let connectedToKafka = false;
+
+        const timeoutID = setTimeout(() => {
+            log.debug("kafkaProducer", "KafkaProducer timeout triggered");
+            connectedToKafka = true;
+            reject(new Error("Timeout"));
+        }, interval * 1000 * 0.8);
+
+        if (saslOptions.mechanism === "None") {
+            saslOptions = undefined;
+        }
+
+        let client = new Kafka({
+            brokers: brokers,
+            clientId: clientId,
+            sasl: saslOptions,
+            retry: {
+                retries: 0,
+            },
+            ssl: ssl,
+        });
+
+        let producer = client.producer({
+            allowAutoTopicCreation: allowAutoTopicCreation,
+            retry: {
+                retries: 0,
+            }
+        });
+
+        producer.connect().then(
+            () => {
+                try {
+                    producer.send({
+                        topic: topic,
+                        messages: [{
+                            value: message,
+                        }],
+                    });
+                    connectedToKafka = true;
+                    clearTimeout(timeoutID);
+                    resolve("Message sent successfully");
+                } catch (e) {
+                    connectedToKafka = true;
+                    producer.disconnect();
+                    clearTimeout(timeoutID);
+                    reject(new Error("Error sending message: " + e.message));
+                }
+            }
+        ).catch(
+            (e) => {
+                connectedToKafka = true;
+                producer.disconnect();
+                clearTimeout(timeoutID);
+                reject(new Error("Error in producer connection: " + e.message));
+            }
+        );
+
+        producer.on("producer.network.request_timeout", (_) => {
+            clearTimeout(timeoutID);
+            reject(new Error("producer.network.request_timeout"));
+        });
+
+        producer.on("producer.disconnect", (_) => {
+            if (!connectedToKafka) {
+                clearTimeout(timeoutID);
+                reject(new Error("producer.disconnect"));
+            }
+        });
+    });
+};
+
 /**
 * Use NTLM Auth for a http request.
 * @param {Object} options The http request options