<commit_before># Helper functions for running benchmarks

#globals
BENCH_CURRENT_NUM_RUN="0"
BENCH_BIN_PATH="\$HOME/share/sw/bin"

# Outputs a list of defined benchmark suites separated by spaces
get_bench_suites() {
  local defined_benchs=""
  # Search for benchmark suite definition files
  for bench_file in $ALOJA_REPO_PATH/shell/common/benchmark_*.sh ; do
    local base_name="${bench_file##*/benchmark_}"
    defined_benchs+="${base_name:0:(-3)} "
  done

  #echo -e "sleep HiBench2 HiBench2-min HiBench2-1TB Hecuba-WordCount HiBench3 HiBench3-min"
  echo -e "${defined_benchs:0:(-1)}" #remove trailing space
}

# Enabled benchmarks
[ ! "$BENCH_SUITES" ] && BENCH_SUITES="$(get_bench_suites)"

# prints usage and exits
usage() {

  # Colorize when interactive
  if [ -t 1 ] ; then
    local reset="\033[0m" #"$(tput sgr0)"
    local red="$(tput setaf 1)"
    local green="$(tput setaf 2)"
    local yellow="$(tput setaf 3)"
    local cyan="$(tput setaf 6)"
    local white="$(tput setaf 7)"
  fi

  echo -e "${yellow}\nALOJA-BENCH, script to run benchmarks and collect results
${white}Usage:
$0 [-C clusterName <uses aloja_cluster.conf if present or not specified>]
[-n net <IB|ETH>]
[-d disk <SSD|HDD|RL{1,2,3}|R{1,2,3}>]
[-b \"benchmark suite\" <$BENCH_SUITES>]
[-r replicaton <positive int>]
[-m max mappers and reducers <positive int>]
[-i io factor <positive int>] [-p port prefix <3|4|5>]
[-I io.file <positive int>]
[-l list of benchmarks <space separated string>]
[-c compression <0 (dissabled)|1|2|3>]
[-z <block size in bytes>]
[-s save prepare]
[-N don't delete files]
[-S leave services running (implies -N)]
[-t execution type (e.g: default, experimental)]
[-e extrae (instrument execution)]

${cyan}example: $0 -C vagrant-99 -n ETH -d HDD -r 1 -m 12 -i 10 -p 3 -b HiBench2-min -I 4096 -l wordcount -c 1
$reset"
  exit 1;
}

# parses command line options
get_options() {

  OPTIND=1 #A POSIX variable, reset in case getopts has been used previously in the shell.

  while getopts "h?:C:b:r:n:d:m:i:p:l:I:c:z:s:D:tNS" opt; do
      case "$opt" in
      h|\?)
        usage
        ;;
      C)
        clusterName=$OPTARG
        ;;
      n)
        NET=$OPTARG
        [ "$NET" == "IB" ] || [ "$NET" == "ETH" ] || usage
        ;;
      d)
        DISK=$OPTARG
        defaultDisk=0
        ;;
      b)
        BENCH_SUITE=$OPTARG
        #[ "$BENCH_SUITE" == "HiBench2" ] || [ "$BENCH_SUITE" == "HiBench2-min" ] || [ "$BENCH_SUITE" == "HiBench2-1TB" ] || [ "$BENCH_SUITE" == "HiBench3" ] || [ "$BENCH_SUITE" == "HiBench3HDI" ] || [ "$BENCH_SUITE" == "HiBench3-min" ] || [ "$BENCH_SUITE" == "sleep" ] || [ "$BENCH_SUITE" == "Big-Bench" ] || [ "$BENCH_SUITE" == "TPCH" ] || usage
        if ! inList "$BENCH_SUITES" "$BENCH_SUITE" ; then
          logger "ERROR: supplied benchmark $BENCH_SUITE not enabled in list: $BENCH_SUITE_SUITES"
          usage
        fi
        ;;
      r)
        REPLICATION=$OPTARG
        ((REPLICATION > 0)) || usage
        ;;
      m)
        MAX_MAPS=$OPTARG
        ((MAX_MAPS > 0)) || usage
        ;;
      i)
        IO_FACTOR=$OPTARG
        ((IO_FACTOR > 0)) || usage
        ;;
      I)
        IO_FILE=$OPTARG
        ((IO_FILE > 0)) || usage
        ;;
      p)
        PORT_PREFIX=$OPTARG
        ((PORT_PREFIX > 0 && PORT_PREFIX < 6)) || usage
        ;;
      c)
        if [ "$OPTARG" == "0" ] ; then
          COMPRESS_GLOBAL=0
          COMPRESS_TYPE=0
        elif [ "$OPTARG" == "1" ] ; then
          COMPRESS_GLOBAL=1
          COMPRESS_TYPE=1
          COMPRESS_CODEC_GLOBAL=org.apache.hadoop.io.compress.DefaultCodec
        elif [ "$OPTARG" == "2" ] ; then
          COMPRESS_GLOBAL=1
          COMPRESS_TYPE=2
          COMPRESS_CODEC_GLOBAL=com.hadoop.compression.lzo.LzoCodec
        elif [ "$OPTARG" == "3" ] ; then
          COMPRESS_GLOBAL=1
          COMPRESS_TYPE=3
          COMPRESS_CODEC_GLOBAL=org.apache.hadoop.io.compress.SnappyCodec
        fi
        ;;
      l)
        BENCH_LIST=$OPTARG
        ;;
      z)
        BLOCK_SIZE=$OPTARG
        ;;
      s)
        SAVE_BENCH=1
        ;;
      t)
        EXEC_TYPE=$OPTARG
        ;;
      N)
        BENCH_KEEP_FILES="1"
        DELETE_HDFS=""
        ;;
      S)
        BENCH_LEAVE_SERVICES="1"
        ;;
      D)
        LIMIT_DATA_NODES=$OPTARG
        echo "LIMIT_DATA_NODES $LIMIT_DATA_NODES"
        ;;
      e)
          INSTRUMENTATION=1
        ;;
      esac
  done

  shift $((OPTIND-1))

  [ "$1" = "--" ] && shift

}


# Temple functions, reimplement in benchmark if needed

benchmark_suite_config() {
  logger "DEBUG: No specific ${FUNCNAME[0]} defined for $BENCH_SUITE"
}

# Iterate the specified benchmarks in the suite
benchmark_suite_run() {
  logger "INFO: Running $BENCH_SUITE"

  for bench in $BENCH_LIST ; do

    bench_input_dir="$BENCH_SUITE/$bench/input"
    bench_output_dir="$BENCH_SUITE/$bench/output"

    # Prepare run (in case defined)
    function_call "benchmark_prepare_$bench"

    BENCH_CURRENT_NUM_RUN="1" #reset the global counter

    # Iterate at least one time
    while true; do
      [ "$BENCH_NUM_RUNS" ] && logger "INFO: Starting iteration $BENCH_CURRENT_NUM_RUN of $BENCH_NUM_RUNS"
      # Bench Run
      function_call "benchmark_$bench"

      # Validate (eg. teravalidate)
      function_call "benchmark_validate_$bench"

      # Clean-up HDFS space (in case necessary)
      #clean_HDFS "$bench_name" "$BENCH_SUITE"

      # Check if requested to iterate multiple times
      if [ ! "$BENCH_NUM_RUNS" ] || [[ "$BENCH_CURRENT_NUM_RUN" -ge "$BENCH_NUM_RUNS" ]] ; then
        break
      else
        BENCH_CURRENT_NUM_RUN="$((BENCH_CURRENT_NUM_RUN + 1))"
      fi
    done

  done

  logger "INFO: DONE executing $BENCH_SUITE"
}

benchmark_suite_save() {
  logger "DEBUG: No specific ${FUNCNAME[0]} defined for $BENCH_SUITE"
}

benchmark_suite_cleanup() {
  logger "DEBUG: No specific ${FUNCNAME[0]} defined for $BENCH_SUITE"
}

########## END TEMPLATE FUNCTIONS

loggerb(){
  stamp=$(date '+%s')
  echo "${stamp} : $1"
  #log to zabbix
  #zabbix_sender "hadoop.status $stamp $1"
}

get_date_folder(){
  echo "$(date +%Y%m%d_%H%M%S)"
}

get_user_bin_path() {
  echo -e "export PATH=$BENCH_BIN_PATH:\$PATH;"
}

# Wraps supplied command with DSH interactive options
# $1 DSH connection string
# $1 cmd
execute_interactive() {
 local DSH_string="$1"
 local cmd="$2"
 $DSH_string -o -t -o -t -- "stty -echo -onlcr; bash -i -c '$cmd'"
}

# Tests if the supplied hostname can coincides with any node in the cluster
# NOTE: if you cluster doesnt pass this function you should overwrite it with and specific implementation in your benchark defs
# $1 hostname to check
test_in_cluster() {
  local hostname="$1"
  local coincides=1 #return code when not found

  local node_names="$(get_node_names)"

  if [ "$node_names" ] ; then
    for node in $node_names ; do #pad the sequence with 0s
      [[ "$hostname" == "$node"* ]] && coincides=0
    done
  else
    die "Cannot determine nodeNames for cluster $clusterName"
  fi

  return $coincides
}

#$1 port prefix (optional)
get_aloja_dir() {
 if [ "$1" ] ; then
  echo "${BENCH_FOLDER}_$PORT_PREFIX"
 else
  echo "${BENCH_FOLDER}"
 fi
}

# Return a list of
# $1 disk type
get_specified_disks() {
  local disk="$1"
  local dir

  if [ "$disk" == "SSD" ] || [ "$disk" == "HDD" ] || [ "$disk" == "RAM" ] || [ "$disk" == "NFS" ] || [ "$disk" == "NVE" ]; then
    dir="${BENCH_DISKS["$disk"]}"
  elif [[ "$disk" =~ .+[0-9] ]] ; then #if last char is a number

    if [[ "$1" =~ [a-zA-Z]+[0-9][0-9] ]] ; then #if last 2 chars are a numbers
      local disks="${1:(-2)}"
      local disks_type="${1:0:(-2)}"
    elif [[ "$1" =~ .+[0-9] ]] ; then
      local disks="${1:(-1)}"
      local disks_type="${1:0:(-1)}"
    fi

    for disk_number in $(seq 1 $disks) ; do
      dir+="${BENCH_DISKS["${disks_type}${disk_number}"]}\n"
    done
    dir="${dir:0:(-2)}" #remove trailing \n
  else
    die "Incorrect disk specified: $disk"
  fi

  echo -e "$dir"
}

# Returns the tmp disk in cases when mixing local and remote disks (eg. RL1)
#$1 disk type
get_tmp_disk() {
  local dir

  if [ "$1" == "SSD" ] || [ "$1" == "HDD" ] || [ "$1" == "RAM" ]  || [ "$1" == "NFS" ] || [ "$disk" == "NVE" ]; then
    dir="${BENCH_DISKS["$DISK"]}"
  elif [[ "$1" =~ .+[0-9] ]] ; then #if last char is a number

    if [[ "$1" =~ [a-zA-Z]+[0-9][0-9] ]] ; then #if last 2 chars are a numbers
      local disks="${1:(-2)}"
      local disks_type="${1:0:(-2)}"
    elif [[ "$1" =~ .+[0-9] ]] ; then
      local disks="${1:(-1)}"
      local disks_type="${1:0:(-1)}"
    fi

    if [ "$disks_type" == "RL" ] ; then
      dir="${BENCH_DISKS["HDD"]}"
    elif [ "$disks_type" == "HS" ] ; then
      dir="${BENCH_DISKS["SSD"]}"
    elif [ "$disks_type" == "ST" ] ; then
      dir="${BENCH_DISKS["TMP"]}"
    elif [ "$disks_type" == "SR" ] ; then
      dir="${BENCH_DISKS["TMP_RAM"]}"
    elif [ "$disks_type" == "NFS" ] ; then # on NFS use local as /tmp
      dir="${BENCH_DISKS["HDD"]}"
    elif [ "$disks_type" == "NVE" ] ; then
      dir="${BENCH_DISKS["NVE"]}"
    elif [ "$disks_type" == "HN" ] ; then # HDD + NVE cache
      dir="${BENCH_DISKS["NVE"]}"
    else
      dir="${BENCH_DISKS["${disks_type}1"]}"
    fi
  fi

  if [ "$dir" ] ; then
    echo -e "$dir"
  else
    die "Cannot determine tmp disk"
  fi
}

