refactor: replace the collectd mapper with local flows by didier-wenzek · Pull Request #4136 · thin-edge/thin-edge.io

didier-wenzek · 2026-04-16T16:08:17Z

Proposed changes

The goal of this PR is to deprecate the collectd mapper providing the tools to implement the same using flows

Flow script translating collectd messages into te measurements
Flow transformer batching messages
Collectd Flow in https://github.com/thin-edge/tedge-flows-examples or https://github.com/thin-edge/tedge-collectd-setup
Rewrite system tests to use a flow-based collectd mapper
Update sections of the docs that use collectd to illustrate device monitoring
Deprecate the batcher actor
Deprecate the collectd mapper

Types of changes

Bugfix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Improvement (general improvements like code refactoring that doesn't explicitly fix a bug or add any new functionality)
Documentation Update (if none of the other choices apply)
Breaking change (fix or feature that would cause existing functionality to not work as expected)

Paste Link to the issue

Checklist

I have read the CONTRIBUTING doc
I have signed the CLA (in all commits with git commit -s. You can activate automatic signing by running just prepare-dev once)
I ran just format as mentioned in CODING_GUIDELINES
I used just check as mentioned in CODING_GUIDELINES
I have added tests that prove my fix is effective or that my feature works
I have added necessary documentation (if appropriate)

Further comments

github-actions · 2026-04-16T16:48:43Z

Robot Results

✅ Passed	❌ Failed	⏭️ Skipped	Total	Pass %	⏱️ Duration
896	1	4	897	99.89	2h43m45.296348s

Failed Tests

Name	Message	⏱️ Duration	Suite
Connect to Cumulocity MQTT Service endpoint basic auth	`tedge connect c8y` returned an unexpected exit code stdout: stderr: Connecting to Cumulocity: mapper configuration file: /etc/tedge/mappers/c8y/mapper.toml device id: TST_pass_absolute_ambush cloud profile: <none> cloud host: qaenvironment.eu-latest.cumulocity.com:9883 auth type: Basic credentials path: /etc/tedge/credentials.toml bridge: built-in service manager: systemd mosquitto version: 2.0.11 proxy: Not configured Updating bridge rule templates... ✓ Creating device in Cumulocity cloud... ✗ error: Connection error while creating device in Cumulocity: Connection refused, return code: `NotAuthorized` [RETRY] FAIL on 3. retry.	48.168 s	`Tedge Connect Mqtt Service`

codecov · 2026-04-21T11:31:51Z

Codecov Report

❌ Patch coverage is 91.22807% with 20 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
crates/common/batcher/src/flows.rs	91.97%	10 Missing and 1 partial ⚠️
crates/extensions/tedge_flows/src/flow.rs	43.75%	9 Missing ⚠️

📢 Thoughts on this report? Let us know!

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

didier-wenzek · 2026-04-21T11:24:31Z

-Check running collectd
-    Service Should Be Running    collectd
-
-Is collectd publishing MQTT messages?
-    ${messages}=    Should Have MQTT Messages    topic=collectd/#    minimum=1    maximum=None
-


These checks have been moved to the suite setup. They are not system tests but prerequisite checks.

didier-wenzek · 2026-04-21T14:45:17Z

@@ -0,0 +1,18 @@
+// Translate an array of thin-edge measurements into a single message with grouped measurements


I consider promoting this script to a builtin transformer: grouping measurements is a quite common use case.

That would be a good addition. The input topic would also have to be in the tedge measurements format, right?

The input topic would also have to be in the tedge measurements format, right?

I still have to see what's the best approach to decompose the flow into independent steps that can be re-used / re-combined.

didier-wenzek · 2026-04-21T14:53:00Z

+        let batch_config = BatchConfigBuilder::new()
+            .event_jitter(500)
+            .delivery_jitter(400) // Heuristic delay that should work out well on a Rpi
+            .message_leap_limit(500)
+            .build();


These configuration parameters are simply hairy to get correct, the batching algorithm being complicated and fragile with two independent clocks (batching time and event time). This complexity is furthermore useless in practice, the best way to avoid too many collectd measurements being to reduce the collection frequency.

didier-wenzek · 2026-04-22T15:05:11Z

+    fn event_time(&self) -> OffsetDateTime {
+        self.timestamp
+            .map(|t| t.into())
+            .unwrap_or_else(OffsetDateTime::now_utc)
+    }


This defeats the purpose of the batcher: i.e using the processing time to batch message, ignoring the event time.

I will fix that with a flow config giving a JSON path to the message event time.

Finally, this code will be let unchanged, i.e. the timestamp of the message being used as the event time.

For that purpose the flow engine has been improved. Now, message timestamps can be set by a script and used by the subsequent steps.

didier-wenzek · 2026-04-22T15:08:12Z

+    type Key = String;
+
+    fn key(&self) -> Self::Key {
+        self.topic.clone()


Using the topic as a key make the batcher transformer a bit complicated to use: the previous step needs to be aware that the topic is used as a key by the batcher.

I will fix that with a way to configure the transformer with a path to the key.

rina23q

I guess still code is in progress, so I reviewed the documentation changes.

rina23q · 2026-04-22T16:52:33Z

 ## Enabling the systemd watchdog feature for a tedge service

-Enabling systemd watchdog for a %%te%% service (tedge-agent, tedge-mapper-c8y/az/collectd) is a two-step process.
+Enabling systemd watchdog for a %%te%% service (tedge-agent, tedge-mapper-c8y/az) is a two-step process.


(unrelated) Out of curiosity, The watchdog feature doesn't work also for aws and local mappers?
I found this document is actually unlisted, so maybe it doesn't matter how precise information we have here.

rina23q · 2026-04-22T17:04:07Z

 Run `tedge_mapper --debug aws` to log more debug messages
 :::

-### Device monitoring logs {#device-logs}


Probably we can also remove the section "Collectd logs" below (inside "Third-party component logs"). Instead, we can mention how to see the log of collectd in the new repository for the collectd flow.

rina23q · 2026-04-22T17:12:52Z

+Install Collectd
    Execute Command
    ...    sudo apt-get update && sudo apt-get install -y --no-install-recommends collectd-core libmosquitto1
    Execute Command    sudo cp /etc/tedge/contrib/collectd/collectd.conf /etc/collectd/collectd.conf


I guess this collectd.conf file is also a candidate to move out of the thin-edge.io repository later.

albinsuresh · 2026-04-23T15:40:46Z

    ${start_time}=    Get Unix Timestamp
-    Execute Command
-    ...    tedge mqtt pub collectd/localhost/temperature/temp1 "`date +%s.%N`:50" && tedge mqtt pub collectd/localhost/temperature/temp2 "`date +%s.%N`:40" && tedge mqtt pub collectd/localhost/pressure/pres1 "`date +%s.%N`:10" && tedge mqtt pub collectd/localhost/pressure/pres2 "`date +%s.%N`:20"
+    Execute Command    tedge mqtt pub collectd/localhost/temperature/temp1 "`date +%s.%N`:50"


IIRC, the chaining of the pub commands in the former version was done to minimise the timestamp drift between the first measurement and the last measurement, to make sure that they get batched together. So, this version might end up being flaky.

This was the point that we always wanted to improve, at least by making the batch window configurable, so that we don't have to rely on fragile tricks like that.

So, this version might end up being flaky.

This is not what I observe:

$ invoke flake-finder --test-name "Check grouping of measurements" --iterations 25 --outputdir output_ff --clean ------------------------------ Overall: PASSED Results: 25 iterations, 25 passed, 0 failed Elapsed time: 0:10:34.572869

This was the point that we always wanted to improve, at least by making the batch window configurable,

I will do that in this PR.

albinsuresh · 2026-04-23T18:04:43Z

+    let grouped_measurements = {}
+
+    for (let measurement of measurements) {
+        Object.assign(grouped_measurements, measurement.payload)


That would get the timestamp wrong, no? The timestamp of the last measurement would replace the former ones, while the expectation is for the batch to keep the timestamp of the first measurement that triggered the batch.

I would not say this is wrong. Not your expectation apparently, but as meaningful.

Anyway, this script is only used for testing purposes and will be replaced by a builtin transformer.

An the joke side. According this comment, messages must happen in a blink to be batched! So first or last timestamp should not be a big deal ;-)

albinsuresh · 2026-04-23T18:06:19Z

@@ -0,0 +1,18 @@
+// Translate an array of thin-edge measurements into a single message with grouped measurements


That would be a good addition. The input topic would also have to be in the tedge measurements format, right?

The fact that the batcher is always initialized with default values highlights that something is wrong. It's really difficult to pick sensible values. And in practice nobody care. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

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

The following sections have still to be deeply rewritten: - Getting Started / A tour of thin-edge.io / Monitor the device - Getting Started / Monitoring - Operate Devices / Troubleshooting / Device Monitoring Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

A value created in JS using `new Date()` cannot be directly returned to the flow engine. A script has then to invoke `valueOf()` method to assign a date to a message timestamp. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

didier-wenzek temporarily deployed to Test Pull Request April 16, 2026 16:08 — with GitHub Actions Inactive

didier-wenzek added theme:monitoring Theme: Service monitoring and watchdogs theme:flows labels Apr 16, 2026

didier-wenzek had a problem deploying to Test Auto April 16, 2026 16:17 — with GitHub Actions Failure

didier-wenzek force-pushed the refactor/replace-collectd-mapper-with-flows branch from a8ba280 to e460f13 Compare April 20, 2026 14:15

didier-wenzek temporarily deployed to Test Pull Request April 20, 2026 14:15 — with GitHub Actions Inactive

didier-wenzek temporarily deployed to Test Auto April 20, 2026 14:26 — with GitHub Actions Inactive

didier-wenzek force-pushed the refactor/replace-collectd-mapper-with-flows branch from e460f13 to e3a4f91 Compare April 21, 2026 11:22

didier-wenzek temporarily deployed to Test Pull Request April 21, 2026 11:22 — with GitHub Actions Inactive

didier-wenzek temporarily deployed to Test Pull Request April 21, 2026 11:56 — with GitHub Actions Inactive

didier-wenzek commented Apr 21, 2026

View reviewed changes

didier-wenzek had a problem deploying to Test Auto April 21, 2026 12:05 — with GitHub Actions Failure

didier-wenzek temporarily deployed to Test Pull Request April 21, 2026 13:34 — with GitHub Actions Inactive

didier-wenzek temporarily deployed to Test Auto April 21, 2026 13:57 — with GitHub Actions Inactive

didier-wenzek temporarily deployed to Test Pull Request April 21, 2026 14:38 — with GitHub Actions Inactive

didier-wenzek marked this pull request as ready for review April 21, 2026 14:41

didier-wenzek requested review from a team, albinsuresh, jarhodes314, reubenmiller and rina23q as code owners April 21, 2026 14:41

didier-wenzek temporarily deployed to Test Auto April 21, 2026 14:49 — with GitHub Actions Inactive

didier-wenzek commented Apr 21, 2026

View reviewed changes

didier-wenzek force-pushed the refactor/replace-collectd-mapper-with-flows branch from dedbc3a to e00dafc Compare April 22, 2026 13:29

didier-wenzek requested a deployment to Test Pull Request April 22, 2026 13:29 — with GitHub Actions Waiting

didier-wenzek commented Apr 22, 2026

View reviewed changes

rina23q reviewed Apr 22, 2026

View reviewed changes

didier-wenzek force-pushed the refactor/replace-collectd-mapper-with-flows branch from e00dafc to 9bb4852 Compare April 23, 2026 08:28

didier-wenzek requested a deployment to Test Pull Request April 23, 2026 08:28 — with GitHub Actions Waiting

didier-wenzek temporarily deployed to Test Pull Request April 23, 2026 08:30 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto April 23, 2026 09:01 — with GitHub Actions Failure

didier-wenzek temporarily deployed to Test Pull Request April 23, 2026 13:42 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto April 23, 2026 13:52 — with GitHub Actions Failure

albinsuresh reviewed Apr 23, 2026

View reviewed changes

didier-wenzek added 14 commits April 24, 2026 14:25

refactor: remove dead code

fee9770

The fact that the batcher is always initialized with default values highlights that something is wrong. It's really difficult to pick sensible values. And in practice nobody care. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

The batcher flow transformer batch messages over time windows

dc2822f

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

Use flows to ingest collectd measurements

26636e8

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

breaking change! deprecate tedge-mapper-collectd

687e158

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

breaking change! deprecate collectd_ext

44526bb

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

breaking change! deprecate tedge-mapper-collectd packaging

4bb3c10

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

Remove batcher dependency on actors

3466f4a

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

fixup! breaking change! deprecate tedge-mapper-collectd packaging

727357d

remove collectd from tedge diag plugins

56cb40a

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

fixup! docs: remove references to tedge-mapper-collectd

2aa6bd4

fixup! fixup! docs: remove references to tedge-mapper-collectd

b0e2540

fixup! Return to the flow engine message timestamps assigned by a script

080f3a6

didier-wenzek force-pushed the refactor/replace-collectd-mapper-with-flows branch from 89338e2 to 080f3a6 Compare April 24, 2026 12:25

didier-wenzek requested a deployment to Test Pull Request April 24, 2026 12:25 — with GitHub Actions Waiting

		@@ -0,0 +1,18 @@
		// Translate an array of thin-edge measurements into a single message with grouped measurements

Conversation

didier-wenzek commented Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Proposed changes

Types of changes

Paste Link to the issue

Checklist

Further comments

Uh oh!

github-actions Bot commented Apr 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Robot Results

Failed Tests

Uh oh!

codecov Bot commented Apr 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rina23q left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

albinsuresh Apr 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

didier-wenzek commented Apr 16, 2026 •

edited

Loading

github-actions Bot commented Apr 16, 2026 •

edited

Loading

codecov Bot commented Apr 21, 2026 •

edited

Loading

albinsuresh Apr 23, 2026 •

edited

Loading