header增加dag_run_id字段

L1/pipelines/csst-msc-l1-mbi/message-router/messageRoute.py

@@ -14,19 +14,22 @@ class messageRoute():
 
         if headers =="null" or "from_job" not in headers:
             print("received redis-cli message")
-            messageRoute.appready(message)
+            headersstr = json.loads(headers)
+            dag_run_id=headersstr["dag_run_id"]
+            messageRoute.appready(dag_run_id,message)
         else:
             try:
                 headersstr = json.loads(headers)
                 from_job=headersstr["from_job"]
                 #from_ip=headersstr["from_ip"]
-                print("from_job :"+from_job)
-                messageRoute.sendsinkjobs(from_job,message)
+                dag_run_id=headersstr["dag_run_id"]
+                print("from_job :"+from_job+" dag_run_id :"+dag_run_id)
+                messageRoute.sendsinkjobs(dag_run_id,from_job,message)
               
             except json.JSONDecodeError as e:
                 print("Invalid JSON format in headers:", e)
     @classmethod
-    def appready(self,message):
+    def appready(self,dag_run_id,message):
         #解析对应的DAG文件
         with open('/dag-yaml/csst-msc-l1-mbi.yml', 'r', encoding='utf-8') as file:
             config = yaml.safe_load(file)
@@ -55,10 +58,10 @@ class messageRoute():
                 if upstream_tasks is None:
                     sink_job=config['tasks'][subkey].get('image')
                     print("The header job is "+sink_job)
-                    messageRoute.sendmsg(sink_job,message)
+                    messageRoute.sendmsg(dag_run_id,sink_job,message)
 
     @classmethod
-    def sendsinkjobs(self,from_job,message):
+    def sendsinkjobs(self,dag_run_id,from_job,message):
         with open('/dag-yaml/csst-msc-l1-mbi.yml', 'r') as file:
             config = yaml.safe_load(file)
 
@@ -68,7 +71,7 @@ class messageRoute():
             if key_to_check in config and isinstance(config[key_to_check], dict):
                 subkey_count = len(config[key_to_check])
 
-            print("The key "+key_to_check+" has "+str(subkey_count)+" subkeys.")
+            print("The key "+key_to_check+" has "+str(subkey_count)+" subkeys. dag_run_id is "+dag_run_id)
 
             if subkey_count == 0:
                 print("DAG file is not correct.")
@@ -92,12 +95,14 @@ class messageRoute():
                 else:
                     if target_value in upstream_tasks:
                         sink_job=config[key_to_check][subkey].get('image')
-                        messageRoute.sendmsg(sink_job,message)
+                        messageRoute.sendmsg(dag_run_id,sink_job,message)
 
     @classmethod
-    def sendmsg(self,job,message):
+    def sendmsg(self,dag_run_id,job,message):
         #执行解包操作
-        command = f"scalebox task add -sink-job={job} {message}"
+        print("sendmsg dag_run_id is "+dag_run_id)
+        command = f"scalebox task add --header dag_run_id={dag_run_id} --sink-job={job} {message}"
+        print("command : "+command)
         result=subprocess.run(command, shell=True)
         if result.returncode == 0:
             print(f"send message {message} to {job}")