实现在工程云的测试，增加优先级设置

修改yaml和json接口格式

L1/dockerfiles/redis-cli/Dockerfile

@@ -7,9 +7,10 @@ RUN apt-get update \
 ENTRYPOINT []
 
 #ENV REDIS_SERVER=10.0.0.9
-ENV REDIS_SERVER=172.24.232.12
+ENV REDIS_SERVER=192.168.25.55
 ENV REDIS_PORT=26379
 ENV REDIS_PWD=csst__2025
+ENV REDIS_DB=1
 
 COPY --from=hub.cstcloud.cn/scalebox/agent /usr/local /usr/local
 

L1/dockerfiles/redis-cli/Makefile

@@ -9,8 +9,8 @@ dist:
 push:
 	docker push $(IMAGE_NAME)
 run-bash:
-	docker run -it --entrypoint bash -e PGHOST=172.24.232.14:5432 -e GRPC_SERVER=172.24.232.14 -v ${PWD}/config:/config $(IMAGE_NAME)
+	docker run -it --entrypoint bash -e PGHOST=192.168.25.205:5432 -e GRPC_SERVER=192.168.25.205 -v ${PWD}/config:/config $(IMAGE_NAME)
 run:
-	docker run -d -e PGHOST=172.24.232.14:5432 -e GRPC_SERVER=172.24.232.14 -v ${PWD}/config:/config $(IMAGE_NAME)
+	docker run -d -e PGHOST=192.168.25.205:5432 -e GRPC_SERVER=192.168.25.205 -v ${PWD}/config:/config -v /nfs/scalebox/redis-cli:/logs $(IMAGE_NAME)
 down:
 	docker stop $(IMAGE_NAME)

L1/dockerfiles/redis-cli/config/appconfig.txt

-csst-msc-l1-mbi 1
+csst-msc-l1-mbi 21
\ No newline at end of file

L1/dockerfiles/redis-cli/run.sh

@@ -2,12 +2,37 @@
 
 # json串
 # "{\"dag_id\": \"csst-msc-l1-mbi\", \"dag_run_id\": \"202411111505581036\", \"message\": {\"obsid\": \"11009101682009\", \"chipid\": \"24\"}}"
+# {
+#     "priority": 1,
+#     "dag_id": "csst-msc-l1-mbi",
+#     "dag_run_id": "12345",
+#     "dataset": "csst-msc-c9-25sqdeg-v3",
+#     "obs_type": "WIDE",
+#     "project_id": "none",
+#     "obs_id": "10100543790",
+#     "chip_id": "09",
+#     "batch_id": "csst-msc-c9-25-sqdeg-v3-rdx1"
+# }
+
+# 20250621
+# {
+#     "dag_group": "csst_dag.cli.msc_l1",
+#     "dag_group_run": "195244ff176f923aec9a9328c75ecaeb4a8c4345",
+#     "dag": "csst-msc-l1-mbi",
+#     "dag_run": "c89d7e7a022e6f0cdf1daff921c29dbce0ac7c01",
+#     "batch_id": "inttest",
+#     "priority": 1,
+#     "dataset": "csst-msc-c9-25sqdeg-v3",
+#     "obs_type": "WIDE",
+#     "obs_group": "W2",
+#     "obs_id": "10100232366",
+#     "detector": "09"
+# }
 while true
 do
     # obsid=$(redis-cli -h ${REDIS_SERVER} -p ${REDIS_PORT} rpop single-image-reduction:obsid)
-    msg=$(redis-cli -h ${REDIS_SERVER} -p ${REDIS_PORT} -a ${REDIS_PWD} rpop csst_data_list)
-    # msg=$(redis-cli -h 172.24.232.14 -a 123456 lpop csst_data_list_zxl)
-    #redis-cli -h 10.3.10.28 -p 26379 -a "csst__2025" rpop csst_data_list
+    msg=$(redis-cli -h ${REDIS_SERVER} -p ${REDIS_PORT} -a ${REDIS_PWD} -n ${REDIS_DB} rpop csst_data_list)
+
     echo "msg : "$msg
 
     if [ -z "$msg" ]; then
@@ -16,7 +41,7 @@ do
     else
         echo "msg is not empty" $msg
         # 提取dag_id 对应app_id
