feat: add `RabbitMQ` monitor (#5199)
Co-authored-by: Frank Elsinga <frank@elsinga.de>pull/5181/head^2
Suven-p
1 month ago
committed by
GitHub
parent
a7e9bdd43e
commit
c01494ec33
@ -0,0 +1,17 @@
|
|||||||
|
exports.up = function (knex) {
|
||||||
|
return knex.schema.alterTable("monitor", function (table) {
|
||||||
|
table.text("rabbitmq_nodes");
|
||||||
|
table.string("rabbitmq_username");
|
||||||
|
table.string("rabbitmq_password");
|
||||||
|
});
|
||||||
|
|
||||||
|
};
|
||||||
|
|
||||||
|
exports.down = function (knex) {
|
||||||
|
return knex.schema.alterTable("monitor", function (table) {
|
||||||
|
table.dropColumn("rabbitmq_nodes");
|
||||||
|
table.dropColumn("rabbitmq_username");
|
||||||
|
table.dropColumn("rabbitmq_password");
|
||||||
|
});
|
||||||
|
|
||||||
|
};
|
||||||
@ -0,0 +1,67 @@
|
|||||||
|
const { MonitorType } = require("./monitor-type");
|
||||||
|
const { log, UP, DOWN } = require("../../src/util");
|
||||||
|
const { axiosAbortSignal } = require("../util-server");
|
||||||
|
const axios = require("axios");
|
||||||
|
|
||||||
|
class RabbitMqMonitorType extends MonitorType {
|
||||||
|
name = "rabbitmq";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @inheritdoc
|
||||||
|
*/
|
||||||
|
async check(monitor, heartbeat, server) {
|
||||||
|
let baseUrls = [];
|
||||||
|
try {
|
||||||
|
baseUrls = JSON.parse(monitor.rabbitmqNodes);
|
||||||
|
} catch (error) {
|
||||||
|
throw new Error("Invalid RabbitMQ Nodes");
|
||||||
|
}
|
||||||
|
|
||||||
|
heartbeat.status = DOWN;
|
||||||
|
for (let baseUrl of baseUrls) {
|
||||||
|
try {
|
||||||
|
// Without a trailing slash, path in baseUrl will be removed. https://example.com/api -> https://example.com
|
||||||
|
if ( !baseUrl.endsWith("/") ) {
|
||||||
|
baseUrl += "/";
|
||||||
|
}
|
||||||
|
const options = {
|
||||||
|
// Do not start with slash, it will strip the trailing slash from baseUrl
|
||||||
|
url: new URL("api/health/checks/alarms/", baseUrl).href,
|
||||||
|
method: "get",
|
||||||
|
timeout: monitor.timeout * 1000,
|
||||||
|
headers: {
|
||||||
|
"Accept": "application/json",
|
||||||
|
"Authorization": "Basic " + Buffer.from(`${monitor.rabbitmqUsername || ""}:${monitor.rabbitmqPassword || ""}`).toString("base64"),
|
||||||
|
},
|
||||||
|
signal: axiosAbortSignal((monitor.timeout + 10) * 1000),
|
||||||
|
// Capture reason for 503 status
|
||||||
|
validateStatus: (status) => status === 200 || status === 503,
|
||||||
|
};
|
||||||
|
log.debug("monitor", `[${monitor.name}] Axios Request: ${JSON.stringify(options)}`);
|
||||||
|
const res = await axios.request(options);
|
||||||
|
log.debug("monitor", `[${monitor.name}] Axios Response: status=${res.status} body=${JSON.stringify(res.data)}`);
|
||||||
|
if (res.status === 200) {
|
||||||
|
heartbeat.status = UP;
|
||||||
|
heartbeat.msg = "OK";
|
||||||
|
break;
|
||||||
|
} else if (res.status === 503) {
|
||||||
|
heartbeat.msg = res.data.reason;
|
||||||
|
} else {
|
||||||
|
heartbeat.msg = `${res.status} - ${res.statusText}`;
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
if (axios.isCancel(error)) {
|
||||||
|
heartbeat.msg = "Request timed out";
|
||||||
|
log.debug("monitor", `[${monitor.name}] Request timed out`);
|
||||||
|
} else {
|
||||||
|
log.debug("monitor", `[${monitor.name}] Axios Error: ${JSON.stringify(error.message)}`);
|
||||||
|
heartbeat.msg = error.message;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
module.exports = {
|
||||||
|
RabbitMqMonitorType,
|
||||||
|
};
|
||||||
@ -0,0 +1,53 @@
|
|||||||
|
const { describe, test } = require("node:test");
|
||||||
|
const assert = require("node:assert");
|
||||||
|
const { RabbitMQContainer } = require("@testcontainers/rabbitmq");
|
||||||
|
const { RabbitMqMonitorType } = require("../../server/monitor-types/rabbitmq");
|
||||||
|
const { UP, DOWN, PENDING } = require("../../src/util");
|
||||||
|
|
||||||
|
describe("RabbitMQ Single Node", {
|
||||||
|
skip: !!process.env.CI && (process.platform !== "linux" || process.arch !== "x64"),
|
||||||
|
}, () => {
|
||||||
|
test("RabbitMQ is running", async () => {
|
||||||
|
// The default timeout of 30 seconds might not be enough for the container to start
|
||||||
|
const rabbitMQContainer = await new RabbitMQContainer().withStartupTimeout(60000).start();
|
||||||
|
const rabbitMQMonitor = new RabbitMqMonitorType();
|
||||||
|
const connectionString = `http://${rabbitMQContainer.getHost()}:${rabbitMQContainer.getMappedPort(15672)}`;
|
||||||
|
|
||||||
|
const monitor = {
|
||||||
|
rabbitmqNodes: JSON.stringify([ connectionString ]),
|
||||||
|
rabbitmqUsername: "guest",
|
||||||
|
rabbitmqPassword: "guest",
|
||||||
|
};
|
||||||
|
|
||||||
|
const heartbeat = {
|
||||||
|
msg: "",
|
||||||
|
status: PENDING,
|
||||||
|
};
|
||||||
|
|
||||||
|
try {
|
||||||
|
await rabbitMQMonitor.check(monitor, heartbeat, {});
|
||||||
|
assert.strictEqual(heartbeat.status, UP);
|
||||||
|
assert.strictEqual(heartbeat.msg, "OK");
|
||||||
|
} finally {
|
||||||
|
rabbitMQContainer.stop();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("RabbitMQ is not running", async () => {
|
||||||
|
const rabbitMQMonitor = new RabbitMqMonitorType();
|
||||||
|
const monitor = {
|
||||||
|
rabbitmqNodes: JSON.stringify([ "http://localhost:15672" ]),
|
||||||
|
rabbitmqUsername: "rabbitmqUser",
|
||||||
|
rabbitmqPassword: "rabbitmqPass",
|
||||||
|
};
|
||||||
|
|
||||||
|
const heartbeat = {
|
||||||
|
msg: "",
|
||||||
|
status: PENDING,
|
||||||
|
};
|
||||||
|
|
||||||
|
await rabbitMQMonitor.check(monitor, heartbeat, {});
|
||||||
|
assert.strictEqual(heartbeat.status, DOWN);
|
||||||
|
});
|
||||||
|
|
||||||
|
});
|
||||||
Loading…
Reference in new issue