# Simple helper to append the tmp disk path
# $1 disk name
get_all_disks() {
  local disk_name="$1"
  [ ! "$disk_name" ] && die "No disk specified to get_all_disks(). Cannot continue."

  local all_disks="$(get_specified_disks "$disk_name")
$(get_tmp_disk "$disk_name")"

  #remove duplicate lines
  all_disks="$(remove_duplicate_lines "$all_disks")"

  echo -e "$all_disks"
}

# Returns the main benchmark path (useful for multidisk setups)
# $1 disk type
get_initial_disk() {
  if [ "$1" == "SSD" ] || [ "$1" == "HDD" ] || [ "$1" == "RAM" ] || [ "$1" == "NFS" ] || [ "$1" == "NVE" ] ; then
    local dir="${BENCH_DISKS["$DISK"]}"
  elif [[ "$1" =~ [a-zA-Z]+[0-9][0-9] ]] ; then #if last 2 chars are a numbers
    local disks="${1:(-2)}"
    local disks_type="${1:0:(-2)}"
    #set the first dir
    local dir="${BENCH_DISKS["${disks_type}1"]}"
  elif [[ "$1" =~ .+[0-9] ]] ; then #if last char is a number
    local disks="${1:(-1)}"
    local disks_type="${1:0:(-1)}"
    #set the first dir
    local dir="${BENCH_DISKS["${disks_type}1"]}"
  fi
  echo -e "$dir"
}

# Check if the supplied list of variables are not null
# $1 variable or list of vars
test_variable_set() {
  local variables="$1"
  for variable in $variables ; do
    if [ ! "${!variable}" ] ; then
      die "Required variable $variable is null, please check the config"
    fi
  done
}

# Performs some basic validations
# $1 DISK
validate() {
  local disk="$1"

  # First check if some required variables are set
  test_variable_set "BENCH_LOCAL_DIR BENCH_SHARE_DIR"

  if [ "$clusterType" != "PaaS" ]; then
    # Check whether we are in the right cluster
    if [ "${checkClusterMembership}" = "1" ] && ! test_in_cluster "$(hostname)" ; then
      die "host $(hostname) does not belong to specified cluster $clusterName\nMake sure you run this script from within a cluster"
    fi

    if ! inList "$CLUSTER_NETS" "$NET" ; then
      die "Disk type $NET not supported for $clusterName\nSupported: $NET"
    fi

    # Disk validations
    if ! inList "$CLUSTER_DISKS" "$DISK" ; then
      die "Disk type $DISK not supported for $clusterName\nSupported: $CLUSTER_DISKS"
    fi

    # Check that we got the dynamic disk location correctly
    if [ ! "$(get_initial_disk "$disk")" ] ; then
      die "cannot determine $DISK path.  Output: $(get_initial_disk "$disk")"
    fi

    # Iterate all defined and tmp disks to see if we can write to them
    local disks="$(get_all_disks "$disk" )"
    logger "DEBUG: testing write permissions for mount(s) for config $disk: $(nl2char "$disks" " ")"

    for disk_tmp in $disks ; do
      logger "DEBUG: testing write permissions in $disk_tmp"
      local touch_file="$disk_tmp/aloja.touch"
      #if file exists test if we can delete it
      if [ -f "$touch_file" ] ; then
        rm "$touch_file" || die "Cannot delete files in $disk_tmp"
      fi
      touch "$touch_file" || die "Cannot write files in $disk_tmp"
      rm "$touch_file" || die "Cannot delete files in $disk_tmp"
    done
  fi

  if [ "$clusterType" != "PaaS" ]; then
    # Check whether we are in the right cluster
    if [ "${checkClusterMembership}" = "1" ] && ! test_in_cluster "$(hostname)" ; then
      die "host $(hostname) does not belong to specified cluster $clusterName\nMake sure you run this script from within a cluster"
    fi
  else
    logger "INFO: Skipping some validations in PaaS"
  fi
}

# Groups initialization phases
initialize() {

  # initialize cluster node names and connect string
  initialize_node_names
  # set the name for the job run
  set_job_config
  # check if all nodes are up
  test_nodes_connection
  # check if ~/share is correctly mounted
  test_share_dir
}

#old code moved here
# TODO cleanup
initialize_node_names() {
  local extra_node_names

  #For infiniband tests
  if [ "$NET" == "IB" ] ; then
    [ ! "$defaultProvider" == "vagrant" ] && IFACE="ib0" #vagrant we use for testing IB config
    master_name="$(get_master_name_IB)"
    node_names="$(get_node_names_IB)"
    extra_node_names="$(get_extra_node_names)"
  else
    #IFACE should be already setup
    master_name="$(get_master_name)"
    node_names="$(get_node_names)"
    extra_node_names="$(get_extra_node_names)"
  fi

  NUMBER_OF_DATA_NODES="$numberOfNodes"

  if [ ! -z "$LIMIT_DATA_NODES" ] ; then
    node_iteration=0
    for node in $node_names ; do
      if [ ! -z "$nodes_tmp" ] ; then
        node_tmp="$node_tmp\n$node"
      else
        node_tmp="$node"
      fi
      [[ $node_iteration -ge $LIMIT_DATA_NODES ]]  && break;
      node_iteration=$((node_iteration+1))
    done

    node_name=$(echo -e "$nodes_tmp")
    NUMBER_OF_DATA_NODES="$LIMIT_DATA_NODES"
  fi

  DSH="dsh -M -c -m "
  DSH_EXTRA="$DSH"

  DSH_MASTER="dsh -H -m $master_name"

  DSH="$DSH $(nl2char "$node_names" ",") "

  # TODO deprecate this var
  DSH_C="$DSH -c " #concurrent

  DSH_SLAVES="${DSH_C/"$master_name,"/}" #remove master name and trailing coma

  # Instrument additional nodes
  DSH_ALL="$DSH"

  if [ "$extra_node_names" ] ; then
    logger "WARNING: extra nodes requested for instrumentation"
    DSH_EXTRA+=" $(nl2char "$extra_node_names" ",") "
    # TODO temporary to test
    DSH_ALL="${DSH:0:(-1)},$(nl2char "$extra_node_names" ",")"
  fi
}

# Tests cluster nodes for a defined condition
# $1 condition string
# $2 severity of error
test_nodes() {
  local condition="$1"
  local severity="$2"

  [ ! "$severity" ] && severity="ERROR"

  local node_output="$($DSH "$condition && echo '$testKey' 2>&1"|sort )"
  local num_OK="$(echo -e "$node_output"|grep "$testKey"|wc -l)"

  local KO_output="$(echo -e "$node_output"|grep -v "$testKey")"

  local num_nodes="$(get_num_nodes)"
  if (( num_OK != num_nodes )) ; then
    logger "${severity}: Cannot execute: $condition in all nodes. Num OK: $num_OK Num KO: $((num_nodes - num_OK))
DEBUG output:
$node_output"
    return 1
  else
    # all is good
    return 0
  fi
}

# Tests cluster nodes for NOT HAVING a defined condition
# $1 condition string
# $2 severity of error
test_nodes_inverse() {
  local condition="$1"
  local severity="$2"

  [ ! "$severity" ] && severity="ERROR"

  local node_output="$($DSH "$condition && echo '$testKey'"|sort 2>&1)"
  local num_OK="$(echo -e "$node_output"|grep "$testKey"|wc -l)"
  local num_nodes="$(get_num_nodes)"
  if (( num_OK > 0 )) ; then
    logger "${severity}: Found condition: $condition in nodes. Num OK: $num_OK Num KO: $((num_nodes - num_OK))
DEBUG Output:
$node_output"
    return 1
  else
    # all is good
    return 0
  fi
}

# Tests if defined nodes are accesible vis SSH
test_nodes_connection() {
  logger "INFO: Testing connectivity to nodes"
  if test_nodes "hostname" ; then
    logger "INFO: All $(get_num_nodes) nodes are accesible via SSH"
  else
    die "Cannot connect via SSH to all nodes"
  fi
}

# Sends kill signal to processes listening in a particular port for the whole cluster
# $1 TCP port
kill_on_port() {
 local port="$1"
 test_nodes "fuser -k -n tcp $port"
}

# Tries to mount shared folder
# $1 shared folder
mount_share() {
  shared_folder="$1"

  if [ ! "$noSudo" ] ; then
    logger "WARNING: attempting to remount $shared_folder"
    $DSH "
if [ ! -f '$shared_folder/safe_store' ] ; then
  sudo umount -f '$shared_folder';
  sudo mount '$shared_folder';
  sudo mount -a;
fi
"

  fi
}

# Tests if nodes have the shared dir correctly mounted
# $1 if to exit (for retries)
test_share_dir() {
  local no_retry="$1"
  local test_file="$homePrefixAloja/$userAloja/share/safe_store"

  logger "INFO: Testing if ~/share mounted correctly"
  if test_nodes "ls '$test_file'" ; then
    logger "INFO: All $(get_num_nodes) nodes have the ~/share dir correctly mounted"
  else
    if [ "$no_retry" ] ; then
      die "~/share dir not mounted correctly"
    else #try again
      mount_share "$homePrefixAloja/$userAloja/share/"
      test_share_dir "no_retry"
    fi
  fi
}

#old code moved here
# TODO cleanup
set_job_config() {
  # Output directory name
  CONF="${NET}_${DISK}_b${BENCH_SUITE}_S${BENCH_SCALE_FACTOR}_D${NUMBER_OF_DATA_NODES}_${clusterName}"

  JOB_NAME="$(get_date_folder)_$CONF"

  JOB_PATH="$BENCH_SHARE_DIR/jobs_$clusterName/$JOB_NAME"
  #LOG_PATH="$JOB_PATH/log_${JOB_NAME}.log"
  #LOG="2>&1 |tee -a $LOG_PATH"

  #create dir to save files in one host
  $DSH_MASTER "mkdir -p $JOB_PATH;"

  # Automatically log all output to file
  log_all_output "$JOB_PATH/${0##*/}"

  logger "STARTING RUN $JOB_NAME"
  logger "INFO: Job path: $JOB_PATH"
  logger "INFO: Conf: $CONF"
  logger "INFO: Benchmark Suite: $BENCH_SUITE"
  logger "INFO: Benchmarks to execute: $BENCH_LIST"
  logger "DEBUG: DSH: $DSH\n"
  #logger "INFO: DSH_C: $DSH_C"
  #logger "INFO: DSH_SLAVES: $DSH_SLAVES"
}

# Set some OS requirements (e.g., to dissable swapping)
update_OS_config() {
  if [ ! "$noSudo" ] && [ "$EXECUTE_HIBENCH" ]; then
    $DSH "
sudo sysctl -w vm.swappiness=0 > /dev/null;
sudo sysctl vm.panic_on_oom=1 > /dev/null;
sudo sysctl -w fs.file-max=65536 > /dev/null;
sudo service ufw stop 2>&1 > /dev/null;
"
  fi
}

get_apps_path() {
  echo -e "aplic2/apps"
}

# Get the main path for the benchmark
get_local_bench_path() {
  echo -e "$(get_initial_disk "$DISK")/$(get_aloja_dir "$PORT_PREFIX")"
}

