OBU OTA Firmware Request Logging

docker-compose-obu-ota-server.yml

@@ -18,6 +18,13 @@ services:
 
       OTA_USERNAME: ${OTA_USERNAME}
       OTA_PASSWORD: ${OTA_PASSWORD}
+
+      PG_DB_HOST: ${PG_DB_HOST}
+      PG_DB_NAME: ${PG_DB_NAME}
+      PG_DB_USER: ${PG_DB_USER}
+      PG_DB_PASS: ${PG_DB_PASS}
+
+      MAX_COUNT: ${MAX_COUNT}
     volumes:
       - ./resources/ota/firmwares:/firmwares
     logging:

resources/kubernetes/obu-ota-server.yaml

@@ -154,6 +154,22 @@ spec:
                 secretKeyRef:
                   name: some-api-password-name
                   key: some-api-password-key
+            - name: MAX_COUNT
+              value: ''
+            - name: PG_DB_HOST
+              value: ''
+            - name: PG_DB_NAME
+              value: ''
+            - name: PG_DB_USER
+              valueFrom:
+                secretKeyRef:
+                  name: some-postgres-secret-user
+                  key: some-postgres-secret-key
+            - name: PG_DB_PASS
+              valueFrom:
+                secretKeyRef:
+                  name: some-postgres-secret-password
+                  key: some-postgres-secret-key
           volumeMounts:
             - name: cv-manager-service-key
               mountPath: /home/secret

resources/sql_scripts/CVManager_CreateTables.sql

@@ -401,4 +401,25 @@ CREATE TABLE IF NOT EXISTS public.snmp_msgfwd_config
 		ON DELETE NO ACTION
 );
 
+CREATE SEQUENCE public.obu_ota_request_id_seq
+   INCREMENT 1
+   START 1
+   MINVALUE 1
+   MAXVALUE 2147483647
+   CACHE 1;
+
+
+CREATE TABLE IF NOT EXISTS public.obu_ota_requests (
+   request_id integer NOT NULL DEFAULT nextval('obu_ota_request_id_seq'::regclass),
+	obu_sn character varying(128) NOT NULL,
+	request_datetime timestamp NOT NULL,
+	origin_ip inet NOT NULL,
+   obu_firmware_version varchar(128) NOT NULL,
+   requested_firmware_version varchar(128) NOT NULL,
+	error_status bit(1) NOT NULL,
+   error_message varchar(128) NOT NULL,
+   manufacturer int4 NOT NULL,
+	CONSTRAINT fk_manufacturer FOREIGN KEY (manufacturer) REFERENCES public.manufacturers(manufacturer_id)
+);
+
 CREATE SCHEMA IF NOT EXISTS keycloak;

resources/sql_scripts/update_scripts/obu_ota_requests.sql

@@ -0,0 +1,20 @@
+CREATE SEQUENCE public.obu_ota_request_id_seq
+   INCREMENT 1
+   START 1
+   MINVALUE 1
+   MAXVALUE 2147483647
+   CACHE 1;
+
+
+CREATE TABLE IF NOT EXISTS public.obu_ota_requests (
+   request_id integer NOT NULL DEFAULT nextval('obu_ota_request_id_seq'::regclass),
+	obu_sn character varying(128) NOT NULL,
+	request_datetime timestamp NOT NULL,
+	origin_ip inet NOT NULL,
+   obu_firmware_version varchar(128) NOT NULL,
+   requested_firmware_version varchar(128) NOT NULL,
+	error_status bit(1) NOT NULL,
+   error_message varchar(128) NOT NULL,
+   manufacturer int4 NOT NULL,
+	CONSTRAINT fk_manufacturer FOREIGN KEY (manufacturer) REFERENCES public.manufacturers(manufacturer_id)
+);

sample.env

@@ -227,6 +227,9 @@ NGINX_ENCRYPTION="plain"
 SERVER_CERT_FILE="ota_server.crt"
 SERVER_KEY_FILE="ota_server.key"
 
+# Max number of succesfull firmware upgrades to keep in the database per device SN
+MAX_COUNT = 10
+
 # ---------------------------------------------------------------------
 
 

services/addons/images/obu_ota_server/README.md

@@ -50,6 +50,16 @@ The following environmental variables must be set:
 
 <b>OTA_PASSWORD:</b> Password to be used with basic authentication
 
