|
|
@@ -58,7 +58,8 @@ def run_job(ini_path: str,
|
|
|
'-start-date', start_date,
|
|
|
'-stop-date', stop_date,
|
|
|
]
|
|
|
- gen_rc = _run_local_or_remote(gen_argv, worker_host, current_host, stdout=stdout, stderr=stderr)
|
|
|
+ gen_rc = _run_local_or_remote(gen_argv, worker_host, current_host, base_dir,
|
|
|
+ stdout=stdout, stderr=stderr)
|
|
|
if gen_rc != 0:
|
|
|
raise RuntimeError('生成 DataX json 失败: ' + ini_path)
|
|
|
|
|
|
@@ -70,16 +71,21 @@ def run_job(ini_path: str,
|
|
|
os.path.join(datax_home, 'bin', 'datax.py'),
|
|
|
json_path,
|
|
|
]
|
|
|
- return _run_local_or_remote(exec_argv, worker_host, current_host, stdout=stdout, stderr=stderr)
|
|
|
+ return _run_local_or_remote(exec_argv, worker_host, current_host, base_dir,
|
|
|
+ stdout=stdout, stderr=stderr)
|
|
|
|
|
|
|
|
|
-def _run_local_or_remote(argv, worker_host: str, current_host: str,
|
|
|
+def _run_local_or_remote(argv, worker_host: str, current_host: str, base_dir: str,
|
|
|
stdout=None, stderr=None) -> int:
|
|
|
"""
|
|
|
- 本机 = subprocess.run 直跑;远端 = ssh worker_host '<cmd>' 传字符串。
|
|
|
+ 本机:subprocess.run 直跑 + PYTHONPATH 注入 base_dir(让 python -m dw_base.datax.cli 能找到包)
|
|
|
+ 远端:ssh worker_host 'cd <base_dir> && <cmd>'(cwd 为项目根,同样保证 -m 能找到 dw_base)
|
|
|
stdout/stderr 默认继承父进程;batch 层可传文件句柄做每任务独立日志。
|
|
|
"""
|
|
|
if worker_host == current_host:
|
|
|
- return subprocess.run(argv, stdout=stdout, stderr=stderr).returncode
|
|
|
- remote_cmd = ' '.join(shlex.quote(a) for a in argv)
|
|
|
+ env = os.environ.copy()
|
|
|
+ existing = env.get('PYTHONPATH', '')
|
|
|
+ env['PYTHONPATH'] = base_dir + (os.pathsep + existing if existing else '')
|
|
|
+ return subprocess.run(argv, stdout=stdout, stderr=stderr, env=env).returncode
|
|
|
+ remote_cmd = 'cd ' + shlex.quote(base_dir) + ' && ' + ' '.join(shlex.quote(a) for a in argv)
|
|
|
return subprocess.run(['ssh', worker_host, remote_cmd], stdout=stdout, stderr=stderr).returncode
|