get_local_tmp_path() {
  echo -e "$(get_tmp_disk "$DISK")/$(get_aloja_dir "$PORT_PREFIX")"
}

get_local_apps_path() {
  echo -e "$BENCH_LOCAL_DIR/$(get_apps_path)"
}

get_local_configs_path() {
  echo -e "$BENCH_LOCAL_DIR/aplic2/configs"
}

get_base_apps_path() {
  echo -e "$BENCH_SHARE_DIR/$(get_apps_path)"
}

get_base_tarballs_path() {
  echo -e "$BENCH_SHARE_DIR/aplic2/tarballs"
}

get_base_configs_path() {
  echo -e "$ALOJA_REPO_PATH/config/bench/config_files"
}

# Installs binaries and configs
# TODO needs improvement
install_requires() {
  if [ "${#BENCH_REQUIRED_FILES[@]}" ] ; then
    #logger "INFO: Checking if need to download/copy files to node local dirs at: $(get_local_apps_path)"
    for required_file in "${!BENCH_REQUIRED_FILES[@]}" ; do
      logger "INFO: Checking if to download/copy $required_file"
      local base_name="${BENCH_REQUIRED_FILES["$required_file"]##*/}"

      # For github repos, add other exceptions to file names that might repeat here
      if [[ "$base_name" =~ "master."* ]] ; then
        # Use the array key index name
        base_name="${required_file}.${base_name#*.}"
      fi

      # test if we need to download first to share dir
      local test_action="$($DSH_MASTER "[ -f '$(get_base_tarballs_path)/$base_name' ] && echo '$testKey'")"
      if [[ ! "$test_action" == *"$testKey"* ]] ; then
        logger "INFO: Downloading $required_file"
        $DSH_MASTER "
mkdir -p '$(get_base_tarballs_path)' && wget --progress=dot -e dotbytes=10M '${BENCH_REQUIRED_FILES["$required_file"]}' -O '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name'"

        # test if download was succesful
        local test_action="$($DSH_MASTER "[ -f '$(get_base_tarballs_path)/$base_name' ] && echo '$testKey'")"
        if [[ ! "$test_action" == *"$testKey"* ]] ; then
          die "Could not download $required_file from ${BENCH_REQUIRED_FILES["$required_file"]}"
        fi
      fi

      $DSH "
if [ ! -d '$(get_local_apps_path)/$required_file' ] ; then
  mkdir -p '$(get_local_apps_path)/';
  cd '$(get_local_apps_path)/';
  echo 'DEBUG: need to uncompress $(get_base_tarballs_path)/$base_name to $(get_local_apps_path)/$required_file';
  if [[ '$base_name' == *'.tar.gz' || '$base_name' == *'.tgz' ]] ; then
    tar -xzf '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name';
  elif [[ '$base_name' == *'.tar.bz2' ]] ; then
    tar -xjf '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name';
  elif [[ '$base_name' == *'.zip' ]] ; then
    unzip -q -o '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name';
  else
    echo 'ERROR: unknown file extension for $base_name';
  fi
else
  echo 'DEBUG: local dir $(get_local_apps_path)/$required_file exists'
fi
"
    done
  else
    logger "INFO: No required files to download/copy specified"
  fi
}

# Installs binaries and configs
# TODO needs merging with install packages
install_bench_packages() {
  if [ "$BENCH_REQUIRED_PACKAGES" ] ; then
    if [[ "$vmOSType" == "Ubuntu" || "$vmOSType" == "Debian" ]] ; then
      logger "INFO: Testing to see if packages: $BENCH_REQUIRED_PACKAGES are installed"
      if ! test_nodes "dpkg -s $BENCH_REQUIRED_PACKAGES" ; then
        if [ ! "$noSudo" ] ; then
          logger "INFO: Attempting to install: $BENCH_REQUIRED_PACKAGES"
          if ! test_nodes "dpkg -s $BENCH_REQUIRED_PACKAGES 2> /dev/null" ; then
            $DSH "sudo apt-get update;
export DEBIAN_FRONTEND=noninteractive;
sudo apt-get -o Dpkg::Options::='--force-confold' install -y --force-yes $BENCH_REQUIRED_PACKAGES"

            logger "WARNING: failed to install one or more packages: $BENCH_REQUIRED_PACKAGES"
          fi
        else
          logger "WARNING: no sudo selected, cannot install needed packages in: $BENCH_REQUIRED_PACKAGES"
        fi
      else
        logger "INFO: Required packages bench: $BENCH_REQUIRED_PACKAGES are correctly installed"
      fi
    elif [[ "$vmOSType" == "Fedora" || "$vmOSType" == "RHEL" || "$vmOSType" == "CentOS" ]] ; then
      for package in $BENCH_REQUIRED_PACKAGES ; do
        if [ ! "$($DSH "which $package; 2> /dev/null")" ] ; then
          logger "INFO: Attempting to install: $BENCH_REQUIRED_PACKAGES"
          $DSH "sudo yum install -y $BENCH_REQUIRED_PACKAGES"
          break
        fi
      done
    else
      logger "WARNING: not a Debian or Fedora based system, not checking if to install packages: $BENCH_REQUIRED_PACKAGES"
    fi
  fi
}


# Rsyncs specified config folders in aplic2/configs/
install_configs() {
  if [ "$BENCH_CONFIG_FOLDERS" ] ; then
    for config_folder in $BENCH_CONFIG_FOLDERS ; do
      local full_config_folder_path="$(get_base_configs_path)/$config_folder"
      if [ -d "$full_config_folder_path" ] ; then
        logger "INFO: Synching configs from $config_folder"
        $DSH "rsync -aur --delete '$full_config_folder_path' '$(get_local_configs_path)' "
      else
        die "Cannot find config folder in $full_config_folder_path"
      fi
    done
  else
    logger "DEBUG: No config folder specified to copy"
  fi
}

install_files() {
  install_requires
  install_configs
  install_bench_packages
}

