Skip to content

poc: tedge MEA DB #3725

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Draft
wants to merge 50 commits into
base: main
Choose a base branch
from
Draft

poc: tedge MEA DB #3725

wants to merge 50 commits into from

Conversation

didier-wenzek
Copy link
Contributor

@didier-wenzek didier-wenzek commented Jul 7, 2025 •
edited
Loading

Proposed changes

Thin-edge currently delegates to the MQTT broker all message-persistence requirements

  • It's work well but lacks flexibility
  • If a network outage is long enough to fill the broker database, there is no way to choose which messages can be discarded

Can thin-edge be extended with a Round Robin Database (or similar) to accumulate MEA messages?
Adding the ability to publish consolidated data to the cloud or a data lake.

Idea

It would be nice if users where able to define themselves persistence rules for messages that cannot be processed right now

  • Keep all recent measurements (say from the last hour)
  • Reduce the numbers of measurements if too many (say be keeping 5 minutes averages for the current day)
  • Archive coarse-grained average for a longer period of time
  • Discard messages definitely too old

Demo

  • Use the generic mapper and user-provided rules to filter, transform and aggregate messages
  • Persist messages in an MEA database storing series of messages
  • Drain messages out of a series when too old
  • Process drained messages with the generic mapper: aggregate, discard ...

Next steps

Connect the builtin bridge to the MEA DB.

  • Consume local messages no more directly from MQTT but from MEA series
  • Drain messages from the DB when acknowledged by the cloud
  • Let messages pile up when the network is down
  • Use user-defined rules to reduce the number of pending messages if the outage lasts

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

didier-wenzek and others added 30 commits June 27, 2025 16:10
@didier-wenzek
Add a new mapper: tedge-gen-mapper
a8ee90f 
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>
@didier-wenzek
Sketch message transformation pipelines
4e1613c 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Run JS transformation stages
e58c8cc 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Update deprecated call tempfile into_path -> keep
97c46a8 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Avoid to deserialize MQTT at each pipeline stage.
1b59365 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add JS example: drop stragglers
046db7e 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add ability to configure each stages
cbc345a 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add JS example: collectd pipeline
c86ee6c 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add JS example: thin-edge JSON to c8y JSON
c56c290 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Fix tests broken by cargo update
5fa5cdd 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Update JS filter config using MQTT
09887c5 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Filters can defer messages up to the end of a time window
42a3fff 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@jarhodes314 @didier-wenzek
Tedge-gen-mapper dynamically reloads pipelines and filters
b61f9d0 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@didier-wenzek
Tedge-gen-mapper dynamically subscribes to MQTT topics
e0a367c 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Test JS example: collectd pipeline
7d29614
@jarhodes314 @didier-wenzek
Tidy JS example
edb5e0f 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@jarhodes314 @didier-wenzek
Relay error messages from Javascript to the generic mapper logs
35510ee 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@jarhodes314 @didier-wenzek
fixup! fixup! Add JS example: collectd pipline
04ec637 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@jarhodes314 @didier-wenzek
fixup! fixup! Add JS example: collectd pipline
d9b2ed5 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@didier-wenzek
Fix memory leak
2e797ab 
The first design was naive (i.e. loading the javascript module
for each and every function call) and leading to a memory leak
(loading a module with the same name keeps the previous version of the
module in memory).

The new design fixes the issue. The memory is stable while translating to c8y
600 tedge measurements per second during 20 minutes (mode debug)

```shell
$ watch ps -p 1382888 -o args,%cpu,etimes,times,%mem,rss,vsz
COMMAND                     %CPU ELAPSED     TIME %MEM   RSS    VSZ
tedge-mapper gen            47.4    1322      627  0.0 41440 1359476

$ tedge mqtt sub 'test/output' --duration 1000 | pv | wc -l
89.3MiB 0:16:40 [91.4KiB/s] [  <=>                                                                                                                            ]
586311
```

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add benchmark options to tedge mqtt pub
e827a12 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add gen-mapper system tests
c19623f 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add a reference guide
61270b4 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Implement a circuit breaker filter
4edc775 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add a filter to compute average over a time window
96a056f 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Extract message processing from gen-mapper actor
44873ae 
The goal is to be able to use the message processor from within the
tedge cli command, notably for testing pipelines and filters.

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Add cli commands to test user-defined mappings
819e44c 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Simplify gen-mapper subscription updates
a104b2b 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Gen mapper dynamically load/reload/remove pipelines and filters
1cf271c 
Still to be improved: some event are processed twice
- `mv pipeline.toml /tmp` leads to update then remove
- `rm pipeline.toml` correctly triggers only remove
- `cp /tmp/pipeline.toml .` triggers creation and update

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Index pipelines using full paths not just filenames
9024cef 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek
Gen-mapper pipelines can drain data out MeaDB
b184a82 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek didier-wenzek marked this pull request as draft July 7, 2025 13:17
@didier-wenzek
Gen-mapper pipelines can persist data in MeaDB
cba0b4b 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@jarhodes314
add initial mea database implementation to the generic mapper
663d408 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@didier-wenzek
Add example: pipeline using mea-db
fdd578d 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@jarhodes314
Fix segfault
3e8bbf4 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@jarhodes314
Remove oldest message timestamp cache
a6dc1ec 
Signed-off-by: James Rhodes <jarhodes314@gmail.com>
@didier-wenzek
Distinguish two message sources: MQTT vs DB
8b9ce32 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@Bravo555
Add db_dump bin to read and display db contents
44968f7 
Signed-off-by: Marcel Guzik <marcel.guzik@cumulocity.com>
@didier-wenzek
Fix cascading pipelines
4ec15fb 
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
@didier-wenzek didier-wenzek self-assigned this Jul 9, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@reubenmiller reubenmiller Awaiting requested review from reubenmiller reubenmiller is a code owner

@Bravo555 Bravo555 Awaiting requested review from Bravo555 Bravo555 is a code owner

@albinsuresh albinsuresh Awaiting requested review from albinsuresh albinsuresh is a code owner

@jarhodes314 jarhodes314 Awaiting requested review from jarhodes314 jarhodes314 is a code owner

@rina23q rina23q Awaiting requested review from rina23q rina23q is a code owner

At least 1 approving review is required to merge this pull request.

Assignees

@didier-wenzek didier-wenzek

Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
4 participants
@didier-wenzek @jarhodes314 @reubenmiller @Bravo555