Redirect

.gencode_hash.txt

@@ -12,6 +12,7 @@ f48026471ae3cd7867bce416dc21c2fb728f48d8476a8d6e95f6acaf1d8b6cf3  gencode/docs/i
 43019239edfdad71647c40d254b9b3aec68f3deb0c808ba636530c1cf9985fe0  gencode/docs/state.html
 d39d7fe37a41c74a40080af7b0a429d201ab1fdff7444428c4b98eb7b38c332b  gencode/java/udmi/schema/Asset.java
 0825a5cec83003bb0a6488c4ed7010a04ae0d3848ef36fe01bb4e6718ba7b96d  gencode/java/udmi/schema/Aux.java
+df118b8e5abebac38579027b5a17192163f0cc97e55a8e44b847b1a965263636  gencode/java/udmi/schema/Blob.java
 d5adb804697243f97cdd8589750401654f3fab075a9aeac4f2851e46695ef11d  gencode/java/udmi/schema/BlobBlobsetConfig.java
 2c03651cb2ecda072b1418222eebb5560185669f8ffdd03021ad5ad8ff7ba3b0  gencode/java/udmi/schema/BlobBlobsetState.java
 d2c5b5aae8db27b68104fc83a1f38de0a3f1b5d683f2b13599adf24e96c7d124  gencode/java/udmi/schema/BlobsetConfig.java

.github/workflows/testing.yml

@@ -65,10 +65,15 @@ jobs:
         run: |
           # Simple checks that a redirect happened and failed
           set +x
-          fgrep registries/ZZ-TRI-FECTA/devices/GAT-123 pubber.out
-          fgrep 'system.config.parse success' pubber.out
-          fgrep registries/missing/devices/GAT-123 pubber.out
-          fgrep 'Not authorized to connect' pubber.out
+          fgrep registries/ZZ-TRI-FECTA/devices/GAT-123 pubber.out.1
+          fgrep 'system.config.parse success' pubber.out.1
+          fgrep registries/missing/devices/GAT-123 pubber.out.1
+          fgrep 'Not authorized to connect' pubber.out.1
+          # Now check the redirect-by-config setup
+          fgrep registries/ZZ-TRI-FECTA/devices/GAT-123 pubber.out.2
+          fgrep 'system.config.parse success' pubber.out.2
+          fgrep registries/reconfigure/devices/GAT-123 pubber.out.2
+          fgrep 'Not authorized to connect' pubber.out.2
       - name: pubber logs
         if: ${{ always() }}
         run: |
@@ -78,6 +83,7 @@ jobs:
     name: Sequence tests
     runs-on: ubuntu-latest
     timeout-minutes: 15
+    needs: redirect  # Access to UDMI-REFLECTOR is mutually exclusive
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-java@v1

bin/gencode_java

@@ -31,8 +31,8 @@ echo Generating code in $OUTDIR
 $JARBIN $JOPTS --source . --target $OUTDIR
 
 # There is no way to specify enum constants in generated code, so just hack it in manually.