check_aplic_updates() {
  #only copy files if version has changed (to save time)
  logger "INFO: Checking if to generate source dirs $BENCH_SHARE_DIR/aplic/aplic_version == $BENCH_SOURCE_DIR/aplic_version"
  for node in $node_names ; do
    logger "INFO:  for host $node"
    if [ "$(ssh "$node" "[ "\$\(cat $BENCH_SHARE_DIR/aplic/aplic_version\)" == "\$\(cat $BENCH_SOURCE_DIR/aplic_version 2\> /dev/null \)" ] && echo 'OK' || echo 'KO'" )" != "OK" ] ; then
      logger "INFO: At least host $node did not have source dirs. Generating source dirs for ALL hosts"

      if [ ! "$(ssh "$node" "[ -d \"$BENCH_SHARE_DIR/aplic\" ] && echo 'OK' || echo 'KO'" )" != "OK" ] ; then
        #logger "Downloading initial aplic dir from dropbox"
        #$DSH "wget -nv https://www.dropbox.com/s/ywxqsfs784sk3e4/aplic.tar.bz2?dl=1 -O $BASE_DIR/aplic.tar.bz2"

        $DSH "rsync -aur --force $BENCH_SHARE_DIR/aplic.tar.bz2 /tmp/"

        logger "INFO: Uncompressing aplic"
        $DSH  "mkdir -p $BENCH_SOURCE_DIR/; cd $BENCH_SOURCE_DIR/../; tar -C $BENCH_SOURCE_DIR/../ -jxf /tmp/aplic.tar.bz2; "  #rm aplic.tar.bz2;
      fi

      logger "Rsynching files"
      $DSH "mkdir -p $BENCH_SOURCE_DIR; rsync -aur --force $BENCH_SHARE_DIR/aplic/* $BENCH_SOURCE_DIR/"
      break #dont need to check after one is missing
    else
      logger "INFO:  Host $node up to date"
    fi
  done

  #if [ "$(cat $BENCH_SHARE_DIR/aplic/aplic_version)" != "$(cat $BENCH_SOURCE_DIR/aplic_version)" ] ; then
  #  logger "INFO: Generating source dirs"
  #  $DSH "mkdir -p $BENCH_SOURCE_DIR; cp -ru $BENCH_SHARE_DIR/aplic/* $BENCH_SOURCE_DIR/"
  #  #$DSH "cp -ru $BENCH_SOURCE_DIR/${HADOOP_VERSION}-home $BENCH_SOURCE_DIR/${HADOOP_VERSION}" #rm -rf $BENCH_SOURCE_DIR/${HADOOP_VERSION};
  #elsefi
  #  logger "INFO: Source dirs up to date"
  #fi

}

# Exports a var and path to the cluster
# $1 varname
# $2 path
export_var_path() {
  : # WiP
}

zabbix_sender(){
  :
  #echo "al-1001 $1" | /home/pristine/share/aplic/zabbix/bin/zabbix_sender -c /home/pristine/share/aplic/zabbix/conf/zabbix_agentd_az.conf -T -i - 2>&1 > /dev/null
  #>> $LOG_PATH

##For zabbix monitoring make sure IB ports are available
#ssh_tunnel="ssh -N -L al-1001:30070:al-1001-ib0:30070 -L al-1001:30030:al-1001-ib0:30030 al-1001"
##first make sure we kill any previous, even if we don't need it
#pkill -f "ssh -N -L"
##"$ssh_tunnel"
#
#if [ "${NET}" == "IB" ] ; then
#  $ssh_tunnel &
#fi

}

# Copies specified perf mon binaries to bench path, so that they can be started
# and specially killed easily
set_monit_binaries() {
  if [ "$BENCH_PERF_MONITORS" ] ; then
    local perf_mon_bin_path
    local perf_mon_bench_path="$(get_local_bench_path)/aplic"

    if [ "$vmType" != "windows" ] ; then
      for perf_mon in $BENCH_PERF_MONITORS ; do

        if ! inList "$BENCH_PERF_NON_BINARY" "$perf_mon" ; then
          logger "INFO: Setting up performance monitor: $perf_mon"

          # Get the path of file from the aloja repo
          if [ "$perf_mon" == "cachestat" ] ; then
            perf_mon_bin_path="$ALOJA_REPO_PATH/aloja-tools/src/cachestat.sh"

            if [ "$noSudo" ] ; then
              logger "WARNING: $perf_mon requires sudo, skipping setting it up."
              continue
            fi
          # JavaStat actually uses pidstat
          elif [ "$perf_mon" == "JavaStat" ] ; then
            perf_mon_bin_path="$($DSH_MASTER "$(get_user_bin_path) which 'pidstat'")"
          # Get it from the user path (normal case)
          else
            # we need to include our custom bin path as SSH is not interactive
            perf_mon_bin_path="$($DSH_MASTER "$(get_user_bin_path) which '$perf_mon'")"
          fi

          if [ -f "$perf_mon_bin_path" ] ; then
            log_DEBUG "Copying $perf_mon binary from: $perf_mon_bin_path to $perf_mon_bench_path"
            $DSH "mkdir -p '$perf_mon_bench_path'; cp '$perf_mon_bin_path' '$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX';"

            if [ "$(get_extra_node_names)" ] ; then
              $DSH_EXTRA "mkdir -p '$(get_extra_node_folder)/aplic'; cp '$perf_mon_bin_path' '$(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX'"
            fi
          else
            log_ERR "Cannot find $perf_mon binary on the system at: $perf_mon_bin_path"
            log_DEBUG "Perf monitor bin path: $($DSH_MASTER 'ls "$perf_mon_bin_path" 2>&1')"
          fi
        else
          logger "INFO: Setting up script-style perfomance monitor: $perf_mon"
        fi
      done
    else
      logger "WARNING: no extra perf monitors set for Windows"
    fi
  else
    logger "WARNING: No peformance monitors (e.g., vmstat) have been selected"
  fi
}

# Before starting monitors always check if they are already running
start_monit() {
  restart_monit
}

# Stops monitors (if any) and starts them
restart_monit(){
  if [ "$BENCH_PERF_MONITORS" ] ; then
    local perf_mon_bin_path
    local perf_mon_bench_path="$(get_local_bench_path)/aplic"

    if [ "$vmType" != "windows" ] ; then
      logger "INFO: Restarting perf monit"
      stop_monit #in case there is any running

      # Make sure we clean monits on abnormal exit
      if [ "$BENCH_PERF_MONITORS" ] ; then
        update_traps "stop_monit;" "update_logger"
      fi

      for perf_mon in $BENCH_PERF_MONITORS ; do
        run_monit "$perf_mon"
      done
      #logger "DEBUG: perf monitors ready"
    fi
  fi
}

# Starts the specified perf_mon
# They execute in background (&) to start them as close as possible in time
# $1 perf_mon
run_monit() {
  local perf_mon="$1"
  local perf_mon_bin="$(get_local_bench_path)/aplic/${perf_mon}_$PORT_PREFIX"

  if [ "$perf_mon" == "sar" ] ; then
    if [ "$clusterType" != "PaaS" ]; then
      $DSH "$(get_user_bin_path) $perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -o $(get_local_bench_path)/sar-\$(hostname).sar $BENCH_PERF_INTERVAL >/dev/null &" & #2>&1

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "$(get_user_bin_path) $(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -o $(get_extra_node_folder)/sar-\$(hostname).sar $BENCH_PERF_INTERVAL >/dev/null &" & #2>&1
      fi
    else
      $DSH "$(get_user_bin_path) sar -o $(get_local_bench_path)/sar-\$(hostname).sar $BENCH_PERF_INTERVAL >/dev/null &" & # 2>&1
    fi
  elif [ "$perf_mon" == "vmstat" ] ; then
    if [ "$clusterType" != "PaaS" ]; then
      $DSH "$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -n $BENCH_PERF_INTERVAL >> $(get_local_bench_path)/vmstat-\$(hostname).log &" &

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "$(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -n $BENCH_PERF_INTERVAL >> $(get_extra_node_folder)/vmstat-\$(hostname).log &" &
      fi
    else
      $DSH "vmstat -n $BENCH_PERF_INTERVAL >> $(get_local_bench_path)/vmstat-\$(hostname).log &" &
    fi
  # For iostat use PAT's syntax
  elif [ "$perf_mon" == "iostat" ] ; then
    if [ "$clusterType" != "PaaS" ]; then
      $DSH "$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -x -k -y -d $BENCH_PERF_INTERVAL | awk -v host=\$(hostname) '(!/^$/){now=strftime(\"%s \");if(/Device:/){print \"HostName\",\"TimeStamp\", \$0} else{ if(\$0 && !/Linux/) print host, now \$0}}; fflush()' >> $(get_local_bench_path)/iostat-\$(hostname).log &" &

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "$(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -x -k -y -d $BENCH_PERF_INTERVAL | awk -v host=\$(hostname) '(!/^$/){now=strftime(\"%s \");if(/Device:/){print \"HostName\",\"TimeStamp\", \$0} else{ if(\$0 && !/Linux/) print host, now \$0}}; fflush()' >> $(get_extra_node_folder)/iostat-\$(hostname).log &"  &
      fi
    else
      $DSH "iostat -x -k -y -d $BENCH_PERF_INTERVAL | awk -v host=\$(hostname) '(!/^$/){now=strftime(\"%s \");if(/Device:/){print \"HostName\",\"TimeStamp\", \$0} else{ if(\$0 && !/Linux/) print host, now \$0}}; fflush()' >> $(get_local_bench_path)/iostat-\$(hostname).log &" &
      #iostat -x -k -d $SAMPLING_INTERVAL
    fi
  # To count Java processes (for PAT export)
  elif [ "$perf_mon" == "MapRed" ] ; then
    $DSH "
(
echo MapCount ReduceCount ContainerCount TezCount JavaCount ProcCount
while :
do
  processes=\"\$(ps fauxwww)\";
  echo \"\$(echo -e \"\$processes\"|grep [j]ava|grep _m_ |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava| grep _r_ |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava| grep container_ |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava| grep container_|grep tez |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava|wc -l ) \$(echo -e \"\$processes\" |wc -l )\"
  sleep $BENCH_PERF_INTERVAL
done
) | awk -v host=\$(hostname) '(!/^\$/){now=strftime(\"%s \");if(\$0 && !/Linux/) if (/MapCount/){print \"HostName\",\"TimeStamp\",\$0} else {print host,now \$0}}; fflush()' > $(get_local_bench_path)/MapRed-\$(hostname).log &" &

  # To count Java processes (for PAT export)
  elif [ "$perf_mon" == "JavaStat" ] ; then
    local pidstat_cmd="java"
    $DSH "
$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -rudh -p ALL -C init | awk -v host=\$(hostname) '(/Time/){\$1=\$2=\"\"; print \"HostName\",\"TimeStamp\", \$0}; fflush()' > $(get_local_bench_path)/JavaStat-\$(hostname).log
$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -rudh -p ALL -C $pidstat_cmd $BENCH_PERF_INTERVAL | awk -v cmd='$pidstat_cmd' -v host=\$(hostname) '(!/^\$/ && !/Time/ && !/CPU/){if (\$NF == cmd){now=strftime(\"%s\"); \$1=\"\"; print host, now, \$0}; fflush()}' >> $(get_local_bench_path)/JavaStat-\$(hostname).log &
" &

  # iotop, requires sudo and interval only 1 sec supported
  elif [ "$perf_mon" == "iotop" ] ; then
    if [ -z "$noSudo" ] || [ "$BENCH_PERF_INTERVAL" == "1" ]; then
      if [ "$clusterType" != "PaaS" ]; then
        local iotop_log="$(get_local_bench_path)/iotop-\$(hostname).log"
        $DSH "touch $iotop_log; sudo $perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -btoqqk >> $iotop_log &" &

        if [ "$(get_extra_node_names)" ] ; then
          $DSH_EXTRA "touch $(get_extra_node_folder)/iotop-\$(hostname).log; sudo $(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -btoqqk >> $iotop_log &" &
        fi
      else
        $DSH "touch $iotop_log; sudo iotop  >> $iotop_log &" &
      fi
    else
      logger "WARNING: iotop requires root and sudo is disabled for cluster OR BENCH_PERF_INTERVAL != 1 (set to: $BENCH_PERF_INTERVAL), skipping..."
    fi
  # dstat
  elif [ "$perf_mon" == "dstat" ] ; then
     # Removed for ubuntu
     # -T --cpu-adv --top-cpu-adv -l -d --aio --disk-avgqu --disk-avgrq --disk-svctm --disk-tps --disk-util --disk-wait --top-bio-adv --top-io-adv --md-status -n --net-packets  -gimprsy --cpu-use --fs --top-int --top-latency -ipc -lock --mem-adv --top-mem --raw --unix --vm-adv --bits --nocolor --noheader --profile --power --proc-count --thermal --noheaders
     #--cpu-adv --disk-avgqu --disk-avgrq --disk-svctm --disk-wait --md-status  --cpu-use --mem-adv --vm-adv --bits --thermal --top-io-adv --top-int

    if [ "$clusterType" != "PaaS" ]; then
      local dstat_log="$(get_local_bench_path)/dstat-\$(hostname).log"
      $DSH "$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -T --cpu --top-cpu-adv -l -d --aio --disk-tps --disk-util --top-bio-adv --top-io-adv -n --net-packets  -gimprsy --fs --top-int --top-latency -ipc -lock --top-mem --raw --unix --nocolor --noheader --profile --power --proc-count --noheaders  $BENCH_PERF_INTERVAL >> $dstat_log &" &
    else
      $DSH "dstat -T --cpu --top-cpu-adv -l -d --aio --disk-tps --disk-util --top-bio-adv -n --net-packets  -gimprsy --fs --top-latency -ipc -lock --top-mem --raw --unix --nocolor --noheader --profile --power --proc-count --noheaders  $BENCH_PERF_INTERVAL >> $dstat_log &" &
    fi
  # perf
  elif [ "$perf_mon" == "perf" ] ; then

    # Enable CPU trace data (if we have root)
    if [ -z "$noSudo" ] ; then
      $DSH "sudo echo '0' > /proc/sys/kernel/perf_event_paranoid"
    fi
    # TODO: https://github.com/intel-hadoop/PAT/blob/master/PAT/WORKER_scripts/instruments/perf

  # cachestat
  elif [ "$perf_mon" == "cachestat" ] ; then
      $DSH "sudo $perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -n -t $BENCH_PERF_INTERVAL > $(get_local_bench_path)/cachestat-\$(hostname).log &" & #2>&1

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "sudo $(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -n -t $BENCH_PERF_INTERVAL > $(get_extra_node_folder)/cachestat-\$(hostname).log &" & #2>&1
      fi
  else
    die "Specified perf mon $perf_mon not implemented"
  fi


  wait #for the bg processes

  # BWM not used any more
  #$DSH_C "$bwm -o csv -I bond0,eth0,eth1,eth2,eth3,ib0,ib1 -u bytes -t 1000 >> $(get_local_bench_path)/bwm-\$(hostname).log &"
}

# Kill possibly running perf mons
stop_monit(){
  if [ "$BENCH_PERF_MONITORS" ] ; then
    if [ "$vmType" != "windows" ]; then
      logger "INFO: Stoping monit (in case necessary). Monitors: $BENCH_PERF_MONITORS"
      for perf_mon in $BENCH_PERF_MONITORS ; do

        local requires_sudo=""
        if [[ "$perf_mon" == "iotop" || "$perf_mon" == "cachestat" ]] ; then
          requires_sudo="sudo"
        fi

        local perf_mon_bin="$(get_local_bench_path)/aplic/${perf_mon}_$PORT_PREFIX"

        if ! inList "$BENCH_PERF_NON_BINARY" "$perf_mon" ; then
          $DSH "$requires_sudo pkill -9 -f '[${perf_mon_bin:0:1}]${perf_mon_bin:1}' 2> /dev/null" #& # [] for it not to match itself in ssh
        elif [ "$perf_mon" == "MapRed" ] ; then
          $DSH "$requires_sudo pkill -9 -f [M]apCount" # [] for it not to match itself in ssh
        elif [ "$perf_mon" == "JavaStat" ] ; then
          $DSH "$requires_sudo pkill -9 pidstat" # TODO improve to use custom naming
        fi

        if [ "$(get_extra_node_names)" ] ; then
          $DSH_EXTRA "$requires_sudo pkill -9 -f '[${perf_mon_bin:0:1}]${perf_mon_bin:1}' 2> /dev/null" #&
        fi

        # TODO this is something temporal for PaaS clusters
        if [ "$clusterType" == "PaaS" ]; then
          $DSH "killall -9 sadc; killall -9 vmstat; killall -9 iostat; killall -9 pidstat; pkill -9 -f [M]apCount; pgrep -f '[M]apCount'|xargs kill -9; $requires_sudo kill -9 iotop 2> /dev/null"  2> /dev/null
        fi
      done
      #logger "DEBUG: perf monitors ready"
    fi
  fi

  #wait #for the bg processes
}
<commit_msg>BENCH: Implemented perf monitor to drop caches at specified interval<commit_after># Helper functions for running benchmarks

#globals
BENCH_CURRENT_NUM_RUN="0"
BENCH_BIN_PATH="\$HOME/share/sw/bin"

# Outputs a list of defined benchmark suites separated by spaces
get_bench_suites() {
  local defined_benchs=""
  # Search for benchmark suite definition files
  for bench_file in $ALOJA_REPO_PATH/shell/common/benchmark_*.sh ; do
    local base_name="${bench_file##*/benchmark_}"
    defined_benchs+="${base_name:0:(-3)} "
  done

  #echo -e "sleep HiBench2 HiBench2-min HiBench2-1TB Hecuba-WordCount HiBench3 HiBench3-min"
  echo -e "${defined_benchs:0:(-1)}" #remove trailing space
}

# Enabled benchmarks
[ ! "$BENCH_SUITES" ] && BENCH_SUITES="$(get_bench_suites)"

# prints usage and exits
usage() {

  # Colorize when interactive
  if [ -t 1 ] ; then
    local reset="\033[0m" #"$(tput sgr0)"
    local red="$(tput setaf 1)"
    local green="$(tput setaf 2)"
    local yellow="$(tput setaf 3)"
    local cyan="$(tput setaf 6)"
    local white="$(tput setaf 7)"
  fi

  echo -e "${yellow}\nALOJA-BENCH, script to run benchmarks and collect results
${white}Usage:
$0 [-C clusterName <uses aloja_cluster.conf if present or not specified>]
[-n net <IB|ETH>]
[-d disk <SSD|HDD|RL{1,2,3}|R{1,2,3}>]
[-b \"benchmark suite\" <$BENCH_SUITES>]
[-r replicaton <positive int>]
[-m max mappers and reducers <positive int>]
[-i io factor <positive int>] [-p port prefix <3|4|5>]
[-I io.file <positive int>]
[-l list of benchmarks <space separated string>]
[-c compression <0 (dissabled)|1|2|3>]
[-z <block size in bytes>]
[-s save prepare]
[-N don't delete files]
[-S leave services running (implies -N)]
[-t execution type (e.g: default, experimental)]
[-e extrae (instrument execution)]

${cyan}example: $0 -C vagrant-99 -n ETH -d HDD -r 1 -m 12 -i 10 -p 3 -b HiBench2-min -I 4096 -l wordcount -c 1
$reset"
  exit 1;
}

# parses command line options
get_options() {

  OPTIND=1 #A POSIX variable, reset in case getopts has been used previously in the shell.

  while getopts "h?:C:b:r:n:d:m:i:p:l:I:c:z:s:D:tNS" opt; do
      case "$opt" in
      h|\?)
        usage
        ;;
      C)
        clusterName=$OPTARG
        ;;
      n)
        NET=$OPTARG
        [ "$NET" == "IB" ] || [ "$NET" == "ETH" ] || usage
        ;;
      d)
        DISK=$OPTARG
        defaultDisk=0
        ;;
      b)
        BENCH_SUITE=$OPTARG
        #[ "$BENCH_SUITE" == "HiBench2" ] || [ "$BENCH_SUITE" == "HiBench2-min" ] || [ "$BENCH_SUITE" == "HiBench2-1TB" ] || [ "$BENCH_SUITE" == "HiBench3" ] || [ "$BENCH_SUITE" == "HiBench3HDI" ] || [ "$BENCH_SUITE" == "HiBench3-min" ] || [ "$BENCH_SUITE" == "sleep" ] || [ "$BENCH_SUITE" == "Big-Bench" ] || [ "$BENCH_SUITE" == "TPCH" ] || usage
        if ! inList "$BENCH_SUITES" "$BENCH_SUITE" ; then
          logger "ERROR: supplied benchmark $BENCH_SUITE not enabled in list: $BENCH_SUITE_SUITES"
          usage
        fi
        ;;
      r)
        REPLICATION=$OPTARG
        ((REPLICATION > 0)) || usage
        ;;
      m)
        MAX_MAPS=$OPTARG
        ((MAX_MAPS > 0)) || usage
        ;;
      i)
        IO_FACTOR=$OPTARG
        ((IO_FACTOR > 0)) || usage
        ;;
      I)
        IO_FILE=$OPTARG
        ((IO_FILE > 0)) || usage
        ;;
      p)
        PORT_PREFIX=$OPTARG
        ((PORT_PREFIX > 0 && PORT_PREFIX < 6)) || usage
        ;;
      c)
        if [ "$OPTARG" == "0" ] ; then
          COMPRESS_GLOBAL=0
          COMPRESS_TYPE=0
        elif [ "$OPTARG" == "1" ] ; then
          COMPRESS_GLOBAL=1
          COMPRESS_TYPE=1
          COMPRESS_CODEC_GLOBAL=org.apache.hadoop.io.compress.DefaultCodec
        elif [ "$OPTARG" == "2" ] ; then
          COMPRESS_GLOBAL=1
          COMPRESS_TYPE=2
          COMPRESS_CODEC_GLOBAL=com.hadoop.compression.lzo.LzoCodec
        elif [ "$OPTARG" == "3" ] ; then
          COMPRESS_GLOBAL=1
          COMPRESS_TYPE=3
          COMPRESS_CODEC_GLOBAL=org.apache.hadoop.io.compress.SnappyCodec
        fi
        ;;
      l)
        BENCH_LIST=$OPTARG
        ;;
      z)
        BLOCK_SIZE=$OPTARG
        ;;
      s)
        SAVE_BENCH=1
        ;;
      t)
        EXEC_TYPE=$OPTARG
        ;;
      N)
        BENCH_KEEP_FILES="1"
        DELETE_HDFS=""
        ;;
      S)
        BENCH_LEAVE_SERVICES="1"
        ;;
      D)
        LIMIT_DATA_NODES=$OPTARG
        echo "LIMIT_DATA_NODES $LIMIT_DATA_NODES"
        ;;
      e)
          INSTRUMENTATION=1
        ;;
      esac
  done

  shift $((OPTIND-1))

  [ "$1" = "--" ] && shift

}


