Support different clouds in built-in bridge #2821

jarhodes314 · 2024-04-12T09:19:28Z

Proposed changes

This adds support for AWS and Azure to the built-in bridge, as well as some clean session configuration to ensure we don't lose messages.

Previously the rules were Cumulocity rules, where all we needed to do was remove and add local prefixes, so a large section of the changes are around supporting more complex mapping logic, so a BridgeRule struct has been added to support the same configurations as mosquitto.

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)
I ran cargo fmt as mentioned in CODING_GUIDELINES
I used cargo clippy 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 · 2024-04-12T09:29:42Z

Codecov Report

Attention: Patch coverage is 82.37986% with 154 lines in your changes are missing coverage. Please review.

Project coverage is 77.8%. Comparing base (77a9f56) to head (6621920).
Report is 7 commits behind head on main.

Additional details and impacted files

Files	Coverage Δ
crates/core/c8y_api/src/smartrest/topic.rs	`96.2% <100.0%> (+<0.1%)`	⬆️
crates/core/tedge/src/bridge/aws.rs	`98.1% <100.0%> (+<0.1%)`	⬆️
crates/core/tedge/src/bridge/azure.rs	`98.1% <100.0%> (+<0.1%)`	⬆️
crates/core/tedge/src/cli/common.rs	`0.0% <ø> (ø)`
...rates/extensions/c8y_firmware_manager/src/tests.rs	`92.5% <100.0%> (ø)`
crates/extensions/c8y_mapper_ext/src/config.rs	`47.4% <ø> (ø)`
crates/extensions/c8y_mapper_ext/src/converter.rs	`83.5% <100.0%> (+<0.1%)`	⬆️
...s/c8y_mapper_ext/src/operations/config_snapshot.rs	`89.3% <100.0%> (ø)`
...ons/c8y_mapper_ext/src/operations/config_update.rs	`92.6% <100.0%> (ø)`
...s/c8y_mapper_ext/src/operations/firmware_update.rs	`90.5% <100.0%> (ø)`
... and 15 more

... and 7 files with indirect coverage changes

github-actions · 2024-04-12T10:02:15Z

Robot Results

✅ Passed	❌ Failed	⏭️ Skipped	Total	Pass %	⏱️ Duration
433	0	3	433	100	0s

albinsuresh

I didn't fully get the benefit of using raw &str form of Topics and TopicFilters in the newly added APIs. It felt like an eventual conversion to a Topic is unavoidable and hence the benefits of using Cow<&str> might not be significant.

Also, as discussed offline, the AWS bridge also needs to be added.

crates/extensions/tedge_mqtt_bridge/src/lib.rs

albinsuresh · 2024-04-21T17:30:52Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+    match (prefix, target) {
+        (prefix, target) if prefix.is_empty() => target,
+        (prefix, target) if target.is_empty() => prefix.clone(),
+        (prefix, Cow::Borrowed(target)) => format!("{prefix}{target}").into(),


Since this is the case that's gonna be used 90% of the time (since we don't passthrough any of the messages AFAIK), I'm, not sure if this overhead of using Cow here and in BridgeRule is worth it. But, that's no argument to undo something that's already written in an optimal way, which isn't overly complex either. So, let it be. At least, it was good to learn a practical use-case of Cow.

This does make a difference (albeit not a particularly notable one). Most of the Cumulocity rules make use of this, for instance. They are &'static str to begin with as the topics don't change, and the prefix is empty.

It's perhaps also worth pointing out that all we care about here is not adding a prefix, as removing a prefix is trivially done by borrowing a portion of the string slice and therefore never involves cloning.

