Add reflector utility

.github/workflows/testing.yml

@@ -78,6 +78,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/reset_config

@@ -0,0 +1,23 @@
+#!/bin/bash -e
+
+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
+
+device_config=/tmp/${device_id}_config.json
+cp $site_dir/devices/${device_id}/out/generated_config.json $device_config
+now_date=$(date -Ins -u | sed -E 's/.{6}\+.*/Z/' | tr , .)
+echo Setting config timestamp $now_date
+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,6 +33,8 @@ if [[ -n $pids ]]; then
     kill $pids
 fi
 
+bin/reset_config $site_path $project_id $device_id
+
 echo Writing pubber output to $PUBBER_OUT
 echo bin/pubber $site_path $project_id $device_id $serial_no
 

bin/test_validator

@@ -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

validator/bin/reflector

@@ -0,0 +1,23 @@
+#!/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 $*"
+$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;

validator/src/main/java/com/google/daq/mqtt/util/Common.java

@@ -0,0 +1,43 @@
+package com.google.daq.mqtt.util;
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.core.JsonParser.Feature;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.util.ISO8601DateFormat;
+import java.util.List;
+import java.util.MissingFormatArgumentException;
+
+/**
+ * Collection of common constants and minor utilities.
+ */
+public abstract class Common {
+
+  public static final String STATE_QUERY_TOPIC = "query/state";
+  public static final String TIMESTAMP_ATTRIBUTE = "timestamp";
+  public static final String NO_SITE = "--";
+  public static final ObjectMapper OBJECT_MAPPER =
+      new ObjectMapper()
+          .enable(Feature.ALLOW_COMMENTS)
+          .enable(SerializationFeature.INDENT_OUTPUT)
+          .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
+          .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+          .setDateFormat(new ISO8601DateFormat())
+          .setSerializationInclusion(Include.NON_NULL);
+  public static final String JSON_SUFFIX = ".json";
+  public static final String GCP_REFLECT_KEY_PKCS8 = "validator/rsa_private.pkcs8";
+
+  /**
+   * Remove the next item from the list in an exception-safe way.
+   *
+   * @param argList list of arguments
+   * @return removed argument
+   */
+  public static String removeNextArg(List<String> argList) {
+    if (argList.isEmpty()) {
+      throw new MissingFormatArgumentException("Missing argument");
+    }
+    return argList.remove(0);
+  }
+}

validator/src/main/java/com/google/daq/mqtt/validator/Reflector.java

