remote: Create the remote package

This commit moves functions related to remote execution, and to starting model runs, from `utils` to their own package `base/remote`.

It also contains a major refactor of the `start.model.runs` code that makes this code more modular and, hopefully, easier to maintain.

base/remote/.Rbuildignore

@@ -0,0 +1,2 @@
+^.*\.Rproj$
+^\.Rproj\.user$

base/remote/DESCRIPTION

@@ -0,0 +1,13 @@
+Package: PEcAn.remote
+Type: Package
+Title: PEcAn model execution utilities
+Version: 0.1.0
+Author: Alexey Shiklomanov, Rob Kooper, Shawn Serbin, David LeBauer
+Maintainer: Alexey Shiklomanov <ashiklom@bu.edu>
+Description: This package contains utilities for communicating with and executing code on local and remote hosts.
+    In particular, it has PEcAn-specific utilities for starting ecosystem model runs.
+License: FreeBSD + file LICENSE
+Encoding: UTF-8
+LazyData: true
+Roxygen: list(markdown = TRUE)
+RoxygenNote: 6.0.1

base/remote/LICENSE

@@ -0,0 +1,29 @@
+University of Illinois/NCSA Open Source License
+
+Copyright (c) 2012, University of Illinois, NCSA.  All rights reserved.
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the 
+"Software"), to deal with the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+- Redistributions of source code must retain the above copyright
+  notice, this list of conditions and the following disclaimers.
+- Redistributions in binary form must reproduce the above copyright
+  notice, this list of conditions and the following disclaimers in the
+  documentation and/or other materials provided with the distribution.
+- Neither the names of University of Illinois, NCSA, nor the names
+  of its contributors may be used to endorse or promote products
+  derived from this Software without specific prior written permission.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE CONTRIBUTORS OR COPYRIGHT HOLDERS BE LIABLE FOR
+ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF 
+CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS WITH THE SOFTWARE.
+

base/remote/NAMESPACE

@@ -0,0 +1,21 @@
+# Generated by roxygen2: do not edit by hand
+
+export(check_model_run)
+export(is.localhost)
+export(kill.tunnel)
+export(open_tunnel)
+export(qsub_get_jobid)
+export(qsub_run_finished)
+export(remote.copy.from)
+export(remote.copy.to)
+export(remote.copy.update)
+export(remote.execute.R)
+export(remote.execute.cmd)
+export(runModule.start.model.runs)
+export(setup_modellauncher)
+export(stamp_finished)
+export(stamp_started)
+export(start.model.runs)
+export(start_qsub)
+export(start_serial)
+export(test_remote)

base/remote/R/check_model_run.R

@@ -0,0 +1,17 @@
+#' Check if model run was successful
+#'
+#' @param out Output from model execution, as a character.
+#' @inheritParams start.model.runs
+#'
+#' @return
+#' @export
+check_model_run <- function(out, stop.on.error = TRUE) {
+  if ("ERROR IN MODEL RUN" %in% out) {
+    msg <- paste0("Model run aborted with the following error:\n", out)
+    if (stop.on.error) {
+      PEcAn.logger::logger.severe(msg)
+    } else {
+      PEcAn.logger::logger.error(msg)
+    }
+  }
+}

base/remote/R/check_qsub_status.R

