feat: flow enabled c8y mapper by didier-wenzek · Pull Request #3934 · thin-edge/thin-edge.io

didier-wenzek · 2026-01-21T10:49:25Z

Proposed changes

The c8y mapper can be extended using flows.

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 · 2026-01-21T11:01:46Z

Codecov Report

❌ Patch coverage is 81.08527% with 244 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
crates/extensions/c8y_mapper_ext/src/mea/events.rs	66.36%	30 Missing and 7 partials ⚠️
crates/extensions/c8y_mapper_ext/src/converter.rs	82.55%	27 Missing and 3 partials ⚠️
crates/core/tedge_mapper/src/c8y/mapper.rs	22.22%	21 Missing ⚠️
...ates/extensions/c8y_mapper_ext/src/entity_cache.rs	86.00%	18 Missing and 3 partials ⚠️
crates/extensions/c8y_mapper_ext/src/mea/health.rs	72.30%	15 Missing and 3 partials ⚠️
...extensions/c8y_mapper_ext/src/mea/message_cache.rs	65.38%	17 Missing and 1 partial ⚠️
...ates/extensions/tedge_flows/src/js_lib/kv_store.rs	82.00%	14 Missing and 4 partials ⚠️
crates/extensions/c8y_mapper_ext/src/flows.rs	90.47%	3 Missing and 11 partials ⚠️
.../extensions/c8y_mapper_ext/src/mea/measurements.rs	75.00%	12 Missing and 1 partial ⚠️
crates/extensions/c8y_mapper_ext/src/mea/alarms.rs	87.77%	7 Missing and 4 partials ⚠️
... and 13 more

📢 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.

github-actions · 2026-01-21T11:22:37Z

Robot Results

✅ Passed	❌ Failed	⏭️ Skipped	Total	Pass %	⏱️ Duration
831	0	3	831	100	2h44m34.215053s

albinsuresh

LGTM. Huge achievement, considering the complexity of this mapper with a lot of legacy code that isn't the most intuitive. There are a few points that could be improved, but as future iterations.

albinsuresh · 2026-02-06T12:19:14Z

