feat: translate collect metrics into tegde measurements

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

Author: didier-wenzek

.release-please-manifest.json

@@ -14,5 +14,6 @@
   "flows/tedge-events": "1.0.0",
   "flows/tedge-measurement-batch": "1.0.0",
   "flows/x509-cert-issuer": "1.0.0",
-  "flows/tedge-compat": "0.1.0"
+  "flows/tedge-compat": "0.1.0",
+  "flows/collectd": "0.0.1"
 }

flows/collectd/README.md

@@ -0,0 +1,9 @@
+## collectd
+
+This flows converts messages received from collectd into thin-edge measurements
+
+### Description
+
+Messages are received from collectd over the MQTT topics `collectd/$HOSTNAME/$GROUP/$MEASUREMENT`.
+
+These messages are CSV encoded (using `':'` column separators) and contains a unix timestamp followed by the measurement value.

flows/collectd/flow.toml

@@ -0,0 +1,8 @@
+[input]
+mqtt.topics = [
+    "collectd/+/+/+"
+]
+
+[[steps]]
+script = "lib/main.js"
+config.topic = "te/device/main///m/collectd"

flows/collectd/package.json

@@ -0,0 +1,26 @@
+{
+  "name": "collectd",
+  "version": "0.0.1",
+  "description": "Translate collectd metrics into thin-edge measurements",
+  "source": "src/main.ts",
+  "module": "lib/main.js",
+  "type": "module",
+  "tedge": {
+    "mapper": "local"
+  },
+  "scripts": {
+    "build": "esbuild src/main.ts --target=es2018 --bundle --outfile=lib/main.js --format=esm",
+    "test": "jest --coverageProvider=v8 --coverage"
+  },
+  "author": "thin-edge.io",
+  "license": "Apache-2.0",
+  "devDependencies": {
+    "@types/jest": "^30.0.0",
+    "esbuild": "^0.25.5",
+    "esbuild-register": "^3.6.0",
+    "jest": "^30.0.4",
+    "jest-esbuild": "^0.4.0",
+    "prettier": "3.6.2",
+    "typescript": "^5.8.3"
+  }
+}

flows/collectd/params.toml.template

@@ -0,0 +1,2 @@
+# enable debug messages
+debug = false

flows/collectd/src/main.ts

@@ -0,0 +1,42 @@
+import { Message, Context, decodePayload } from "../../common/tedge";
+
+export interface Config {
+  topic?: string;
+}
+
+export interface FlowContext extends Context {
+  config: Config;
+}
+
+export function onMessage(message: Message, context: FlowContext): Message[] {
+  const output = [];
+  const { topic = "te/device/main///m/collectd" } = context.config;
+
+  const groups = message.topic.split("/");
+  const data = decodePayload(message.payload).split(":");
+
+  if (groups.length < 4) {
+    throw new Error("Not a collectd topic");
+  }
+  if (data.length < 2) {
+    throw new Error("Not a collectd payload");
+  }
+
+  let group = groups[2];
+  let measurement = groups[3];
+  let time = data[0];
+  let values = data.slice(1);
+
+  values.forEach((value, index) => {
+    let key = measurement;
+    if (values.length > 1) {
+      key = `${measurement}_val${index}`;
+    }
+    output.push({
+      topic,
+      payload: `{"time": ${time}, "${group}": {"${key}": ${value}}}`,
+    });
+  });
+
+  return output;
+}

flows/collectd/tests/main.test.ts

@@ -0,0 +1,53 @@
+import { expect, test, describe } from "@jest/globals";
+import * as tedge from "../../common/tedge";
+import * as flow from "../src/main";
+
+describe("map collectd messages", () => {
+  test("simple message", () => {
+    const output = flow.onMessage(
+      {
+        time: new Date("2026-01-01"),
+        topic: "collectd/localhost/temperature/temp1",
+        payload: "1776866602.745439166:23.7",
+      },
+      tedge.createContext({}),
+    );
+    expect(output).toHaveLength(1);
+    expect(output[0].topic).toBe("te/device/main///m/collectd");
+    const payload = tedge.decodeJsonPayload(output[0].payload);
+    expect(payload).toEqual({
+      time: 1776866602.745439166,
+      temperature: { temp1: 23.7 },
+    });
+  });
+
+  test("combined values", () => {
+    const output = flow.onMessage(
+      {
+        time: new Date("2026-01-01"),
+        topic: "collectd/localhost/temperature/temp",
+        payload: "1776866602.745439166:23.7:24.0:25.8",
+      },
+      tedge.createContext({}),
+    );
+    expect(output).toHaveLength(3);
+
+    expect(output[0].topic).toBe("te/device/main///m/collectd");
+    expect(tedge.decodeJsonPayload(output[0].payload)).toEqual({
+      time: 1776866602.745439166,
+      temperature: { temp_val0: 23.7 },
+    });
+
+    expect(output[1].topic).toBe("te/device/main///m/collectd");
+    expect(tedge.decodeJsonPayload(output[1].payload)).toEqual({
+      time: 1776866602.745439166,
+      temperature: { temp_val1: 24.0 },
+    });
+
+    expect(output[2].topic).toBe("te/device/main///m/collectd");
+    expect(tedge.decodeJsonPayload(output[2].payload)).toEqual({
+      time: 1776866602.745439166,
+      temperature: { temp_val2: 25.8 },
+    });
+  });
+});