Weekly Conflict Monitor Report Emails

.github/workflows/ci.yml

@@ -7,7 +7,7 @@ on:
       - cdot-release*
   pull_request:
     branches:
-      - "**" # Run on PRs to any branch
+      - '**' # Run on PRs to any branch
 
 jobs:
   test_change_limit:
@@ -128,8 +128,8 @@ jobs:
       - name: Set up JDK 21
         uses: actions/setup-java@v4
         with:
-          distribution: "adopt"
-          java-version: "21"
+          distribution: 'adopt'
+          java-version: '21'
 
       - name: Build and Test with Maven
         run: |
@@ -190,6 +190,54 @@ jobs:
         run: |
           rm -rf $GITHUB_WORKSPACE/services/intersection-api/api/target
 
+  build_rsu_status_monitor:
+    needs: test_change_limit
+    if: ${{ always() && !failure() && !cancelled() }}
+    runs-on: ubuntu-latest
+    container:
+      image: maven:3.9.9-eclipse-temurin-22
+      options: --user root
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v4
+
+      - name: Cache Maven packages
+        uses: actions/cache@v4
+        with:
+          path: ~/.m2
+          key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}-${{ hashFiles('**/settings.xml') }}-${{ runner.os }}-java-22
+          restore-keys: |
+            ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}-${{ runner.os }}-java-22
+            ${{ runner.os }}-maven-${{ runner.os }}-java-22
+            ${{ runner.os }}-maven
+
+      - name: Install Dependencies
+        env:
+          MAVEN_GITHUB_TOKEN: ${{ secrets.MAVEN_GITHUB_TOKEN }}
+          MAVEN_GITHUB_ORG: ${{ github.repository_owner }}
+        run: |
+          cd $GITHUB_WORKSPACE/services/rsu-status-monitor/rsu-status-monitor
+          mvn clean install -DskipTests -s ./settings.xml
+          mvn dependency:go-offline -s ./settings.xml
+
+      - name: Run tests
+        env:
+          MAVEN_GITHUB_TOKEN: ${{ secrets.MAVEN_GITHUB_TOKEN }}
+          MAVEN_GITHUB_ORG: ${{ github.repository_owner }}
+        run: |
+          cd $GITHUB_WORKSPACE/services/rsu-status-monitor/rsu-status-monitor
+          mvn verify -s ./settings.xml
+
+      - name: Archive Test Results
+        uses: actions/upload-artifact@v4
+        with:
+          name: rsu-status-monitor-test-results
+          path: $GITHUB_WORKSPACE/services/rsu-status-monitor/rsu-status-monitor/target/surefire-reports
+
+      - name: Clean up
+        run: |
+          rm -rf $GITHUB_WORKSPACE/services/rsu-status-monitor/rsu-status-monitor/target
+
   sonar:
     if: github.event.pull_request.base.repo.owner.login == 'usdot-jpo-ode'
     needs: [build_api, webapp]
@@ -224,8 +272,8 @@ jobs:
 
       - uses: actions/setup-java@v4
         with:
-          distribution: "temurin"
-          java-version: "17"
+          distribution: 'temurin'
+          java-version: '17'
 
       - name: Setup SonarScanner
         uses: warchant/setup-sonar-scanner@v7

.gitmodules

@@ -1,4 +1,3 @@
 [submodule "jpo-utils"]
 	path = jpo-utils
-	url = https://github.com/usdot-jpo-ode/jpo-utils
-	branch = master
+	url = https://github.com/neaeraconsulting/jpo-utils

docker-compose-intersection.yml

@@ -50,6 +50,16 @@ services:
       INTERSECTION_API_ENABLE_API: ${INTERSECTION_API_ENABLE_API}
       INTERSECTION_API_ENABLE_EMAILER: ${INTERSECTION_API_ENABLE_EMAILER}
       INTERSECTION_API_ENABLE_REPORTS: ${INTERSECTION_API_ENABLE_REPORTS}