@@ -0,0 +1,140 @@
+package com.google.daq.mqtt.validator;
+
+import static com.google.daq.mqtt.util.Common.GCP_REFLECT_KEY_PKCS8;
+import static com.google.daq.mqtt.util.Common.NO_SITE;
+import static com.google.daq.mqtt.util.Common.removeNextArg;
+
+import com.google.bos.iot.core.proxy.IotReflectorClient;
+import com.google.daq.mqtt.util.CloudIotConfig;
+import com.google.daq.mqtt.util.CloudIotManager;
+import com.google.daq.mqtt.util.ConfigUtil;
+import java.io.File;
+import java.io.FileInputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * General utility for working with UDMI Reflector messages.
+ */
+public class Reflector {
+
+  private final List<String> reflectCommands;
+  private String projectId;
+  private String siteDir;
+  private CloudIotConfig cloudIotConfig;
+  private File baseDir;
+  private IotReflectorClient client;
+  private String deviceId;
+
+  /**
+   * Create an instance of the Reflector class.
+   *
+   * @param argsList command-line arguments
+   */
+  public Reflector(List<String> argsList) {
+    reflectCommands = parseArgs(argsList);
+  }
+
+  /**
+   * Let's go.
+   *
+   * @param args command-line arguments
+   */
+  public static void main(String[] args) {
+    Reflector reflector = new Reflector(Arrays.asList(args));
+    reflector.initialize();
+    reflector.reflect();
+    reflector.shutdown();
+  }
+
+  private void shutdown() {
+    try {
+      Thread.sleep(5000);
+    } catch (Exception e) {
+      throw new RuntimeException("While sleeping", e);
+    }
+    client.close();
+  }
+
+  private void reflect() {
+    System.err.printf("Reflecting %d directives...%n", reflectCommands.size());
+    while (!reflectCommands.isEmpty()) {
+      reflect(reflectCommands.remove(0));
+    }
+    System.err.println("Done with all reflective directives!");
+  }
+
+  private void reflect(String directive) {
+    System.err.printf("Reflecting %s%n", directive);
+    String[] parts = directive.split(":", 2);
+    if (parts.length != 2) {
+      throw new RuntimeException("Expected topic:file format reflect directive " + directive);
+    }
+    File file = new File(parts[1]);
+    reflect(parts[0], file);
+  }
+
+  private void reflect(String topic, File dataFile) {
+    try (FileInputStream fis = new FileInputStream(dataFile)) {
+      String data = new String(fis.readAllBytes());
+      reflect(topic, data);
+    } catch (Exception e) {
+      throw new RuntimeException("While processing input file " + dataFile.getAbsolutePath(), e);
+    }
+  }
+
+  private void reflect(String topic, String data) {
+    client.publish(deviceId, topic, data);
+  }
+
+  private void initialize() {
+    String keyFile = new File(siteDir, GCP_REFLECT_KEY_PKCS8).getAbsolutePath();
+    System.err.println("Loading reflector key file from " + keyFile);
+    client = new IotReflectorClient(projectId, cloudIotConfig, keyFile);
+  }
+
+  private List<String> parseArgs(List<String> argsList) {
+    List<String> listCopy = new ArrayList<>(argsList);
+    while (!listCopy.isEmpty()) {
+      String option = removeNextArg(listCopy);
+      try {
+        switch (option) {
+          case "-p":
+            projectId = removeNextArg(listCopy);
+            break;
+          case "-s":
+            setSiteDir(removeNextArg(listCopy));
+            break;
+          case "-d":
+            deviceId = removeNextArg(listCopy);
+            break;
+          default:
+            listCopy.add(option);
+            return listCopy; // default case is the remaining list of reflection directives
+        }
+      } catch (Exception e) {
+        throw new RuntimeException("While processing option " + option, e);
+      }
+    }
+    throw new IllegalArgumentException("No reflect directives specified!");
+  }
+
+  /**
+   * Set the site directory to use for this run.
+   *
+   * @param siteDir site model directory
+   */
+  public void setSiteDir(String siteDir) {
+    if (NO_SITE.equals(siteDir)) {
+      siteDir = null;
+      baseDir = new File(".");
+    } else {
+      this.siteDir = siteDir;
+      baseDir = new File(siteDir);
+      File cloudConfig = new File(siteDir, "cloud_iot_config.json");
+      cloudIotConfig = CloudIotManager.validate(ConfigUtil.readCloudIotConfig(cloudConfig),
+          projectId);
+    }
+  }
+}

validator/src/main/java/com/google/daq/mqtt/validator/SequenceValidator.java

@@ -12,6 +12,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.daq.mqtt.util.CloudIotConfig;
+import com.google.daq.mqtt.util.Common;
 import com.google.daq.mqtt.util.ConfigUtil;
 import com.google.daq.mqtt.util.ValidatorConfig;
 import java.io.ByteArrayOutputStream;
@@ -148,6 +149,7 @@ public abstract class SequenceValidator {
     resultSummary.delete();
     System.err.println("Writing results to " + resultSummary.getAbsolutePath());
 
+    System.err.printf("Loading reflector key file from %s%n", new File(key_file).getAbsolutePath());
     System.err.printf("Validating against device %s serial %s%n", deviceId, serialNo);
     client = new IotReflectorClient(projectId, cloudIotConfig, key_file);
     setReflectorState();
@@ -454,7 +456,7 @@ private String writeLogEntry(Entry logEntry, String filename) {
   }
 
   protected void queryState() {
-    client.publish(deviceId, Validator.STATE_QUERY_TOPIC, EMPTY_MESSAGE);
+    client.publish(deviceId, Common.STATE_QUERY_TOPIC, EMPTY_MESSAGE);
   }
 
   /**