Implementation of a print friendly method for selected `aws.ec2` functions

chk_init.R

chk_init <- function(is_initialized, i, check = c("both", "running")){
  while(!all(is_initialized)) {

    for(ii in seq_along(i)) {

      # Current instance
      i_ii <- instance_status(i[[ii]])

      # Initially, we don't get any information
      if(length(i_ii) == 0) {
        next()
      }

      # First check if we are at least running
      if(!is_running[ii]) {
        if(unlist(i_ii$item$instanceState$name) == "running") {
          is_running[ii] <- TRUE
          message("Instance ", ii, " is running. Now initialzing.")
          if(check[1] == "running")
             break
        }
      }

      # Then check if we are initialized
      if(!is_initialized[ii]) {
        if(unlist(i_ii$item$instanceStatus$status) == "ok") {
          is_initialized[ii] <- TRUE
          message("Instance ", ii, " is initialized.")
        }
      }

    }
  }
}

connect_aws_cl.R

connect_aws_cl <- function(workers_num = 1, public_ip, ssh_private_key_file, install = c('future', 'purrr', 'furrr', 'doParallel')){
  
  if(!require(glue)) install.packages("glue")
  library(glue)
  
  if(!missing(install)){
    ## Set up .libPaths() for the 'ubuntu' user and
    ## install packages
    r_args <- c(
      "-e", shQuote("local({p <- Sys.getenv('R_LIBS_USER'); dir.create(p, recursive = TRUE, showWarnings = FALSE); .libPaths(p)})"),
      "-e", gsub("\\n", "", shQuote(glue("list.of.packages <- c({paste0(\"'\", {install},\"'\", collapse = \",\")});
                          new.packages <- list.of.packages[!(list.of.packages %in% installed.packages()[,'Package'])];
                          if(length(new.packages)) install.packages(new.packages)")))
      #"-e", shQuote(glue("install.packages(c({paste0(\"'\", install,\"'\", collapse = \",\")}))"))
    )
  }
  
  makeClusterPSOCK(
    # Number of workers 
    workers = rep(public_ip, workers_num),
    
    makeNode = makeNodePSOCK,
    
    ## User name (always 'ubuntu')
    user = "ubuntu",
    
    ## Use private SSH key registered with AWS
    rshopts = c(
      "-o", "StrictHostKeyChecking=no",
      "-o", "IdentitiesOnly=yes",
      "-i", ssh_private_key_file
    ),
    rscript_args = r_args,
    dryrun = FALSE,
    verbose = T
  )
}

get_ip.R

get_ip <- function(i){
  vapply(
    i,
    function(i_ii) {
      i_di <- describe_instances(i_ii)
      i_di[[1]]$instancesSet[[1]]$networkInterfaceSet$privateIpAddressesSet$association$publicIp
    },
    FUN.VALUE = character(1)
  )
}

run_instance_spot.R

run_instances_spot <- 
  function(
    Action = "RunInstances",
    image,
    type,
    min = 1,
    max = min,
    keypair = NULL,
    subnet = NULL,
    sgroup = NULL,
    userdata = NULL,
    shutdown = c("stop", "terminate"),
    token = NULL,
    tags = NULL,
    spot_options = NULL,
    query_extra = list(),
    launch_template = NULL,
    dryrun = F,
    ...
  ) {
    
    query <- list(Action = Action, 
                  ImageId = aws.ec2:::get_imageid(image),
                  InstanceType = type,
                  MinCount = min,
                  MaxCount = max)
    if (!is.null(keypair)) {
      query$KeyName <- aws.ec2:::get_keypairname(keypair)
    }
    if (!is.null(subnet)) {
      query$SubnetId <- aws.ec2:::get_subnetid(subnet)
    }
    if (!is.null(sgroup)) {
      if (inherits(sgroup, "ec2_security_group")) {
        sgroup <- list(aws.ec2:::get_sgid(sgroup))
      } else if (is.character(sgroup)) {
        sgroup <- as.list(aws.ec2:::get_sgid(sgroup))
      } else {
        sgroup <- lapply(sgroup, aws.ec2:::get_sgid)
      }
      names(sgroup) <- paste0("SecurityGroupId.", seq_along(sgroup))
      query <- c(query, sgroup)
    }
    if (!is.null(userdata)) {
      query$UserData <- base64enc::base64encode(userdata)
    }
    if (!is.null(shutdown)) {
      query$InstanceInitiatedShutdownBehavior <- match.arg(shutdown)
    }
    if (!is.null(token)) {
      query$ClientToken <- token
    }
    if (!is.null(spot_options)) {
      query$InstanceMarketOptions.MarketType <- "spot"
      if (length(spot_options)) {
        names(spot_options) <- paste0("InstanceMarketOptions.SpotOptions.",names(spot_options))
        query <- c(query, spot_options) 
      }
    }
    if (!is.null(launch_template)) {
      names(launch_template) <- "LaunchTemplate.LaunchTemplateId"
      query <- c(query, launch_template)
    }
    if (!is.null(tags)) {
      tags <- .tag_specification(resource_type = "instance", tags)
      query <- c(query, tags)
    }
    query <- c(query, query_extra)
    browser()
    r <- ec2HTTP(query = query, ...)
    if(dryrun){
      print(query)
      print(r)
      return(NULL)
    }
    return(lapply(r$instancesSet, `class<-`, "ec2_instance"))
  }

print.ec2_instance <- function(x, ...) {
  cat("instanceId    : ", x$instanceId[[1]], "\n")
  cat("imageId       : ", x$imageId[[1]], "\n")
  cat("instanceType  : ", x$instanceType[[1]], "\n")
  cat("instanceState : ", paste0(x$instanceState$name[[1]], " (", x$instanceState$code[[1]], ")"), "\n")
  cat("subnetId      : ", x$subnetId[[1]], "\n")
  cat("vpcId         : ", x$vpcId[[1]], "\n")
  # cat("ipAddress: ", x$ipAddress, "\n")
  invisible(x)
}

tidy_describe.R

tidy_describe <- function(df, important = TRUE, pretty_print = T){
  df_class<- class(df[[1]])
  
  if(!grepl("ec2", df_class)) stop("Object not of class ec2.*. See aws.ec2 for details")
  
  to_df <- function(df){
    df[[1]] %>% 
      unlist %>% 
      data.frame(Field = names(.), Value = .) %>% 
      tbl_df
    }
  
  if(df_class == "ec2_image"){
    x <- df %>% to_df 
    if(important)
      x <- x %>% filter(Field %in% c("name", "imageId", "imageOwnerId", "creationDate", "description"))
  }
  
  ### Change to have a pretty print option
  if(df_class == "ec2_security_group"){
    x <- df %>% to_df 
    if(important){
      x_ports <- df[[1]]$ipPermissions %>%  
        map(~.x %>% unlist %>% t %>% data.frame(.)) %>% 
        reduce(plyr::rbind.fill)
    
      x <- cbind(
        x %>% 
          filter(grepl("groupName|Owner|groupDescription", Field)) %>% spread(Field, Value), 
        x_ports
      )
    }
    return(x)
  }

  
  if(df_class == "ec2_reservation"){
    x <- df %>% to_df
    if(important)
      x <- x %>% filter(Field %in% c("instancesSet.instanceType", "instancesSet.cpuOptions.coreCount", "instancesSet.cpuOptions.threadsPerCore", "instancesSet.instanceId", "instancesSet.imageId", "ownerId", "instancesSet.launchTime", "instancesSet.ipAddress", "instancesSet.privateIpAddress", "instancesSet.placement.availabilityZone")) 
  }
  
  if(df_class == "ec2_instance_status"){
    x <- df %>% to_df
    if(important)
      x <- x %>% filter(Field %in% c("instanceId", "availabilityZone", "instanceState.code", "instanceState.name")) 
  }
  x <- x %>% 
    mutate(Field = basename(gsub("\\.", "/", Field)))
  
  if(pretty_print){
    cat("--------------------------------------\n")
    cat("                Summary               \n")
    cat("--------------------------------------\n")
    for(i in 1:nrow(x)){
      y <- x %>% data.frame()
        cat(y[i, 'Field'], ":", y[i, 'Value'], "\n")
    }
    
    return(message("\nTo only return tibble: pretty_print = FALSE"))
  }
  x <- x %>% 
    tidyr::spread(., Field, Value) %>% 
    select(matches("instanceId"), everything())
  return(x)
}

All of this based on:
https://gist.github.com/DavisVaughan/5aac4a2757c0947a499d25d28a8ca89b