+        let pending_messages = self.cache.take();
+        for message in pending_messages.into_iter() {
+            if message.topic.starts_with(entity_topic) {


With this logic, if there are some cached messages for some entities (e.g: child1 and child2) in this cache, and when messages are received for another entity (e.g: child0) that is already registered, we'll be unnecessarily draining the cache first, only to put them all back. And we repeat that for every single message.

We could make this logic more efficient, by changing the cache from a simple RingBuffer<Message> to HasMap<EntityTopicId, RingBuffer<Message>> so that this step only consumes the messages for the current entity and leave the ones from other entities untouched. But, this is just a matter of efficiency and not correctness. So, it can be improved in a later PR.

With this logic, if there are some cached messages for some entities (e.g: child1 and child2) in this cache, and when messages are received for another entity (e.g: child0) that is already registered, we'll be unnecessarily draining the cache first, only to put them all back. And we repeat that for every single message.

The normal case is to have an empty cache and if not empty this will be for not too long, the main purpose of this cache being to solve races between registration and mea messages.

I see one annoying case though. A measurement which has been sent by mistake on behalf of a non-existing entity on a device configured not to use auto-registration. No registration message will ever be sent and the measurement sent by mistake will stay in the cache for ever. Worse, this measurement will be considered again and again for all cache drains.

In other word, I see more a cache pollution problem than a performance problem.

We could make this logic more efficient, by changing the cache from a simple RingBuffer<Message> to HasMap<EntityTopicId, RingBuffer<Message>> so that this step only consumes the messages for the current entity and leave the ones from other entities untouched. But, this is just a matter of efficiency and not correctness. So, it can be improved in a later PR.

Indeed, this would solve the issue of the unknown-source measurement polluting the drain process for all messages and all entities.

We could make this logic more efficient, by changing the cache from a simple RingBuffer<Message> to HasMap<EntityTopicId, RingBuffer<Message>> so that this step only consumes the messages for the current entity and leave the ones from other entities untouched.

Done af732fe

albinsuresh · 2026-02-06T12:20:31Z

+        let message_topic = te.topic_for(
+            mapper,
+            &Channel::Status {
+                component: "entities".to_string(),
+            },
+        );


Decision deferred for later.

jarhodes314 · 2026-02-06T17:47:50Z

+        context: &FlowContextHandle,
+    ) -> Result<Vec<Message>, FlowError> {
+        match self.mqtt_schema.entity_channel_of(&message.topic) {
+            Ok((_, Channel::Status { component })) if component == "entities" => {


Do we also run into issues here if we have multiple c8y mappers running simultaneously (e.g. with multiple profiles)? Should we not be checking the service topic ID is for this mapper, and ignoring birth messages for other mappers?

Agree. This is related to #3934 (comment)

As the birth messages are related to a c8y mapper instance, we have to use a c8y related topic for those. Precisely this needs to be a sub-topic of c8y.bridge.topic_prefix for the appropriate profile.

Actually, the current code is working because the flows using this cache subscribe to a c8y mapper instance topic (aka te/device/main/service/tedge-mapper-c8y/status/entities).

This can made more obvious with a check on the entity sending the birth message (instead of ignoring is it with _).

Independently , we can also change the topic as proposed by @albinsuresh feat: flow enabled c8y mapper #3934 (comment)

This can made more obvious with a check on the entity sending the birth message (instead of ignoring is it with _).

I think this is wise. It feels like a bit of a hack to rely on "we don't subscribe to other conflicting topics" to ensure we only respond to this mapper's messages.

This can made more obvious with a check on the entity sending the birth message (instead of ignoring is it with _).

I think this is wise. It feels like a bit of a hack to rely on "we don't subscribe to other conflicting topics" to ensure we only respond to this mapper's messages.

Done: 61d7259

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

This is a first step toward a mapper context populated by the mapper using a specific implementation (say to read the entity cache), while the context for a script or a flow is a simple key value store. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

so the flow context can be populated by the mapper, notably with entity cache entries. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

The goal is to group the methods acting on the content of the entity cache, putting apart all the business logic related to entity metadata. The next step will be to connect the EntityCache to the FlowContext, populating the latter with entity metadata. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

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

Starting only with `Mqtt { qos: QoS, retain: bool }`, the plan begin to add `Process { command: String }` and, possibly, `Http { url: String }` Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

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

preventing tedge flows to start. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

reubenmiller

Approved. Everything works nicely and this allows users to also edit existing tedge-mapper-c8y logic (which was not possible before)

didier-wenzek temporarily deployed to Test Pull Request January 21, 2026 10:49 — with GitHub Actions Inactive

didier-wenzek added theme:c8y Theme: Cumulocity related topics theme:flows labels Jan 21, 2026

didier-wenzek had a problem deploying to Test Auto January 21, 2026 10:59 — with GitHub Actions Failure

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

didier-wenzek changed the title ~~feat! flow enabled c8y mapper~~ feat: flow enabled c8y mapper Jan 21, 2026

didier-wenzek had a problem deploying to Test Auto January 21, 2026 14:14 — with GitHub Actions Failure

albinsuresh reviewed Jan 22, 2026

View reviewed changes

Comment thread tests/RobotFramework/tests/cumulocity/flows/custom-measurements.js Outdated

didier-wenzek temporarily deployed to Test Pull Request January 22, 2026 14:04 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 22, 2026 14:15 — with GitHub Actions Failure

didier-wenzek force-pushed the feat/flow-enabled-c8y-mapper branch from c1e9cf1 to 666848b Compare January 22, 2026 14:47

didier-wenzek temporarily deployed to Test Pull Request January 22, 2026 14:47 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 22, 2026 15:01 — with GitHub Actions Failure

didier-wenzek had a problem deploying to Test Pull Request January 22, 2026 16:01 — with GitHub Actions Failure

didier-wenzek temporarily deployed to Test Pull Request January 22, 2026 20:17 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 22, 2026 20:28 — with GitHub Actions Failure

didier-wenzek force-pushed the feat/flow-enabled-c8y-mapper branch from 38bdaf8 to 01fba3f Compare January 23, 2026 09:58

didier-wenzek temporarily deployed to Test Pull Request January 23, 2026 09:58 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 23, 2026 10:06 — with GitHub Actions Failure

didier-wenzek temporarily deployed to Test Pull Request January 23, 2026 11:07 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 23, 2026 11:18 — with GitHub Actions Failure

didier-wenzek commented Jan 23, 2026

View reviewed changes

Comment thread crates/extensions/c8y_mapper_ext/src/actor.rs Outdated

didier-wenzek temporarily deployed to Test Pull Request January 23, 2026 14:19 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 23, 2026 14:32 — with GitHub Actions Failure

didier-wenzek temporarily deployed to Test Pull Request January 23, 2026 15:59 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 23, 2026 16:08 — with GitHub Actions Failure

didier-wenzek temporarily deployed to Test Pull Request January 26, 2026 13:05 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto January 26, 2026 13:14 — with GitHub Actions Failure

albinsuresh approved these changes Feb 6, 2026

View reviewed changes

jarhodes314 reviewed Feb 6, 2026

View reviewed changes

reubenmiller reviewed Feb 9, 2026

View reviewed changes

Comment thread tests/RobotFramework/tests/cumulocity/flows/c8y_flows.robot Outdated

jarhodes314 approved these changes Feb 12, 2026

View reviewed changes

didier-wenzek added 24 commits February 13, 2026 09:45

The c8y mapper runs user provided flows

8f2cceb

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

refactoring: expose the flow context as a KV store

5deaaf8

so the flow context can be populated by the mapper, notably with entity cache entries. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

The context flow of the c8y mapper holds entity metadata

bd6fe4b

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

A builtin transformer can get/set flow context values

8218706

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

The c8y mapper uses flows to process measurements

44d1132

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

refactoring: a flow step might update its state on messages

4e75d10

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

The c8y mapper uses flows to process alarms

c89a791

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

A transport property can be attached to flow messages

31d1ecb

Starting only with `Mqtt { qos: QoS, retain: bool }`, the plan begin to add `Process { command: String }` and, possibly, `Http { url: String }` Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

The c8y mapper uses flows to process events

a40ff5a

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

The c8y mapper publishes birth messages for new entities

cd17b62

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

refactoring: removed dead code: mod size_threshold

daabf35

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

Builtin flows can be disabled with tedge config c8y.topics

83d8405

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

The c8y mapper uses flows to process health status

36081b6

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

tests: remove race condition when disabling a flow

34f5ebd

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

refactoring: remove unused c8y mapper dependency on a timer

e6c0c41

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

docs: The c8y mapper is defined using regular flow

90af9e6

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

Entity-cache preparation step for mea transformers

ce02edd

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

fix: store entity display name and type in flow context

91b4a8b

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

naming: use kebab-case for transformation step names

e4aa991

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

tests: generated builtin flows are well-formed toml flow specs

77d04b4

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

Disabled builtin flows can be restored when updating the config

95216fc

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

fix: tedge connect creating mapper directory owned by root

c088d33

preventing tedge flows to start. Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

reubenmiller approved these changes Feb 13, 2026

View reviewed changes

didier-wenzek mentioned this pull request Mar 23, 2026

fix: tedge flows on context update #4069

Merged

14 tasks

Conversation

didier-wenzek commented Jan 21, 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!

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

Uh oh!

Codecov Report

Uh oh!

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

Uh oh!

Robot Results

Uh oh!

Uh oh!

Uh oh!

albinsuresh left a comment

Choose a reason for hiding this comment

Uh oh!

albinsuresh Feb 6, 2026

Choose a reason for hiding this comment

Uh oh!

didier-wenzek Feb 6, 2026

Choose a reason for hiding this comment

Uh oh!

didier-wenzek Feb 6, 2026

Choose a reason for hiding this comment

Uh oh!

albinsuresh Feb 6, 2026

Choose a reason for hiding this comment

Uh oh!

jarhodes314 Feb 6, 2026

Choose a reason for hiding this comment

Uh oh!

didier-wenzek Feb 9, 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!

didier-wenzek Feb 9, 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!

jarhodes314 Feb 9, 2026

Choose a reason for hiding this comment

Uh oh!

didier-wenzek Feb 9, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

reubenmiller left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

didier-wenzek commented Jan 21, 2026 •

edited

Loading

codecov Bot commented Jan 21, 2026 •

edited

Loading

github-actions Bot commented Jan 21, 2026 •

edited

Loading

didier-wenzek Feb 9, 2026 •

edited

Loading

didier-wenzek Feb 9, 2026 •

edited

Loading