-echo Copying Level.java
-cp -n $ROOT_DIR/etc/Level.java $OUTDIR/udmi/schema/Level.java
+echo Copying shared constants $ROOT_DIR/etc/*.java
+cp -n $ROOT_DIR/etc/*.java $OUTDIR/udmi/schema/
 
 echo Cleaning up __ classes...
 find $OUTDIR -name \*__\*.java | xargs rm

bin/reset_config

@@ -0,0 +1,26 @@
+#!/bin/bash -ex
+
+ROOT=$(realpath $(dirname $0)/..)
+
+if [[ $# != 3 ]]; then
+    echo Usage: $0 site_dir project_id device_id
+    false
+fi
+
+site_dir=$(realpath $1)
+project_id=$2
+device_id=$3
+shift 3
+cd ${ROOT}
+
+pwd
+
+device_config=/tmp/${device_id}_config.json
+cp ${site_dir}/devices/${device_id}/out/generated_config.json ${device_config}
+now_date=$(python3 -c 'import datetime; print(datetime.datetime.utcnow().isoformat() + "Z")')
+echo Setting config timestamp ${now_date}
+sponge < /dev/null || echo sponge command not found, might need workaround for osx or other
+jq .timestamp=\"${now_date}\" < ${device_config} | sponge ${device_config}
+
+echo Resetting device ${device_id} config...
+validator/bin/reflector ${site_dir} ${project_id} ${device_id} update/config:${device_config}

bin/setup_base

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-sudo apt-get install -y hxtools
+sudo apt-get install -y hxtools moreutils
 
 python3 -m venv venv
 

bin/test_redirect

@@ -33,10 +33,19 @@ if [[ -n $pids ]]; then
     kill $pids
 fi
 
-echo Writing pubber output to $PUBBER_OUT
+bin/reset_config $site_path $project_id $device_id
+
 echo bin/pubber $site_path $project_id $device_id $serial_no
 
+echo Writing pubber output to $PUBBER_OUT.1
 result=0
-timeout 5m bin/pubber $site_path $project_id $device_id $serial_no redirectRegistry=missing > $PUBBER_OUT 2>&1 || result=$?
-
+timeout 5m bin/pubber $site_path $project_id $device_id $serial_no redirectRegistry=missing > $PUBBER_OUT.1 2>&1 || result=$?
 echo Pubber exit code $result
+
+echo Writing pubber output to $PUBBER_OUT.2
+bin/pubber $site_path $project_id $device_id $serial_no > $PUBBER_OUT.2 2>&1 &
+
+# Wait for initial connection, then reset config to redirect
+sleep 20
+bin/reset_config $site_path $project_id $device_id #redirect_config.json
+sleep 20

bin/test_validator

@@ -3,7 +3,7 @@
 # Force consistent sort order
 export LC_ALL=C
 
-ROOT_DIR=$(dirname $0)/..
+ROOT_DIR=$(realpath $(dirname $0)/..)
 cd $ROOT_DIR
 
 if [[ $# != 1 ]]; then
@@ -38,6 +38,8 @@ vpid=$!
 sleep 10
 echo Started validator pid $vpid
 
+bin/reset_config $site_path $project_id $device_id
+
 pubber/bin/build
 
 echo Writing pubber output to $PUBBER_OUT
@@ -65,7 +67,7 @@ sleep 30
 
 pids=`ps ax | fgrep pubber | fgrep java | awk '{print $1}'`
 echo Killing pids $vpid $pids
-kill $vpid $pids
+kill $vpid $pids || true
 
 outfiles=`find $site_path/out/devices -name \*.out | sort` || true
 if [[ -z $outfiles ]]; then

bin/validator

@@ -23,6 +23,9 @@ fi
 echo Building validator...
 $ROOT_DIR/validator/bin/build > /dev/null
 
+echo Files built
+(cd $ROOT_DIR; find validator/build -type f) || true
+
 echo Running tools version `(cd $ROOT_DIR; git describe)`
 
 if [[ -n $subscription ]]; then

etc/Blob.java

@@ -0,0 +1,8 @@
+package udmi.schema;
+
+// This class is manually curated and then copied into the gencode directory. Look for the
+// proper source and don't be fooled! These are general constants for blob management.
+public abstract class Blob {
+    public static final String FINAL_PHASE = "final";
+    public static final String IOT_CONFIG_BLOB = "_iot_config";
+}

pubber/src/main/java/daq/pubber/Pubber.java

@@ -1,6 +1,8 @@
 package daq.pubber;
 
 import static java.util.stream.Collectors.toMap;
+import static udmi.schema.Blob.FINAL_PHASE;
+import static udmi.schema.Blob.IOT_CONFIG_BLOB;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -43,6 +45,7 @@
 import org.apache.http.ConnectionClosedException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import udmi.schema.BlobBlobsetConfig;
 import udmi.schema.CloudModel.Auth_type;
 import udmi.schema.Config;
 import udmi.schema.DiscoveryConfig;
@@ -136,6 +139,8 @@ public class Pubber {
   private Consumer<String> onDone;
   private boolean publishingLog;
   private Map<String, Metadata> allMetadata;
+  private String appliedEndpoint;
+  private EndpointConfiguration extractedEndpoint;
 
   /**
    * Start an instance from a configuration file.
@@ -486,7 +491,7 @@ private void sendMessages() {
       updatePoints();
       sendDeviceMessage();
       flushDirtyState();
-      // Some things can't be done from a on-message callback, so do them here instead.
+      // Some things can't be done from an on-message callback, so do them here instead.
       maybeRedirectEndpoint();
     } catch (Exception e) {
       error("Fatal error during execution", e);
@@ -576,6 +581,7 @@ private void initializeMqtt() {
     }
     Preconditions.checkState(mqttPublisher == null, "mqttPublisher already defined");
     ensureKeyBytes();
+    appliedEndpoint = toJson(configuration.endpoint);
     mqttPublisher = new MqttPublisher(configuration, this::publisherException);
     if (configuration.gatewayId != null) {
       mqttPublisher.registerHandler(configuration.gatewayId, CONFIG_TOPIC,
@@ -587,13 +593,22 @@ private void initializeMqtt() {
         this::configHandler, Config.class);
   }
 
+  private String toJson(Object configuration) {
+    try {
+      return OBJECT_MAPPER.writeValueAsString(configuration);
+    } catch (Exception e) {
+      throw new RuntimeException("While converting object to string", e);
+    }
+  }
+
   private void ensureKeyBytes() {
     if (configuration.keyBytes != null) {
       return;
     }
     Preconditions.checkNotNull(configuration.keyFile, "configuration keyFile not defined");
     info("Loading device key bytes from " + configuration.keyFile);
     configuration.keyBytes = getFileBytes(configuration.keyFile);
+    configuration.keyFile = null;
   }
 
   private String getDeviceKeyPrefix() {
@@ -705,29 +720,64 @@ private void processConfigUpdate(Config config) {
       actualInterval = updateSystemConfig(config.pointset);
       updatePointsetConfig(config.pointset);
       updateDiscoveryConfig(config.discovery);
+      extractedEndpoint = extractEndpointBlobConfig();
     } else {
       info(getTimestamp() + " defaulting empty config");
       actualInterval = DEFAULT_REPORT_SEC * 1000;
     }
     maybeRestartExecutor(actualInterval);
   }
 
+  private EndpointConfiguration extractEndpointBlobConfig() {
+    try {
+      String iotConfig = extractConfigBlob(IOT_CONFIG_BLOB);
+      if (iotConfig == null) {
+        return null;
+      }
+      return OBJECT_MAPPER.readValue(iotConfig, EndpointConfiguration.class);
+    } catch (Exception e) {
+      throw new RuntimeException("While extracting endpoint blob config", e);
+    }
+  }
+
   private void maybeRedirectEndpoint() {
     String redirectRegistry = configuration.options.redirectRegistry;
-    if (redirectRegistry == null || redirectRegistry.equals(configuration.endpoint.registryId)
-        || configLatch.getCount() > 0) {
+    if (extractedEndpoint != null && !toJson(extractedEndpoint).equals(appliedEndpoint)) {
+      info("New config blob endpoint detected");
+      configuration.endpoint = extractedEndpoint;
+    } else if (redirectRegistry != null && configLatch.getCount() <= 0
+        && !redirectRegistry.equals(configuration.endpoint.registryId)) {
+      info("Mismatched redirectRegistry detected");
+      configuration.endpoint.registryId = redirectRegistry;
+    } else {
       return;
     }
     try {
       disconnectMqtt();
-      configuration.endpoint.registryId = redirectRegistry;
       initializeMqtt();
       startConnection(onDone);
     } catch (Exception e) {
       throw new RuntimeException("While redirecting connection endpoint", e);
     }
   }
 
+  private String extractConfigBlob(String blobName) {
+    try {
+      if (deviceConfig == null || deviceConfig.blobset == null
+          || deviceConfig.blobset.blobs == null) {
+        return null;
+      }
+      BlobBlobsetConfig blobBlobsetConfig = deviceConfig.blobset.blobs.get(blobName);
+      if (blobBlobsetConfig != null && FINAL_PHASE.equals(blobBlobsetConfig.phase)
+          && blobBlobsetConfig.base64 != null) {
+        return new String(Base64.getDecoder().decode(blobBlobsetConfig.base64));
+      }
+      return null;
+    } catch (Exception e) {
+      throw new RuntimeException("While extracting config blob " + blobName, e);
+    }
+  }
+
   private void updateDiscoveryConfig(DiscoveryConfig discovery) {
     DiscoveryConfig discoveryConfig = discovery == null ? new DiscoveryConfig() : discovery;
     if (deviceState.discovery == null) {

validator/bin/build

@@ -14,7 +14,7 @@ cd $ROOT
 echo Building validataor in $PWD
 
 rm -rf build
-./gradlew shadow $check
+./gradlew --info shadow $check
 
 ls -l build/libs/validator-1.0-SNAPSHOT-all.jar
 

validator/bin/reflector

@@ -0,0 +1,24 @@
+#!/bin/bash -e
+
+ROOT=$(realpath $(dirname $0)/../..)
+
+if [[ $# < 4 ]]; then
+    echo Usage: $0 site_dir project_id device_id directive [directives...]
+    echo "  Directive is something like update/config:sites/udmi_site_model/devices/AHU-1/out/generated_config.json"
+    false
+fi
+
+site_dir=$(realpath $1)
+project_id=$2
+device_id=$3
+shift 3
+cd $ROOT
+
+validator/bin/build
+
+jarfile=validator/build/libs/validator-1.0-SNAPSHOT-all.jar
+mainclass=com.google.daq.mqtt.validator.Reflector
+
+cmd="java -cp $jarfile $mainclass -p $project_id -s $site_dir -d $device_id $*"
+echo $cmd
+$cmd

validator/src/main/java/com/google/bos/iot/core/proxy/IotReflectorClient.java

@@ -59,7 +59,7 @@ public IotReflectorClient(String projectId, CloudIotConfig iotConfig, String key
       mqttPublisher = new MqttPublisher(projectId, cloudRegion, UDMS_REFLECT,
           siteName, keyBytes, IOT_KEY_ALGORITHM, this::messageHandler, this::errorHandler);
     } catch (Exception e) {
-      throw new RuntimeException("While connecting subscription " + subscriptionId, e);
+      throw new RuntimeException("While connecting MQTT endpoint " + subscriptionId, e);
     }
 
     active = true;

validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java

@@ -1,6 +1,6 @@
 package com.google.daq.mqtt.registrar;
 
-import static com.google.daq.mqtt.validator.Validator.NO_SITE;
+import static com.google.daq.mqtt.util.Common.NO_SITE;
 
 import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -9,7 +9,6 @@
 import com.fasterxml.jackson.databind.util.ISO8601DateFormat;
 import com.github.fge.jsonschema.core.load.configuration.LoadingConfiguration;
 import com.github.fge.jsonschema.core.load.download.URIDownloader;
-import com.github.fge.jsonschema.core.report.ProcessingReport;
 import com.github.fge.jsonschema.main.JsonSchema;
 import com.github.fge.jsonschema.main.JsonSchemaFactory;
 import com.google.api.services.cloudiot.v1.model.Device;
@@ -30,7 +29,6 @@
 import java.io.FilenameFilter;
 import java.io.InputStream;
 import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
 import java.math.BigInteger;
 import java.net.URI;
 import java.time.Duration;