@@ -0,0 +1,32 @@
+#' Check if qsub run finished
+#'
+#' @param run run ID, as an integer
+#' @param qstat (string) qstat command for checking job status
+#' @inheritParams remote.execute.cmd
+#'
+#' @return `TRUE` if run is marked as DONE, otherwise FALSE.
+#' @export
+qsub_run_finished <- function(run, host, qstat) {
+  if (is.na(run)) {
+    PEcAn.logger::logger.warn("Job", run, "encountered an error during submission.",
+                              "NOTE that the job will be stamped as 'finished' in BETY.")
+    return(FALSE)
+  }
+  run_id_string <- format(run, scientific = FALSE)
+  check <- gsub("@JOBID", run, qstat)
+  cmd_list <- strsplit(check, " (?=([^\"']*\"[^\"']*\")*[^\"']*$)", perl = TRUE)
+  cmd <- cmd_list[[1]]
+  args <- cmd_list[-1]
+  if (is.localhost(host)) {
+    out <- system2(cmd, args, stdout = TRUE, stderr = TRUE)
+  } else {
+    out <- remote.execute.cmd(host = host, cmd = cmd, args = args, stderr = TRUE)
+  }
+
+  if (length(out) > 0 && substring(out, nchar(out) - 3) == "DONE") {
+    PEcAn.logger::logger.debug("Job", run, "for run", run_id_string, "finished")
+    return(TRUE)
+  } else {
+    return(FALSE)
+  }
+}

base/remote/R/is.localhost.R

@@ -0,0 +1,21 @@
+#' Check if host is local
+#'
+#' Given the hostname is this the localhost. This returns true if either
+#' the value is localhost, or the value is the same as the fqdn.
+#'
+#' @title Check if local host
+#' @param host the hostname to be checked
+#' @return true if the host is the local host name
+#' @author Rob Kooper
+#' @export
+#' @examples
+#' is.localhost(fqdn())
+is.localhost <- function(host) {
+  if (is.character(host)) {
+    return((host == "localhost") || (host == PEcAn.utils::fqdn()))
+  } else if (is.list(host)) {
+    return((host$name == "localhost") || (host$name == PEcAn.utils::fqdn()))
+  } else {
+    return(FALSE)
+  }
+} # is.localhost

base/utils/R/open.tunnel.R → base/remote/R/open.tunnel.R

@@ -1,63 +1,65 @@
-#' Title
+#' Open an SSH tunnel
 #'
 #' @param remote_host name of remote server to connect to (e.g. geo.bu.edu)
 #' @param tunnel_dir  directory to store tunnel file in, typically from settings$host
 #' @param user        username on remote_host
 #' @param password    password on remote_host
 #' @param wait.time   how long to give system to connect before deleting password (seconds)
+#' @param tunnel_script Path to sshtunnel.sh script file for opening tunnel
 #'
 #' @return
 #' @export
 #'
 #' @examples
