merge Release/2024 q3 into master

docker-compose-multi-instance.yml

@@ -90,7 +90,7 @@ services:
       - ./docker/mongo/b_create_indexes.js:/docker-entrypoint-initdb.d/b_create_indexes.js
     healthcheck:
       test: |
-        test $$(mongosh --username ${MONGO_INITDB_ROOT_USERNAME} --password ${MONGO_INITDB_ROOT_PASSWORD}  --quiet --eval "try { rs.initiate({ _id: 'rs0', members: [{ _id: 0, host: '${DB_HOST_IP}' }] }).ok } catch (_) { rs.status().ok }") -eq 1
+        test $$(mongosh --username ${MONGO_INITDB_ROOT_USERNAME} --password ${MONGO_INITDB_ROOT_PASSWORD} --quiet --eval "db.runCommand({ping:1}).ok") -eq 1
       interval: 10s
       start_period: 30s
     entrypoint:

docker-compose.yml

@@ -77,13 +77,13 @@ services:
     volumes:
       - mongodb_data_container:/data/db
       - ./docker/mongo/manage-volume-cron:/docker-entrypoint-initdb.d/manage-volume-cron
-      - ./docker/mongo/keyfile.txt:/data/keyfile.txt
+      - ./docker/mongo/keyfile.txt:/data/keyfile-import.txt
       - ./docker/mongo/a_init_replicas.js:/docker-entrypoint-initdb.d/a_init_replicas.js
       - ./docker/mongo/b_create_indexes.js:/docker-entrypoint-initdb.d/b_create_indexes.js
       - ./docker/mongo/manage_volume.js:/docker-entrypoint-initdb.d/manage_volume.js
     healthcheck:
       test: | 
-        test $$(mongosh --username ${MONGO_INITDB_ROOT_USERNAME} --password ${MONGO_INITDB_ROOT_PASSWORD}  --quiet --eval "try { rs.initiate({ _id: 'rs0', members: [{ _id: 0, host: '${DB_HOST_IP}' }] }).ok } catch (_) { rs.status().ok }") -eq 1
+        test $$(mongosh --username ${MONGO_INITDB_ROOT_USERNAME} --password ${MONGO_INITDB_ROOT_PASSWORD}  --quiet --eval "db.runCommand({ping:1}).ok") -eq 1
       interval: 10s
       start_period: 60s
     entrypoint:
@@ -98,6 +98,7 @@ services:
         dos2unix /etc/cron.d/manage-volume-cron
         chmod 644 /etc/cron.d/manage-volume-cron
         systemctl restart cron
+        cp /data/keyfile-import.txt /data/keyfile.txt
         chmod 400 /data/keyfile.txt
         chown 999:999 /data/keyfile.txt
 

docker/connect/connect_start.sh

