feat: 支持 v2 DagRunGroup 格式并修复目录权限 (8a728034) · Commits · csst-cicd / csst-pipeline-portal

.gitignore

+1 −0

Original line number	Diff line number	Diff line
		@@ -63,3 +63,4 @@ data/
		# Generated during Ansible variable import
		docker-celery-3.0.5/api_gateway/ccds/
		docker-celery-3.0.5/api_gateway/csst-dfs-client/
		external/csst-dag

docker-celery-3.0.5/ansible/deploy.yml

+13 −0

Original line number	Diff line number	Diff line
		@@ -257,6 +257,19 @@
		path: "{{ _deploy_dir }}/deploy_configs/{{ _env }}/dag.env"
		state: absent

		- name: Ensure worker shared directories are writable by airflow user
		file:
		path: "{{ item }}"
		state: directory
		owner: "50000"
		group: "0"
		mode: "0775"
		loop:
		- "{{ _deploy_dir }}/volumes/logs"
		- "{{ _deploy_dir }}/dags"
		- "{{ _deploy_dir }}/volumes/plugins"
		- "{{ _deploy_dir }}/volumes/config"

		- name: Start Worker services (celery-worker, filebeat, monitoring)
		shell: \|
		export HARBOR_PROJECT="{{ harbor_project \| default('') }}"

docker-celery-3.0.5/ansible/update.yml

+13 −0

Original line number	Diff line number	Diff line
		@@ -246,6 +246,19 @@
		state: absent
		ignore_errors: yes

		- name: Ensure worker shared directories are writable by airflow user
		file:
		path: "{{ item }}"
		state: directory
		owner: "50000"
		group: "0"
		mode: "0775"
		loop:
		- "{{ _deploy_dir }}/volumes/logs"
		- "{{ _deploy_dir }}/dags"
		- "{{ _deploy_dir }}/volumes/plugins"
		- "{{ _deploy_dir }}/volumes/config"

		- name: Render docker compose extra_hosts override on worker (optional)
		template:
		src: templates/docker-compose.extra-hosts.override.yml.j2

docker-celery-3.0.5/api_gateway/main.py

+26 −13

