poc: Generic Mapper using JavaScript by didier-wenzek · Pull Request #3626 · thin-edge/thin-edge.io

didier-wenzek · 2025-05-15T17:01:15Z

Proposed changes

This POC explores an alternative to #3517

A thin-edge mapper that can be extended by users with mapping rules implemented in JavaScript.

Features:

Chain stateless filters along pipelines: MQTT sub| filter-1 | filter-2 | ... | filter-n | MQTT pub
Configure filters
Update filter config using MQTT
Implement state-full filters (deferring messages up to the end of a time window)
Combine user-provided JS filters with built-in Rust filters

Criteria:

Expressive power?
Dev tooling ?
Performance? Memory footprint?
Debuggability?

Examples:

Add missing timestamps
Discard measurements with timestamp out of the expected range (too old or in the future)
Translate thin-edge json measurment into cumulocity json
Add measurement units - read from topic metadata
Translate Collectd data into thin-edge JSON
Group messages received during the same time window
Compute 5-minute rolling average

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

codecov · 2025-05-15T17:10:49Z

Codecov Report

Attention: Patch coverage is 80.18868% with 21 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
crates/extensions/tedge_mqtt_ext/src/lib.rs	80.23%	13 Missing and 4 partials ⚠️
crates/common/mqtt_channel/src/connection.rs	50.00%	2 Missing ⚠️
crates/core/plugin_sm/src/operation_logs.rs	0.00%	0 Missing and 1 partial ⚠️
crates/core/tedge_mapper/src/lib.rs	0.00%	1 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.

This commit is a very first step, scaffolding a generic mapper. The aim is to let users define their own mapping rules to tranform, filter or enrich data received from various sources. The idea is to form pipelines of user-provided transformation functions that: - consume messages from MQTT, - stream these messages along the transformation stages, - publish back to MQTT the resulting messages. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

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

…mapper Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

NB This doesn't currently work since dynamic subscriptions aren't hooked up fully. tedge_mqtt_ext needs some further modifications to support sending the received messages on newly subscribed topics to the actor that requested the subscription Signed-off-by: James Rhodes <jarhodes314@gmail.com>

didier-wenzek · 2025-05-23T09:19:24Z

crates/extensions/tedge_gen_mapper/src/actor.rs

