Use message properties

refresh_stationws.py

@@ -22,7 +22,7 @@ logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 min_refresh_interval = 600
-delay_refresh = 60
+delay_refresh = 300
 last_refresh = datetime.now() - timedelta(seconds=delay_refresh+1)
 
 rabbitmq_host = os.getenv("AMQP_HOST", "resif-mq.u-ga.fr")
@@ -37,7 +37,7 @@ pguri = os.getenv(
     "postgres://resifinvdev@resif-pgpreprod.u-ga.fr:5432/resifInv-Preprod",
 )
 
-k8s_config_file = os.getenv("KUBE_CONFIG_FILE", "~/.kube/config.yaml")
+k8s_config_file = os.getenv("KUBE_CONFIG_FILE", "~/.kube/config")
 k8s_deployment = os.getenv("K8S_POSTGRES_DEPLOYMENT", "staging-ws-station-postgres")
 k8s_namespace = os.getenv("K8S_NAMESPACE", "staging")
 
@@ -110,14 +110,10 @@ def notify_post_integration(msg):
 
 def on_message(ch, method, properties, body):
     global last_refresh
-    logger.info(f"Received {body}")
-    # parse message of pattern: {"timestamp": 16930587, "msg": "NET.STA"}
+    logger.info(f"Received {body} with properties {properties.headers}")
     try:
-        message = json.loads(body)
-    except Exception:
-        logger.exception("Message could not be parsed: %s", body)
-    try:
-        integration_arrival = datetime.fromtimestamp(int(message['timestamp']))
+        integration_arrival = datetime.fromtimestamp(int(properties.headers['timestamp']))
+        logger.info(integration_arrival)
     except ValueError:
         logger.exception("Expected message body to be a unix epoch in seconds.")
         return 1
@@ -127,7 +123,7 @@ def on_message(ch, method, properties, body):
     # Since we already refreshed all the views
     if integration_arrival < last_refresh:
         logger.info("Discard this message arrived at %s prior to last refresh %s", integration_arrival, last_refresh)
-        notify_post_integration(message['msg'])
+        notify_post_integration(body)
         return 0
 
     if not should_run_now(integration_arrival):