# Temple functions, re implement in benchmark if needed

benchmark_suite_config() {
  logger "DEBUG: No specific ${FUNCNAME[0]} defined for $BENCH_SUITE"
}

# Iterate the specified benchmarks in the suite
benchmark_suite_run() {
  logger "INFO: Running $BENCH_SUITE"

  for bench in $BENCH_LIST ; do

    bench_input_dir="$BENCH_SUITE/$bench/input"
    bench_output_dir="$BENCH_SUITE/$bench/output"

    # Prepare run (in case defined)
    function_call "benchmark_prepare_$bench"

    BENCH_CURRENT_NUM_RUN="1" #reset the global counter

    # Iterate at least one time
    while true; do
      [ "$BENCH_NUM_RUNS" ] && logger "INFO: Starting iteration $BENCH_CURRENT_NUM_RUN of $BENCH_NUM_RUNS"
      # Bench Run
      function_call "benchmark_$bench"

      # Validate (eg. teravalidate)
      function_call "benchmark_validate_$bench"

      # Clean-up HDFS space (in case necessary)
      #clean_HDFS "$bench_name" "$BENCH_SUITE"

      # Check if requested to iterate multiple times
      if [ ! "$BENCH_NUM_RUNS" ] || [[ "$BENCH_CURRENT_NUM_RUN" -ge "$BENCH_NUM_RUNS" ]] ; then
        break
      else
        BENCH_CURRENT_NUM_RUN="$((BENCH_CURRENT_NUM_RUN + 1))"
      fi
    done

  done

  logger "INFO: DONE executing $BENCH_SUITE"
}

benchmark_suite_save() {
  logger "DEBUG: No specific ${FUNCNAME[0]} defined for $BENCH_SUITE"
}

benchmark_suite_cleanup() {
  logger "DEBUG: No specific ${FUNCNAME[0]} defined for $BENCH_SUITE"
}

########## END TEMPLATE FUNCTIONS

loggerb(){
  stamp=$(date '+%s')
  echo "${stamp} : $1"
  #log to zabbix
  #zabbix_sender "hadoop.status $stamp $1"
}

get_date_folder(){
  echo "$(date +%Y%m%d_%H%M%S)"
}

get_user_bin_path() {
  echo -e "export PATH=$BENCH_BIN_PATH:\$PATH;"
}

# Wraps supplied command with DSH interactive options
# $1 DSH connection string
# $1 cmd
execute_interactive() {
 local DSH_string="$1"
 local cmd="$2"
 $DSH_string -o -t -o -t -- "stty -echo -onlcr; bash -i -c '$cmd'"
}

# Tests if the supplied hostname can coincides with any node in the cluster
# NOTE: if you cluster doesnt pass this function you should overwrite it with and specific implementation in your benchark defs
# $1 hostname to check
test_in_cluster() {
  local hostname="$1"
  local coincides=1 #return code when not found

  local node_names="$(get_node_names)"

  if [ "$node_names" ] ; then
    for node in $node_names ; do #pad the sequence with 0s
      [[ "$hostname" == "$node"* ]] && coincides=0
    done
  else
    die "Cannot determine nodeNames for cluster $clusterName"
  fi

  return $coincides
}

#$1 port prefix (optional)
get_aloja_dir() {
 if [ "$1" ] ; then
  echo "${BENCH_FOLDER}_$PORT_PREFIX"
 else
  echo "${BENCH_FOLDER}"
 fi
}

# Return a list of
# $1 disk type
get_specified_disks() {
  local disk="$1"
  local dir

  if [ "$disk" == "SSD" ] || [ "$disk" == "HDD" ] || [ "$disk" == "RAM" ] || [ "$disk" == "NFS" ] || [ "$disk" == "NVE" ]; then
    dir="${BENCH_DISKS["$disk"]}"
  elif [[ "$disk" =~ .+[0-9] ]] ; then #if last char is a number

    if [[ "$1" =~ [a-zA-Z]+[0-9][0-9] ]] ; then #if last 2 chars are a numbers
      local disks="${1:(-2)}"
      local disks_type="${1:0:(-2)}"
    elif [[ "$1" =~ .+[0-9] ]] ; then
      local disks="${1:(-1)}"
      local disks_type="${1:0:(-1)}"
    fi

    for disk_number in $(seq 1 $disks) ; do
      dir+="${BENCH_DISKS["${disks_type}${disk_number}"]}\n"
    done
    dir="${dir:0:(-2)}" #remove trailing \n
  else
    die "Incorrect disk specified: $disk"
  fi

  echo -e "$dir"
}

# Returns the tmp disk in cases when mixing local and remote disks (eg. RL1)
#$1 disk type
get_tmp_disk() {
  local dir

  if [ "$1" == "SSD" ] || [ "$1" == "HDD" ] || [ "$1" == "RAM" ]  || [ "$1" == "NFS" ] || [ "$disk" == "NVE" ]; then
    dir="${BENCH_DISKS["$DISK"]}"
  elif [[ "$1" =~ .+[0-9] ]] ; then #if last char is a number

    if [[ "$1" =~ [a-zA-Z]+[0-9][0-9] ]] ; then #if last 2 chars are a numbers
      local disks="${1:(-2)}"
      local disks_type="${1:0:(-2)}"
    elif [[ "$1" =~ .+[0-9] ]] ; then
      local disks="${1:(-1)}"
      local disks_type="${1:0:(-1)}"
    fi

    if [ "$disks_type" == "RL" ] ; then
      dir="${BENCH_DISKS["HDD"]}"
    elif [ "$disks_type" == "HS" ] ; then
      dir="${BENCH_DISKS["SSD"]}"
    elif [ "$disks_type" == "ST" ] ; then
      dir="${BENCH_DISKS["TMP"]}"
    elif [ "$disks_type" == "SR" ] ; then
      dir="${BENCH_DISKS["TMP_RAM"]}"
    elif [ "$disks_type" == "NFS" ] ; then # on NFS use local as /tmp
      dir="${BENCH_DISKS["HDD"]}"
    elif [ "$disks_type" == "NVE" ] ; then
      dir="${BENCH_DISKS["NVE"]}"
    elif [ "$disks_type" == "HN" ] ; then # HDD + NVE cache
      dir="${BENCH_DISKS["NVE"]}"
    else
      dir="${BENCH_DISKS["${disks_type}1"]}"
    fi
  fi

  if [ "$dir" ] ; then
    echo -e "$dir"
  else
    die "Cannot determine tmp disk"
  fi
}

