YARN-3852. Add docker container support to container-executor. Contributed by Abin Shahab.

hadoop-yarn-project/CHANGES.txt

@@ -150,6 +150,9 @@ Release 2.8.0 - UNRELEASED
     YARN-3656. LowCost: A Cost-Based Placement Agent for YARN Reservations. 
     (Jonathan Yaniv and Ishai Menache via curino)
 
+    YARN-3852. Add docker container support to container-executor
+    (Abin Shahab via vvasudev)
+
   IMPROVEMENTS
 
     YARN-644. Basic null check is not performed on passed in arguments before

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/configuration.c

@@ -291,27 +291,23 @@ char ** get_values(const char * key) {
   return extract_values(value);
 }
 
-/**
+char ** extract_values_delim(char *value, const char *delim) {
- * Extracts array of values from the '%' separated list of values.
- */
-char ** extract_values(char *value) {
   char ** toPass = NULL;
   char *tempTok = NULL;
   char *tempstr = NULL;
   int size = 0;
   int toPassSize = MAX_SIZE;
-
   //first allocate any array of 10
   if(value != NULL) {
     toPass = (char **) malloc(sizeof(char *) * toPassSize);
-    tempTok = strtok_r((char *)value, "%", &tempstr);
+    tempTok = strtok_r((char *)value, delim, &tempstr);
     while (tempTok != NULL) {
       toPass[size++] = tempTok;
       if(size == toPassSize) {
         toPassSize += MAX_SIZE;
         toPass = (char **) realloc(toPass,(sizeof(char *) * toPassSize));
       }
-      tempTok = strtok_r(NULL, "%", &tempstr);
+      tempTok = strtok_r(NULL, delim, &tempstr);
     }
   }
   if (toPass != NULL) {
@@ -320,6 +316,13 @@ char ** extract_values(char *value) {
   return toPass;
 }
 
+/**
+ * Extracts array of values from the '%' separated list of values.
+ */
+char ** extract_values(char *value) {
+  extract_values_delim(value, "%");
+}
+
 // free an entry set of values
 void free_values(char** values) {
   if (*values != NULL) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/configuration.h

@@ -46,6 +46,8 @@ char ** get_values(const char* key);
 // Extracts array of values from the comma separated list of values.
 char ** extract_values(char *value);
 
+char ** extract_values_delim(char *value, const char *delim);
+
 // free the memory returned by get_values
 void free_values(char** values);
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c

@@ -207,16 +207,20 @@ static int write_pid_to_file_as_nm(const char* pid_file, pid_t pid) {
   uid_t user = geteuid();
   gid_t group = getegid();
   if (change_effective_user(nm_uid, nm_gid) != 0) {
+    fprintf(ERRORFILE, "Could not change to effective users %d, %d\n", nm_uid, nm_gid);
+    fflush(ERRORFILE);
     return -1;
   }
 
   char *temp_pid_file = concatenate("%s.tmp", "pid_file_path", 1, pid_file);
-
+  fprintf(LOGFILE, "Writing to tmp file %s\n", temp_pid_file);
+  fflush(LOGFILE);
   // create with 700
   int pid_fd = open(temp_pid_file, O_WRONLY|O_CREAT|O_EXCL, S_IRWXU);
   if (pid_fd == -1) {
     fprintf(LOGFILE, "Can't open file %s as node manager - %s\n", temp_pid_file,
            strerror(errno));
+    fflush(LOGFILE);
     free(temp_pid_file);
     return -1;
   }
@@ -229,6 +233,7 @@ static int write_pid_to_file_as_nm(const char* pid_file, pid_t pid) {
   if (written == -1) {
     fprintf(LOGFILE, "Failed to write pid to file %s as node manager - %s\n",
        temp_pid_file, strerror(errno));
+    fflush(LOGFILE);
     free(temp_pid_file);
     return -1;
   }
@@ -238,6 +243,7 @@ static int write_pid_to_file_as_nm(const char* pid_file, pid_t pid) {
   if (rename(temp_pid_file, pid_file)) {
     fprintf(LOGFILE, "Can't move pid file from %s to %s as node manager - %s\n",
         temp_pid_file, pid_file, strerror(errno));
+    fflush(LOGFILE);
     unlink(temp_pid_file);
     free(temp_pid_file);
     return -1;
@@ -848,12 +854,15 @@ static int copy_file(int input, const char* in_filename,
 		     const char* out_filename, mode_t perm) {
   const int buffer_size = 128*1024;
   char buffer[buffer_size];
+
   int out_fd = open(out_filename, O_WRONLY|O_CREAT|O_EXCL|O_NOFOLLOW, perm);
   if (out_fd == -1) {
     fprintf(LOGFILE, "Can't open %s for output - %s\n", out_filename, 
             strerror(errno));
+    fflush(LOGFILE);
     return -1;
   }
+
   ssize_t len = read(input, buffer, buffer_size);
   while (len > 0) {
     ssize_t pos = 0;
@@ -1026,7 +1035,325 @@ int initialize_app(const char *user, const char *app_id,
   return -1;
 }
 
-int launch_container_as_user(const char *user, const char *app_id, 
+char* parse_docker_command_file(const char* command_file) {
+  int i = 0;
+  size_t len = 0;
+  char *line = NULL;
+  ssize_t read;
+  FILE *stream;
+  stream = fopen(command_file, "r");
+  if (stream == NULL) {
+   fprintf(ERRORFILE, "Cannot open file %s - %s",
+                 command_file, strerror(errno));
+   fflush(ERRORFILE);
+   exit(ERROR_OPENING_FILE);
+  }
+  if ((read = getline(&line, &len, stream)) == -1) {
+     fprintf(ERRORFILE, "Error reading command_file %s\n", command_file);
+     fflush(ERRORFILE);
+     exit(ERROR_READING_FILE);
+  }
+  fclose(stream);
+
+  return line;
+}
+
+int run_docker(const char *command_file) {
+  char* docker_command = parse_docker_command_file(command_file);
+  char* docker_binary = get_value(DOCKER_BINARY_KEY);
+  char* docker_command_with_binary = calloc(sizeof(char), PATH_MAX);
+  sprintf(docker_command_with_binary, "%s %s", docker_binary, docker_command);
+  char **args = extract_values_delim(docker_command_with_binary, " ");
+
+  int exit_code = -1;
+  if (execvp(docker_binary, args) != 0) {
+    fprintf(ERRORFILE, "Couldn't execute the container launch with args %s - %s",
+              docker_binary, strerror(errno));
+      fflush(LOGFILE);
+      fflush(ERRORFILE);
+      free(docker_binary);
+      free(args);
+      free(docker_command_with_binary);
+      free(docker_command);
+      exit_code = DOCKER_RUN_FAILED;
+  }
+  exit_code = 0;
+  return exit_code;
+}
+
+int create_script_paths(const char *work_dir,
+  const char *script_name, const char *cred_file,
+  char** script_file_dest, char** cred_file_dest,
+  int* container_file_source, int* cred_file_source ) {
+  int exit_code = -1;
+
+  *script_file_dest = get_container_launcher_file(work_dir);
+  if (script_file_dest == NULL) {
+    exit_code = OUT_OF_MEMORY;
+    fprintf(ERRORFILE, "Could not create script_file_dest");
+    fflush(ERRORFILE);
+    return exit_code;
+  }
+
+  *cred_file_dest = get_container_credentials_file(work_dir);
+  if (NULL == cred_file_dest) {
+    exit_code = OUT_OF_MEMORY;
+    fprintf(ERRORFILE, "Could not create cred_file_dest");
+    fflush(ERRORFILE);
+    return exit_code;
+  }
+  // open launch script
+  *container_file_source = open_file_as_nm(script_name);
+  if (*container_file_source == -1) {
+    exit_code = INVALID_NM_ROOT_DIRS;
+    fprintf(ERRORFILE, "Could not open container file");
+    fflush(ERRORFILE);
+    return exit_code;
+  }
+  // open credentials
+  *cred_file_source = open_file_as_nm(cred_file);
+  if (*cred_file_source == -1) {
+    exit_code = INVALID_ARGUMENT_NUMBER;
+    fprintf(ERRORFILE, "Could not open cred file");
+    fflush(ERRORFILE);
+    return exit_code;
+  }
+
+  exit_code = 0;
+  return exit_code;
+}
+
+int create_local_dirs(const char * user, const char *app_id,
+                       const char *container_id, const char *work_dir,
+                       const char *script_name, const char *cred_file,
+                       char* const* local_dirs,
+                       char* const* log_dirs, int effective_user,
+                       char* script_file_dest, char* cred_file_dest,
+                       int container_file_source, int cred_file_source) {
+  int exit_code = -1;
+  // create the user directory on all disks
+  int result = initialize_user(user, local_dirs);
+  if (result != 0) {
+    fprintf(ERRORFILE, "Could not create user dir");
+    fflush(ERRORFILE);
+    return result;
+  }
+
+  // initializing log dirs
+  int log_create_result = create_log_dirs(app_id, log_dirs);
+  if (log_create_result != 0) {
+    fprintf(ERRORFILE, "Could not create log dirs");
+    fflush(ERRORFILE);
+    return log_create_result;
+  }
+  if (effective_user == 1) {
+    if (change_effective_user(user_detail->pw_uid, user_detail->pw_gid) != 0) {
+      fprintf(ERRORFILE, "Could not change to effective users %d, %d\n", user_detail->pw_uid, user_detail->pw_gid);
+      fflush(ERRORFILE);
+      goto cleanup;
+    }
+  } else {
+   // give up root privs
+    if (change_user(user_detail->pw_uid, user_detail->pw_gid) != 0) {
+      exit_code = SETUID_OPER_FAILED;
+      goto cleanup;
+    }
+  }
+  // Create container specific directories as user. If there are no resources
+  // to localize for this container, app-directories and log-directories are
+  // also created automatically as part of this call.
+  if (create_container_directories(user, app_id, container_id, local_dirs,
+                                   log_dirs, work_dir) != 0) {
+    fprintf(ERRORFILE, "Could not create container dirs");
+    fflush(ERRORFILE);
+    goto cleanup;
+  }
+
+  // 700
+  if (copy_file(container_file_source, script_name, script_file_dest,S_IRWXU) != 0) {
+    fprintf(ERRORFILE, "Could not create copy file %d %s\n", container_file_source, script_file_dest);
+    fflush(ERRORFILE);
+    exit_code = INVALID_COMMAND_PROVIDED;
+    goto cleanup;
+  }
+
+  // 600
+  if (copy_file(cred_file_source, cred_file, cred_file_dest,
+        S_IRUSR | S_IWUSR) != 0) {
+    exit_code = UNABLE_TO_EXECUTE_CONTAINER_SCRIPT;
+    fprintf(ERRORFILE, "Could not copy file");
+    fflush(ERRORFILE);
+    goto cleanup;
+  }
+
+  if (chdir(work_dir) != 0) {
+    fprintf(ERRORFILE, "Can't change directory to %s -%s\n", work_dir,
+      strerror(errno));
+      fflush(ERRORFILE);
+    goto cleanup;
+  }
+  exit_code = 0;
+  cleanup:
+  return exit_code;
+}
+
+int launch_docker_container_as_user(const char * user, const char *app_id,
+                              const char *container_id, const char *work_dir,
+                              const char *script_name, const char *cred_file,
+                              const char *pid_file, char* const* local_dirs,
+                              char* const* log_dirs, const char *command_file,
+                              const char *resources_key,
+                              char* const* resources_values) {
+  int exit_code = -1;
+  char *script_file_dest = NULL;
+  char *cred_file_dest = NULL;
+  char *exit_code_file = NULL;
+  char *docker_command_with_binary[PATH_MAX];
+  char *docker_wait_command[PATH_MAX];
+  char *docker_inspect_command[PATH_MAX];
+  char *docker_rm_command[PATH_MAX];
+  int container_file_source =-1;
+  int cred_file_source = -1;
+
+  char *docker_command = parse_docker_command_file(command_file);
+  char *docker_binary = get_value(DOCKER_BINARY_KEY);
+  if (docker_binary == NULL) {
+    docker_binary = "docker";
+  }
+  exit_code = create_script_paths(
+    work_dir, script_name, cred_file, &script_file_dest, &cred_file_dest,
+    &container_file_source, &cred_file_source);
+  if (exit_code != 0) {
+    fprintf(ERRORFILE, "Could not create script path\n");
+    fflush(ERRORFILE);
+    goto cleanup;
+  }
+  uid_t user_uid = geteuid();
+  gid_t user_gid = getegid();
+
+  exit_code = create_local_dirs(user, app_id, container_id,
+    work_dir, script_name, cred_file, local_dirs, log_dirs,
+    1, script_file_dest, cred_file_dest,
+    container_file_source, cred_file_source);
+  if (exit_code != 0) {
+    fprintf(ERRORFILE, "Could not create local files and directories %d %d\n", container_file_source, cred_file_source);
+    fflush(ERRORFILE);
+    goto cleanup;
+  }
+
+  exit_code_file = get_exit_code_file(pid_file);
+  if (NULL == exit_code_file) {
+    exit_code = OUT_OF_MEMORY;
+    fprintf(ERRORFILE, "Container out of memory");
+    fflush(ERRORFILE);
+    goto cleanup;
+  }
+
+  if (change_effective_user(0, user_gid) != 0) {
+    fprintf(ERRORFILE, "Could not change to effective users %d, %d\n", 0, user_gid);
+    fflush(ERRORFILE);
+    goto cleanup;
+  }
+
+  sprintf(docker_command_with_binary, "%s %s", docker_binary, docker_command);
+
+  FILE* start_docker = popen(docker_command_with_binary, "r");
+  if (pclose (start_docker) != 0)
+  {
+    fprintf (ERRORFILE,
+     "Could not invoke docker %s.\n", docker_command_with_binary);
+    fflush(ERRORFILE);
+    exit_code = UNABLE_TO_EXECUTE_CONTAINER_SCRIPT;
+    goto cleanup;
+  }
+
+  sprintf(docker_inspect_command,
+    "%s inspect --format {{.State.Pid}} %s",
+    docker_binary, container_id);
+
+  FILE* inspect_docker = popen(docker_inspect_command, "r");
+  int pid = 0;
+  fscanf (inspect_docker, "%d", &pid);
+  if (pclose (inspect_docker) != 0)
+  {
+    fprintf (ERRORFILE,
+     "Could not inspect docker %s.\n", docker_inspect_command);
+    fflush(ERRORFILE);
+    exit_code = UNABLE_TO_EXECUTE_CONTAINER_SCRIPT;
+    goto cleanup;
+  }
+
+  if (pid != 0) {
+    // cgroups-based resource enforcement
+    if (resources_key != NULL && ! strcmp(resources_key, "cgroups")) {
+     // write pid to cgroups
+     char* const* cgroup_ptr;
+     for (cgroup_ptr = resources_values; cgroup_ptr != NULL &&
+          *cgroup_ptr != NULL; ++cgroup_ptr) {
+       if (strcmp(*cgroup_ptr, "none") != 0 &&
+             write_pid_to_cgroup_as_root(*cgroup_ptr, pid) != 0) {
+         exit_code = WRITE_CGROUP_FAILED;
+         goto cleanup;
+       }
+     }
+    }
+    // write pid to pidfile
+    if (pid_file == NULL
+        || write_pid_to_file_as_nm(pid_file, (pid_t)pid) != 0) {
+      exit_code = WRITE_PIDFILE_FAILED;
+      fprintf(ERRORFILE, "Could not write pid to %s", pid_file);
+      fflush(ERRORFILE);
+      goto cleanup;
+    }
+
+    sprintf(docker_wait_command,
+      "%s wait %s", docker_binary, container_id);
+
+    FILE* wait_docker = popen(docker_wait_command, "r");
+    fscanf (wait_docker, "%d", &exit_code);
+    if (pclose (wait_docker) != 0) {
+      fprintf (ERRORFILE,
+       "Could not attach to docker is container dead? %s.\n", docker_wait_command);
+      fflush(ERRORFILE);
+    }
+  }
+
+  sprintf(docker_rm_command,
+    "%s rm %s", docker_binary, container_id);
+  FILE* rm_docker = popen(docker_rm_command, "w");
+  if (pclose (rm_docker) != 0)
+  {
+    fprintf (ERRORFILE,
+     "Could not remove container %s.\n", docker_rm_command);
+    fflush(ERRORFILE);
+    exit_code = UNABLE_TO_EXECUTE_CONTAINER_SCRIPT;
+    goto cleanup;
+  }
+
+cleanup:
+  if (exit_code_file != NULL && write_exit_code_file(exit_code_file, exit_code) < 0) {
+    fprintf (ERRORFILE,
+      "Could not write exit code to file %s.\n", exit_code_file);
+    fflush(ERRORFILE);
+  }
+#if HAVE_FCLOSEALL
+  fcloseall();
+#else
+  // only those fds are opened assuming no bug
+  fclose(LOGFILE);
+  fclose(ERRORFILE);
+  fclose(stdin);
+  fclose(stdout);
+  fclose(stderr);
+#endif
+  free(exit_code_file);
+  free(script_file_dest);
+  free(cred_file_dest);
+  return exit_code;
+}
+
+
+int launch_container_as_user(const char *user, const char *app_id,
                    const char *container_id, const char *work_dir,
                    const char *script_name, const char *cred_file,
                    const char* pid_file, char* const* local_dirs,
@@ -1037,31 +1364,21 @@ int launch_container_as_user(const char *user, const char *app_id,
   char *cred_file_dest = NULL;
   char *exit_code_file = NULL;
 
-  script_file_dest = get_container_launcher_file(work_dir);
+
-  if (script_file_dest == NULL) {
-    exit_code = OUT_OF_MEMORY;
-    goto cleanup;
-  }
-  cred_file_dest = get_container_credentials_file(work_dir);
-  if (NULL == cred_file_dest) {
-    exit_code = OUT_OF_MEMORY;
-    goto cleanup;
-  }
   exit_code_file = get_exit_code_file(pid_file);
   if (NULL == exit_code_file) {
     exit_code = OUT_OF_MEMORY;
     goto cleanup;
   }
 
-  // open launch script
+  int container_file_source =-1;
-  int container_file_source = open_file_as_nm(script_name);
+  int cred_file_source = -1;
-  if (container_file_source == -1) {
+  exit_code = create_script_paths(
-    goto cleanup;
+    work_dir, script_name, cred_file, &script_file_dest, &cred_file_dest,
-  }
+    &container_file_source, &cred_file_source);
-
+  if (exit_code != 0) {
-  // open credentials
+    fprintf(ERRORFILE, "Could not create local files and directories");
-  int cred_file_source = open_file_as_nm(cred_file);
+    fflush(ERRORFILE);
-  if (cred_file_source == -1) {
     goto cleanup;
   }
 
@@ -1088,7 +1405,6 @@ int launch_container_as_user(const char *user, const char *app_id,
 
   // cgroups-based resource enforcement
   if (resources_key != NULL && ! strcmp(resources_key, "cgroups")) {
-
     // write pid to cgroups
     char* const* cgroup_ptr;
     for (cgroup_ptr = resources_values; cgroup_ptr != NULL && 
@@ -1101,42 +1417,13 @@ int launch_container_as_user(const char *user, const char *app_id,
     }
   }
 
-  // create the user directory on all disks
+  exit_code = create_local_dirs(user, app_id, container_id,
-  int result = initialize_user(user, local_dirs);
+    work_dir, script_name, cred_file, local_dirs, log_dirs,
-  if (result != 0) {
+    0, script_file_dest, cred_file_dest,
-    return result;
+    container_file_source, cred_file_source);
-  }
+  if (exit_code != 0) {
-
+    fprintf(ERRORFILE, "Could not create local files and directories");
-  // initializing log dirs
+    fflush(ERRORFILE);
-  int log_create_result = create_log_dirs(app_id, log_dirs);
-  if (log_create_result != 0) {
-    return log_create_result;
-  }
-
-  // give up root privs
-  if (change_user(user_detail->pw_uid, user_detail->pw_gid) != 0) {
-    exit_code = SETUID_OPER_FAILED;
-    goto cleanup;
-  }
-
-  // Create container specific directories as user. If there are no resources
-  // to localize for this container, app-directories and log-directories are
-  // also created automatically as part of this call.
-  if (create_container_directories(user, app_id, container_id, local_dirs,
-                                   log_dirs, work_dir) != 0) {
-    fprintf(LOGFILE, "Could not create container dirs");
-    goto cleanup;
-  }
-
-
-  // 700
-  if (copy_file(container_file_source, script_name, script_file_dest,S_IRWXU) != 0) {
-    goto cleanup;
-  }
-
-  // 600
-  if (copy_file(cred_file_source, cred_file, cred_file_dest,
-        S_IRUSR | S_IWUSR) != 0) {
     goto cleanup;
   }
 
@@ -1151,24 +1438,20 @@ int launch_container_as_user(const char *user, const char *app_id,
   fclose(stderr);
 #endif
   umask(0027);
-  if (chdir(work_dir) != 0) {
+
-    fprintf(LOGFILE, "Can't change directory to %s -%s\n", work_dir,
-	    strerror(errno));
-    goto cleanup;
-  }
   if (execlp(script_file_dest, script_file_dest, NULL) != 0) {
-    fprintf(LOGFILE, "Couldn't execute the container launch file %s - %s", 
+    fprintf(LOGFILE, "Couldn't execute the container launch file %s - %s",
             script_file_dest, strerror(errno));
     exit_code = UNABLE_TO_EXECUTE_CONTAINER_SCRIPT;
     goto cleanup;
   }
   exit_code = 0;
 
- cleanup:
+  cleanup:
-  free(exit_code_file);
+    free(exit_code_file);
-  free(script_file_dest);
+    free(script_file_dest);
-  free(cred_file_dest);
+    free(cred_file_dest);
-  return exit_code;
+    return exit_code;
 }
 
 int signal_container_as_user(const char *user, int pid, int sig) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h

@@ -25,6 +25,7 @@ enum command {
   LAUNCH_CONTAINER = 1,
   SIGNAL_CONTAINER = 2,
   DELETE_AS_USER = 3,
+  LAUNCH_DOCKER_CONTAINER = 4
 };
 
 enum errorcodes {
@@ -55,7 +56,10 @@ enum errorcodes {
   SETSID_OPER_FAILED = 25,
   WRITE_PIDFILE_FAILED = 26,
   WRITE_CGROUP_FAILED = 27,
-  TRAFFIC_CONTROL_EXECUTION_FAILED = 28
+  TRAFFIC_CONTROL_EXECUTION_FAILED = 28,
+  DOCKER_RUN_FAILED=29,
+  ERROR_OPENING_FILE = 30,
+  ERROR_READING_FILE = 31
 };
 
 enum operations {
@@ -67,7 +71,9 @@ enum operations {
   RUN_AS_USER_INITIALIZE_CONTAINER = 6,
   RUN_AS_USER_LAUNCH_CONTAINER = 7,
   RUN_AS_USER_SIGNAL_CONTAINER = 8,
-  RUN_AS_USER_DELETE = 9
+  RUN_AS_USER_DELETE = 9,
+  RUN_AS_USER_LAUNCH_DOCKER_CONTAINER = 10,
+  RUN_DOCKER = 11
 };
 
 #define NM_GROUP_KEY "yarn.nodemanager.linux-container-executor.group"
@@ -79,6 +85,7 @@ enum operations {
 #define MIN_USERID_KEY "min.user.id"
 #define BANNED_USERS_KEY "banned.users"
 #define ALLOWED_SYSTEM_USERS_KEY "allowed.system.users"
+#define DOCKER_BINARY_KEY "docker.binary"
 #define TMP_DIR "tmp"
 
 extern struct passwd *user_detail;
@@ -109,6 +116,14 @@ int initialize_app(const char *user, const char *app_id,
                    const char *credentials, char* const* local_dirs,
                    char* const* log_dirs, char* const* args);
 
+int launch_docker_container_as_user(const char * user, const char *app_id,
+                              const char *container_id, const char *work_dir,
+                              const char *script_name, const char *cred_file,
+                              const char *pid_file, char* const* local_dirs,
+                              char* const* log_dirs,
+                              const char *command_file,const char *resources_key,
+                              char* const* resources_values);
+
 /*
  * Function used to launch a container as the provided user. It does the following :
  * 1) Creates container work dir and log dir to be accessible by the child
@@ -241,3 +256,9 @@ int traffic_control_read_state(char *command_file);
  * calling process.
  */
 int traffic_control_read_stats(char *command_file);
+
+
+/**
+ * Run a docker command passing the command file as an argument
+ */
+int run_docker(const char *command_file);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/main.c

@@ -49,16 +49,19 @@ static void display_usage(FILE *stream) {
       "       container-executor --tc-modify-state <command-file>\n" \
       "       container-executor --tc-read-state <command-file>\n" \
       "       container-executor --tc-read-stats <command-file>\n" \
+      "       container-executor --run-docker <command-file>\n" \
       "       container-executor <user> <yarn-user> <command> <command-args>\n"  \
       "       where command and command-args: \n" \
       "            initialize container:  %2d appid tokens nm-local-dirs nm-log-dirs cmd app...\n" \
       "            launch container:      %2d appid containerid workdir container-script " \
                               "tokens pidfile nm-local-dirs nm-log-dirs resources optional-tc-command-file\n" \
+      "            launch docker container:      %2d appid containerid workdir container-script " \
+                              "tokens pidfile nm-local-dirs nm-log-dirs docker-command-file resources optional-tc-command-file\n" \
       "            signal container:      %2d container-pid signal\n" \
       "            delete as user:        %2d relative-path\n" ;
 
 
-  fprintf(stream, usage_template, INITIALIZE_CONTAINER, LAUNCH_CONTAINER,
+  fprintf(stream, usage_template, INITIALIZE_CONTAINER, LAUNCH_CONTAINER, LAUNCH_DOCKER_CONTAINER,
           SIGNAL_CONTAINER, DELETE_AS_USER);
 }
 
@@ -160,6 +163,7 @@ static struct {
   const char *dir_to_be_deleted;
   int container_pid;
   int signal;
+  const char *docker_command_file;
 } cmd_input;
 
 static int validate_run_as_user_commands(int argc, char **argv, int *operation);
@@ -227,6 +231,16 @@ static int validate_arguments(int argc, char **argv , int *operation) {
     return 0;
   }
 
+  if (strcmp("--run-docker", argv[1]) == 0) {
+    if (argc != 3) {
+      display_usage(stdout);
+      return INVALID_ARGUMENT_NUMBER;
+    }
+    optind++;
+    cmd_input.docker_command_file = argv[optind++];
+    *operation = RUN_DOCKER;
+    return 0;
+  }
   /* Now we have to validate 'run as user' operations that don't use
     a 'long option' - we should fix this at some point. The validation/argument
     parsing here is extensive enough that it done in a separate function */
@@ -252,7 +266,9 @@ static int validate_run_as_user_commands(int argc, char **argv, int *operation)
   fprintf(LOGFILE, "main : run as user is %s\n", cmd_input.run_as_user_name);
   fprintf(LOGFILE, "main : requested yarn user is %s\n", cmd_input.yarn_user_name);
   fflush(LOGFILE);
-
+  char * resources = NULL;// key,value pair describing resources
+  char * resources_key = NULL;
+  char * resources_value = NULL;
   switch (command) {
   case INITIALIZE_CONTAINER:
     if (argc < 9) {
@@ -268,6 +284,46 @@ static int validate_run_as_user_commands(int argc, char **argv, int *operation)
 
     *operation = RUN_AS_USER_INITIALIZE_CONTAINER;
     return 0;
+ case LAUNCH_DOCKER_CONTAINER:
+    //kill me now.
+    if (!(argc == 14 || argc == 15)) {
+      fprintf(ERRORFILE, "Wrong number of arguments (%d vs 14 or 15) for launch docker container\n",
+       argc);
+      fflush(ERRORFILE);
+      return INVALID_ARGUMENT_NUMBER;
+    }
+
+    cmd_input.app_id = argv[optind++];
+    cmd_input.container_id = argv[optind++];
+    cmd_input.current_dir = argv[optind++];
+    cmd_input.script_file = argv[optind++];
+    cmd_input.cred_file = argv[optind++];
+    cmd_input.pid_file = argv[optind++];
+    cmd_input.local_dirs = argv[optind++];// good local dirs as a comma separated list
+    cmd_input.log_dirs = argv[optind++];// good log dirs as a comma separated list
+    cmd_input.docker_command_file = argv[optind++];
+    resources = argv[optind++];// key,value pair describing resources
+    resources_key = malloc(strlen(resources));
+    resources_value = malloc(strlen(resources));
+    if (get_kv_key(resources, resources_key, strlen(resources)) < 0 ||
+      get_kv_value(resources, resources_value, strlen(resources)) < 0) {
+      fprintf(ERRORFILE, "Invalid arguments for cgroups resources: %s",
+                         resources);
+      fflush(ERRORFILE);
+      free(resources_key);
+      free(resources_value);
+      return INVALID_ARGUMENT_NUMBER;
+    }
+    //network isolation through tc
+    if (argc == 15) {
+      cmd_input.traffic_control_command_file = argv[optind++];
+    }
+
+    cmd_input.resources_key = resources_key;
+    cmd_input.resources_value = resources_value;
+    cmd_input.resources_values = extract_values(resources_value);
+    *operation = RUN_AS_USER_LAUNCH_DOCKER_CONTAINER;
+    return 0;
 
   case LAUNCH_CONTAINER:
     //kill me now.
@@ -286,9 +342,9 @@ static int validate_run_as_user_commands(int argc, char **argv, int *operation)
     cmd_input.pid_file = argv[optind++];
     cmd_input.local_dirs = argv[optind++];// good local dirs as a comma separated list
     cmd_input.log_dirs = argv[optind++];// good log dirs as a comma separated list
-    char * resources = argv[optind++];// key,value pair describing resources
+    resources = argv[optind++];// key,value pair describing resources
-    char * resources_key = malloc(strlen(resources));
+    resources_key = malloc(strlen(resources));
-    char * resources_value = malloc(strlen(resources));
+    resources_value = malloc(strlen(resources));
 
     if (get_kv_key(resources, resources_key, strlen(resources)) < 0 ||
         get_kv_value(resources, resources_value, strlen(resources)) < 0) {
@@ -385,6 +441,9 @@ int main(int argc, char **argv) {
   case TRAFFIC_CONTROL_READ_STATS:
     exit_code = traffic_control_read_stats(cmd_input.traffic_control_command_file);
     break;
+  case RUN_DOCKER:
+    exit_code = run_docker(cmd_input.docker_command_file);
+    break;
   case RUN_AS_USER_INITIALIZE_CONTAINER:
     exit_code = set_user(cmd_input.run_as_user_name);
     if (exit_code != 0) {
@@ -398,6 +457,34 @@ int main(int argc, char **argv) {
                             extract_values(cmd_input.log_dirs),
                             argv + optind);
     break;
+  case RUN_AS_USER_LAUNCH_DOCKER_CONTAINER:
+     if (cmd_input.traffic_control_command_file != NULL) {
+        //apply tc rules before switching users and launching the container
+        exit_code = traffic_control_modify_state(cmd_input.traffic_control_command_file);
+        if( exit_code != 0) {
+          //failed to apply tc rules - break out before launching the container
+          break;
+        }
+      }
+
+      exit_code = set_user(cmd_input.run_as_user_name);
+      if (exit_code != 0) {
+        break;
+      }
+
+      exit_code = launch_docker_container_as_user(cmd_input.yarn_user_name,
+                      cmd_input.app_id,
+                      cmd_input.container_id,
+                      cmd_input.current_dir,
+                      cmd_input.script_file,
+                      cmd_input.cred_file,
+                      cmd_input.pid_file,
+                      extract_values(cmd_input.local_dirs),
+                      extract_values(cmd_input.log_dirs),
+                      cmd_input.docker_command_file,
+                      cmd_input.resources_key,
+                      cmd_input.resources_values);
+      break;
   case RUN_AS_USER_LAUNCH_CONTAINER:
     if (cmd_input.traffic_control_command_file != NULL) {
       //apply tc rules before switching users and launching the container