fix (job run-input) : create asset subdir (#702)

* fix (job run-input) : create asset subdir

* fix : add JOB_NAME in executor env

* fix : tests for base_dag

Author: smarchone

client/cmd/job/run_input.go

@@ -191,6 +191,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/airflow2/resources/base_dag.py

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

ext/scheduler/airflow2/resources/expected_compiled_template.py

@@ -92,6 +92,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='foo'),
 ]
 
 init_env_vars = [