Handle legacy state messages (#710)

.gencode_hash.txt

@@ -12,7 +12,7 @@ d40bfc9f4a30c56986435dc08f1e5f42401e5ac043359a1e359011c913cad673  gencode/docs/c
 71fe25d3b17d1dc87b52c049a3235c20527f73e12ffa53fa4800a15e53f73732  gencode/docs/event_pointset.html
 acd50976d1afd771d55177db9a1b9452a0c6a35e3e15b2e444c799991294808d  gencode/docs/event_system.html
 816481f69d3b1bdeb2224eaad6e3751a991d20eb98294d89f888b1323505209c  gencode/docs/event_validation.html
-21a66523e339c2d6424761e7dbcc9bf14a1a294a684361ff060a91792e6d4fc0  gencode/docs/metadata.html
+ae5762dbe2304e44cd742e7236d2de6987bdc1384b21093490cfcd7e33a5509f  gencode/docs/metadata.html
 4db86e0b979a1873d73da07d68a432df39282cb8b94ea509514d2b411411d55e  gencode/docs/monitoring.html
 180b32717db748e164a185b163ef9a97aa83d9d6add306283d5b9852d04af947  gencode/docs/persistent_device.html
 5d039d607af9ec75ee552dfe36b16c702687ea16f5663f41fc49b4533b86e00d  gencode/docs/properties.html
@@ -78,7 +78,7 @@ ccc7c234dd522a91d387d55573677681bbc5a6926f0fbf1101c2e9f607cafd63  gencode/java/u
 39f4cf5f89dc52f24ea02246e17668ebc85ed0fab351e69138e477bcc048d79c  gencode/java/udmi/schema/MappingEvent.java
 052b6b7f9ca8173901d9d2fae314f5858e034f20efa5d184780b800bd870cb2b  gencode/java/udmi/schema/MappingEventEntity.java
 d2bf4eea0ca3df47b9ffe31481a52170e2d2bc3a0e7f2eab582e93cc20ccc886  gencode/java/udmi/schema/MappingState.java
-7668ef7a34380a13454de3e027972cab9fdafd858b8af6be90116702701556aa  gencode/java/udmi/schema/Metadata.java
+476f542353b69649eea8a63fe53a01d83a1d62a33eae140241b107525024dda5  gencode/java/udmi/schema/Metadata.java
 df64e4ddbf543ac70e7c2af9d3fbc20ffe3dff68c6718aa9ceadab7f64d3d171  gencode/java/udmi/schema/Metrics.java
 14722df90406cdcd29c7c946e5fcdf6a3e513eb879a7fbb493801fc0d9093504  gencode/java/udmi/schema/Monitoring.java
 b3b7ef6cae004d5fdb4052f860df0aa583df87bb798a027fc769aeaf16489789  gencode/java/udmi/schema/MonitoringMetric.java
@@ -164,7 +164,7 @@ ddf849bfeb2b87d071cefd5e6feacabc57375a7fff6d72b6d42ffb89f33c859b  gencode/python
 d89500d1e1c88ac754ddc5abd13135dd140d40c60568ca9294f6fe384b0b12ef  gencode/python/udmi/schema/event_system.py
 4361f48caba6a948a5916f67c8168215721cb3cc2384649f0336804785f01f1d  gencode/python/udmi/schema/event_validation.py
 633c85b0999d969310918bb03e9a40a08960f420b862340045a0290f8f5c4fe5  gencode/python/udmi/schema/event_validation_device.py
-30dd2d2029ba3abd6f958edd105033f7fba73fb302895c429f56ccb35e9d755b  gencode/python/udmi/schema/metadata.py
+c7304b2070e6d930b4d101263b90768b036f7fb7e352a6e5147b83a5523987f4  gencode/python/udmi/schema/metadata.py
 6e763776c8fc5b5a980b490873b1ee7613e15c5b1dee151278f7eb941e897901  gencode/python/udmi/schema/model_cloud.py
 805cc8dcb29732d1965bbd533b12d2bc4966d584f05b8a478dd2cac98fd99d52  gencode/python/udmi/schema/model_discovery.py
 44d057dffcac3a994e7b894f8e1ebb643ba49d7dbfbb0bfea9cc04f97af0bc47  gencode/python/udmi/schema/model_discovery_family.py

.github/workflows/testing.yml

@@ -2,6 +2,7 @@ name: UDMI test suite
 
 on:
   push:
+  workflow_dispatch:
   pull_request:
   schedule:
     - cron: '0 13 * * *'
@@ -40,7 +41,7 @@ jobs:
         run: bin/test_mosquitto
       - name: bin/run_tests udmis_tests
         run: bin/run_tests udmis_tests
-      - name: all test output
+      - name: test output
         if: ${{ always() }}
         run: more out/test_* | cat
       - name: Log in to the github registry

bin/pull_message

@@ -21,20 +21,20 @@ gcloud --format=json --project=$project_id pubsub subscriptions pull $subscripti
 subType=$(jq -r '.[0].message.attributes.subType' $out_file 2> /dev/null)
 subFolder=$(jq -r '.[0].message.attributes.subFolder' $out_file 2> /dev/null)
 deviceId=$(jq -r '.[0].message.attributes.deviceId' $out_file 2> /dev/null)
-timestamp=$(jq -r '.[0].message.data' $out_file 2>/dev/null | base64 --decode | jq -r .timestamp 2> /dev/null) || true
-data=$(jq -r .[0].message.data /tmp/captured.json  | base64 --decode)
-
-out_folder=$ROOT/out/devices/$deviceId
-mkdir -p $out_folder
+registryId=$(jq -r '.[0].message.attributes.deviceRegistryId' $out_file 2> /dev/null)
+timestamp=$(jq -r '.[0].message.publishTime' $out_file 2> /dev/null)
+data=$(jq -r .[0].message.data $out_file | base64 --decode)
 
 if [[ $subType == null ]]; then
     subType=event
 fi
 
 if [[ $subFolder != null ]]; then
-    echo $subFolder $subType $deviceId $timestamp
-    out_file=$out_folder/${timestamp}_${subFolder}_${subType}.json
+    out_file=$ROOT/out/registries/$registryId/devices/$deviceId/${timestamp}_${subFolder}_${subType}.json
+    echo $out_file
+    mkdir -p $(dirname $out_file)
     echo $data | jq . > $out_file
 else
     echo No matching messages found.
+    false
 fi

bin/run_tests

@@ -88,7 +88,7 @@ case "$1" in
 esac
 
 if [[ -s $failures ]]; then
-    echo ====== Test failures: $(tr '\n' ' ' <$failures)
+    echo ====== Failures for $1: $(tr '\n' ' ' <$failures)
     false
 else
     echo ====== All tests passed for $1

bin/test_regclean

@@ -101,6 +101,10 @@ fi
 echo $pubber_config:
 cat $pubber_config
 
+echo Corrupting site model to check error handling...
+mkdir -p $site_path/devices/XXX-1
+echo { > $site_path/devices/XXX-1/metadata.json
+
 echo Clean out the registry to make sure devices get removed...
 bin/registrar $site_arg $registrar_project -d
 

bin/test_registrar

@@ -19,7 +19,7 @@ exit_status=clean
 echo Found ${clean_devices} clean devices.
 [ "${clean_devices}" == $EXPECTED_CLEAN ] || exit_status=expected_clean
 
-device_files=$(ls -d ${TEST_SITE}/devices/*)
+device_files=$(ls -d ${TEST_SITE}/devices/* | fgrep -v XXX)
 
 # Individual tests for generated config values
 echo
@@ -58,6 +58,7 @@ function check_norm {
 
 sm_devices=$(check_norm)
 echo Devices with correct site_metadata: ${sm_devices}
+
 [ "${sm_devices}" == "${device_files}" ] || exit_status=site_metadata
 
 mv ${TEST_SITE}/site_metadata.json ${TEST_SITE}/_site_metadata.json

bin/test_sites

@@ -29,7 +29,7 @@ shift $((OPTIND-1))
 function redact_files {
     for file in $@; do
         sed -E -i \
-            -e 's-While loading.*udmi/tests/-REDACTED/-' \
+            -e 's-oading .*udmi/tests/-REDACTED/-' \
             -e 's/at .*main\(.*\.java:[0-9]+\)/REDACTED/' \
             -e '/at .*\(.*\.java:[0-9]+\)/d' \
             -e '/\.\.\. [0-9]+ more/d' \

bin/test_trace

@@ -28,7 +28,8 @@ site_out=$site_model/out
 # Use a static/constant value to make output stable.
 export UDMI_TOOLS=test_trace
 
-rm -rf $site_out
+# Clean things out, including some potentially corrupt state.
+rm -rf $site_out $site_model/devices/XXX-1
 validator/bin/validate -- schema trace $trace_in $site_model
 
 echo Redacting output files...

common/src/main/java/com/google/udmi/util/GeneralUtils.java

@@ -415,7 +415,7 @@ public static void writeString(File metadataFile, String metadataString) {
   }
 
   public static String multiTrim(String message) {
-    return Arrays.stream(message.split("\n"))
+    return Arrays.stream(ofNullable(message).orElse("").split("\n"))
         .map(String::trim).collect(Collectors.joining(" "));
   }
 }

validator/src/main/java/com/google/daq/mqtt/registrar/MessageDowngrader.java → common/src/main/java/com/google/udmi/util/MessageDowngrader.java

@@ -1,4 +1,4 @@
-package com.google.daq.mqtt.registrar;
+package com.google.udmi.util;
 
 import static com.google.udmi.util.Common.VERSION_KEY;
 

validator/src/main/java/com/google/daq/mqtt/util/MessageUpgrader.java → common/src/main/java/com/google/udmi/util/MessageUpgrader.java

@@ -1,20 +1,17 @@
-package com.google.daq.mqtt.util;
+package com.google.udmi.util;
 
 import static com.google.udmi.util.Common.VERSION_KEY;
-import static com.google.udmi.util.GeneralUtils.CSV_JOINER;
+import static com.google.udmi.util.GeneralUtils.OBJECT_MAPPER_RAW;
 import static com.google.udmi.util.GeneralUtils.ifNotNullThen;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.node.TextNode;
-import com.google.udmi.util.GeneralUtils;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
 
 /**
  * Container class for upgrading UDMI messages from older versions.
@@ -27,6 +24,7 @@ public class MessageUpgrader {
   public static final String METADATA_SCHEMA = "metadata";
   private static final String TARGET_FORMAT = "%d.%d.%d";
   private final ObjectNode message;
+  private final JsonNode original;
   private final String schemaName;
   private final int major;
   private int patch;
@@ -41,6 +39,7 @@ public class MessageUpgrader {
   public MessageUpgrader(String schemaName, JsonNode message) {
     this.message = (ObjectNode) message;
     this.schemaName = schemaName;
+    this.original = message.deepCopy();
 
     JsonNode version = message.get(VERSION_KEY);
     String verStr =
@@ -57,18 +56,31 @@ public MessageUpgrader(String schemaName, JsonNode message) {
     }
   }
 
+  public MessageUpgrader(String schemaName, Object originalMessage) {
+    this(schemaName, OBJECT_MAPPER_RAW.valueToTree(originalMessage));
+  }
+
+  public boolean wasUpgraded() {
+    return !original.equals(message);
+  }
+
+  /**
+   * Update message to the latest standard, if necessary.
+   */
+  public Object upgrade() {
+    return upgrade(false);
+  }
+
   /**
    * Update message to the latest standard.
    *
    * @param forceUpgrade true to force a complete upgrade pass irrespective of original version
-   * @return true if the message has been altered
    */
-  public boolean upgrade(boolean forceUpgrade) {
+  public Object upgrade(boolean forceUpgrade) {
     if (major != 1) {
       throw new IllegalArgumentException("Starting major version " + major);
     }
 
-    final JsonNode original = message.deepCopy();
     boolean upgraded = false;
 
     if (forceUpgrade || minor < 0) {
@@ -102,11 +114,10 @@ public boolean upgrade(boolean forceUpgrade) {
     }
 
     if (upgraded && message.has(VERSION_KEY)) {
-      ((ObjectNode) message).put(VERSION_KEY,
-          String.format(TARGET_FORMAT, major, minor, patch));
+      message.put(VERSION_KEY, String.format(TARGET_FORMAT, major, minor, patch));
     }
 
-    return !original.equals(message);
+    return message;
   }
 
   private void upgrade_1_3_14() {
@@ -193,8 +204,8 @@ private void upgradeFirmware(ObjectNode system) {
   private TextNode sanitizeFirmwareVersion(JsonNode version) {
     if (version.isArray()) {
       List<String> values = new ArrayList<>();
-      Iterator<JsonNode> elements = ((ArrayNode) version).elements();
-      elements.forEachRemaining(item -> values.add(((TextNode) item).asText()));
+      Iterator<JsonNode> elements = version.elements();
+      elements.forEachRemaining(item -> values.add(item.asText()));
       String collect = values.stream().collect(Collectors.joining(", "));
       return new TextNode(collect);
     }