diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml
index d8240bc97a..02d9ac56c2 100644
--- a/.github/workflows/testing.yml
+++ b/.github/workflows/testing.yml
@@ -49,12 +49,13 @@ jobs:
- name: Setup udmis container build
if: ${{ github.event_name == 'push' }}
run: |
+ set -x
revhash=$(git rev-parse $GITHUB_REF)
IMAGE_TAG=g${revhash:0:9}
PUSH_REPO=$PUSH_REGISTRY/${{ github.repository }}
PUSH_TAG=$PUSH_REPO:$IMAGE_TAG
echo PUSH_TAG=$PUSH_TAG >> $GITHUB_ENV
- udmis/bin/container prep --no-check $PUSH_REPO
+ udmis/bin/container prep --no-check ${PUSH_REPO%/*}
echo Pushing built container as $PUSH_TAG | tee -a $GITHUB_STEP_SUMMARY
- name: Build and push docker image
if: ${{ github.event_name == 'push' }}
@@ -295,7 +296,6 @@ jobs:
cd sites/udmi_site_model/out
find . -type f | sort | xargs more | cat
- name: itemized sequencer tests
- if: ${{ always() }}
run: |
bin/test_itemized $GCP_TARGET_PROJECT
- name: itemized output logs
diff --git a/common/src/main/java/com/google/udmi/util/GeneralUtils.java b/common/src/main/java/com/google/udmi/util/GeneralUtils.java
index 26d2fb2246..14524838e5 100644
--- a/common/src/main/java/com/google/udmi/util/GeneralUtils.java
+++ b/common/src/main/java/com/google/udmi/util/GeneralUtils.java
@@ -49,8 +49,9 @@ public class GeneralUtils {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
.enable(SerializationFeature.INDENT_OUTPUT)
.setDateFormat(new ISO8601DateFormat())
+ .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
.setSerializationInclusion(JsonInclude.Include.NON_NULL);
- private static final ObjectMapper OBJECT_MAPPER_RAW =
+ public static final ObjectMapper OBJECT_MAPPER_RAW =
OBJECT_MAPPER.copy()
.enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS)
.enable(Feature.ALLOW_TRAILING_COMMA)
@@ -59,7 +60,7 @@ public class GeneralUtils {
.setSerializationInclusion(Include.NON_NULL);
public static final ObjectMapper OBJECT_MAPPER_STRICT =
OBJECT_MAPPER_RAW.copy()
- .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+ .enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
.disable(SerializationFeature.INDENT_OUTPUT);
private static final String SEPARATOR = "\n ";
diff --git a/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml b/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml
index 59670b69c6..d05214369a 100644
--- a/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml
+++ b/udmis/.idea/runConfigurations/UdmiServicePod_configured.xml
@@ -19,4 +19,4 @@
-
+
\ No newline at end of file
diff --git a/udmis/bin/container b/udmis/bin/container
index c85b0bfc22..ee98c296da 100755
--- a/udmis/bin/container
+++ b/udmis/bin/container
@@ -6,7 +6,7 @@ DROOT=.
function usage {
echo Error: $*
- echo Usage: $0 { prep, build, shell, run, push, deploy, update, status, logs, stop } [--no-check] [--selfie] [repo]
+ echo Usage: $0 { prep, build, shell, run, deploy, update, status, logs, stop } [--no-check] [--selfie] [repo]
echo Project: $GCP_PROJECT
echo Try starting with: $0 build
exit 1
@@ -18,8 +18,6 @@ shift || usage missing command
GCP_PROJECT=$(gcloud config get project)
REPOSITORY=gcr.io/$GCP_PROJECT
-mkdir -p tmp
-
IMAGE=udmis
if [[ $1 == "--no-check" ]]; then
@@ -60,7 +58,8 @@ current_user=$USER@$HOSTNAME
revparse=`git rev-parse HEAD`
-udmi_ref=$REPOSITORY:g${revparse:0:9}
+udmi_ver=g${revparse:0:9}
+udmi_ref=$REPOSITORY/udmi:$udmi_ver
version=`git describe || echo $udmi_ref`
@@ -68,7 +67,7 @@ RUNARGS="--rm -ti -v $PWD/var:/udmi -v $HOME/.config:/root/.config --tmpfs /tmp"
TEMPLATES=$(cd etc; ls k8s_*.yaml)
if [[ $cmd == prep || $cmd == build ]]; then
- rm -rf var tmp && mkdir -p var
+ rm -rf var tmp && mkdir -p var tmp
bin/build
build_time=`date --utc -Imin -r $LIBFILE`
cp etc/prod_pod.json var/
@@ -84,29 +83,32 @@ EOF
echo Next try: $0 build
fi
-if [[ $cmd == prep ]]; then
- true
-elif [[ $cmd == build ]]; then
+if [[ $cmd == build ]]; then
echo Building Dockerfile.$IMAGE
docker build -f Dockerfile.$IMAGE -t $IMAGE $DROOT
- echo Next try: $0 push
-elif [[ $cmd == run ]]; then
- docker run $RUNARGS $IMAGE
-elif [[ $cmd == shell ]]; then
- docker run $RUNARGS $IMAGE bash
-elif [[ $cmd == push ]]; then
- IMAGENAME=$REPOSITORY/$IMAGE
- docker tag $IMAGE $IMAGENAME
- docker push $IMAGENAME
- hash=$(docker images --digests $IMAGENAME | fgrep latest | awk '{print $3}')
- ihash=$IMAGENAME@$hash
+fi
+
+if [[ $cmd == build && $REPOSITORY != local ]]; then
+ docker tag $IMAGE $udmi_ref
+ docker push $udmi_ref
+ hash=$(docker images --digests ${udmi_ref%:*} | fgrep $udmi_ver | awk '{print $3}')
+ ihash=$udmi_ref@$hash
for file in $TEMPLATES; do
if fgrep -q @IMAGE-$IMAGE@ etc/$file; then
sed < etc/$file > tmp/$file -e "s^@IMAGE-$IMAGE@^$ihash^"
echo Updated tmp/$file with image $ihash
fi
done
- echo Next try: $0 deploy
+fi
+
+if [[ $cmd == prep ]]; then
+ echo Next try: $0 build
+elif [[ $cmd == build ]]; then
+ echo Next try: $0 '{run, deploy, update}'
+elif [[ $cmd == run ]]; then
+ docker run $RUNARGS $IMAGE
+elif [[ $cmd == shell ]]; then
+ docker run $RUNARGS $IMAGE bash
elif [[ $cmd == deploy ]]; then
kubectl apply -f tmp/k8s_pod.yaml
echo Next try: $0 status
diff --git a/udmis/bin/deploy b/udmis/bin/deploy
deleted file mode 100755
index 1735976d30..0000000000
--- a/udmis/bin/deploy
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/bin/bash -e
-
-if [[ $# != 1 ]]; then
- echo Usage: $0 project_id
- false
-fi
-
-project_id=$1
-shift
-
-ROOT=$(dirname $0)/..
-cd $ROOT
-
-bin/setup $project_id
-
-bin/container build
-bin/container push
-bin/container deploy
-sleep 10
-kubectl get pods
diff --git a/udmis/bin/deploy b/udmis/bin/deploy
new file mode 120000
index 0000000000..77c430a5ac
--- /dev/null
+++ b/udmis/bin/deploy
@@ -0,0 +1 @@
+update
\ No newline at end of file
diff --git a/udmis/bin/update b/udmis/bin/update
index d0b26ec379..fc873822e7 100755
--- a/udmis/bin/update
+++ b/udmis/bin/update
@@ -8,13 +8,13 @@ fi
project_id=$1
shift
+CMD=$(basename $0)
ROOT=$(dirname $0)/..
cd $ROOT
bin/setup $project_id
bin/container build
-bin/container push
-bin/container update
+bin/container $CMD
sleep 10
kubectl get pods
diff --git a/udmis/etc/prod_pod.json b/udmis/etc/prod_pod.json
index 3b3cfc5b93..487171373f 100644
--- a/udmis/etc/prod_pod.json
+++ b/udmis/etc/prod_pod.json
@@ -24,12 +24,13 @@
},
"clearblade-iot-core": {
"provider": "clearblade",
- "project_id": "${CLEARBLADE_PROJECT}"
+ "project_id": "${CLEARBLADE_PROJECT}",
+ "options": "distributor=stately"
},
"gcp-iot-core": {
"provider": "gcp",
"project_id": "${GCP_PROJECT}",
- "options": "${GCP_IOT_OPTIONS}"
+ "options": "${GCP_IOT_OPTIONS},distributor=stately"
}
},
"distributors": {
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java
index 1ace7685f0..5dcbd05eaf 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/ClearBladeIotAccessProvider.java
@@ -7,10 +7,10 @@
import static com.google.udmi.util.GeneralUtils.friendlyStackTrace;
import static com.google.udmi.util.GeneralUtils.ifNotNullGet;
import static com.google.udmi.util.GeneralUtils.ifNotNullThen;
+import static com.google.udmi.util.GeneralUtils.ifNotTrueThen;
import static com.google.udmi.util.GeneralUtils.ifTrueThen;
import static com.google.udmi.util.GeneralUtils.isTrue;
import static com.google.udmi.util.JsonUtil.getDate;
-import static com.google.udmi.util.JsonUtil.getTimestamp;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
import static java.util.Optional.ofNullable;
@@ -53,10 +53,7 @@
import com.google.common.collect.BiMap;
import com.google.common.collect.ImmutableBiMap;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
import com.google.udmi.util.GeneralUtils;
-import java.time.Duration;
-import java.time.Instant;
import java.util.AbstractMap.SimpleEntry;
import java.util.Base64;
import java.util.Date;
@@ -66,7 +63,6 @@
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -103,8 +99,11 @@ public class ClearBladeIotAccessProvider extends IotAccessBase {
* Create a new instance for interfacing with GCP IoT Core.
*/
public ClearBladeIotAccessProvider(IotAccess iotAccess) {
+ super(iotAccess);
projectId = getProjectId(iotAccess);
ifTrueThen(isEnabled(), this::fetchRegistryRegions);
+ ifNotTrueThen(isEnabled(),
+ () -> warn("Clearblade access provided disabled because project id is null or empty"));
}
private static Credential convertIot(DeviceCredential device) {
@@ -147,6 +146,27 @@ protected DeviceManagerClient getDeviceManagerClient() {
return new DeviceManagerClient();
}
+ @NotNull
+ protected Set getRegistriesForRegion(String region) {
+ try {
+ DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
+ ListDeviceRegistriesRequest request = ListDeviceRegistriesRequest.Builder.newBuilder()
+ .setParent(LocationName.of(projectId, region).getLocationFullName())
+ .build();
+ ListDeviceRegistriesResponse response = deviceManagerClient.listDeviceRegistries(request);
+ requireNonNull(response, "get registries response is null");
+ List deviceRegistries = response.getDeviceRegistriesList();
+ Set registries =
+ ofNullable(deviceRegistries).orElseGet(ImmutableList::of).stream()
+ .map(registry -> registry.toBuilder().getId())
+ .collect(Collectors.toSet());
+ debug("Fetched " + registries.size() + " registries for region " + region);
+ return registries;
+ } catch (Exception e) {
+ throw new RuntimeException("While fetching registries for region " + region, e);
+ }
+ }
+
@Override
protected boolean isEnabled() {
return !isNullOrEmpty(projectId);
@@ -276,6 +296,32 @@ private CloudModel deleteDevice(String registryId, Device device) {
}
}
+ @NotNull
+ private HashMap fetchDevices(String deviceRegistryId, String gatewayId) {
+ String location = getRegistryLocation(deviceRegistryId);
+ DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
+ GatewayListOptions gatewayListOptions = ifNotNullGet(gatewayId, this::getGatewayListOptions);
+ String registryFullName =
+ RegistryName.of(projectId, location, deviceRegistryId).getRegistryFullName();
+ String pageToken = null;
+ HashMap collect = new HashMap<>();
+ do {
+ DevicesListRequest request = DevicesListRequest.Builder.newBuilder().setParent(
+ registryFullName)
+ .setGatewayListOptions(gatewayListOptions)
+ .setPageToken(pageToken)
+ .build();
+ DevicesListResponse response = deviceManagerClient.listDevices(request);
+ requireNonNull(response, "DeviceRegistriesList fetch failed");
+ Map responseMap =
+ response.getDevicesList().stream().map(ClearBladeIotAccessProvider::convertToEntry)
+ .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+ collect.putAll(responseMap);
+ pageToken = response.getNextPageToken();
+ } while (pageToken != null);
+ return collect;
+ }
+
private String getDeviceName(String registryId, String deviceId) {
return DeviceName.of(projectId, getRegistryLocation(registryId), registryId, deviceId)
.toString();
@@ -298,27 +344,6 @@ private String getProjectId(IotAccess iotAccess) {
}
}
- @NotNull
- protected Set getRegistriesForRegion(String region) {
- try {
- DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
- ListDeviceRegistriesRequest request = ListDeviceRegistriesRequest.Builder.newBuilder()
- .setParent(LocationName.of(projectId, region).getLocationFullName())
- .build();
- ListDeviceRegistriesResponse response = deviceManagerClient.listDeviceRegistries(request);
- requireNonNull(response, "get registries response is null");
- List deviceRegistries = response.getDeviceRegistriesList();
- Set registries =
- ofNullable(deviceRegistries).orElseGet(ImmutableList::of).stream()
- .map(registry -> registry.toBuilder().getId())
- .collect(Collectors.toSet());
- debug("Fetched " + registries.size() + " registries for region " + region);
- return registries;
- } catch (Exception e) {
- throw new RuntimeException("While fetching registries for region " + region, e);
- }
- }
-
private String getRegistryLocation(String registry) {
return getRegistryRegion(registry);
}
@@ -362,32 +387,6 @@ private CloudModel listRegistryDevices(String deviceRegistryId, String gatewayId
}
}
- @NotNull
- private HashMap fetchDevices(String deviceRegistryId, String gatewayId) {
- String location = getRegistryLocation(deviceRegistryId);
- DeviceManagerClient deviceManagerClient = getDeviceManagerClient();
- GatewayListOptions gatewayListOptions = ifNotNullGet(gatewayId, this::getGatewayListOptions);
- String registryFullName =
- RegistryName.of(projectId, location, deviceRegistryId).getRegistryFullName();
- String pageToken = null;
- HashMap collect = new HashMap<>();
- do {
- DevicesListRequest request = DevicesListRequest.Builder.newBuilder().setParent(
- registryFullName)
- .setGatewayListOptions(gatewayListOptions)
- .setPageToken(pageToken)
- .build();
- DevicesListResponse response = deviceManagerClient.listDevices(request);
- requireNonNull(response, "DeviceRegistriesList fetch failed");
- Map responseMap =
- response.getDevicesList().stream().map(ClearBladeIotAccessProvider::convertToEntry)
- .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
- collect.putAll(responseMap);
- pageToken = response.getNextPageToken();
- } while (pageToken != null);
- return collect;
- }
-
private void unbindDevice(String registryId, String gatewayId, String proxyId) {
try {
debug(format("Unbind %s: %s from %s", registryId, proxyId, gatewayId));
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java
index 9c0da43cd8..10530654e4 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/DynamicIotAccessProvider.java
@@ -38,6 +38,7 @@ public class DynamicIotAccessProvider extends IotAccessBase {
* Create a new instance for interfacing with multiple providers.
*/
public DynamicIotAccessProvider(IotAccess iotAccess) {
+ super(iotAccess);
providerList = Arrays.asList(iotAccess.project_id.split(","));
}
@@ -135,6 +136,11 @@ public void sendCommandBase(String registryId, String deviceId, SubFolder folder
getProviderFor(registryId).sendCommandBase(registryId, deviceId, folder, message);
}
+ @Override
+ public void updateRegistryRegions(Map regions) {
+ providers.values().forEach(provider -> provider.updateRegistryRegions(regions));
+ }
+
@Override
public void setProviderAffinity(String registryId, String deviceId, String providerId) {
if (providerId != null) {
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java
index b8ccc65899..2f77d9aeed 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/GcpIotAccessProvider.java
@@ -94,8 +94,8 @@ public class GcpIotAccessProvider extends IotAccessBase {
* TODO: Need to implement page tokens for all requisite API calls.
*/
public GcpIotAccessProvider(IotAccess iotAccess) {
- String options = variableSubstitution(iotAccess.options, null);
- if (!ENABLED_OPTION.equals(options)) {
+ super(iotAccess);
+ if (!options.containsKey(ENABLED_OPTION)) {
warn("access provider disabled, missing option '%s'", ENABLED_OPTION);
projectId = null;
cloudIotService = null;
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java b/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java
index fe6efafcb7..478919bfec 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/IotAccessBase.java
@@ -11,14 +11,17 @@
import static java.util.Objects.requireNonNull;
import static java.util.Optional.ofNullable;
+import com.google.bos.udmi.service.core.DistributorPipe;
import com.google.bos.udmi.service.core.ProcessorBase.PreviousParseException;
import com.google.bos.udmi.service.pod.ContainerBase;
+import com.google.bos.udmi.service.pod.UdmiServicePod;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalAmount;
+import java.util.Arrays;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
@@ -28,9 +31,11 @@
import java.util.stream.Collectors;
import org.jetbrains.annotations.NotNull;
import udmi.schema.CloudModel;
+import udmi.schema.Envelope;
import udmi.schema.Envelope.SubFolder;
import udmi.schema.IotAccess;
import udmi.schema.IotAccess.IotProvider;
+import udmi.schema.UdmiState;
/**
* Generic interface for accessing iot device management.
@@ -53,8 +58,14 @@ public abstract class IotAccessBase extends ContainerBase {
IotProvider.GCP, GcpIotAccessProvider.class,
IotProvider.LOCAL, LocalIotAccessProvider.class
);
+ final Map options;
private CompletableFuture> registryRegions;
private Instant regionRetry = Instant.now();
+ private DistributorPipe distributor;
+
+ public IotAccessBase(IotAccess iotAccess) {
+ options = parseOptions(iotAccess);
+ }
/**
* Factory constructor for new instances.
@@ -79,6 +90,17 @@ private static String getBackoffKey(String registryId, String deviceId) {
protected abstract Entry fetchConfig(String registryId, String deviceId);
+ /**
+ * Update the cached registry regions with any incremental updates.
+ */
+ public void updateRegistryRegions(Map regions) {
+ try {
+ registryRegions.get().putAll(regions);
+ } catch (Exception e) {
+ throw new RuntimeException("While updating registry regions", e);
+ }
+ }
+
protected Map fetchRegistryRegions() {
Map regionMap = CLOUD_REGIONS.stream().flatMap(
region -> getRegistriesForRegion(region).stream()
@@ -104,9 +126,13 @@ protected String getRegistryRegion(String registryId) {
protected synchronized String populateRegistryRegions(String registryId) {
if (regionRetry.isBefore(Instant.now())) {
+ regionRetry = Instant.now().plus(REGION_RETRY_BACKOFF);
+ Map previousRegions =
+ ifNotNullGet(registryRegions, regions -> regions.getNow(ImmutableMap.of()));
registryRegions = new CompletableFuture<>();
registryRegions.complete(fetchRegistryRegions());
- regionRetry = Instant.now().plus(REGION_RETRY_BACKOFF);
+ Map currentRegions = registryRegions.getNow(ImmutableMap.of());
+ ifNotNullThen(previousRegions, () -> disseminateDifference(previousRegions, currentRegions));
}
return getCompletedRegistryRegion(registryId);
}
@@ -126,6 +152,17 @@ private String checkedUpdate(String registryId, String deviceId, Long version, S
return updateConfig(registryId, deviceId, updated, version);
}
+ private void disseminateDifference(Map previousRegions,
+ Map currentRegions) {
+ Map newRegions = currentRegions.entrySet().stream()
+ .filter(entry -> !previousRegions.containsKey(entry.getKey())).collect(
+ Collectors.toMap(Entry::getKey, Entry::getValue));
+ UdmiState udmiState = new UdmiState();
+ udmiState.regions = newRegions;
+ Envelope envelope = new Envelope();
+ distributor.distribute(envelope, udmiState);
+ }
+
private String getCompletedRegistryRegion(String registryId) {
try {
return registryRegions.get().get(registryId);
@@ -169,6 +206,7 @@ private String safeMunge(Function munger, Entry co
@Override
public void activate() {
super.activate();
+ distributor = UdmiServicePod.maybeGetComponent((String) options.get("distributor"));
if (isEnabled()) {
populateRegistryRegions(REFLECT_REGISTRY);
}
@@ -258,6 +296,16 @@ public AbortLoopException(String message) {
}
}
+ Map parseOptions(IotAccess iotAccess) {
+ String options = variableSubstitution(iotAccess.options, null);
+ if (options == null) {
+ return ImmutableMap.of();
+ }
+ String[] parts = options.split(",");
+ return Arrays.stream(parts).map(String::trim).map(option -> option.split("=", 2))
+ .collect(Collectors.toMap(x -> x[0], x -> x.length > 1 ? x[1] : true));
+ }
+
abstract String fetchRegistryMetadata(String registryId, String metadataKey);
}
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java
index ec2ed41b41..a9cee3bb50 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/LocalIotAccessProvider.java
@@ -30,6 +30,7 @@ public class LocalIotAccessProvider extends IotAccessBase {
* Create a new instance for interfacing with multiple providers.
*/
public LocalIotAccessProvider(IotAccess iotAccess) {
+ super(iotAccess);
}
@Override
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java b/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java
index c29223aa63..d076cd701b 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/core/DistributorPipe.java
@@ -58,10 +58,10 @@ public void shutdown() {
/**
* Distribute a message (broadcast).
*/
- public void distribute(Envelope envelope, UdmiState toolState) {
+ public void distribute(Envelope envelope, Object toolState) {
try {
- debug("Distributing %s for %s %s", toolState.getClass().getSimpleName(),
- envelope.deviceId, envelope.transactionId);
+ debug("Distributing %s for %s/%s", toolState.getClass().getSimpleName(),
+ envelope.deviceRegistryId, envelope.deviceId);
envelope.gatewayId = clientId;
dispatcher.publish(dispatcher.makeMessageBundle(envelope, toolState));
} catch (Exception e) {
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java b/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java
index 53f1c33291..db1ddc2f54 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/core/ProcessorBase.java
@@ -108,6 +108,7 @@ protected void exceptionHandler(Exception e) {
String message = Common.getExceptionMessage(e);
String payload = friendlyStackTrace(e);
error(format("Received message exception: %s", payload));
+ e.printStackTrace();
BundleException bundleException = new BundleException(message, toStringMap(envelope), payload);
reflectError(SubType.EVENT, bundleException);
}
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java b/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java
index 5ec2b6a715..425a1499d5 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/core/ReflectProcessor.java
@@ -16,6 +16,7 @@
import static com.google.udmi.util.JsonUtil.fromStringStrict;
import static com.google.udmi.util.JsonUtil.loadFileStrictRequired;
import static com.google.udmi.util.JsonUtil.stringify;
+import static com.google.udmi.util.JsonUtil.stringifyTerse;
import static com.google.udmi.util.JsonUtil.toMap;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
@@ -28,6 +29,7 @@
import java.io.File;
import java.util.HashMap;
import java.util.Map;
+import java.util.stream.Collectors;
import udmi.schema.CloudModel;
import udmi.schema.CloudModel.Operation;
import udmi.schema.Envelope;
@@ -44,6 +46,7 @@
public class ReflectProcessor extends ProcessorBase {
public static final String PAYLOAD_KEY = "payload";
+ public static final String HOSTNAME = System.getenv("HOSTNAME");
static final String DEPLOY_FILE = "var/deployed_version.json";
static final SetupUdmiConfig DEPLOYED_CONFIG =
loadFileStrictRequired(SetupUdmiConfig.class, new File(DEPLOY_FILE));
@@ -137,13 +140,19 @@ private void processReflection(Envelope reflection, Envelope envelope,
Map payload) {
debug("Processing reflection %s/%s %s", envelope.subType, envelope.subFolder,
envelope.transactionId);
- iotAccess.setProviderAffinity(envelope.deviceRegistryId, envelope.deviceId, reflection.source);
+ updateProviderAffinity(envelope, reflection.source);
CloudModel result = getReflectionResult(envelope, payload);
ifNotNullThen(result,
v -> debug("Reflection result %s: %s", envelope.transactionId, envelope.subType));
ifNotNullThen(result, v -> sendReflectCommand(reflection, envelope, result));
}
+ private void processStateUpdate(Envelope envelope, StateUpdate stateUpdate) {
+ debug("Updating last_start b/c of device state query %s/%s",
+ envelope.deviceRegistryId, envelope.deviceId);
+ updateLastStart(envelope, stateUpdate);
+ }
+
private CloudModel queryCloudDevice(Envelope attributes) {
return iotAccess.fetchDevice(attributes.deviceRegistryId, attributes.deviceId);
}
@@ -170,12 +179,6 @@ private CloudModel queryDeviceState(Envelope attributes) {
}
}
- private void processStateUpdate(Envelope envelope, StateUpdate stateUpdate) {
- debug("Updating last_start b/c of device state query %s/%s",
- envelope.deviceRegistryId, envelope.deviceId);
- updateLastStart(envelope, stateUpdate);
- }
-
private CloudModel reflectModel(Envelope attributes, CloudModel request) {
return requireNonNull(
iotAccess.modelDevice(attributes.deviceRegistryId, attributes.deviceId, request),
@@ -221,6 +224,7 @@ private void reflectStateHandler(Envelope envelope, UdmiState toolState) {
udmiConfig.last_state = toolState.timestamp;
udmiConfig.setup = new SetupUdmiConfig();
copyFields(DEPLOYED_CONFIG, udmiConfig.setup, false);
+ udmiConfig.setup.hostname = HOSTNAME;
udmiConfig.setup.udmi_version = UDMI_VERSION;
udmiConfig.setup.functions_min = FUNCTIONS_VERSION_MIN;
udmiConfig.setup.functions_max = FUNCTIONS_VERSION_MAX;
@@ -246,6 +250,18 @@ private void sendReflectCommand(Envelope reflection, Envelope message, Object pa
iotAccess.sendCommand(reflectRegistry, deviceRegistry, SubFolder.UDMI, stringify(message));
}
+ private void updateProviderAffinity(Envelope envelope, String source) {
+ debug("Setting affinity for %s/%s to %s", envelope.deviceRegistryId, envelope.deviceId, source);
+ iotAccess.setProviderAffinity(envelope.deviceRegistryId, envelope.deviceId, source);
+ }
+
+ private void updateRegistryRegions(Map regions) {
+ debug("Updating registry regions: ",
+ regions.entrySet().stream().map(entry -> format("%s:%s", entry.getKey(), entry.getValue()))
+ .collect(Collectors.joining(", ")));
+ iotAccess.updateRegistryRegions(regions);
+ }
+
@Override
public void activate() {
debug("Deployment configuration: " + stringify(DEPLOYED_CONFIG));
@@ -253,8 +269,9 @@ public void activate() {
}
void updateAwareness(Envelope envelope, UdmiState toolState) {
- debug("Processing UdmiState for %s/%s %s", envelope.deviceId, envelope.source,
- stringify(toolState));
- iotAccess.setProviderAffinity(envelope.deviceRegistryId, envelope.deviceId, envelope.source);
+ debug("Processing UdmiState for %s/%s: %s", envelope.deviceRegistryId, envelope.deviceId,
+ stringifyTerse(toolState));
+ ifNotNullThen(toolState.setup, setup -> updateProviderAffinity(envelope, envelope.source));
+ ifNotNullThen(toolState.regions, this::updateRegistryRegions);
}
}
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java
index b03a6f0df4..9261e0baf5 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageBase.java
@@ -26,13 +26,10 @@
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -198,8 +195,8 @@ private void messageLoop(String id) {
return;
}
envelope = bundle.envelope;
- debug("Processing %s/%s %s %s -> %s", envelope.subType, envelope.subFolder,
- envelope.transactionId, queueIdentifier(), dispatcher);
+ debug("Processing %s %s/%s %s", this, envelope.subType, envelope.subFolder,
+ envelope.transactionId);
if (ERROR_MESSAGE_MARKER.equals(envelope.transactionId)) {
throw new RuntimeException("Exception due to test-induced error");
}
@@ -254,7 +251,7 @@ private void receiveException(Map attributesMap, String messageS
public void activate(Consumer bundleConsumer) {
dispatcher = bundleConsumer;
ensureSourceQueue();
- debug("Handling %s to %08x", this, Objects.hash(bundleConsumer));
+ debug("Handling %s", this);
handleQueue();
activated = true;
}
@@ -315,7 +312,7 @@ public void terminate() {
@Override
public String toString() {
- return format("MessagePipe %s", queueIdentifier());
+ return format("MessagePipe %s => %s", queueIdentifier(), Objects.hash(dispatcher));
}
/**
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java
index d24919ae3e..1d5c579afa 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/messaging/impl/MessageDispatcherImpl.java
@@ -68,7 +68,7 @@ public class MessageDispatcherImpl extends ContainerBase implements MessageDispa
private final MessagePipe messagePipe;
private final Map messageEnvelopes = new ConcurrentHashMap<>();
- private final Map, Consumer