# Simple helper to append the tmp disk path
# $1 disk name
get_all_disks() {
  local disk_name="$1"
  [ ! "$disk_name" ] && die "No disk specified to get_all_disks(). Cannot continue."

  local all_disks="$(get_specified_disks "$disk_name")
$(get_tmp_disk "$disk_name")"

  #remove duplicate lines
  all_disks="$(remove_duplicate_lines "$all_disks")"

  echo -e "$all_disks"
}

# Returns the main benchmark path (useful for multidisk setups)
# $1 disk type
get_initial_disk() {
  if [ "$1" == "SSD" ] || [ "$1" == "HDD" ] || [ "$1" == "RAM" ] || [ "$1" == "NFS" ] || [ "$1" == "NVE" ] ; then
    local dir="${BENCH_DISKS["$DISK"]}"
  elif [[ "$1" =~ [a-zA-Z]+[0-9][0-9] ]] ; then #if last 2 chars are a numbers
    local disks="${1:(-2)}"
    local disks_type="${1:0:(-2)}"
    #set the first dir
    local dir="${BENCH_DISKS["${disks_type}1"]}"
  elif [[ "$1" =~ .+[0-9] ]] ; then #if last char is a number
    local disks="${1:(-1)}"
    local disks_type="${1:0:(-1)}"
    #set the first dir
    local dir="${BENCH_DISKS["${disks_type}1"]}"
  fi
  echo -e "$dir"
}

# Check if the supplied list of variables are not null
# $1 variable or list of vars
test_variable_set() {
  local variables="$1"
  for variable in $variables ; do
    if [ ! "${!variable}" ] ; then
      die "Required variable $variable is null, please check the config"
    fi
  done
}

# Performs some basic validations
# $1 DISK
validate() {
  local disk="$1"

  # First check if some required variables are set
  test_variable_set "BENCH_LOCAL_DIR BENCH_SHARE_DIR"

  if [ "$clusterType" != "PaaS" ]; then
    # Check whether we are in the right cluster
    if [ "${checkClusterMembership}" = "1" ] && ! test_in_cluster "$(hostname)" ; then
      die "host $(hostname) does not belong to specified cluster $clusterName\nMake sure you run this script from within a cluster"
    fi

    if ! inList "$CLUSTER_NETS" "$NET" ; then
      die "Disk type $NET not supported for $clusterName\nSupported: $NET"
    fi

    # Disk validations
    if ! inList "$CLUSTER_DISKS" "$DISK" ; then
      die "Disk type $DISK not supported for $clusterName\nSupported: $CLUSTER_DISKS"
    fi

    # Check that we got the dynamic disk location correctly
    if [ ! "$(get_initial_disk "$disk")" ] ; then
      die "cannot determine $DISK path.  Output: $(get_initial_disk "$disk")"
    fi

    # Iterate all defined and tmp disks to see if we can write to them
    local disks="$(get_all_disks "$disk" )"
    logger "DEBUG: testing write permissions for mount(s) for config $disk: $(nl2char "$disks" " ")"

    for disk_tmp in $disks ; do
      logger "DEBUG: testing write permissions in $disk_tmp"
      local touch_file="$disk_tmp/aloja.touch"
      #if file exists test if we can delete it
      if [ -f "$touch_file" ] ; then
        rm "$touch_file" || die "Cannot delete files in $disk_tmp"
      fi
      touch "$touch_file" || die "Cannot write files in $disk_tmp"
      rm "$touch_file" || die "Cannot delete files in $disk_tmp"
    done
  fi

  if [ "$clusterType" != "PaaS" ]; then
    # Check whether we are in the right cluster
    if [ "${checkClusterMembership}" = "1" ] && ! test_in_cluster "$(hostname)" ; then
      die "host $(hostname) does not belong to specified cluster $clusterName\nMake sure you run this script from within a cluster"
    fi
  else
    logger "INFO: Skipping some validations in PaaS"
  fi
}

# Groups initialization phases
initialize() {

  # initialize cluster node names and connect string
  initialize_node_names
  # set the name for the job run
  set_job_config
  # check if all nodes are up
  test_nodes_connection
  # check if ~/share is correctly mounted
  test_share_dir
}

#old code moved here
# TODO cleanup
initialize_node_names() {
  local extra_node_names

  #For infiniband tests
  if [ "$NET" == "IB" ] ; then
    [ ! "$defaultProvider" == "vagrant" ] && IFACE="ib0" #vagrant we use for testing IB config
    master_name="$(get_master_name_IB)"
    node_names="$(get_node_names_IB)"
    extra_node_names="$(get_extra_node_names)"
  else
    #IFACE should be already setup
    master_name="$(get_master_name)"
    node_names="$(get_node_names)"
    extra_node_names="$(get_extra_node_names)"
  fi

  NUMBER_OF_DATA_NODES="$numberOfNodes"

  if [ ! -z "$LIMIT_DATA_NODES" ] ; then
    node_iteration=0
    for node in $node_names ; do
      if [ ! -z "$nodes_tmp" ] ; then
        node_tmp="$node_tmp\n$node"
      else
        node_tmp="$node"
      fi
      [[ $node_iteration -ge $LIMIT_DATA_NODES ]]  && break;
      node_iteration=$((node_iteration+1))
    done

    node_name=$(echo -e "$nodes_tmp")
    NUMBER_OF_DATA_NODES="$LIMIT_DATA_NODES"
  fi

  DSH="dsh -M -c -m "
  DSH_EXTRA="$DSH"

  DSH_MASTER="dsh -H -m $master_name"

  DSH="$DSH $(nl2char "$node_names" ",") "

  # TODO deprecate this var
  DSH_C="$DSH -c " #concurrent

  DSH_SLAVES="${DSH_C/"$master_name,"/}" #remove master name and trailing coma

  # Instrument additional nodes
  DSH_ALL="$DSH"

  if [ "$extra_node_names" ] ; then
    logger "WARNING: extra nodes requested for instrumentation"
    DSH_EXTRA+=" $(nl2char "$extra_node_names" ",") "
    # TODO temporary to test
    DSH_ALL="${DSH:0:(-1)},$(nl2char "$extra_node_names" ",")"
  fi
}

# Tests cluster nodes for a defined condition
# $1 condition string
# $2 severity of error
test_nodes() {
  local condition="$1"
  local severity="$2"

  [ ! "$severity" ] && severity="ERROR"

  local node_output="$($DSH "$condition && echo '$testKey' 2>&1"|sort )"
  local num_OK="$(echo -e "$node_output"|grep "$testKey"|wc -l)"

  local KO_output="$(echo -e "$node_output"|grep -v "$testKey")"

  local num_nodes="$(get_num_nodes)"
  if (( num_OK != num_nodes )) ; then
    logger "${severity}: Cannot execute: $condition in all nodes. Num OK: $num_OK Num KO: $((num_nodes - num_OK))
DEBUG output:
$node_output"
    return 1
  else
    # all is good
    return 0
  fi
}

# Tests cluster nodes for NOT HAVING a defined condition
# $1 condition string
# $2 severity of error
test_nodes_inverse() {
  local condition="$1"
  local severity="$2"

  [ ! "$severity" ] && severity="ERROR"

  local node_output="$($DSH "$condition && echo '$testKey'"|sort 2>&1)"
  local num_OK="$(echo -e "$node_output"|grep "$testKey"|wc -l)"
  local num_nodes="$(get_num_nodes)"
  if (( num_OK > 0 )) ; then
    logger "${severity}: Found condition: $condition in nodes. Num OK: $num_OK Num KO: $((num_nodes - num_OK))
DEBUG Output:
$node_output"
    return 1
  else
    # all is good
    return 0
  fi
}

# Tests if defined nodes are accesible vis SSH
test_nodes_connection() {
  logger "INFO: Testing connectivity to nodes"
  if test_nodes "hostname" ; then
    logger "INFO: All $(get_num_nodes) nodes are accesible via SSH"
  else
    die "Cannot connect via SSH to all nodes"
  fi
}

# Sends kill signal to processes listening in a particular port for the whole cluster
# $1 TCP port
kill_on_port() {
 local port="$1"
 test_nodes "fuser -k -n tcp $port"
}

# Tries to mount shared folder
# $1 shared folder
mount_share() {
  shared_folder="$1"

  if [ ! "$noSudo" ] ; then
    logger "WARNING: attempting to remount $shared_folder"
    $DSH "
if [ ! -f '$shared_folder/safe_store' ] ; then
  sudo umount -f '$shared_folder';
  sudo mount '$shared_folder';
  sudo mount -a;
fi
"

  fi
}

# Tests if nodes have the shared dir correctly mounted
# $1 if to exit (for retries)
test_share_dir() {
  local no_retry="$1"
  local test_file="$homePrefixAloja/$userAloja/share/safe_store"

  logger "INFO: Testing if ~/share mounted correctly"
  if test_nodes "ls '$test_file'" ; then
    logger "INFO: All $(get_num_nodes) nodes have the ~/share dir correctly mounted"
  else
    if [ "$no_retry" ] ; then
      die "~/share dir not mounted correctly"
    else #try again
      mount_share "$homePrefixAloja/$userAloja/share/"
      test_share_dir "no_retry"
    fi
  fi
}

#old code moved here
# TODO cleanup
set_job_config() {
  # Output directory name
  CONF="${NET}_${DISK}_b${BENCH_SUITE}_S${BENCH_SCALE_FACTOR}_D${NUMBER_OF_DATA_NODES}_${clusterName}"

  JOB_NAME="$(get_date_folder)_$CONF"

  JOB_PATH="$BENCH_SHARE_DIR/jobs_$clusterName/$JOB_NAME"
  #LOG_PATH="$JOB_PATH/log_${JOB_NAME}.log"
  #LOG="2>&1 |tee -a $LOG_PATH"

  #create dir to save files in one host
  $DSH_MASTER "mkdir -p $JOB_PATH;"

  # Automatically log all output to file
  log_all_output "$JOB_PATH/${0##*/}"

  logger "STARTING RUN $JOB_NAME"
  logger "INFO: Job path: $JOB_PATH"
  logger "INFO: Conf: $CONF"
  logger "INFO: Benchmark Suite: $BENCH_SUITE"
  logger "INFO: Benchmarks to execute: $BENCH_LIST"
  logger "DEBUG: DSH: $DSH\n"
  #logger "INFO: DSH_C: $DSH_C"
  #logger "INFO: DSH_SLAVES: $DSH_SLAVES"
}

# Set some OS requirements (e.g., to dissable swapping)
update_OS_config() {
  if [ ! "$noSudo" ] && [ "$EXECUTE_HIBENCH" ]; then
    $DSH "
sudo sysctl -w vm.swappiness=0 > /dev/null;
sudo sysctl vm.panic_on_oom=1 > /dev/null;
sudo sysctl -w fs.file-max=65536 > /dev/null;
sudo service ufw stop 2>&1 > /dev/null;
"
  fi
}

get_apps_path() {
  echo -e "aplic2/apps"
}

# Get the main path for the benchmark
get_local_bench_path() {
  echo -e "$(get_initial_disk "$DISK")/$(get_aloja_dir "$PORT_PREFIX")"
}

get_local_tmp_path() {
  echo -e "$(get_tmp_disk "$DISK")/$(get_aloja_dir "$PORT_PREFIX")"
}

get_local_apps_path() {
  echo -e "$BENCH_LOCAL_DIR/$(get_apps_path)"
}

get_local_configs_path() {
  echo -e "$BENCH_LOCAL_DIR/aplic2/configs"
}

get_base_apps_path() {
  echo -e "$BENCH_SHARE_DIR/$(get_apps_path)"
}

get_base_tarballs_path() {
  echo -e "$BENCH_SHARE_DIR/aplic2/tarballs"
}