+      INTERSECTION_API_ENABLE_REPORT_EMAIL: ${INTERSECTION_API_ENABLE_REPORT_EMAIL}
+
+      INTERSECTION_EMAIL_BROKER: ${INTERSECTION_EMAIL_BROKER:-SMTP}
+      INTERSECTION_SENDER_EMAIL: ${INTERSECTION_SENDER_EMAIL:-donotreply@cvmanager.com}
+      UNSUBSCRIBE_SECRET_KEY: ${UNSUBSCRIBE_SECRET_KEY:-replace_me}
+      CV_MANAGER_FRONT_END_URI: ${CV_MANAGER_FRONT_END_URI:-http://localhost:${WEBAPP_PORT:-3000}}
+      INTERSECTION_SMTP_SERVER_IP: ${INTERSECTION_SMTP_SERVER_IP:-localhost}
+      INTERSECTION_SMTP_SERVER_PORT: ${INTERSECTION_SMTP_SERVER_PORT:-1025}
+      SENDGRID_API_KEY: ${SENDGRID_API_KEY}
+      POSTMARK_API_KEY: ${POSTMARK_API_KEY}
 
       BASE_ROUTE_PREFIX: ${INTERSECTION_API_ROUTE_PREFIX:-/}
     entrypoint:

docker-compose-rsu-status-monitor.yml

@@ -0,0 +1,32 @@
+services:
+  rsu_status_monitor:
+    profiles:
+      - rsu_status_monitor
+    image: rsu-status-monitor:latest
+    build:
+      context: ./services
+      dockerfile: Dockerfile.rsu_status_monitor
+      args:
+        MAVEN_GITHUB_TOKEN: ${MAVEN_GITHUB_TOKEN:?error}
+        MAVEN_GITHUB_ORG: ${MAVEN_GITHUB_ORG:?error}
+    ports:
+      - '8086:8080'
+      - '42300:42300/udp'
+    environment:
+      LOGGING_LEVEL: ${RSU_STATUS_LOGGING_LEVEL}
+      KAFKA_CONCURRENCY: ${KAFKA_CONCURRENCY:-1}
+      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_BOOTSTRAP_SERVERS}
+      SPRING_PROFILES_ACTIVE: ${SPRING_PROFILES_ACTIVE:-default}
+      POSTGRES_SERVER_URL: jdbc:postgresql://${PG_DB_HOST}
+      POSTGRES_DB: ${PG_DB_NAME:-postgres}
+      POSTGRES_USER: ${PG_DB_USER:-postgres}
+      POSTGRES_PASSWORD: ${PG_DB_PASS:-postgres}
+      RSU_STATUS_MONITOR_ENABLE_MONITORING: ${RSU_STATUS_MONITOR_ENABLE_MONITORING:-true}
+      INTERSECTION_API_MONITOR_INTERVAL: ${INTERSECTION_API_MONITOR_INTERVAL:-300000}
+    depends_on:
+      cvmanager_postgres:
+        required: false
+        condition: service_started
+      kafka-setup:
+        required: false
+        condition: service_started

docker-compose.yml

@@ -3,6 +3,7 @@ include:
   - docker-compose-intersection.yml
   - docker-compose-conflictmonitor.yml
   - docker-compose-obu-ota-server.yml
+  - docker-compose-rsu-status-monitor.yml
 
 services:
   cvmanager_api:

resources/sql_scripts/CVManager_SampleData.sql