+                            }
+                        },
+                        Some(InputMessage::FsWatchEvent(e)) => {
+                            tracing::warn!("TODO do something with {e:?}")


We can definitely ignore anything that is is related to a file with an unknown extension (such as .collectd.toml.swp and collectd.toml~).

Deletion of a pipeline should be okay, unless if we try to unsubscribe to the input topics.

Deletion of a script is more problematic, if still used by one of the pipeline.

didier-wenzek · 2025-05-23T13:03:52Z

crates/common/mqtt_channel/src/config.rs

    ///
    /// Default: An empty topic list
-    pub subscriptions: TopicFilter,
+    pub subscriptions: Arc<Mutex<TopicFilter>>,


A bit unusual to have interior mutability for a config struct - something granted as immutable.

didier-wenzek · 2025-05-23T14:03:07Z

crates/extensions/tedge_mqtt_ext/src/lib.rs

+        client: mqtt_channel::AsyncClient,
+        subscriptions: Arc<Mutex<TopicFilter>>,


I would combine these two fields of an mqtt_chanel::Connection in a single one (say a subscriber) that handles the whole logic below to compute the new subscription set and update mqtt_channel::AsyncClient.

didier-wenzek · 2025-05-23T14:23:30Z

crates/extensions/tedge_mqtt_ext/src/lib.rs

+                    let to_sub = to_sub.filters();
+                    if !to_sub.is_empty() {
+                        info!("Updating MQTT subscription to include {to_sub:?}");
+                        client.subscribe_many(to_sub).await.unwrap();


NB This doesn't currently work since dynamic subscriptions aren't hooked up fully. tedge_mqtt_ext needs some further modifications to support sending the received messages on newly subscribed topics to the actor that requested the subscription

Okay, I understand this comment now. The MQTT subscription is correctly updated but the messages are not forwarded by this actor which ToPeers::peer_senders subscriptions list has not been updated.

Indeed, not easy to update from the FromPeers event loop this peer_senders list which is owned by the ToPeers event loop. A new channel, would be not enough as these peer Senders are anonymous. Which one would have to be updated?

Here is a proposal to fix that: 43932cf

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

didier-wenzek · 2025-06-05T11:39:57Z

Closing this POC which is superseded by #3650.

The main reason being the executable size (59M - release build, dynamically linked, stripped) compared to the POC using QuickJS (17M - release build, statically linked, stripped) and tedge without any JS support (16M- release build, statically linked, stripped)

didier-wenzek had a problem deploying to Test Pull Request May 15, 2025 17:01 — with GitHub Actions Failure

didier-wenzek force-pushed the poc/tedge-js-mapper branch from cbab55d to c87474d Compare May 16, 2025 07:43

didier-wenzek had a problem deploying to Test Pull Request May 16, 2025 07:43 — with GitHub Actions Failure

didier-wenzek added 4 commits May 16, 2025 15:26

Sketch message transformation pipelines

fb46fe9

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

Run JS transformation stages

5ab0433

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

Update deprecated call tempfile into_path -> keep

188fd01

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

didier-wenzek force-pushed the poc/tedge-js-mapper branch from c87474d to 188fd01 Compare May 16, 2025 15:12

didier-wenzek had a problem deploying to Test Pull Request May 16, 2025 15:12 — with GitHub Actions Failure

Avoid to deserialize MQTT at each pipeline stage.

3400e3b

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

didier-wenzek had a problem deploying to Test Pull Request May 19, 2025 10:26 — with GitHub Actions Failure

Add JS example: drop stragglers

d0a0cc2

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

didier-wenzek had a problem deploying to Test Pull Request May 19, 2025 14:17 — with GitHub Actions Failure

Add ability to configure each stages

9d93d4e

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

didier-wenzek had a problem deploying to Test Pull Request May 19, 2025 15:42 — with GitHub Actions Failure

Add JS example: collectd pipeline

34ec01a

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

didier-wenzek had a problem deploying to Test Pull Request May 20, 2025 08:49 — with GitHub Actions Failure

Add JS example: thin-edge JSON to c8y JSON

47caf11

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

didier-wenzek had a problem deploying to Test Pull Request May 20, 2025 12:04 — with GitHub Actions Failure

Fix tests broken by cargo update

3ac138a

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

didier-wenzek had a problem deploying to Test Pull Request May 20, 2025 13:48 — with GitHub Actions Failure

Update JS filter config using MQTT

31648c0

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

didier-wenzek had a problem deploying to Test Pull Request May 20, 2025 15:57 — with GitHub Actions Failure

Filters can defer messages up to the end of a time window

8765160

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

didier-wenzek had a problem deploying to Test Pull Request May 20, 2025 17:00 — with GitHub Actions Failure

Add an initial attempt at reloading filters/pipelines in the generic …

f4b6c2d

…mapper Signed-off-by: James Rhodes <jarhodes314@gmail.com>

jarhodes314 had a problem deploying to Test Pull Request May 21, 2025 16:09 — with GitHub Actions Failure

jarhodes314 added 2 commits May 22, 2025 16:44

mostly add support for dynamic mqtt subscriptions

226d83d

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

jarhodes314 had a problem deploying to Test Pull Request May 22, 2025 15:46 — with GitHub Actions Failure

didier-wenzek commented May 23, 2025

View reviewed changes

Let actors dynamically update their MQTT subscriptions

43932cf

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

didier-wenzek had a problem deploying to Test Pull Request May 23, 2025 15:52 — with GitHub Actions Failure

fixup! use dynamic subscriptions to update mqtt subs on pipeline updates

7ad9c4f

didier-wenzek had a problem deploying to Test Pull Request May 26, 2025 08:11 — with GitHub Actions Failure

Tedge-gen-mapper dynamically subscribes to MQTT topics

1f21181

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

didier-wenzek had a problem deploying to Test Pull Request May 26, 2025 12:21 — with GitHub Actions Failure

didier-wenzek mentioned this pull request May 28, 2025

feat: tedge flows, a configurable mapper using JavaScript #3650

Merged

32 tasks

jarhodes314 mentioned this pull request Jun 2, 2025

refactor: support dynamic MQTT subscriptions in mqtt-channel/tedge-mqtt-ext #3661

Merged

11 tasks

didier-wenzek closed this Jun 5, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

poc: Generic Mapper using JavaScript#3626

poc: Generic Mapper using JavaScript#3626
didier-wenzek wants to merge 18 commits intothin-edge:mainfrom
didier-wenzek:poc/tedge-js-mapper

didier-wenzek commented May 15, 2025 •

edited

Loading

Uh oh!

codecov bot commented May 15, 2025 •

edited

Loading

Uh oh!

didier-wenzek May 23, 2025

Uh oh!

didier-wenzek May 23, 2025

Uh oh!

didier-wenzek May 23, 2025

Uh oh!

didier-wenzek May 23, 2025

Uh oh!

didier-wenzek May 23, 2025

Uh oh!

didier-wenzek commented Jun 5, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		client: mqtt_channel::AsyncClient,
		subscriptions: Arc<Mutex<TopicFilter>>,

Conversation

didier-wenzek commented May 15, 2025 • 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!

codecov bot commented May 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

didier-wenzek May 23, 2025

Choose a reason for hiding this comment

Uh oh!

didier-wenzek May 23, 2025

Choose a reason for hiding this comment

Uh oh!

didier-wenzek May 23, 2025

Choose a reason for hiding this comment

Uh oh!

didier-wenzek May 23, 2025

Choose a reason for hiding this comment

Uh oh!

didier-wenzek May 23, 2025

Choose a reason for hiding this comment

Uh oh!

didier-wenzek commented Jun 5, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

didier-wenzek commented May 15, 2025 •

edited

Loading

codecov bot commented May 15, 2025 •

edited

Loading