merge from master to feature/longrunjob

Author: zfengms

.lintr

@@ -1 +1 @@
-exclusions: list("R/validators.R")
+exclusions: list("R/validationUtilities.R")

CHANGELOG.md

@@ -1,12 +1,26 @@
 # Change Log
-## [0.5.2] 2017-10-30
+## [0.6.1] 2017-11-10
 ### Added
 - Support for users to delete job and terminate job
 ### Changed
 - Add retry to get job result
 - Add errorHandling and wait option to job metadata
 - Save job metadata to job result storage blob
 
+## [0.6.0] 2017-11-03
+### Added
+- Support for users to run custom versions of R via Docker containers
+- GitHub and BioConductor support as parameters in the foreach
+
+### Changed
+- [BREAKING CHANGE] Host OS distribution is now Debian instead of CentOS
+- [BREAKING CHANGE] Command line no longer updates the environment of R
+- [BREAKING CHANGE] Default version of R changed from MRO 3.3.2 to latest version of CRAN R
+
+### Fixed
+- Packages installed in foreach are only present and visible to a single foreach loop and then deleted from the cluster
+- Linux clients would get stuck waiting for the job to finish when using the .packages() option in the foreach loop
+
 ## [0.5.1] 2017-09-28
 ### Added
 - Support for users to get job and job results for long running job

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: doAzureParallel
 Type: Package
 Title: doAzureParallel
-Version: 0.5.0
+Version: 0.6.1
 Author: Brian Hoang
 Maintainer: Brian Hoang <brhoan@microsoft.com>
 Description: The project is for data experts who use R at scale. The project
@@ -17,7 +17,7 @@ Depends:
     foreach (>= 1.4.3),
     iterators (>= 1.0.8)
 Imports:
-    rAzureBatch (>= 0.5.1),
+    rAzureBatch (>= 0.5.3),
     jsonlite,
     rjson,
     xml2,
@@ -27,5 +27,5 @@ Suggests:
     caret,
     plyr,
     lintr
-Remotes: Azure/rAzureBatch@v0.5.1
+Remotes: Azure/rAzureBatch@v0.5.3
 RoxygenNote: 6.0.1

R/cluster.R

@@ -86,9 +86,11 @@ generateClusterConfig <- function(fileName) {
                                 max = 3),
         autoscaleFormula = "QUEUE"
       ),
+      containerImage = "rocker/tidyverse:latest",
       rPackages = list(
         cran = vector(),
         github = vector(),
+        bioconductor = vector(),
         githubAuthenticationToken = ""
       ),
       commandLine = vector()
@@ -143,6 +145,7 @@ makeCluster <-
 
     installCranCommand <- NULL
     installGithubCommand <- NULL
+    installBioconductorCommand <- NULL
 
     if (!is.null(poolConfig$rPackages) &&
         !is.null(poolConfig$rPackages$cran) &&
@@ -158,21 +161,62 @@ makeCluster <-
         getPoolPackageInstallationCommand("github", poolConfig$rPackages$github)
     }
 