+<b>PG_DB_USER:</b> PostgreSQL access username.
+
+<b>PG_DB_PASS:</b> PostgreSQL access password.
+
+<b>PG_DB_NAME:</b> PostgreSQL database name.
+
+<b>PG_DB_HOST:</b> PostgreSQL hostname, make sure to include port number.
+
+<b>MAX_COUNT:</b> Max number of succesfull firmware upgrades to keep in the database per device SN.
+
 ### GCP required variables <a name = "gcp-requirements"></a>
 
 <b>BLOB_STORAGE_BUCKET:</b> Cloud blob storage bucket for firmware storage.

services/addons/images/obu_ota_server/commsignia_manifest.py

@@ -32,7 +32,7 @@
 )
 
 
-def add_contents(server: str, firmware_list: List):
+def add_contents(server: str, firmware_list: List) -> dict:
     manifest = copy.deepcopy(document)
 
     for firmware in firmware_list:

services/addons/images/obu_ota_server/obu_ota_server.py

@@ -1,23 +1,23 @@
+from typing import Any
 from fastapi import FastAPI, Request, Response, HTTPException, Depends
 from fastapi.security import HTTPBasic, HTTPBasicCredentials
-from fastapi.middleware.cors import CORSMiddleware
-from common import gcs_utils
+from common import gcs_utils, pgquery
 import commsignia_manifest
 import os
 import glob
 import aiofiles
 from starlette.responses import Response
 import logging
+from datetime import datetime
+import asyncio
 
 app = FastAPI()
 log_level = "INFO" if "LOGGING_LEVEL" not in os.environ else os.environ["LOGGING_LEVEL"]
-
 logging.basicConfig(format="%(levelname)s:%(message)s", level=log_level)
-
 security = HTTPBasic()
 
 
-def authenticate_user(credentials: HTTPBasicCredentials = Depends(security)):
+def authenticate_user(credentials: HTTPBasicCredentials = Depends(security)) -> str:
     correct_username = os.getenv("OTA_USERNAME")
     correct_password = os.getenv("OTA_PASSWORD")
     if (
@@ -40,7 +40,7 @@ async def read_root(request: Request):
     }
 
 
-def get_firmware_list():
+def get_firmware_list() -> list:
     blob_storage_provider = os.getenv("BLOB_STORAGE_PROVIDER", "DOCKER")
     files = []
     file_extension = ".tar.sig"
@@ -53,7 +53,7 @@ def get_firmware_list():
 
 
 @app.get("/firmwares/commsignia", dependencies=[Depends(authenticate_user)])
-async def get_manifest(request: Request):
+async def get_manifest(request: Request) -> dict[str, Any]:
     try:
         files = get_firmware_list()
         logging.debug(f"get_manifest :: Files: {files}")
@@ -64,7 +64,7 @@ async def get_manifest(request: Request):
         raise HTTPException(status_code=500, detail=str(e))
 
 
-def get_firmware(firmware_id: str, local_file_path: str):
+def get_firmware(firmware_id: str, local_file_path: str) -> bool:
     try:
         blob_storage_provider = os.getenv("BLOB_STORAGE_PROVIDER", "DOCKER")
         # checks if firmware exists locally
@@ -78,11 +78,11 @@ def get_firmware(firmware_id: str, local_file_path: str):
                 return gcs_utils.download_gcp_blob(firmware_id, local_file_path)
         return True
     except Exception as e:
-        logging.error(f"Error getting firmware: {e}")
+        logging.error(f"parse_range_header: Error getting firmware: {e}")
         raise HTTPException(status_code=500, detail="Error getting firmware")
 
 
-def parse_range_header(range_header):
+def parse_range_header(range_header: str) -> tuple[int, int | None]:
     start, end = 0, None
     try:
         if range_header:
@@ -99,7 +99,9 @@ def parse_range_header(range_header):
     return start, end
 
 
-async def read_file(file_path, start, end):
+async def read_file(
+    file_path: str, start: int, end: int | None
+) -> tuple[bytes, int, int]:
     try:
         async with aiofiles.open(file_path, mode="rb") as file:
             file_size = os.path.getsize(file_path)
@@ -109,29 +111,83 @@ async def read_file(file_path, start, end):
             data = await file.read(end - start)
         return data, file_size, end
     except Exception as e:
-        logging.error(f"Error reading file: {e}")
+        logging.error(f"read_file: Error reading file: {e}")
         raise HTTPException(status_code=500, detail="Error reading file")
 
 
