fixes for uploader + cdr

- uploaded will now load cdr json and add cog_id, system and system_version.
- timeout on ack for RabbitMQ is now 2 hours

50-criticalmaas.conf

@@ -0,0 +1 @@
+consumer_timeout = 7200000

CHANGELOG.md

@@ -4,6 +4,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## [0.1.0] - 2024-04-16
+
+### Changed
+- uploaded will now load cdr json and add cog_id, system and system_version.
+
+### Fixed
+- timeout on ack for RabbitMQ is now 2 hours
+
 ## [0.0.7] - 2024-04-15
 
 This is the inital release of the UIUC CDR Processing steps. The pipeline is in another git repository.

docker-compose.yml

@@ -49,6 +49,7 @@ services:
       RABBITMQ_DEFAULT_PASS: "${RABBITMQ_PASSWORD:-guest}"
     volumes:
       - rabbitmq:/var/lib/rabbitmq
+      - ./50-criticalmaas.conf:/etc/rabbitmq/conf.d/50-criticalmaas.conf:ro
 
   # ----------------------------------------------------------------------
   # CDR HOOK
@@ -114,6 +115,8 @@ services:
       - /logs/logs.latest
       - --output
       - /output
+      - --feedback
+      - /feedback
       - --amqp
       - "amqp://${RABBITMQ_USERNAME}:${RABBITMQ_PASSWORD}@rabbitmq/%2F"
       - --model
@@ -123,6 +126,7 @@ services:
       - "data:/data"
       - "logs:/logs"
       - "output:/output"
+      - "feedback:/feedback"
 
   # ----------------------------------------------------------------------
   # DATA PROCESSING PIPELINE
@@ -145,6 +149,7 @@ services:
 volumes:
   traefik:
   rabbitmq:
+  feedback:
   data:
   logs:
   output:

uploader/uploader.py

@@ -3,6 +3,7 @@
 import os
 import pika
 import requests
+from requests.exceptions import RequestException
 import threading
 
 # rabbitmq uri
@@ -25,9 +26,17 @@ def run(self):
             logging.debug(f"Uploading data for {data['cog_id']} from {file}")
             headers = {'Authorization': f'Bearer {cdr_token}', 'Content-Type': 'application/json'}
             with open(file, 'rb') as f:
-                response = requests.post(f'{cdr_url}/v1/maps/publish/features', data=f, headers=headers)
-                logging.debug(response.text)
-                response.raise_for_status()
+                cdr_data = json.load(f)
+            # TODO this needs to be in pipeline.py
+            cdr_data['cog_id'] = data['cog_id']
+            cdr_data['system'] = data['system']
+            cdr_data['system_version'] = data['version']
+            response = requests.post(f'{cdr_url}/v1/maps/publish/features', data=json.dumps(cdr_data), headers=headers)
+            logging.debug(response.text)
+            response.raise_for_status()
+        except RequestException as e:
+            logging.exception(f"Request Error {response.text}.")
+            self.exception = e
         except Exception as e:
             logging.exception("Error processing pipeline request.")
             self.exception = e