-    packages <- NULL
-    if (!is.null(installCranCommand)) {
-      packages <- installCranCommand
+    if (!is.null(poolConfig$rPackages) &&
+        !is.null(poolConfig$rPackages$bioconductor) &&
+        length(poolConfig$rPackages$bioconductor) > 0) {
+      installBioconductorCommand <-
+        getPoolPackageInstallationCommand("bioconductor", poolConfig$rPackages$bioconductor)
     }
 
-    if (!is.null(installGithubCommand) && is.null(packages)) {
-      packages <- installGithubCommand
+    packages <- c()
+    if (!is.null(installCranCommand)) {
+      packages <- c(installCranCommand, packages)
+    }
+    if (!is.null(installGithubCommand)) {
+      packages <- c(installGithubCommand, packages)
+    }
+    if (!is.null(installBioconductorCommand)) {
+      packages <- c(installBioconductorCommand, packages)
     }
-    else if (!is.null(installGithubCommand) && !is.null(packages)) {
-      packages <- c(installCranCommand, installGithubCommand)
+
+    if (length(packages) == 0) {
+      packages <- NULL
     }
 
     commandLine <- NULL
+
+    # install docker and create docker container
+    dockerImage <- "rocker/tidyverse:latest"
+    if (!is.null(poolConfig$containerImage)) {
+      dockerImage <- poolConfig$containerImage
+    }
+
+    config$containerImage <- dockerImage
+    installAndStartContainerCommand <- paste("cluster_setup.sh",
+                                             dockerImage,
+                                             sep = " ")
+
+    containerInstallCommand <- c(
+      paste0(
+        "wget https://raw.githubusercontent.com/Azure/doAzureParallel/",
+        "master/inst/startup/cluster_setup.sh"),
+      "chmod u+x cluster_setup.sh",
+      paste0(
+        "wget https://raw.githubusercontent.com/Azure/doAzureParallel/",
+        "master/inst/startup/install_bioconductor.R"),
+      "chmod u+x install_bioconductor.R",
+      installAndStartContainerCommand
+    )
+
     if (!is.null(poolConfig$commandLine)) {
-      commandLine <- poolConfig$commandLine
+      commandLine <- c(containerInstallCommand, poolConfig$commandLine)
+    }
+
+    if (!is.null(packages)) {
+      # install packages
+      commandLine <-
+        c(commandLine,
+          dockerRunCommand(dockerImage, packages, NULL, FALSE, FALSE))
     }
 
     environmentSettings <- NULL
@@ -189,17 +233,17 @@ makeCluster <-
     }
 
     if (!is.null(poolConfig[["pool"]])) {
-      validateDeprecatedClusterConfig(clusterSetting)
+      validation$isValidDeprecatedClusterConfig(clusterSetting)
       poolConfig <- poolConfig[["pool"]]
     }
     else {
-      validateClusterConfig(clusterSetting)
+      validation$isValidClusterConfig(clusterSetting)
     }
 
     tryCatch({
-      `Validators`$isValidPoolName(poolConfig$name)
+      validation$isValidPoolName(poolConfig$name)
     },
-    error = function(e){
+    error = function(e) {
       stop(paste("Invalid pool name: \n",
                  e))
     })
@@ -219,19 +263,19 @@ makeCluster <-
     if (grepl("PoolBeingDeleted", response)) {
       pool <- rAzureBatch::getPool(poolConfig$name)
 
-      cat(
-        sprintf(
-          paste("Cluster '%s' already exists and is being deleted.",
-                "Another cluster with the same name cannot be created",
-                "until it is deleted. Please wait for the cluster to be deleted",
-                "or create one with a different name"),
-          poolConfig$name
+      cat(sprintf(
+        paste(
+          "Cluster '%s' already exists and is being deleted.",
+          "Another cluster with the same name cannot be created",
+          "until it is deleted. Please wait for the cluster to be deleted",
+          "or create one with a different name"
         ),
-        fill = TRUE
-      )
+        poolConfig$name
+      ),
+      fill = TRUE)
 
       while (areShallowEqual(rAzureBatch::getPool(poolConfig$name)$state,
-                      "deleting")) {
+                             "deleting")) {
         cat(".")
         Sys.sleep(10)
       }

R/commandLineUtilities.R

@@ -0,0 +1,126 @@
+getJobPackageInstallationCommand <- function(type, packages) {
+  script <- ""
+  if (type == "cran") {
+    script <- "Rscript $AZ_BATCH_JOB_PREP_WORKING_DIR/install_cran.R"
+  }
+  else if (type == "github") {
+    script <- "Rscript $AZ_BATCH_JOB_PREP_WORKING_DIR/install_github.R"
+  }
+  else if (type == "bioconductor") {
+    script <-
+      "Rscript $AZ_BATCH_JOB_PREP_WORKING_DIR/install_bioconductor.R"
+  }
+  else {
+    stop("Using an incorrect package source")
+  }
+
+  if (!is.null(packages) && length(packages) > 0) {
+    packageCommands <- paste0(packages, collapse = " ")
+    script <- paste0(script, " ", packageCommands)
+  }
+}
+
+getPoolPackageInstallationCommand <- function(type, packages) {
+  poolInstallationCommand <- character(length(packages))
+
+  sharedPackagesDirectory <- "/mnt/batch/tasks/shared/R/packages"
+
+  libPathsCommand <- paste0('\'.libPaths( c( \\\"',
+                            sharedPackagesDirectory,
+                            '\\\", .libPaths()));')
+
+  installCommand <-
+    paste("Rscript -e \'args <- commandArgs(TRUE)\'",
+          "-e \'options(warn=2)\'")
+
+  # At this point we cannot use install_cran.R and install_github.R because they are not yet available.
+  if (type == "cran") {
+    script <-
+      paste(installCommand,
+            paste("-e",
+                  libPathsCommand,
+                  "install.packages(args[1])\' %s")
+            )
+  }
+  else if (type == "github") {
+    script <-
+      paste(
+        installCommand,
+        paste(
+          "-e",
+          libPathsCommand,
+          "devtools::install_github(args[1])\' %s"
+        )
+      )
+  }
+  else if (type == "bioconductor") {
+    script <- "Rscript /mnt/batch/tasks/startup/wd/install_bioconductor.R %s"
+  }
+  else {
+    stop("Using an incorrect package source")
+  }
+
+  for (i in 1:length(packages)) {
+    poolInstallationCommand[i] <- sprintf(script, packages[i])
+  }
+
+  poolInstallationCommand
+}
+
+dockerRunCommand <-
+  function(containerImage,
+           command,
+           containerName = NULL,
+           runAsDaemon = FALSE,
+           includeEnvironmentVariables = TRUE) {
+    dockerOptions <- paste(
+      "--rm",
+      "-v $AZ_BATCH_NODE_ROOT_DIR:$AZ_BATCH_NODE_ROOT_DIR",
+      "-e AZ_BATCH_NODE_ROOT_DIR=$AZ_BATCH_NODE_ROOT_DIR",
+      "-e AZ_BATCH_NODE_STARTUP_DIR=$AZ_BATCH_NODE_STARTUP_DIR",
+      sep = " "
+    )
+
+    if (runAsDaemon) {
+      dockerOptions <- paste(dockerOptions, "-d", dockerOptions, sep = " ")
+    }
+
+    if (!is.null(containerName)) {
+      dockerOptions <-
+        paste(dockerOptions, "--name", containerName, dockerOptions, sep = " ")
+    }
+
+    if (includeEnvironmentVariables) {
+      dockerOptions <-
+        paste(
+          dockerOptions,
+          "-e AZ_BATCH_TASK_ID=$AZ_BATCH_TASK_ID",
+          "-e AZ_BATCH_JOB_ID=$AZ_BATCH_JOB_ID",
+          "-e AZ_BATCH_TASK_WORKING_DIR=$AZ_BATCH_TASK_WORKING_DIR",
+          "-e AZ_BATCH_JOB_PREP_WORKING_DIR=$AZ_BATCH_JOB_PREP_WORKING_DIR",
+          "-e BLOBXFER_SASKEY=$BLOBXFER_SASKEY",
+          sep = " "
+        )
+    }
+
+    dockerRunCommand <-
+      paste("docker run", dockerOptions, containerImage, command, sep = " ")
+    dockerRunCommand
+  }
+
+linuxWrapCommands <- function(commands = c()) {
+  # Sanitize the vector and don't allow empty values
+  cleanCommands <- commands[lapply(commands, length) > 0]
+
+  commandLine <- ""
+  if (length(cleanCommands) > 0) {
+    # Do not allow absolute paths is enforced in lintr
+    commandLine <-
+      sprintf("/bin/bash -c \"set -e; set -o pipefail; %s wait\"",
+              paste0(paste(
+                cleanCommands, sep = " ", collapse = "; "
+              ), ";"))
+  }
+
+  commandLine
+}