+def removed_old_logs(serialnum: str):
+    try:
+        max_count = int(os.getenv("MAX_COUNT", 10))
+        success_count = pgquery.query_db(
+            f"SELECT COUNT(*) FROM public.obu_ota_requests WHERE obu_sn = '{serialnum}' AND error_status = B'0'"
+        )
+        if success_count[0][0] > max_count:
+            excess_count = success_count[0][0] - max_count
+            oldest_entries = pgquery.query_db(
+                f"SELECT request_id FROM public.obu_ota_requests WHERE obu_sn = '{serialnum}' AND error_status = B'0' ORDER BY request_datetime ASC LIMIT {excess_count}"
+            )
+            oldest_ids = [entry[0] for entry in oldest_entries]
+            pgquery.write_db(
+                f"DELETE FROM public.obu_ota_requests WHERE request_id IN ({','.join(map(str, oldest_ids))})"
+            )
+            logging.debug(
+                f"removed_old_logs: Removed {excess_count} old logs for serialnum: {serialnum}"
+            )
+    except Exception as e:
+        logging.error(f"removed_old_logs: Error removing old entry: {e}")
+
+
+async def log_request(
+    manufacturer: int,
+    request: Request,
+    firmware_id: str,
+    error_status: int,
+    error_message: str,
+):
+    try:
+        query_params = request.query_params
+        serialnum = query_params.get("serialnum")
+        version = query_params.get("version")
+
+        origin_ip = request.client.host
+
+        current_dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        query = f"INSERT INTO public.obu_ota_requests (obu_sn, manufacturer, request_datetime, origin_ip, obu_firmware_version, requested_firmware_version, error_status, error_message) VALUES\
+                ('{serialnum}', {manufacturer}, '{current_dt}', '{origin_ip}', '{version}', '{firmware_id}', B'{error_status}', '{error_message}')"
+        logging.debug(f"Logging request to postgres with insert query: \n{query}")
+        pgquery.write_db(query)
+        removed_old_logs(serialnum)
+    except Exception as e:
+        logging.error(f"log_request: Error logging request: {e} with query: {query}")
+
+
 @app.get(
     "/firmwares/commsignia/{firmware_id}", dependencies=[Depends(authenticate_user)]
 )
 async def get_fw(request: Request, firmware_id: str):
-    file_path = f"/firmwares/{firmware_id}"
+    try:
+        file_path = f"/firmwares/{firmware_id}"
 
-    # Checks if firmware exists locally or downloads it from GCS
-    if not get_firmware(firmware_id, file_path):
-        raise HTTPException(status_code=404, detail="Firmware not found")
+        # Checks if firmware exists locally or downloads it from GCS
+        if not get_firmware(firmware_id, file_path):
+            raise HTTPException(status_code=404, detail="Firmware not found")
 
-    header_start, header_end = parse_range_header(request.headers.get("Range"))
-    data, file_size, end = await read_file(file_path, header_start, header_end)
+        header_start, header_end = parse_range_header(request.headers.get("Range"))
+        data, file_size, end = await read_file(file_path, header_start, header_end)
 
-    headers = {
-        "Content-Range": f"bytes {header_start}-{end-1}/{file_size}",
-        "Content-Length": str(end - header_start),
-        "Accept-Ranges": "bytes",
-    }
+        headers = {
+            "Content-Range": f"bytes {header_start}-{end-1}/{file_size}",
+            "Content-Length": str(end - header_start),
+            "Accept-Ranges": "bytes",
+        }
 
-    return Response(
-        content=data, media_type="application/octet-stream", headers=headers
-    )
+        asyncio.create_task(log_request(1, request, firmware_id, 0, ""))
+        return Response(
+            content=data, media_type="application/octet-stream", headers=headers
+        )
+    except Exception as e:
+        asyncio.create_task(log_request(1, request, firmware_id, 1, e.detail))
+        logging.error(
+            f"get_fw: Error responding with firmware with error: {e} for firmware_id: {firmware_id}"
+        )
+        raise

services/addons/images/obu_ota_server/requirements.txt

@@ -5,3 +5,5 @@ aiofiles==23.2.1
 google-cloud-storage==2.14.0
 python-dateutil==2.8.2
 pytz==2023.3.post1
+sqlalchemy==2.0.21
+pg8000==1.30.2

services/addons/images/obu_ota_server/sample.env

@@ -25,4 +25,13 @@ NGINX_ENCRYPTION="plain"
 
 # SSL file name in path /docker/nginx/ssl/
 SERVER_CERT_FILE="ota_server.crt"
-SERVER_KEY_FILE="ota_server.key"
+SERVER_KEY_FILE="ota_server.key"
+
+# Max number of succesfull firmware upgrades to keep in the database per device SN
+MAX_COUNT = 10
+
+# PostgreSQL database variables
+PG_DB_HOST=""
+PG_DB_NAME=""
+PG_DB_USER=""
+PG_DB_PASS=""