get_base_configs_path() {
  echo -e "$ALOJA_REPO_PATH/config/bench/config_files"
}

# Installs binaries and configs
# TODO needs improvement
install_requires() {
  if [ "${#BENCH_REQUIRED_FILES[@]}" ] ; then
    #logger "INFO: Checking if need to download/copy files to node local dirs at: $(get_local_apps_path)"
    for required_file in "${!BENCH_REQUIRED_FILES[@]}" ; do
      logger "INFO: Checking if to download/copy $required_file"
      local base_name="${BENCH_REQUIRED_FILES["$required_file"]##*/}"

      # For github repos, add other exceptions to file names that might repeat here
      if [[ "$base_name" =~ "master."* ]] ; then
        # Use the array key index name
        base_name="${required_file}.${base_name#*.}"
      fi

      # test if we need to download first to share dir
      local test_action="$($DSH_MASTER "[ -f '$(get_base_tarballs_path)/$base_name' ] && echo '$testKey'")"
      if [[ ! "$test_action" == *"$testKey"* ]] ; then
        logger "INFO: Downloading $required_file"
        $DSH_MASTER "
mkdir -p '$(get_base_tarballs_path)' && wget --progress=dot -e dotbytes=10M '${BENCH_REQUIRED_FILES["$required_file"]}' -O '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name'"

        # test if download was succesful
        local test_action="$($DSH_MASTER "[ -f '$(get_base_tarballs_path)/$base_name' ] && echo '$testKey'")"
        if [[ ! "$test_action" == *"$testKey"* ]] ; then
          die "Could not download $required_file from ${BENCH_REQUIRED_FILES["$required_file"]}"
        fi
      fi

      $DSH "
if [ ! -d '$(get_local_apps_path)/$required_file' ] ; then
  mkdir -p '$(get_local_apps_path)/';
  cd '$(get_local_apps_path)/';
  echo 'DEBUG: need to uncompress $(get_base_tarballs_path)/$base_name to $(get_local_apps_path)/$required_file';
  if [[ '$base_name' == *'.tar.gz' || '$base_name' == *'.tgz' ]] ; then
    tar -xzf '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name';
  elif [[ '$base_name' == *'.tar.bz2' ]] ; then
    tar -xjf '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name';
  elif [[ '$base_name' == *'.zip' ]] ; then
    unzip -q -o '$(get_base_tarballs_path)/$base_name' || rm '$(get_base_tarballs_path)/$base_name';
  else
    echo 'ERROR: unknown file extension for $base_name';
  fi
else
  echo 'DEBUG: local dir $(get_local_apps_path)/$required_file exists'
fi
"
    done
  else
    logger "INFO: No required files to download/copy specified"
  fi
}

# Installs binaries and configs
# TODO needs merging with install packages
install_bench_packages() {
  if [ "$BENCH_REQUIRED_PACKAGES" ] ; then
    if [[ "$vmOSType" == "Ubuntu" || "$vmOSType" == "Debian" ]] ; then
      logger "INFO: Testing to see if packages: $BENCH_REQUIRED_PACKAGES are installed"
      if ! test_nodes "dpkg -s $BENCH_REQUIRED_PACKAGES" ; then
        if [ ! "$noSudo" ] ; then
          logger "INFO: Attempting to install: $BENCH_REQUIRED_PACKAGES"
          if ! test_nodes "dpkg -s $BENCH_REQUIRED_PACKAGES 2> /dev/null" ; then
            $DSH "sudo apt-get update;
export DEBIAN_FRONTEND=noninteractive;
sudo apt-get -o Dpkg::Options::='--force-confold' install -y --force-yes $BENCH_REQUIRED_PACKAGES"

            logger "WARNING: failed to install one or more packages: $BENCH_REQUIRED_PACKAGES"
          fi
        else
          logger "WARNING: no sudo selected, cannot install needed packages in: $BENCH_REQUIRED_PACKAGES"
        fi
      else
        logger "INFO: Required packages bench: $BENCH_REQUIRED_PACKAGES are correctly installed"
      fi
    elif [[ "$vmOSType" == "Fedora" || "$vmOSType" == "RHEL" || "$vmOSType" == "CentOS" ]] ; then
      for package in $BENCH_REQUIRED_PACKAGES ; do
        if [ ! "$($DSH "which $package; 2> /dev/null")" ] ; then
          logger "INFO: Attempting to install: $BENCH_REQUIRED_PACKAGES"
          $DSH "sudo yum install -y $BENCH_REQUIRED_PACKAGES"
          break
        fi
      done
    else
      logger "WARNING: not a Debian or Fedora based system, not checking if to install packages: $BENCH_REQUIRED_PACKAGES"
    fi
  fi
}


# Rsyncs specified config folders in aplic2/configs/
install_configs() {
  if [ "$BENCH_CONFIG_FOLDERS" ] ; then
    for config_folder in $BENCH_CONFIG_FOLDERS ; do
      local full_config_folder_path="$(get_base_configs_path)/$config_folder"
      if [ -d "$full_config_folder_path" ] ; then
        logger "INFO: Synching configs from $config_folder"
        $DSH "rsync -aur --delete '$full_config_folder_path' '$(get_local_configs_path)' "
      else
        die "Cannot find config folder in $full_config_folder_path"
      fi
    done
  else
    logger "DEBUG: No config folder specified to copy"
  fi
}

install_files() {
  install_requires
  install_configs
  install_bench_packages
}