albinsuresh · 2024-04-21T17:38:07Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+
+#[derive(Debug, thiserror::Error)]
+pub enum InvalidBridgeRule {
+    #[error("{0:?} is not a valid MQTT bridge topic prefix as is missing a trailing slash")]


Suggested change

#[error("{0:?} is not a valid MQTT bridge topic prefix as is missing a trailing slash")]

#[error("{0:?} is not a valid MQTT bridge topic prefix as it is missing a trailing slash")]

albinsuresh · 2024-04-21T17:59:24Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+}
+
+#[derive(Debug, Clone)]
+pub struct BridgeRule {


Good to document this struct with some examples for future users not familiar with the Mosquitto bridge. Esp the cases like empty topic filter which wasn't clear to me on a first look.

crates/extensions/tedge_mqtt_bridge/src/lib.rs

albinsuresh · 2024-04-21T18:38:56Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+        fn forwards_unfiltered_topic() {
+            let cloud_topic = "thinedge/devices/my-device/test-connection";
+            let rule =
+                BridgeRule::try_new("".into(), "aws/test-connection".into(), cloud_topic.into())


So, this rule means that it only subscribes to the single topic aws/test-connection and none of the sub-hierarchy like aws/test-connection/#, right? It might be good to add an additional assertion: assert_eq!(rule.apply("aws/test-connection/subtopic"), None) to highlight that.

I'm guessing that the primary use-case for defining such a rule is to allow forwarding an incoming topic to a completely unrelated outgoing topic. You could highlight that in the doc, as it wasn't clear to me on a first look.

albinsuresh · 2024-04-21T18:41:03Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+        }
+
+        #[test]
+        fn rejects_invalid_input_topic() {


Suggested change

fn rejects_invalid_input_topic() {

fn rejects_invalid_input_prefix() {

albinsuresh · 2024-04-21T18:41:15Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+        }
+
+        #[test]
+        fn rejects_invalid_output_topic() {


Suggested change

fn rejects_invalid_output_topic() {

fn rejects_invalid_output_prefix() {

albinsuresh · 2024-04-21T18:45:49Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+    prefix_to_add: Cow<'static, str>,
+}
+
+fn prepend<'a>(target: Cow<'a, str>, prefix: &Cow<'a, str>) -> Cow<'a, str> {


Looking at the usages, I feel that keeping the prefix as the first argument would have looked more intuitive.

I'm going to replace this with the + operator as it's unambiguous and we clearly have opposing opinions here.

albinsuresh · 2024-04-21T18:46:52Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+        }
+    }
+
+    mod single_converter {


Suggested change

mod single_converter {

mod topic_converter {

albinsuresh · 2024-04-21T18:56:25Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+    prefix_to_remove: Cow<'static, str>,
+    prefix_to_add: Cow<'static, str>,


Just an optional suggestion:

Suggested change

prefix_to_remove: Cow<'static, str>,

prefix_to_add: Cow<'static, str>,

incoming_prefix: Cow<'static, str>,

outgoing_prefix: Cow<'static, str>,

I prefer the original names compared to those proposed, even if not perfect.

A BridgeRule can be local_to_remote or remote_to_local, so the incoming and outgoing terms are confusing. The add and remove internal usage hints are okay as these names are only used privately.

Yeah, I also intentionally avoided the names local_prefix and remote_prefix for the same reason. Suggested incoming and outgoing more from the bridge client's perspective, more like input_prefix and output_prefix. But yeah, it was just an optional suggestion and can be ignored.

crates/core/tedge_mapper/src/aws/mapper.rs

didier-wenzek · 2024-04-22T11:33:37Z

crates/core/tedge_mapper/src/aws/mapper.rs

+    // topic to interact with the shadow of the device
+    bridge.forward_from_local("shadow/#", local_prefix, things_prefix.clone())?;
+    bridge.forward_from_remote("shadow/#", local_prefix, things_prefix)?;


How do we ensure this doesn't lead to an infinite loop?

I've added some de-duplication logic, which will detect messages that are identical to ones we have published and avoid republishing them. Using MQTT v3.1, this is I believe the best we can do to avoid infinite loops, as we can't instruct the broker not to send us messages we publish.

On a related note, it may be that we simply don't need this to be bidirectional, or we could split the incoming/outgoing topics. I don't really know anything about the AWS mapper, so I'm not sure how doable this is.

The current logic only affects the bidirectional topics, so there shouldn't be any issues for c8y/az arising from the de-duplication, and most of aws should be completely unaffected. There are also likely more robust ways to achieve this, it currently assumes messages will be delivered in the order sent, even across topics, but we also only have one bidirectional topic for the whole of thin-edge, so at this point it doesn't really matter.

crates/core/tedge_mapper/src/aws/mapper.rs

crates/core/tedge_mapper/src/az/mapper.rs

didier-wenzek · 2024-04-22T12:01:18Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+}
+
+#[derive(Debug, Clone)]
+pub struct BridgeRule {


Suggested change

pub struct BridgeRule {

struct BridgeRule {

didier-wenzek

I still have to review the bulk of the changes, notably in tests/bridge.rs.

Cargo.toml

crates/core/tedge_mapper/Cargo.toml

crates/core/tedge_mapper/src/c8y/mapper.rs

crates/core/tedge_mapper/src/az/mapper.rs

crates/core/tedge_mapper/src/aws/mapper.rs

didier-wenzek · 2024-05-06T11:54:23Z

crates/extensions/tedge_mqtt_bridge/src/topics.rs

+            // TODO should this be an error
+            .unwrap_or_else(|| panic!("Failed to convert {topic:?} with {:?}", self.0))


As long as the bridge rules are hard-coded in the mapper, it's okay to panic here.

didier-wenzek · 2024-05-06T17:05:14Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

        let (local_client, local_event_loop) = AsyncClient::new(local_config, 10);
        let (cloud_client, cloud_event_loop) = AsyncClient::new(cloud_config, 10);


sudo tedge connect c8y --test returned an unexpected exit code stdout: Sending packets to check connection.

I observe on my laptop the same behavior as on the failing system tests: no JWT token is received when requested on c8y/s/uat:

The message published on c8y/s/uat is not even processed by the c8y built-in bridge.

The built-in bridge is even stuck: processing no more messages after a first batch of 11 messages.

Changing the cap parameter from 10 to some crazy number 10000 unblocks the built-in bridge
(this change has to be done for the local_config, the cloud_config and the bidirectional_channel).

I've now understood and fixed the issue here. The issue was the change to the subscriptions. Previously we were subscribing before connecting, but with this PR, we subscribe upon receiving a ConnAck, which means we resubscribe after a dropped connection.

The issue with this is the cloud connection takes a reasonable period of time to come up after starting the mapper, and if during that time 10 messages are sent, this fills up the channel for the cloud event loop. When the cloud then successfully connects, it attempts to subscribe with recv_client.subscribe_many(&topics).await. This attempts to send a message to the event loop to process the subscription request, but the event loop is already full, so this results in a deadlock.

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

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

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

didier-wenzek

Approved.

I really like the use of cargo mutants to improve code coverage. The result gives strong confidence on this tricky piece of code.

$ cargo mutants --test-tool=nextest -p tedge_mqtt_bridge
Found 92 mutants to test
ok       Unmutated baseline in 39.6s build + 0.4s test
 INFO Auto-set test timeout to 20s
92 mutants tested in 6m 48s: 46 caught, 46 unviable

didier-wenzek · 2024-05-13T11:43:55Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+}
+
+#[async_trait::async_trait]
+#[mutants::skip]


So mutants::skip removes one missed test. But is this a good idea? Wouldn't be better to know that test coverage can still be improved?

I would at least add cargo mutants output as a comment

Suggested change

#[mutants::skip]

[mutants::skip] // missed: replace <impl MqttAck for AsyncClient>::ack -> Result<(), ClientError> with Ok(())

didier-wenzek · 2024-05-13T14:06:51Z

crates/extensions/tedge_mqtt_bridge/src/lib.rs

+#[mutants::skip]
 impl Actor for MqttBridgeActor {
    fn name(&self) -> &str {


Suggested change

#[mutants::skip]

impl Actor for MqttBridgeActor {

fn name(&self) -> &str {

impl Actor for MqttBridgeActor {

#[mutants::skip]

fn name(&self) -> &str {

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

jarhodes314 requested a review from didier-wenzek April 12, 2024 09:19

jarhodes314 requested review from albinsuresh and rina23q as code owners April 12, 2024 09:19

jarhodes314 temporarily deployed to Test Pull Request April 12, 2024 09:19 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto April 12, 2024 09:51 — with GitHub Actions Inactive

albinsuresh reviewed Apr 16, 2024

View reviewed changes

jarhodes314 temporarily deployed to Test Pull Request April 16, 2024 13:49 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto April 16, 2024 13:55 — with GitHub Actions Inactive

albinsuresh reviewed Apr 21, 2024

View reviewed changes

didier-wenzek reviewed Apr 22, 2024

View reviewed changes

jarhodes314 had a problem deploying to Test Pull Request May 2, 2024 16:10 — with GitHub Actions Failure

didier-wenzek temporarily deployed to Test Pull Request May 6, 2024 09:21 — with GitHub Actions Inactive

didier-wenzek temporarily deployed to Test Pull Request May 6, 2024 09:50 — with GitHub Actions Inactive

didier-wenzek had a problem deploying to Test Auto May 6, 2024 09:56 — with GitHub Actions Failure

didier-wenzek reviewed May 6, 2024

View reviewed changes

jarhodes314 temporarily deployed to Test Pull Request May 8, 2024 09:57 — with GitHub Actions Inactive

jarhodes314 had a problem deploying to Test Auto May 8, 2024 10:02 — with GitHub Actions Failure

jarhodes314 added 5 commits May 8, 2024 13:14

Support different clouds in built-in bridge

815d499

Update CLI to support multi-cloud built-in bridge

0f4439f

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

Add validation to MQTT topic prefix

6b38eb8

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

Avoid infinite loops when forwarding bidirectionally and add integrat…

0baca35

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

Fix deadlocking bridge on mapper startup

bbca44d

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

jarhodes314 force-pushed the 2794-diff-clouds branch from 4fec17e to bbca44d Compare May 8, 2024 12:18

jarhodes314 temporarily deployed to Test Pull Request May 8, 2024 12:18 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto May 8, 2024 12:24 — with GitHub Actions Inactive

Address review comments

400fdd9

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

Add some tests using cargo-mutants

1e4d222

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

jarhodes314 temporarily deployed to Test Pull Request May 9, 2024 14:43 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto May 10, 2024 09:01 — with GitHub Actions Inactive

Merge remote-tracking branch 'origin/main' into 2794-diff-clouds

25e0ee0

jarhodes314 temporarily deployed to Test Pull Request May 10, 2024 09:02 — with GitHub Actions Inactive

jarhodes314 temporarily deployed to Test Auto May 13, 2024 07:39 — with GitHub Actions Inactive

didier-wenzek approved these changes May 13, 2024

View reviewed changes

Tweak mutant skipping according to review comments

6621920

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

jarhodes314 temporarily deployed to Test Pull Request May 14, 2024 09:05 — with GitHub Actions Inactive

jarhodes314 enabled auto-merge May 14, 2024 11:37

jarhodes314 temporarily deployed to Test Auto May 14, 2024 11:37 — with GitHub Actions Inactive

jarhodes314 added this pull request to the merge queue May 14, 2024

Merged via the queue into thin-edge:main with commit 030efff May 14, 2024
32 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support different clouds in built-in bridge #2821

Support different clouds in built-in bridge #2821

jarhodes314 commented Apr 12, 2024 •

edited

Loading

codecov bot commented Apr 12, 2024 •

edited

Loading

github-actions bot commented Apr 12, 2024 •

edited

Loading

albinsuresh left a comment •

edited

Loading

albinsuresh Apr 21, 2024

jarhodes314 May 8, 2024 •

edited

Loading

albinsuresh Apr 21, 2024

albinsuresh Apr 21, 2024

albinsuresh Apr 21, 2024

albinsuresh Apr 21, 2024

albinsuresh Apr 21, 2024

albinsuresh Apr 21, 2024

jarhodes314 May 9, 2024

albinsuresh Apr 21, 2024

albinsuresh Apr 21, 2024

didier-wenzek Apr 22, 2024

albinsuresh Apr 22, 2024

didier-wenzek Apr 22, 2024

jarhodes314 Apr 23, 2024 •

edited

Loading

didier-wenzek Apr 22, 2024

didier-wenzek left a comment

didier-wenzek May 6, 2024

didier-wenzek May 6, 2024 •

edited

Loading

jarhodes314 May 8, 2024

didier-wenzek left a comment

didier-wenzek May 13, 2024

didier-wenzek May 13, 2024

	#[error("{0:?} is not a valid MQTT bridge topic prefix as is missing a trailing slash")]
	#[error("{0:?} is not a valid MQTT bridge topic prefix as it is missing a trailing slash")]

	fn rejects_invalid_input_topic() {
	fn rejects_invalid_input_prefix() {

	fn rejects_invalid_output_topic() {
	fn rejects_invalid_output_prefix() {

		prefix_to_remove: Cow<'static, str>,
		prefix_to_add: Cow<'static, str>,

		// TODO should this be an error
		.unwrap_or_else(\|\| panic!("Failed to convert {topic:?} with {:?}", self.0))

		let (local_client, local_event_loop) = AsyncClient::new(local_config, 10);
		let (cloud_client, cloud_event_loop) = AsyncClient::new(cloud_config, 10);

	#[mutants::skip]
	[mutants::skip] // missed: replace <impl MqttAck for AsyncClient>::ack -> Result<(), ClientError> with Ok(())

Support different clouds in built-in bridge #2821

Support different clouds in built-in bridge #2821

Conversation

jarhodes314 commented Apr 12, 2024 • edited Loading

Proposed changes

Types of changes

Paste Link to the issue

Checklist

Further comments

codecov bot commented Apr 12, 2024 • edited Loading

Codecov Report

github-actions bot commented Apr 12, 2024 • edited Loading

Robot Results

albinsuresh left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jarhodes314 May 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jarhodes314 Apr 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

didier-wenzek left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

didier-wenzek May 6, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

didier-wenzek left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jarhodes314 commented Apr 12, 2024 •

edited

Loading

codecov bot commented Apr 12, 2024 •

edited

Loading

github-actions bot commented Apr 12, 2024 •

edited

Loading

albinsuresh left a comment •

edited

Loading

jarhodes314 May 8, 2024 •

edited

Loading

jarhodes314 Apr 23, 2024 •

edited

Loading

didier-wenzek May 6, 2024 •

edited

Loading