-open_tunnel <- function(remote_host,user=NULL,password=NULL,tunnel_dir = "~/.pecan/tunnel/",wait.time=15){
-
+open_tunnel <- function(remote_host, user = NULL, password = NULL, tunnel_dir = "~/.pecan/tunnel/",
+                        wait.time = 15, tunnel_script = '~/pecan/web/sshtunnel.sh'){
+
   ## make sure local tunnel directory exists
   dir.create(tunnel_dir)
-  
+
   ## get username if not provided
   if(is.null(user)){
     user <- readline("Username:: ")
   }
-  
+
   ## get password if not provided
   if(is.null(password)){
     password <- getPass::getPass()
   }
-  
-  sshTunnel   <- file.path(tunnel_dir,"tunnel")
-  sshPID      <- file.path(tunnel_dir,"pid")
-  sshPassFile <- file.path(tunnel_dir,"password")
-  
+
+  sshTunnel   <- file.path(tunnel_dir, "tunnel")
+  sshPID      <- file.path(tunnel_dir, "pid")
+  sshPassFile <- file.path(tunnel_dir, "password")
+
   if(file.exists(sshTunnel)){
     PEcAn.logger::logger.warn("Tunnel already exists. If tunnel is not working try calling kill.tunnel then reopen")
     return(TRUE)
   }
-  
+
   ## write password to temporary file
   PEcAn.logger::logger.warn(sshPassFile)
-  write(password,file = sshPassFile)
+  write(password, file = sshPassFile)
 
 #  start <- system(paste0("ssh -nN -o ControlMaster=yes -o ControlPath=",sshTunnel," -l ",user," ",remote_host),wait = FALSE,input = password)
 #  Sys.sleep(5)
 #  end <- system2("send",password)
-  
-  stat <- system(paste("~/pecan/web/sshtunnel.sh",remote_host,user,tunnel_dir),wait=FALSE)
-  
+
+  stat <- system(paste(tunnel_script, remote_host, user, tunnel_dir), wait=FALSE)
+
   ##wait for tunnel to connect
   Sys.sleep(wait.time)
-  
-  if(file.exists(sshPassFile)){
+
+  if (file.exists(sshPassFile)) {
     file.remove(sshPassFile)
     PEcAn.logger::logger.error("Tunnel open failed")
     return(FALSE)
-  }  
-  
-  if(file.exists(sshPID)){
-    pid <- readLines(sshPID,n = -1)
+  }
+
+  if (file.exists(sshPID)) {
+    pid <- readLines(sshPID, n = -1)
     return(as.numeric(pid))
   } else {
     return(TRUE)
   }
-  
+
 }

base/remote/R/qsub_get_jobid.R

@@ -0,0 +1,24 @@
+#' Get Job ID from qsub output
+#'
+#' @inheritParams check_model_run
+#' @inheritParams start.model.runs
+#' @param qsub.jobid (character) Regular expression string for extracting job ID from qsub output.
+#' Usually from `settings$host$qsub.jobid`
+#'
+#' @return
+#' @export
+qsub_get_jobid <- function(out, qsub.jobid, stop.on.error) {
+  qsub_worked <- grepl(qsub.jobid, out)
+  if (!qsub_worked) {
+    msg <- paste0("Job ID not assigned by qsub. The following qsub output may be relevant:\n", out)
+    if (stop.on.error) {
+      PEcAn.logger::logger.severe(msg)
+    } else {
+      PEcAn.logger::logger.error(msg)
+    }
+    jobid <- NA
+  } else {
+    jobid <- sub(qsub.jobid, '\\1', out)
+  }
+  return(jobid)
+}

base/remote/R/remote.copy.from.R

@@ -0,0 +1,48 @@
+#' Copy file/dir from remote server to local server
+#'
+#' Copies the file/dir from the remote server to the local server. If the dst
+#' is a folder it will copy the file into that folder.
+#'
+#' @title Copy file from remote to local
+#' @param host list with server, user and optionally tunnel to use.
+#' @param src remote file/dir to copy
+#' @param dst local file/dir to copy to
+#' @param delete in case of local dir should all non-existent files be removed
+#' @param stderr should stderr be returned
+#' @return output of command executed
+#'
+#' @author Rob Kooper
+#' @export
+#' @examples
+#' \dontrun{
+#'   host <- list(name='geo.bu.edu', user='kooper', tunnel='/tmp/geo.tunnel')
+#'   remote.copy.from(host, '/tmp/kooper', '/tmp/geo.tmp', delete=TRUE)
+#' }
+remote.copy.from <- function(host, src, dst, delete = FALSE, stderr = FALSE) {
+  args <- c("-az", "-q")
+  if (as.logical(delete)) {
+    args <- c(args, "--delete")
+  }
+  if (is.localhost(host)) {
+    args <- c(args, src, dst)
+  } else {
+    tunnel <- host$tunnel
+    if(!is.null(host$data_tunnel)) tunnel <- host$data_tunnel
+    hostname <- host$name
+    if(!is.null(host$data_hostname)) hostname <- host$data_hostname
+    if (!is.null(tunnel)) {
+      if (!file.exists(tunnel)) {
+        PEcAn.logger::logger.severe("Could not find tunnel", tunnel)
+      }
+      args <- c(args, "-e", paste0("ssh -o ControlPath=\"", tunnel, "\"",
+                                   collapse = ""))
+      args <- c(args, paste0(hostname, ":", src), dst)
+    } else if (!is.null(host$user)) {
+      args <- c(args, paste0(host$user, "@", hostname, ":", src), dst)
+    } else {
+      args <- c(args, paste0(hostname, ":", src), dst)
+    }
+  }
+  PEcAn.logger::logger.debug("rsync", shQuote(args))
+  system2("rsync", shQuote(args), stdout = TRUE, stderr = as.logical(stderr))
+} # remote.copy.from

base/remote/R/remote.copy.to.R

@@ -0,0 +1,50 @@
+#' Copy file/dir to remote server from local server
+#'
+#' Copies the file/dir to the remote server from the local server. If the dst
+#' is a folder it will copy the file into that folder.
+#'
+#' @inheritParams remote.execute.cmd
+#' @param src local file/dir to copy
+#' @param dst remote file/dir to copy to
+#' @param delete in case of local dir should all non-existent files be removed
+#' @return output of command executed
+#'
+#' @author Rob Kooper
+#' @export
+#' @examples
+#' \dontrun{
+#'   host <- list(name='geo.bu.edu', user='kooper', tunnel='/tmp/geo.tunnel')
+#'   remote.copy.to(host, '/tmp/kooper', '/tmp/kooper', delete=TRUE)
+#' }
+remote.copy.to <- function(host, src, dst, delete = FALSE, stderr = FALSE) {
+  args <- c("-a", "-q")
+  if (as.logical(delete)) {
+    args <- c(args, "--delete")
+  }
+  if (is.localhost(host)) {
+    args <- c(args, src, dst)
+  } else {
+    tunnel <- host$tunnel
+    if (!is.null(host$data_tunnel)) {
+      tunnel <- host$data_tunnel
+    }
+    hostname <- host$name
+    if (!is.null(host$data_hostname)) {
+      hostname <- host$data_hostname
+    }
+    if (!is.null(tunnel)) {
+      if (!file.exists(tunnel)) {
+        PEcAn.logger::logger.severe("Could not find tunnel", tunnel)
+      }
+      args <- c(args, "-e", paste0("ssh -o ControlPath=\"", tunnel, "\"",
+                                   collapse = ""))
+      args <- c(args, src, paste0(hostname, ":", dst))
+    } else if (!is.null(host$user)) {
+      args <- c(args, src, paste0(host$user, "@", hostname, ":", dst))
+    } else {
+      args <- c(args, src, paste0(hostname, ":", dst))
+    }
+  }
+  PEcAn.logger::logger.debug("rsync", shQuote(args))
+  system2("rsync", shQuote(args), stdout = TRUE, stderr = as.logical(stderr))
+} # remote.copy.to

base/remote/R/remote.copy.update.R

@@ -0,0 +1,41 @@
+#' Copy to remote and update DB
+#' @param input_id
+#' @param remote_dir remote folder path
+#' @param remote_file_name remote file name, no need to provide if it's the same as local
+#' @param host as in settings$host
+#' @param con
+#' @param stderr should stderr be returned
+#' @return remote_id remote dbfile record
+#'
+#' @author Istem Fer
+#' @export
+remote.copy.update <- function(input_id, remote_dir, remote_file_name = NULL, host, con){
+
+  remote.execute.cmd(host, "mkdir", c("-p", remote_dir))
+
+  local_file_record <- db.query(paste("SELECT * from dbfiles where container_id =", input_id), con)
+
+  if(is.null(remote_file_name)){
+    local_file_name <- local_file_record$file_name
+    if(length(local_file_name) > 1){
+      PEcAn.logger::logger.warn(paste0("Multiple file names found in the DB and no remote file name provided. Using the first file name for remote file name: ",
+                         local_file_record$file_name[1]))
+      local_file_name <- local_file_record$file_name[1]
+    }
+    remote_file_name <- local_file_name
+  }
+
+  local_file_path  <- file.path(local_file_record$file_path, local_file_record$file_name)
+  remote_file_path <- file.path(remote_dir, remote_file_name)
+
+  remote.copy.to(host, local_file_path, remote_file_path)
+
+  # update DB record
+  remote_id <- dbfile.insert(in.path = remote_dir, in.prefix = remote_file_name,
+                type = local_file_record$container_type, id = local_file_record$container_id,
+                con = con, hostname = host$name)
+
+
+  return(remote_id)
+
+} # remote.copy.update