check_aplic_updates() {
  #only copy files if version has changed (to save time)
  logger "INFO: Checking if to generate source dirs $BENCH_SHARE_DIR/aplic/aplic_version == $BENCH_SOURCE_DIR/aplic_version"
  for node in $node_names ; do
    logger "INFO:  for host $node"
    if [ "$(ssh "$node" "[ "\$\(cat $BENCH_SHARE_DIR/aplic/aplic_version\)" == "\$\(cat $BENCH_SOURCE_DIR/aplic_version 2\> /dev/null \)" ] && echo 'OK' || echo 'KO'" )" != "OK" ] ; then
      logger "INFO: At least host $node did not have source dirs. Generating source dirs for ALL hosts"

      if [ ! "$(ssh "$node" "[ -d \"$BENCH_SHARE_DIR/aplic\" ] && echo 'OK' || echo 'KO'" )" != "OK" ] ; then
        #logger "Downloading initial aplic dir from dropbox"
        #$DSH "wget -nv https://www.dropbox.com/s/ywxqsfs784sk3e4/aplic.tar.bz2?dl=1 -O $BASE_DIR/aplic.tar.bz2"

        $DSH "rsync -aur --force $BENCH_SHARE_DIR/aplic.tar.bz2 /tmp/"

        logger "INFO: Uncompressing aplic"
        $DSH  "mkdir -p $BENCH_SOURCE_DIR/; cd $BENCH_SOURCE_DIR/../; tar -C $BENCH_SOURCE_DIR/../ -jxf /tmp/aplic.tar.bz2; "  #rm aplic.tar.bz2;
      fi

      logger "Rsynching files"
      $DSH "mkdir -p $BENCH_SOURCE_DIR; rsync -aur --force $BENCH_SHARE_DIR/aplic/* $BENCH_SOURCE_DIR/"
      break #dont need to check after one is missing
    else
      logger "INFO:  Host $node up to date"
    fi
  done

  #if [ "$(cat $BENCH_SHARE_DIR/aplic/aplic_version)" != "$(cat $BENCH_SOURCE_DIR/aplic_version)" ] ; then
  #  logger "INFO: Generating source dirs"
  #  $DSH "mkdir -p $BENCH_SOURCE_DIR; cp -ru $BENCH_SHARE_DIR/aplic/* $BENCH_SOURCE_DIR/"
  #  #$DSH "cp -ru $BENCH_SOURCE_DIR/${HADOOP_VERSION}-home $BENCH_SOURCE_DIR/${HADOOP_VERSION}" #rm -rf $BENCH_SOURCE_DIR/${HADOOP_VERSION};
  #elsefi
  #  logger "INFO: Source dirs up to date"
  #fi

}

# Exports a var and path to the cluster
# $1 varname
# $2 path
export_var_path() {
  : # WiP
}

zabbix_sender(){
  :
  #echo "al-1001 $1" | /home/pristine/share/aplic/zabbix/bin/zabbix_sender -c /home/pristine/share/aplic/zabbix/conf/zabbix_agentd_az.conf -T -i - 2>&1 > /dev/null
  #>> $LOG_PATH

##For zabbix monitoring make sure IB ports are available
#ssh_tunnel="ssh -N -L al-1001:30070:al-1001-ib0:30070 -L al-1001:30030:al-1001-ib0:30030 al-1001"
##first make sure we kill any previous, even if we don't need it
#pkill -f "ssh -N -L"
##"$ssh_tunnel"
#
#if [ "${NET}" == "IB" ] ; then
#  $ssh_tunnel &
#fi

}

# Copies specified perf mon binaries to bench path, so that they can be started
# and specially killed easily
set_monit_binaries() {
  if [ "$BENCH_PERF_MONITORS" ] ; then
    local perf_mon_bin_path
    local perf_mon_bench_path="$(get_local_bench_path)/aplic"

    if [ "$vmType" != "windows" ] ; then
      for perf_mon in $BENCH_PERF_MONITORS ; do

        if ! inList "$BENCH_PERF_NON_BINARY" "$perf_mon" ; then
          logger "INFO: Setting up performance monitor: $perf_mon"

          # Get the path of file from the aloja repo
          if [[ "$perf_mon" == "cachestat" || "$perf_mon" == "drop_cache" ]] ; then
            perf_mon_bin_path="$ALOJA_REPO_PATH/aloja-tools/src/${perf_mon}.sh"

            if [ "$noSudo" ] ; then
              logger "WARNING: $perf_mon requires sudo, skipping setting it up."
              continue
            fi
          # JavaStat actually uses pidstat
          elif [ "$perf_mon" == "JavaStat" ] ; then
            perf_mon_bin_path="$($DSH_MASTER "$(get_user_bin_path) which 'pidstat'")"
          # Get it from the user path (normal case)
          else
            # we need to include our custom bin path as SSH is not interactive
            perf_mon_bin_path="$($DSH_MASTER "$(get_user_bin_path) which '$perf_mon'")"
          fi

          if [ -f "$perf_mon_bin_path" ] ; then
            log_DEBUG "Copying $perf_mon binary from: $perf_mon_bin_path to $perf_mon_bench_path"
            $DSH "mkdir -p '$perf_mon_bench_path'; cp '$perf_mon_bin_path' '$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX';"

            if [ "$(get_extra_node_names)" ] ; then
              $DSH_EXTRA "mkdir -p '$(get_extra_node_folder)/aplic'; cp '$perf_mon_bin_path' '$(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX'"
            fi
          else
            log_ERR "Cannot find $perf_mon binary on the system at: $perf_mon_bin_path"
            log_DEBUG "Perf monitor bin path: $($DSH_MASTER 'ls "$perf_mon_bin_path" 2>&1')"
          fi
        else
          logger "INFO: Setting up script-style perfomance monitor: $perf_mon"
        fi
      done
    else
      logger "WARNING: no extra perf monitors set for Windows"
    fi
  else
    logger "WARNING: No peformance monitors (e.g., vmstat) have been selected"
  fi
}

# Before starting monitors always check if they are already running
start_monit() {
  restart_monit
}

# Stops monitors (if any) and starts them
restart_monit(){
  if [ "$BENCH_PERF_MONITORS" ] ; then
    local perf_mon_bin_path
    local perf_mon_bench_path="$(get_local_bench_path)/aplic"

    if [ "$vmType" != "windows" ] ; then
      logger "INFO: Restarting perf monit"
      stop_monit #in case there is any running

      # Make sure we clean monits on abnormal exit
      if [ "$BENCH_PERF_MONITORS" ] ; then
        update_traps "stop_monit;" "update_logger"
      fi

      for perf_mon in $BENCH_PERF_MONITORS ; do
        run_monit "$perf_mon"
      done
      #logger "DEBUG: perf monitors ready"
    fi
  fi
}

# Starts the specified perf_mon
# They execute in background (&) to start them as close as possible in time
# $1 perf_mon
run_monit() {
  local perf_mon="$1"
  local perf_mon_bin="$(get_local_bench_path)/aplic/${perf_mon}_$PORT_PREFIX"

  if [ "$perf_mon" == "sar" ] ; then
    if [ "$clusterType" != "PaaS" ]; then
      $DSH "$(get_user_bin_path) $perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -o $(get_local_bench_path)/sar-\$(hostname).sar $BENCH_PERF_INTERVAL >/dev/null &" & #2>&1

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "$(get_user_bin_path) $(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -o $(get_extra_node_folder)/sar-\$(hostname).sar $BENCH_PERF_INTERVAL >/dev/null &" & #2>&1
      fi
    else
      $DSH "$(get_user_bin_path) sar -o $(get_local_bench_path)/sar-\$(hostname).sar $BENCH_PERF_INTERVAL >/dev/null &" & # 2>&1
    fi
  elif [ "$perf_mon" == "vmstat" ] ; then
    if [ "$clusterType" != "PaaS" ]; then
      $DSH "$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -n $BENCH_PERF_INTERVAL >> $(get_local_bench_path)/vmstat-\$(hostname).log &" &

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "$(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -n $BENCH_PERF_INTERVAL >> $(get_extra_node_folder)/vmstat-\$(hostname).log &" &
      fi
    else
      $DSH "vmstat -n $BENCH_PERF_INTERVAL >> $(get_local_bench_path)/vmstat-\$(hostname).log &" &
    fi
  # For iostat use PAT's syntax
  elif [ "$perf_mon" == "iostat" ] ; then
    if [ "$clusterType" != "PaaS" ]; then
      $DSH "$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -x -k -y -d $BENCH_PERF_INTERVAL | awk -v host=\$(hostname) '(!/^$/){now=strftime(\"%s \");if(/Device:/){print \"HostName\",\"TimeStamp\", \$0} else{ if(\$0 && !/Linux/) print host, now \$0}}; fflush()' >> $(get_local_bench_path)/iostat-\$(hostname).log &" &

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "$(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -x -k -y -d $BENCH_PERF_INTERVAL | awk -v host=\$(hostname) '(!/^$/){now=strftime(\"%s \");if(/Device:/){print \"HostName\",\"TimeStamp\", \$0} else{ if(\$0 && !/Linux/) print host, now \$0}}; fflush()' >> $(get_extra_node_folder)/iostat-\$(hostname).log &"  &
      fi
    else
      $DSH "iostat -x -k -y -d $BENCH_PERF_INTERVAL | awk -v host=\$(hostname) '(!/^$/){now=strftime(\"%s \");if(/Device:/){print \"HostName\",\"TimeStamp\", \$0} else{ if(\$0 && !/Linux/) print host, now \$0}}; fflush()' >> $(get_local_bench_path)/iostat-\$(hostname).log &" &
      #iostat -x -k -d $SAMPLING_INTERVAL
    fi
  # To count Java processes (for PAT export)
  elif [ "$perf_mon" == "MapRed" ] ; then
    $DSH "
(
echo MapCount ReduceCount ContainerCount TezCount JavaCount ProcCount
while :
do
  processes=\"\$(ps fauxwww)\";
  echo \"\$(echo -e \"\$processes\"|grep [j]ava|grep _m_ |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava| grep _r_ |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava| grep container_ |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava| grep container_|grep tez |wc -l ) \$(echo -e \"\$processes\"|grep [j]ava|wc -l ) \$(echo -e \"\$processes\" |wc -l )\"
  sleep $BENCH_PERF_INTERVAL
done
) | awk -v host=\$(hostname) '(!/^\$/){now=strftime(\"%s \");if(\$0 && !/Linux/) if (/MapCount/){print \"HostName\",\"TimeStamp\",\$0} else {print host,now \$0}}; fflush()' > $(get_local_bench_path)/MapRed-\$(hostname).log &" &

  # To count Java processes (for PAT export)
  elif [ "$perf_mon" == "JavaStat" ] ; then
    local pidstat_cmd="java"
    $DSH "
$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -rudh -p ALL -C init | awk -v host=\$(hostname) '(/Time/){\$1=\$2=\"\"; print \"HostName\",\"TimeStamp\", \$0}; fflush()' > $(get_local_bench_path)/JavaStat-\$(hostname).log
$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -rudh -p ALL -C $pidstat_cmd $(( $BENCH_PERF_INTERVAL + 4 )) | awk -v cmd='$pidstat_cmd' -v host=\$(hostname) '(!/^\$/ && !/Time/ && !/CPU/){if (\$NF == cmd){now=strftime(\"%s\"); \$1=\"\"; print host, now, \$0}; fflush()}' >> $(get_local_bench_path)/JavaStat-\$(hostname).log &
" &

  # iotop, requires sudo and interval only 1 sec supported
  elif [ "$perf_mon" == "iotop" ] ; then
    if [ -z "$noSudo" ] || [ "$BENCH_PERF_INTERVAL" == "1" ]; then
      if [ "$clusterType" != "PaaS" ]; then
        local iotop_log="$(get_local_bench_path)/iotop-\$(hostname).log"
        $DSH "touch $iotop_log; sudo $perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -btoqqk >> $iotop_log &" &

        if [ "$(get_extra_node_names)" ] ; then
          $DSH_EXTRA "touch $(get_extra_node_folder)/iotop-\$(hostname).log; sudo $(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -btoqqk >> $iotop_log &" &
        fi
      else
        $DSH "touch $iotop_log; sudo iotop  >> $iotop_log &" &
      fi
    else
      logger "WARNING: iotop requires root and sudo is disabled for cluster OR BENCH_PERF_INTERVAL != 1 (set to: $BENCH_PERF_INTERVAL), skipping..."
    fi
  # dstat
  elif [ "$perf_mon" == "dstat" ] ; then
     # Removed for ubuntu
     # -T --cpu-adv --top-cpu-adv -l -d --aio --disk-avgqu --disk-avgrq --disk-svctm --disk-tps --disk-util --disk-wait --top-bio-adv --top-io-adv --md-status -n --net-packets  -gimprsy --cpu-use --fs --top-int --top-latency -ipc -lock --mem-adv --top-mem --raw --unix --vm-adv --bits --nocolor --noheader --profile --power --proc-count --thermal --noheaders
     #--cpu-adv --disk-avgqu --disk-avgrq --disk-svctm --disk-wait --md-status  --cpu-use --mem-adv --vm-adv --bits --thermal --top-io-adv --top-int

    if [ "$clusterType" != "PaaS" ]; then
      local dstat_log="$(get_local_bench_path)/dstat-\$(hostname).log"
      $DSH "$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -T --cpu --top-cpu-adv -l -d --aio --disk-tps --disk-util --top-bio-adv --top-io-adv -n --net-packets  -gimprsy --fs --top-int --top-latency -ipc -lock --top-mem --raw --unix --nocolor --noheader --profile --power --proc-count --noheaders  $BENCH_PERF_INTERVAL >> $dstat_log &" &
    else
      $DSH "dstat -T --cpu --top-cpu-adv -l -d --aio --disk-tps --disk-util --top-bio-adv -n --net-packets  -gimprsy --fs --top-latency -ipc -lock --top-mem --raw --unix --nocolor --noheader --profile --power --proc-count --noheaders  $BENCH_PERF_INTERVAL >> $dstat_log &" &
    fi
  # perf
  elif [ "$perf_mon" == "perf" ] ; then

    # Enable CPU trace data (if we have root)
    if [ -z "$noSudo" ] ; then
      $DSH "sudo echo '0' > /proc/sys/kernel/perf_event_paranoid"
    fi
    # TODO: https://github.com/intel-hadoop/PAT/blob/master/PAT/WORKER_scripts/instruments/perf

  # cachestat
  elif [ "$perf_mon" == "cachestat" ] ; then
      $DSH "sudo $perf_mon_bench_path/${perf_mon}_$PORT_PREFIX -n -t $BENCH_PERF_INTERVAL > $(get_local_bench_path)/cachestat-\$(hostname).log &" & #2>&1

      if [ "$(get_extra_node_names)" ] ; then
        $DSH_EXTRA "sudo $(get_extra_node_folder)/aplic/${perf_mon}_$PORT_PREFIX -n -t $BENCH_PERF_INTERVAL > $(get_extra_node_folder)/cachestat-\$(hostname).log &" & #2>&1
      fi
  # drop_cache
  elif [ "$perf_mon" == "drop_cache" ] ; then
      $DSH "$perf_mon_bench_path/${perf_mon}_$PORT_PREFIX $BENCH_PERF_INTERVAL 3 1 > $(get_local_bench_path)/drop_cache-\$(hostname).log &" & #2>&1

      if [ "$(get_extra_node_names)" ] ; then
        : # do nothing
      fi
  else
    die "Specified perf mon $perf_mon not implemented"
  fi


  wait #for the bg processes

  # BWM not used any more
  #$DSH_C "$bwm -o csv -I bond0,eth0,eth1,eth2,eth3,ib0,ib1 -u bytes -t 1000 >> $(get_local_bench_path)/bwm-\$(hostname).log &"
}

# Kill possibly running perf mons
stop_monit(){
  if [ "$BENCH_PERF_MONITORS" ] ; then
    if [ "$vmType" != "windows" ]; then
      logger "INFO: Stoping monit (in case necessary). Monitors: $BENCH_PERF_MONITORS"
      for perf_mon in $BENCH_PERF_MONITORS ; do

        local requires_sudo=""
        if [[ "$perf_mon" == "iotop" || "$perf_mon" == "cachestat" || "$perf_mon" == "drop_cache" ]] ; then
          requires_sudo="sudo"
        fi

        local perf_mon_bin="$(get_local_bench_path)/aplic/${perf_mon}_$PORT_PREFIX"

        if ! inList "$BENCH_PERF_NON_BINARY" "$perf_mon" ; then
          $DSH "$requires_sudo pkill -9 -f '[${perf_mon_bin:0:1}]${perf_mon_bin:1}' 2> /dev/null" #& # [] for it not to match itself in ssh
        elif [ "$perf_mon" == "MapRed" ] ; then
          $DSH "$requires_sudo pkill -9 -f [M]apCount" # [] for it not to match itself in ssh
        elif [ "$perf_mon" == "JavaStat" ] ; then
          $DSH "$requires_sudo pkill -9 pidstat" # TODO improve to use custom naming
        fi

        if [ "$(get_extra_node_names)" ] ; then
          $DSH_EXTRA "$requires_sudo pkill -9 -f '[${perf_mon_bin:0:1}]${perf_mon_bin:1}' 2> /dev/null" #&
        fi

        # TODO this is something temporal for PaaS clusters
        if [ "$clusterType" == "PaaS" ]; then
          $DSH "killall -9 sadc; killall -9 vmstat; killall -9 iostat; killall -9 pidstat; pkill -9 -f [M]apCount; pgrep -f '[M]apCount'|xargs kill -9; $requires_sudo kill -9 iotop 2> /dev/null"  2> /dev/null
        fi
      done
      #logger "DEBUG: perf monitors ready"
    fi
  fi

  #wait #for the bg processes
}

{'fraction_non_alphanumeric': 0.15625694598799733, 'fraction_numerical': 0.009897624308669901, 'mean_word_length': 3.175802576981874, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 159, 'https://': 4, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5429542', 'n_tokens_mistral': 28965, 'n_tokens_neox': 27961, 'n_words': 7139}