@@ -16,6 +16,8 @@ declare -A OdeRawEncodedBSMJson=([name]="topic.OdeRawEncodedBSMJson" [collection
     [convert_timestamp]=false [timefield]="" [use_key]=false [key]="" [add_timestamp]=true)
 declare -A OdeBsmJson=([name]="topic.OdeBsmJson" [collection]="OdeBsmJson"
     [convert_timestamp]=false [timefield]="" [use_key]=false [key]="" [add_timestamp]=true)
+declare -A ProcessedBsm=([name]="topic.ProcessedBsm" [collection]="ProcessedBsm"
+    [convert_timestamp]=false [timefield]="" [use_key]=false [key]="" [add_timestamp]=true)
 
 # Record Map Data
 declare -A OdeMapJson=([name]="topic.DeduplicatedOdeMapJson" [collection]="OdeMapJson"
@@ -89,6 +91,9 @@ declare -A CmMapMinimumDataEvents=([name]="topic.CmMapMinimumDataEvents" [collec
     [convert_timestamp]=true [timefield]="eventGeneratedAt" [use_key]=false [key]="" [add_timestamp]=false)
 declare -A CmSpatBroadcastRateEvents=([name]="topic.CmSpatBroadcastRateEvents" [collection]="CmSpatBroadcastRateEvents"
     [convert_timestamp]=true [timefield]="eventGeneratedAt" [use_key]=false [key]="" [add_timestamp]=false)
+declare -A CmTimestampDeltaEvent=([name]="topic.CmTimestampDeltaEvent" [collection]="CmTimestampDeltaEvent"
+    [convert_timestamp]=true [timefield]="eventGeneratedAt" [use_key]=false [key]="" [add_timestamp]=false)
+
 
 # Record BSM Events
 declare -A CmBsmEvents=([name]="topic.CmBsmEvents" [collection]="CmBsmEvents"
@@ -125,6 +130,8 @@ declare -A CmStopLineStopNotification=([name]="topic.CmStopLineStopNotification"
     [convert_timestamp]=true [timefield]="notificationGeneratedAt" [use_key]=true [key]="key" [add_timestamp]=false)
 declare -A CmStopLinePassageNotification=([name]="topic.CmStopLinePassageNotification" [collection]="CmStopLinePassageNotification"
     [convert_timestamp]=true [timefield]="notificationGeneratedAt" [use_key]=true [key]="key" [add_timestamp]=false)
+declare -A CmTimestampDeltaNotification=([name]="topic.CmTimestampDeltaNotification" [collection]="CmTimestampDeltaNotification"
+    [convert_timestamp]=true [timefield]="notificationGeneratedAt" [use_key]=true [key]="key" [add_timestamp]=false)
 
 function createSink() {
     local -n topic=$1
@@ -209,6 +216,7 @@ function createSink() {
 
 createSink OdeRawEncodedBSMJson
 createSink OdeBsmJson
+createSink ProcessedBsm
 
 createSink OdeMapJson
 createSink ProcessedMap
@@ -245,6 +253,7 @@ createSink CmSpatMinimumDataEvents
 createSink CmMapBroadcastRateEvents
 createSink CmMapMinimumDataEvents
 createSink CmSpatBroadcastRateEvents
+createSink CmTimestampDeltaEvent
 
 createSink CmBsmEvents
 
@@ -262,6 +271,7 @@ createSink CmSignalGroupAlignmentNotification
 createSink CmNotification
 createSink CmStopLineStopNotification
 createSink CmStopLinePassageNotification
+createSink CmTimestampDeltaNotification
 
 
 echo "----------------------------------"

docker/mongo/b_create_indexes.js

@@ -66,6 +66,7 @@ const collections = [
     // GeoJson Converter Data
     {name: "ProcessedMap", ttlField: "recordGeneratedAt", timeField: "properties.timeStamp", intersectionField: "properties.intersectionId"},
     {name: "ProcessedSpat", ttlField: "recordGeneratedAt", timeField: "utcTimeStamp", intersectionField: "intersectionId"},
+    {name: "ProcessedBsm", ttlField: "recordGeneratedAt", timeField: "timeStamp", geoSpatialField: "features.geometry.coordinates"},
 
     // Conflict Monitor Events
     { name: "CmStopLineStopEvent", ttlField: "eventGeneratedAt", timeField: "eventGeneratedAt", intersectionField: "intersectionID" },
@@ -145,6 +146,7 @@ do {
                 createTimeIntersectionIndex(collection);
                 createTimeRsuIpIndex(collection);
                 createTimeIndex(collection);
+		createGeoSpatialIndex(collection);
             }else{
                 missing_collection_count++;
                 console.log("Collection " + collection.name + " does not exist yet");
@@ -318,6 +320,38 @@ function createTimeIntersectionIndex(collection){
     }
 }
 
+function createGeoSpatialIndex(collection){
+    if(geoSpatialIndexExists(collection)){
+        return;
+    }
+
+    if(collection.hasOwnProperty("timeField") && collection.timeField != null && collection.hasOwnProperty("geoSpatialField") && collection.geoSpatialField != null){
+        const collectionName = collection.name;
+        const timeField = collection.timeField;
+        const geoSpatialField = collection.geoSpatialField;
+        console.log("Creating GeoSpatial index for " + collectionName);
+
+        var indexJson = {};
+        indexJson[geoSpatialField] = "2dsphere";
+        indexJson[timeField] = -1;
+
+
+        try {
+            db[collectionName].createIndex(indexJson);
+            console.log("Created time geospatial index for " + collectionName + " using the field: " + timeField + " as the timestamp and : " + geoSpatialField + " as the GeoSpatial Field");
+        } catch (err) {
+            db.runCommand({
+                "collMod": collectionName,
+                "index": {
+                    keyPattern: indexJson
+                }
+            });
+            console.log("Updated time geospatial index for " + collectionName + " using the field: " + timeField + " as the timestamp and : " + geoSpatialField + " as the GeoSpatial Field");
+        }
+    }
+
+}
+
 function ttlIndexExists(collection) {
     return db[collection.name].getIndexes().find((idx) => idx.hasOwnProperty("expireAfterSeconds")) !== undefined;
 }
@@ -333,3 +367,7 @@ function timeRsuIpIndexExists(collection){
 function timeIndexExists(collection){
     return db[collection.name].getIndexes().find((idx) => idx.name == collection.timeField + "_-1") !== undefined;
 }
+
+function geoSpatialIndexExists(collection){
+    return db[collection.name].getIndexes().find((idx) => idx.name == collection.geoSpatialField + "_2dsphere_timeStamp_-1") !== undefined;
+}

docs/KafkaStreamsVisualizations/README.md

@@ -0,0 +1,75 @@
+# Kafka Streams Visualization Tools
+
+Endpoints for describing Kafka Streams internals are available at:
+
+http://localhost:8082/health/topologies
+
+with the Conflict Monitor running in Docker.
+
+## Topology Overviews
+
+The `topologies/simple` endpoints produce GraphViz `dot` files that represent the input and output topics of topologies, connections between topologies, and associated state stores.
+
+To visualize the current state of all Kafka Streams download the `all.dot` file from:
+
+http://localhost:8082/health/topologies/simple/all
+
+To see a simple view of the inputs and outpus of a single topology select one of the `topologies/simple` endpoints, for example:
+
+http://localhost:8082/health/topologies/simple/mapValidation
+
+To create images from the dot files, install the GraphViz command line tool from:
+
+https://graphviz.org/download/
+
+and run the `dot` command, for example:
+
+Create SVG:
+```bash
+dot -Tsvg all.dot > all.svg
+```
+
+Create PNG:
+```bash
+dot -Tpng all.dot > all.png
+```
+
+### Examples
+#### MapValidation Topology Overview
+![MAP Validation Topology Overview](./mapValidation.svg)
+
+#### Spat Validation Topology Overview
+![SPaT Validation Topology Overview](./spatValidation.svg)
+
+#### All Topologies Overview
+
+Open the image in a separate tab to zoom in
+
+![All Topologies Overview](./all.svg)
+
+
+## Topology Details
+
+To view detailed internals for one topology select one of the `topologies/detail` endpoints, for example:
+
+http://localhost:8082/health/topologies/detail/mapValidation
+
+Save the response to a text file.  This file is the result of calling the Topology.describe() method.  This format can be visualized using the online Kafka Streams Topology Visualizer here:
+
+https://zz85.github.io/kafka-streams-viz/
+
+#### Example: MAP Validation Topology Details
+![MAP Validation detailed topology](mapValidation_detail.jpg)
+
+
+
+
+
+
+
+
+
+
+
+
+