-        dag_id=$(echo "$msg" | jq -r '.dag_id')
+        dag_id=$(echo "$msg" | jq -r '.dag')
         echo "dag_id : "$dag_id
         #检索dag_id与app_id对应关系的配置文件appconfig.txt
         app_id=$(awk -v key="$dag_id" '$1 == key {print $2}' /config/appconfig.txt)
@@ -30,20 +55,31 @@ do
         fi
 
         # 提取dag_run_id
-        dag_run_id=$(echo "$msg" | jq -r '.dag_run_id')
+        dag_run_id=$(echo "$msg" | jq -r '.dag_run')
         echo "dag_run_id : "$dag_run_id
 
-        message=$(echo "$msg" | jq -c '.message')
+        # message=$(echo "$msg" | jq -c '.message')
+        message=$(echo "$msg" | tr -d '[:space:]')
         
         # 提取message
         #echo "$message" >> "msgs.txt"
 
-        echo "$(date)    massgae :"$message >> "msgs.txt"
+        echo "$(date)    massage :"$message >> "/logs/msgs.txt"
         #send-message $obsid
         job_name="message-router-csst"
-        scalebox task add --app-id ${app_id} --header dag_run_id=${dag_run_id} --header repeatable=yes --upsert --sink-job ${job_name} ${message}
+
+        priority=$(echo "$msg" | jq -r '.priority')
+        if [[ ! "$priority" =~ ^[1-9]$ ]]; then
+            priority="1"
+        fi  
+        num=10
+        priority=$(expr $num - ${priority%%[^1-9]*})
+        echo "priority : "$priority
+
+        scalebox task add --app-id ${app_id} --header sort_tag=${priority} --header dag_run_id=${dag_run_id} --header repeatable=yes --upsert --sink-job ${job_name} ${message}
         code=$?
 
     fi
     echo $code
+    sleep 0.001
 done
\ No newline at end of file

L1/dockerfiles/redis-cli/test.sh

-redis-cli -h 172.24.232.14 -a 123456 rpush csst_data_list "{\"dag_id\": \"csst-msc-l1-mbi\", \"dag_run_id\": \"202411111841239406\", \"batch_id\": \"msc-093-reduction-v2\", \"message\": {\"obsid\": \"11009101682009\", \"chipid\": \"16\", \"dataset\": \"msc-093\", \"batch_id\": \"msc-093-reduction-v2\"}}"
+# redis-cli -h 172.24.232.14 -a 123456 rpush csst_data_list "{\"dag_id\": \"csst-msc-l1-mbi\", \"dag_run_id\": \"202411111841239406\", \"batch_id\": \"msc-093-reduction-v2\", \"message\": {\"obsid\": \"11009101682009\", \"chipid\": \"16\", \"dataset\": \"msc-093\", \"batch_id\": \"msc-093-reduction-v2\"}}"
\ No newline at end of file
+# redis-cli -h 192.168.25.205 -a 123456 rpush csst_data_list "{\"priority\": 1,\"dag_id\": \"csst-msc-l1-mbi\", \"dag_run_id\": \"12345\", \"dataset\": \"csst-msc-c9-25sqdeg-v3\", \"obs_type\": \"WIDE\", \"project_id\": \"none\", \"obs_id\": \"10100543790\", \"chip_id\": \"09\", \"batch_id\": \"csst-msc-c9-25-sqdeg-v3-rdx1\"}"
+redis-cli -h 192.168.25.205 -p 26379 -a 123456 rpush csst_data_list "{\"dag_group\": \"csst_dag.cli.msc_l1\",\"dag_group_run\": \"195244ff176f923aec9a9328c75ecaeb4a8c4345\",\"dag\": \"csst-msc-l1-mbi\",\"dag_run\": \"c89d7e7a022e6f0cdf1daff921c29dbce0ac7c01\",\"batch_id\": \"inttest\",\"priority\": 1,\"dataset\": \"csst-msc-c9-25sqdeg-v3\",\"obs_type\": \"WIDE\",\"obs_group\": \"W2\",\"obs_id\": \"10100232366\",\"detector\": \"09\"}"
\ No newline at end of file