Original line number	Diff line number	Diff line
		@@ -275,8 +275,12 @@ def delete_user(username: str, db: Session = Depends(get_db), current_user: User
		return {"ok": True}

		class BatchSubmitRequest(BaseModel):
		dag_group_run: Dict[str, Any]
		dag_run_list: List[Dict[str, Any]]
		# v1 fields
		dag_group_run: Optional[Dict[str, Any]] = None
		dag_run_list: Optional[List[Dict[str, Any]]] = None
		# v2 fields
		job: Optional[Dict[str, Any]] = None
		dag_runs: Optional[List[Dict[str, Any]]] = None

		@app.post("/api/tasks/batch-submit", response_model=BatchSubmitResponse)
		def batch_submit_tasks(
		@@ -285,7 +289,8 @@ def batch_submit_tasks(
		db: Session = Depends(get_db)
		):
		"""
		Batch submit tasks. Accepts JSON with dag_group_run and dag_run_list.
		Batch submit tasks. Accepts JSON with dag_group_run and dag_run_list (v1)
		or job and dag_runs (v2).
		Extracts dag and dag_run, stores in PostgreSQL JSONB,
		and triggers Airflow asynchronously.
		"""
		@@ -293,22 +298,25 @@ def batch_submit_tasks(
		db_tasks = []
		seen_task_ids = set()

		group_info = dict(payload.dag_group_run or {})
		tasks = payload.dag_run_list
		# 兼容 v1 和 v2 载荷
		group_info = dict(payload.job or payload.dag_group_run or {})
		tasks = payload.dag_runs or payload.dag_run_list or []

		group_run_id = str(group_info.get("dag_group_run") or "").strip()
		group_run_id = str(group_info.get("dag_run_group") or group_info.get("dag_group_run") or "").strip()
		if not group_run_id:
		group_run_id = str(uuid.uuid4())
		group_info["dag_group_run"] = group_run_id
		group_info["dag_run_group"] = group_run_id

		for task in tasks:
		# Fallback to existing fields if new fields are used
		requested_dag_id = task.get("dag") or task.get("dag_id")
		dag_run_id = task.get("dag_run") or task.get("dag_run_id")
		job_info = task.get("job", {}) if isinstance(task.get("job"), dict) else {}

		requested_dag_id = job_info.get("dag_id") or task.get("dag_id") or task.get("dag")
		dag_run_id = job_info.get("dag_run") or task.get("dag_run_id") or task.get("dag_run")
		requested_priority = task.get("priority")

		if not requested_dag_id or not dag_run_id:
		raise HTTPException(status_code=400, detail="Each task must contain 'dag' and 'dag_run'")
		raise HTTPException(status_code=400, detail="Each task must contain 'dag_id' and 'dag_run' (or v1 equivalent)")

		_, effective_priority, scheduled_dag_id = resolve_scheduled_dag_id(
		requested_dag_id,
		@@ -365,12 +373,17 @@ def get_task_groups(limit: int = 50, offset: int = 0, db: Session = Depends(get_
		"""
		from sqlalchemy import case, func

		# Group by inputs->>'dag_group_run'
		group_col = TaskRecord.inputs['dag_group_run'].astext
		group_col = func.coalesce(
		TaskRecord.inputs['dag_run_group'].astext,
		TaskRecord.inputs['dag_group_run'].astext
		)

		query = db.query(
		group_col.label('dag_group_run'),
		func.max(TaskRecord.inputs['batch_id'].astext).label('batch_id'),
		func.coalesce(
		func.max(TaskRecord.inputs['batch_id'].astext),
		func.max(TaskRecord.inputs['job']['batch_id'].astext)
		).label('batch_id'),
		func.max(TaskRecord.created_at).label('created_at'),
		func.count(TaskRecord.task_id).label('total_tasks'),
		func.sum(case((TaskRecord.status == 'success', 1), else_=0)).label('success_tasks'),

docker-celery-3.0.5/frontend_vue/src/components/SubmissionsView.vue

+65 −1

Original line number	Diff line number	Diff line
		@@ -18,7 +18,12 @@

		<div class="grid gap-5 p-6">
		<label class="flex flex-col gap-2">
		<div class="flex items-center justify-between">
		<span class="text-[11px] font-semibold uppercase tracking-wider text-[#6B7A90]">Airflow DAG</span>
		<el-button v-if="selectedDagId" size="small" type="primary" plain @click="generateTemplate">
		生成 DagRunGroup 模板
		</el-button>
		</div>
		<el-select
		v-model="selectedDagId"
		filterable
		@@ -80,6 +85,65 @@ import { computed, onMounted, ref, watch } from 'vue'
		import axios from 'axios'
		import { ElMessage } from 'element-plus'

		// 生成随机 UUID (简化的 v4 实现)
		function uuidv4() {
		return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {
		const r = Math.random() * 16 \| 0, v = c === 'x' ? r : (r & 0x3 \| 0x8)
		return v.toString(16)
		})
		}

		// 模拟 Python 中的 v2 DagRunGroup 结构
		const generateTemplate = () => {
		const dagId = selectedDagId.value \|\| 'example-dag-id'
		const groupId = uuidv4()
		const runId = uuidv4()
		const now = new Date().toISOString()

		const template = {
		"job": {
		"dag_id": dagId,
		"dag_run_group": groupId,
		"batch_id": "default",
		"created_time": now
		},
		"dag_runs": [
		{
		"job": {
		"dag_id": dagId,
		"dag_run": runId,
		"dag_run_group": groupId,
		"batch_id": "default",
		"created_time": now
		},
		"data": {
		"dataset": "C9",
		"instrument": "MSC",
		"obs_type": "SCI",
		"obs_id": "10100100000",
		"detector": "01",
		"filter": "i"
		},
		"aux": {
		"object": "M31",
		"data_list": ["file1.fits", "file2.fits"],
		"n_file_expected": 2,
		"n_file_found": 2
		},
		"proc": {
		"pmapname": "pmap_v1",
		"extra_kwargs": {
		"use_gpu": true
		}
		}
		}
		]
		}

		jsonText.value = JSON.stringify(template, null, 2)
		ElMessage.success('已生成 v2 DagRunGroup 模板')
		}

		type AirflowDagItem = {
		dag_id: string
		is_paused?: boolean