Skip to content

Commit c45a18f

Browse files
jainrocksjainrocks
authored
Discovery numid stitching to site model (#1209)
1 parent 213cf0d commit c45a18f

File tree

7 files changed

+215
-45
lines changed

7 files changed

+215
-45
lines changed

‎services/src/main/java/com/google/bos/iot/core/reconcile/SourceRepoMessageUtils.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,11 +29,22 @@ public class SourceRepoMessageUtils {
2929
*/
3030
public static Map<String, Object> parseSourceRepoMessageData(PubsubMessage message)
3131
throws IOException {
32+
Map<String, Object> rawMap = parseSourceRepoMessageDataToRawMap(message);
33+
return flattenNestedMap(rawMap, ".");
34+
}
35+
36+
/**
37+
* Parse the received message to get a raw map.
38+
*
39+
* @param message Pub Sub message
40+
*/
41+
public static Map<String, Object> parseSourceRepoMessageDataToRawMap(PubsubMessage message)
42+
throws IOException {
3243
String messageJson = message.getData().toString(StandardCharsets.UTF_8);
3344
TypeReference<Map<String, Object>> typeRef = new TypeReference<>() {
3445
};
3546
Map<String, Object> rawMap = OBJECT_MAPPER.readValue(messageJson, typeRef);
36-
return flattenNestedMap(rawMap, ".");
47+
return rawMap;
3748
}
3849

3950
/**

‎services/src/main/java/com/google/daq/mqtt/mapping/MappingService.java‎

Lines changed: 92 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package com.google.daq.mqtt.mapping;
22

3-
import static com.google.bos.iot.core.reconcile.SourceRepoMessageUtils.parseSourceRepoMessageData;
3+
import static com.google.bos.iot.core.reconcile.SourceRepoMessageUtils.parseSourceRepoMessageDataToRawMap;
44
import static com.google.udmi.util.GeneralUtils.isNotEmpty;
55

66
import com.google.daq.mqtt.registrar.Registrar;
@@ -31,7 +31,9 @@ public class MappingService extends AbstractPollingService {
3131
private static final String DISCOVERY_NODE_DEVICE_ID_FIELD = "deviceId";
3232
private static final String DISCOVERY_TIMESTAMP = "generation";
3333
private static final String TRIGGER_BRANCH = "discovery";
34+
private static final String TRIGGER_BRANCH_IoT = "discovery_iot";
3435
private static final String DEFAULT_TARGET_BRANCH = "main";
36+
private static final String GATEWAY_ID_FIELD = "gatewayId";
3537
private final String projectSpec;
3638

3739
/**
@@ -77,60 +79,113 @@ public MappingService(String projectTarget, String projectSpec, String siteModel
7779

7880
@Override
7981
protected void handleMessage(PubsubMessage message) throws Exception {
80-
Map<String, Object> messageData = parseSourceRepoMessageData(message);
82+
Map<String, Object> messageData = parseSourceRepoMessageDataToRawMap(message);
8183
Integer eventNumber = (Integer) messageData.getOrDefault(EVENT_NUMBER_FIELD, 0);
82-
LOGGER.info("Received event no. from message: {}", eventNumber);
8384
if (eventNumber < 0) {
84-
String mappingFamily = (String) messageData.getOrDefault(FAMILY_FIELD, "");
85-
String registryId = message.getAttributesOrDefault(REGISTRY_ID_FIELD, "");
85+
processDiscoveryComplete(message, messageData);
86+
} else if (message.getAttributesMap().containsKey(GATEWAY_ID_FIELD)) {
87+
stitchDeviceProperties(message, messageData);
88+
}
89+
}
8690

87-
if (registryId.isEmpty()) {
88-
LOGGER.error("Registry Id not found for the message.");
89-
return;
90-
}
91+
private void processDiscoveryComplete(PubsubMessage message, Map<String, Object> messageData)
92+
throws Exception {
93+
LOGGER.info(
94+
"Received event no. from message: {}", messageData.getOrDefault(EVENT_NUMBER_FIELD, 0));
95+
String mappingFamily = (String) messageData.getOrDefault(FAMILY_FIELD, "");
96+
String registryId = message.getAttributesOrDefault(REGISTRY_ID_FIELD, "");
9197

92-
Instant now = Instant.now();
93-
String currentTimestamp = DateTimeFormatter.ISO_INSTANT.format(now);
94-
String discoveryTimestamp = (String) messageData.getOrDefault(DISCOVERY_TIMESTAMP,
95-
currentTimestamp);
96-
String discoveryNodeDeviceId = message.getAttributesOrDefault(
97-
DISCOVERY_NODE_DEVICE_ID_FIELD, "");
98-
if (discoveryNodeDeviceId.isEmpty()) {
99-
LOGGER.error("Discovery Node device Id not found for the message received.");
100-
return;
101-
}
102-
LOGGER.info("Starting Mapping process for registry: {}, family: {}, discoverNode deviceId:"
103-
+ " {}", registryId, mappingFamily, discoveryNodeDeviceId);
98+
if (registryId.isEmpty()) {
99+
LOGGER.error("Registry Id not found for the message.");
100+
return;
101+
}
102+
103+
Instant now = Instant.now();
104+
String currentTimestamp = DateTimeFormatter.ISO_INSTANT.format(now);
105+
String discoveryTimestamp =
106+
(String) messageData.getOrDefault(DISCOVERY_TIMESTAMP, currentTimestamp);
107+
String discoveryNodeDeviceId =
108+
message.getAttributesOrDefault(DISCOVERY_NODE_DEVICE_ID_FIELD, "");
109+
if (discoveryNodeDeviceId.isEmpty()) {
110+
LOGGER.error("Discovery Node device Id not found for the message received.");
111+
return;
112+
}
113+
LOGGER.info(
114+
"Starting Mapping process for registry: {}, family: {}, discoverNode deviceId:" + " {}",
115+
registryId, mappingFamily, discoveryNodeDeviceId);
116+
117+
withRepository(registryId, TRIGGER_BRANCH,
118+
repository -> {
119+
String udmiModelPath = repository.getUdmiModelPath();
120+
(new Registrar())
121+
.processArgs(new ArrayList<>(List.of(udmiModelPath, projectSpec)))
122+
.execute();
123+
MappingAgent mappingAgent =
124+
new MappingAgent(new ArrayList<>(List.of(udmiModelPath, projectSpec)));
125+
mappingAgent.processMapping(
126+
new ArrayList<>(List.of(discoveryNodeDeviceId, discoveryTimestamp, mappingFamily)));
127+
});
128+
}
104129

105-
SourceRepository repository = initRepository(registryId);
106-
if (repository.clone(DEFAULT_TARGET_BRANCH)) {
130+
private void stitchDeviceProperties(PubsubMessage message,
131+
Map<String, Object> messageData)
132+
throws Exception {
133+
String registryId = message.getAttributesOrDefault(REGISTRY_ID_FIELD, "");
134+
135+
if (registryId.isEmpty()) {
136+
LOGGER.error("Registry Id not found for the message.");
137+
return;
138+
}
139+
LOGGER.info("Starting Stitching process for registry: {}", registryId);
140+
141+
withRepository(registryId, TRIGGER_BRANCH_IoT,
142+
repository -> {
143+
Object devicesObject = messageData.get("devices");
144+
if (!(devicesObject instanceof Map)) {
145+
LOGGER.warn("Skipping Processing: Received message without a valid 'devices' map.");
146+
return;
147+
}
148+
String udmiModelPath = repository.getUdmiModelPath();
149+
@SuppressWarnings("unchecked")
150+
Map<String, Map<String, Object>> devices =
151+
(Map<String, Map<String, Object>>) devicesObject;
152+
MappingAgent mappingAgent =
153+
new MappingAgent(new ArrayList<>(List.of(udmiModelPath, projectSpec)));
154+
155+
mappingAgent.stitchProperties(devices);
156+
});
157+
}
158+
159+
private void withRepository(String registryId, String branchPrefix, RepositoryConsumer work)
160+
throws Exception {
161+
SourceRepository repository = initRepository(registryId);
162+
if (repository.clone(DEFAULT_TARGET_BRANCH)) {
163+
try {
107164
String timestamp = LocalDateTime.now()
108165
.format(DateTimeFormatter.ofPattern("yyyyMMdd.HHmmss"));
109166

110-
String exportBranch = String.format("%s/%s/%s", TRIGGER_BRANCH, SERVICE_NAME, timestamp);
167+
String exportBranch = String.format("%s/%s/%s", branchPrefix, SERVICE_NAME, timestamp);
111168
if (!repository.checkoutNewBranch(exportBranch)) {
112-
throw new RuntimeException("Unable to create and checkout export branch "
113-
+ exportBranch);
169+
throw new RuntimeException("Unable to create and checkout export branch " + exportBranch);
114170
}
115171

116-
String udmiModelPath = repository.getUdmiModelPath();
117-
(new Registrar()).processArgs(new ArrayList<>(List.of(udmiModelPath, projectSpec)))
118-
.execute();
119-
MappingAgent mappingAgent = new MappingAgent(new ArrayList<>(
120-
List.of(udmiModelPath, projectSpec)));
121-
mappingAgent.processMapping(new ArrayList<>(List.of(discoveryNodeDeviceId,
122-
discoveryTimestamp, mappingFamily)));
172+
work.accept(repository);
123173

124174
LOGGER.info("Committing and pushing changes to branch {}", exportBranch);
125175
if (!repository.commitAndPush("Merge changes from source: MappingService")) {
126-
throw new RuntimeException("Unable to commit and push changes to branch "
127-
+ exportBranch);
176+
throw new RuntimeException("Unable to commit and push changes to branch " + exportBranch);
128177
}
129178
LOGGER.info("Export operation complete.");
179+
} finally {
130180
repository.delete();
131-
} else {
132-
LOGGER.error("Could not clone repository! PR message was not published!");
133181
}
182+
} else {
183+
LOGGER.error("Could not clone repository! PR message was not published!");
134184
}
135185
}
186+
187+
@FunctionalInterface
188+
private interface RepositoryConsumer {
189+
void accept(SourceRepository repository) throws Exception;
190+
}
136191
}

‎tests/sites/mapping/devices/GAT-123/expected/errors.map‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,8 @@
11
Exceptions for GAT-123
2+
While loading credentials for local device GAT-123
3+
Found 0 credentials
4+
expected files mismatch
5+
Missing files: [rsa_private.pem, rsa_private.pkcs8, rsa_public.pem]
26
While converting device config
37
unrecognized schema version null
48
1 schema violations found

‎tests/sites/mapping/devices/GAT-123/expected/metadata_norm.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,8 @@
11
{
2+
"cloud": {
3+
"auth_type": "RS256",
4+
"num_id": "987654321"
5+
},
26
"discovery": {
37
"families": {
48
"vendor": {}

‎tests/sites/mapping/devices/GAT-123/metadata.json‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,13 @@
11
{
2-
"timestamp": "2024-04-02T17:30:04Z",
3-
"hash": "0001d237",
4-
"discovery": {
5-
"families": {
6-
"vendor": {
7-
}
2+
"timestamp" : "2024-04-02T17:30:04Z",
3+
"hash" : "0001d237",
4+
"cloud" : {
5+
"auth_type" : "RS256",
6+
"num_id" : "987654321"
7+
},
8+
"discovery" : {
9+
"families" : {
10+
"vendor" : { }
811
}
912
}
10-
}
13+
}

‎validator/src/main/java/com/google/daq/mqtt/mapping/MappingAgent.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,7 @@ public class MappingAgent {
8383
private long extrasDeletionTimeInMillis;
8484
private long devicesDeletionTimeInMillis;
8585
private long discoveryEventCompletionTimeInMillis;
86+
private static final String DEVICE_NUM_ID_FIELD = "num_id";
8687

8788
/**
8889
* Create an agent given the configuration.
@@ -452,4 +453,31 @@ public void processMapping(ArrayList<String> argsList) {
452453

453454
System.err.println("Mapping process is completed");
454455
}
456+
457+
/**
458+
* Stitches the properties from a discovery pub/sub message into the site model.
459+
*
460+
* @param devices devices with properties from clearblade
461+
*/
462+
public void stitchProperties(Map<String, Map<String, Object>> devices) {
463+
for (Map.Entry<String, Map<String, Object>> device : devices.entrySet()) {
464+
String deviceId = device.getKey();
465+
Map<String, Object> deviceData = device.getValue();
466+
if (deviceData.containsKey(DEVICE_NUM_ID_FIELD)) {
467+
String numId = (String) deviceData.get(DEVICE_NUM_ID_FIELD);
468+
if (numId != null && !numId.isBlank() && siteModel.deviceExists(deviceId)) {
469+
Metadata metadata = siteModel.getMetadata(deviceId);
470+
if (metadata.cloud == null) {
471+
metadata.cloud = new CloudModel();
472+
}
473+
metadata.cloud.num_id = numId;
474+
siteModel.updateMetadata(deviceId, metadata);
475+
}
476+
}
477+
}
478+
}
479+
480+
public SiteModel getSiteModel() {
481+
return this.siteModel;
482+
}
455483
}

‎validator/src/test/java/com/google/daq/mqtt/mapping/MappingAgentTest.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,20 +4,29 @@
44
import static com.google.daq.mqtt.TestCommon.REGISTRY_ID;
55
import static com.google.udmi.util.SiteModel.MOCK_PROJECT;
66
import static org.junit.Assert.assertEquals;
7+
import static org.junit.Assert.assertNotNull;
8+
import static org.junit.Assert.assertNull;
79

810
import com.google.common.collect.ImmutableList;
911
import com.google.daq.mqtt.util.IotMockProvider.MockAction;
12+
import com.google.udmi.util.SiteModel;
1013
import java.util.ArrayList;
14+
import java.util.HashMap;
1115
import java.util.List;
16+
import java.util.Map;
1217
import org.junit.Test;
18+
import udmi.schema.CloudModel;
19+
import udmi.schema.CloudModel.Auth_type;
1320
import udmi.schema.ExecutionConfiguration;
21+
import udmi.schema.Metadata;
1422

1523
/**
1624
* Test the mapping agent.
1725
*/
1826
public class MappingAgentTest {
1927

2028
private static final String CONFIG_SOURCE = "../tests/sites/mapping/testing_placeholder.json";
29+
private static final String TEST_NUM_ID = "987654321";
2130

2231
@Test
2332
public void initiate_discovery() {
@@ -40,4 +49,60 @@ private ExecutionConfiguration getExecutionConfig() {
4049
executionConfiguration.site_name = REGISTRY_ID;
4150
return executionConfiguration;
4251
}
52+
53+
private Map<String, Map<String, Object>> createDevicesMap(String deviceId, String numId) {
54+
Map<String, Map<String, Object>> devices = new HashMap<>();
55+
Map<String, Object> deviceData = new HashMap<>();
56+
57+
if (numId != null) {
58+
deviceData.put("num_id", numId);
59+
}
60+
devices.put(deviceId, deviceData);
61+
return devices;
62+
}
63+
64+
@Test
65+
public void stitchProperties_happyPath() {
66+
MappingAgent mappingAgent = new MappingAgent(getExecutionConfig());
67+
mappingAgent.stitchProperties(createDevicesMap(GATEWAY_ID, TEST_NUM_ID));
68+
69+
SiteModel siteModel = mappingAgent.getSiteModel();
70+
Metadata metadata = siteModel.getMetadata(GATEWAY_ID);
71+
72+
assertNotNull("Cloud model should have been created.", metadata.cloud);
73+
assertEquals("The num_id was not stitched correctly.", TEST_NUM_ID, metadata.cloud.num_id);
74+
}
75+
76+
@Test
77+
public void stitchProperties_deviceNotInSiteModel() {
78+
MappingAgent mappingAgent = new MappingAgent(getExecutionConfig());
79+
String unknownTest = "unknown-device";
80+
mappingAgent.stitchProperties(createDevicesMap(unknownTest, TEST_NUM_ID));
81+
82+
SiteModel siteModel = mappingAgent.getSiteModel();
83+
Metadata metadata = siteModel.getMetadata(unknownTest);
84+
assertNull(metadata);
85+
}
86+
87+
@Test
88+
public void stitchProperties_preservesExistingCloudData() {
89+
MappingAgent mappingAgent = new MappingAgent(getExecutionConfig());
90+
91+
// Manually add existing cloud data to the site model before the test.
92+
SiteModel siteModel = mappingAgent.getSiteModel();
93+
Metadata metadata = siteModel.getMetadata(GATEWAY_ID);
94+
metadata.cloud = new CloudModel();
95+
metadata.cloud.auth_type = Auth_type.RS_256; // Some pre-existing data
96+
siteModel.updateMetadata(GATEWAY_ID, metadata);
97+
98+
Map<String, Map<String, Object>> devices = createDevicesMap(GATEWAY_ID, TEST_NUM_ID);
99+
mappingAgent.stitchProperties(devices);
100+
101+
Metadata updatedMetadata = siteModel.getMetadata(GATEWAY_ID);
102+
assertNotNull(updatedMetadata.cloud);
103+
assertEquals("The num_id was not stitched correctly.", TEST_NUM_ID,
104+
updatedMetadata.cloud.num_id);
105+
assertEquals("Existing cloud data should be preserved.", "RS256",
106+
updatedMetadata.cloud.auth_type.toString());
107+
}
43108
}

0 commit comments

Comments
 (0)