@@ -72,7 +72,7 @@ INSERT INTO public.snmp_msgfwd_config(
 
 INSERT INTO public.email_type(
 	email_type)
-	VALUES ('Support Requests'), ('Firmware Upgrade Failures'), ('Daily Message Counts');
+	VALUES ('Support Requests'), ('Firmware Upgrade Failures'), ('Daily Message Counts'), ('Conflict Monitor Reports');
 
 INSERT INTO public.intersections(
 	intersection_number, ref_pt, intersection_name)
@@ -84,4 +84,4 @@ INSERT INTO public.intersection_organization(
 
 INSERT INTO public.rsu_intersection(
 	rsu_id, intersection_id)
-	VALUES (1, 1);
+	VALUES (1, 1);

sample.env

@@ -209,6 +209,7 @@ INTERSECTION_API_ENABLE_API=true
 INTERSECTION_API_ENABLE_EMAILER=true
 INTERSECTION_API_ENABLE_REPORTS=true
 INTERSECTION_API_ENABLE_HAAS=true
+INTERSECTION_API_ENABLE_REPORT_EMAIL=true
 
 # Base Path Prefix - Insert the following path into the base route of all REST endpoints in the Intersection API, used for simplifying proxy routing
 # Routes MUST start with a slash

services/Dockerfile.rsu_status_monitor

@@ -0,0 +1,37 @@
+FROM maven:3.9.6-eclipse-temurin-22-jammy AS builder
+
+WORKDIR /home/rsu-status-monitor
+
+COPY ./rsu-status-monitor/rsu-status-monitor/pom.xml .
+COPY ./rsu-status-monitor/rsu-status-monitor/settings.xml .
+
+ARG MAVEN_GITHUB_TOKEN
+ARG MAVEN_GITHUB_ORG
+ENV MAVEN_GITHUB_TOKEN=$MAVEN_GITHUB_TOKEN
+ENV MAVEN_GITHUB_ORG=$MAVEN_GITHUB_ORG
+
+RUN mvn -s settings.xml dependency:resolve
+
+# Copies source code after dependencies are resolved to leverage Docker caching
+COPY ./rsu-status-monitor/rsu-status-monitor/src ./src
+RUN mvn -s settings.xml install -DskipTests
+
+# Create a new stage for runtime
+FROM eclipse-temurin:22-jre-noble
+
+WORKDIR /home
+
+COPY --from=builder /home/rsu-status-monitor/src/main/resources/application.yaml /home
+COPY --from=builder /home/rsu-status-monitor/src/main/resources/application-confluent.yaml /home
+COPY --from=builder /home/rsu-status-monitor/target/rsu-status-monitor.jar /home
+
+ENTRYPOINT ["java", \
+	"-Djava.rmi.server.hostname=$DOCKER_HOST_IP", \
+	"-Dcom.sun.management.jmxremote.port=9090", \
+	"-Dcom.sun.management.jmxremote.rmi.port=9090", \
+	"-Dcom.sun.management.jmxremote", \
+	"-Dcom.sun.management.jmxremote.local.only=true", \
+	"-Dcom.sun.management.jmxremote.authenticate=false", \
+	"-Dcom.sun.management.jmxremote.ssl=false", \
+	"-jar", \
+	"/home/rsu-status-monitor.jar"]

services/intersection-api/api/pom.xml

@@ -294,6 +294,11 @@
             <artifactId>json-unit-assertj</artifactId>
             <version>5.0.0</version>
             <scope>test</scope>
+        </dependency>
+		<dependency>
+            <groupId>com.sun.mail</groupId>
+            <artifactId>jakarta.mail</artifactId>
+            <version>2.0.1</version>
         </dependency>
 	</dependencies>
 	<repositories>

services/intersection-api/api/src/main/java/us/dot/its/jpo/ode/api/accessors/reports/ReportRepository.java

@@ -5,6 +5,8 @@
 import us.dot.its.jpo.ode.api.models.DataLoader;
 import us.dot.its.jpo.ode.api.models.ReportDocument;
 
+import java.util.List;
+
 public interface ReportRepository extends DataLoader<ReportDocument> {
     long count(String reportName, Integer intersectionID, Long startTime, Long endTime);
 
@@ -14,4 +16,6 @@ Page<ReportDocument> findLatest(String reportName, Integer intersectionID, Long
     Page<ReportDocument> find(String reportName, Integer intersectionID, Long startTime, Long endTime,
             boolean includeReportContents, Pageable pageable);
 
+    List<ReportDocument> findAll(String reportName, Integer intersectionID, Long startTime, Long endTime,
+            boolean includeReportContents);
 }

services/intersection-api/api/src/main/java/us/dot/its/jpo/ode/api/accessors/reports/ReportRepositoryImpl.java

@@ -129,4 +129,31 @@ public Page<ReportDocument> find(
     public void add(ReportDocument item) {
         mongoTemplate.insert(item, collectionName);
     }
+
+    public List<ReportDocument> findAll(
+            String reportName,
+            Integer intersectionID,
+            Long startTime,
+            Long endTime,
+            boolean includeReportContents) {
+        Criteria criteria = new IntersectionCriteria()
+                .whereOptional(REPORT_NAME_FIELD, reportName)
+                .whereOptional(INTERSECTION_ID_FIELD, intersectionID);
+
+        // Match exact start and stop times
+        criteria.and("reportStartTime").is(startTime);
+        criteria.and("reportStopTime").is(endTime);
+
+        Sort sort = Sort.by(Sort.Direction.DESC, DATE_FIELD);
+        List<String> excludedFields = new ArrayList<>();
+        if (!includeReportContents) {
+            excludedFields.add("reportContents");
+        }
+        Query query = Query.query(criteria).with(sort);
+        // Exclude fields if necessary
+        if (!excludedFields.isEmpty()) {
+            query.fields().exclude(excludedFields.toArray(new String[0]));
+        }
+        return mongoTemplate.find(query, ReportDocument.class, collectionName);
+    }
 }

services/intersection-api/api/src/main/java/us/dot/its/jpo/ode/api/accessors/rsuState/RsuStateRepository.java

@@ -0,0 +1,14 @@
+package us.dot.its.jpo.ode.api.accessors.rsuState;
+
+import java.util.List;
+import us.dot.its.jpo.ode.api.models.snmp.RsuState;
+
+public interface RsuStateRepository {
+    List<RsuState> retrieveRsuStateWithinTimeInterval(String rsuIP, long start, long end);
+
+    RsuState findLatestByRsuIP(String rsuIP);
+
+    List<RsuState> retrieveRsuStateWithinTimeInterval(String rsuIP, long start, long end,
+            int intervalMinutes);
+
+}

services/intersection-api/api/src/main/java/us/dot/its/jpo/ode/api/accessors/rsuState/RsuStateRepositoryImpl.java

@@ -0,0 +1,86 @@
+package us.dot.its.jpo.ode.api.accessors.rsuState;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.stereotype.Component;
+import java.util.Date;
+import org.bson.Document;
+
+import us.dot.its.jpo.ode.api.models.snmp.RsuState;
+
+import java.time.Instant;
+import java.time.ZoneOffset;
+
+@Component
+public class RsuStateRepositoryImpl implements RsuStateRepository {
+
+    private final MongoTemplate mongoTemplate;
+    private final String collectionName = "RmMonitoringStatusRecords";
+
+    @Autowired
+    public RsuStateRepositoryImpl(MongoTemplate mongoTemplate) {
+        this.mongoTemplate = mongoTemplate;
+    }
+
+    @Override
+    public List<RsuState> retrieveRsuStateWithinTimeInterval(String rsuIP, long start, long end) {
+        Criteria criteria = Criteria.where("rsuIP").is(rsuIP)
+                .and("timestamp").gte(new Date(start)).lte(new Date(end))
+                .andOperator(
+                        Criteria.where("uptime").ne(-1),
+                        Criteria.where("temperature").ne(-1));
+        Query query = Query.query(criteria).with(Sort.by(Sort.Direction.ASC, "timestamp"));
+        return mongoTemplate.find(query, RsuState.class, collectionName);
+    }
+
+    @Override
+    public List<RsuState> retrieveRsuStateWithinTimeInterval(String rsuIP, long start, long end, int intervalMinutes) {
+        long intervalMs = intervalMinutes * 60 * 1000; // Convert minutes to milliseconds
+
+        Criteria criteria = Criteria.where("rsuIP").is(rsuIP)
+                .and("timestamp").gte(new Date(start)).lte(new Date(end))
+                .andOperator(
+                        Criteria.where("uptime").ne(-1),
+                        Criteria.where("temperature").ne(-1));
+
+        Aggregation aggregation = Aggregation.newAggregation(
+                Aggregation.match(criteria), // filter by rsuIP and timestamp
+                Aggregation.project("timestamp", "rsuIP", "temperature", "uptime", "mode")
+                        .andExpression("{$convert: {input: \"$timestamp\", to: \"long\"}}").as("timestampMillis"),
+                Aggregation.project("timestamp", "rsuIP", "temperature", "uptime", "mode", "timestampMillis")
+                        .andExpression("timestampMillis - (timestampMillis % " + intervalMs + ")").as("interval"),
+                Aggregation.group("interval") // Group by interval
+                        .avg("temperature").as("temperature")
+                        .avg("uptime").as("uptime")
+                        .first("rsuIP").as("rsuIP")
+                        .first("mode").as("mode")
+                        .first("timestamp").as("timestamp"),
+                Aggregation.sort(Sort.by(Sort.Direction.ASC, "timestamp")) // Sort by timestamp
+        );
+
+        List<RsuState> results = mongoTemplate.aggregate(aggregation, collectionName, RsuState.class)
+                .getMappedResults();
+
+        return results;
+    }
+
+    @Override
+    public RsuState findLatestByRsuIP(String rsuIP) {
+        Criteria criteria = Criteria.where("rsuIP").is(rsuIP);
+        Query query = Query.query(criteria)
+                .with(Sort.by(Sort.Direction.DESC, "timestamp"))
+                .limit(1); // Limit the results to just one record
+
+        return mongoTemplate.findOne(query, RsuState.class, collectionName);
+    }
+}