fix (scheduler) : create asset subdir in run-input cmd & provide job_…

…name environment variable to executors to maintain backward compatibility (#701)

* fix (scheduler) : create asset subdir in run-input cmd

* fix : add JOB_NAME in executor env

* fix : expected dag

client/cmd/scheduler/run_input.go

@@ -195,6 +195,12 @@ func (j *jobRunInputCommand) writeJobAssetsToFiles(
 	writeToFileFn := utils.WriteStringToFileIndexed()
 	for fileName, fileContent := range jobResponse.Files {
 		filePath := filepath.Join(dirPath, fileName)
+
+		assetSubDir := filepath.Dir(filePath)
+		if err := os.MkdirAll(assetSubDir, fs.FileMode(permission)); err != nil {
+			return fmt.Errorf("failed to create asset sub-directory at %s: %w", assetSubDir, err)
+		}
+
 		if err := writeToFileFn(filePath, fileContent, j.logger.Writer()); err != nil {
 			return fmt.Errorf("failed to write asset file at %s: %w", filePath, err)
 		}

ext/scheduler/airflow/dag/dag.py.tmpl

@@ -114,6 +114,7 @@ asset_volume_mounts = [
 executor_env_vars = [
     k8s.V1EnvVar(name="JOB_LABELS", value='{{.JobDetails.GetLabelsAsString}}'),
     k8s.V1EnvVar(name="JOB_DIR", value=JOB_DIR),
+    k8s.V1EnvVar(name="JOB_NAME", value='{{$.JobDetails.Name}}'),
 ]
 
 init_env_vars = [

ext/scheduler/airflow/dag/expected_dag.py

@@ -84,6 +84,7 @@
 executor_env_vars = [
     k8s.V1EnvVar(name="JOB_LABELS", value='orchestrator=optimus'),
     k8s.V1EnvVar(name="JOB_DIR", value=JOB_DIR),
+    k8s.V1EnvVar(name="JOB_NAME", value='infra.billing.weekly-status-reports'),
 ]
 
 init_env_vars = [