Release 2.5.0

Merge Release 2.5.0

Author: Sean Smith

.flake8

@@ -1,17 +1,13 @@
 [flake8]
-#ignore =
-#    D105, # Missing docstring in magic method
-#    D100, # Missing docstring in public module
-#    D104, # Missing docstring in public package
-#    D107, # Missing docstring in __init__
-#    W503, # line break before binary operator => Conflicts with black style.
-# D103 Missing docstring in public function
-# E402 module level import not at top of file
-# D101 Missing docstring in public class
-# D102 Missing docstring in public method
-# D205 1 blank line required between summary line and description
-# D400 First line should end with a period
-# D401 First line should be in imperative mood
+ignore =
+    D105, # Missing docstring in magic method
+    D100, # Missing docstring in public module
+    D101, # Missing docstring in public class
+    D102, # Missing docstring in public method
+    D103, # Missing docstring in public function
+    D104, # Missing docstring in public package
+    D107, # Missing docstring in __init__
+    W503, # line break before binary operator => Conflicts with black style.
 exclude =
     .tox,
     .git,

CHANGELOG.md

@@ -3,6 +3,40 @@ aws-parallelcluster-node CHANGELOG
 
 This file is used to list changes made in each version of the aws-parallelcluster-node package.
 
+2.5.0
+-----
+
+**ENHANCEMENTS**
+- Slurm:
+  - Add support for scheduling with GPU options. Currently supports the following GPU-related options: `—G/——gpus,
+    ——gpus-per-task, ——gpus-per-node, ——gres=gpu, ——cpus-per-gpu`.
+  - Add gres.conf and slurm_parallelcluster_gres.conf in order to enable GPU options. slurm_parallelcluster_gres.conf
+    is automatically generated by node daemon and contains GPU information from compute instances. If need to specify
+    additional GRES options manually, please modify gres.conf and avoid changing slurm_parallelcluster_gres.conf when
+    possible.
+  - Integrated GPU requirements into scaling logic, cluster will scale automatically to satisfy GPU/CPU requirements
+    for pending jobs. When submitting GPU jobs, CPU/node/task information is not required but preferred in order to
+    avoid ambiguity. If only GPU requirements are specified, cluster will scale up to the minimum number of nodes
+    required to satisfy all GPU requirements.
+  - Slurm daemons will now keep running when cluster is stopped for better stability. However, it is not recommended
+    to submit jobs when the cluster is stopped.
+  - Change jobwatcher logic to consider both GPU and CPU when making scaling decision for slurm jobs. In general,
+    cluster will scale up to the minimum number of nodes needed to satisfy all GPU/CPU requirements.
+- Reduce number of calls to ASG in nodewatcher to avoid throttling, especially at cluster scale-down.
+
+**CHANGES**
+- Increase max number of SQS messages that can be processed by sqswatcher in a single batch from 50 to 200. This
+  improves the scaling time especially with increased ASG launch rates.
+- Increase faulty node termination timeout from 1 minute to 5 in order to give some additional time to the scheduler
+  to recover when under heavy load.
+
+**BUG FIXES**
+- Fix jobwatcher behaviour that was marking nodes locked by the nodewatcher as busy even if they had been removed
+  already from the ASG Desired count. This was causing, in rare circumstances, a cluster overscaling.
+- Better handling of errors occurred when adding/removing nodes from the scheduler config.
+- Fix bug that was causing failures in sqswatcher when ADD and REMOVE event for the same host are fetched together.
+
+
 2.4.1
 -----
 

setup.py

@@ -1,7 +1,7 @@
 # Copyright 2013-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 #
-# Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file except in compliance with the
-# License. A copy of the License is located at
+# Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file except in compliance with
+# the License. A copy of the License is located at
 #
 # http://aws.amazon.com/apache2.0/
 #
@@ -10,7 +10,6 @@
 # limitations under the License.
 
 import os
-import sys
 
 from setuptools import find_packages, setup
 
@@ -29,7 +28,7 @@ def read(fname):
     "nodewatcher = nodewatcher.nodewatcher:main",
     "jobwatcher = jobwatcher.jobwatcher:main",
 ]
-version = "2.4.1"
+version = "2.5.0"
 requires = ["requests>=2.21.0", "boto3>=1.7.55", "retrying>=1.3.3", "configparser>=3.7.4", "paramiko>=2.4.2"]
 
 setup(

src/common/remote_command_executor.py

@@ -18,7 +18,6 @@
 
 from retrying import retry
 
-from common.time_utils import seconds
 from paramiko import AutoAddPolicy, SSHClient
 
 RemoteCommandResult = collections.namedtuple("RemoteCommandResult", ["return_code", "stdout", "stderr"])
@@ -33,6 +32,8 @@ class RemoteCommandExecutionError(Exception):
 class RemoteCommandExecutor:
     """Execute remote commands."""
 
+    DEFAULT_TIMEOUT = 10
+
     def __init__(self, hostname, user, ssh_key_file=None):
         try:
             if not ssh_key_file:
@@ -53,17 +54,19 @@ def __del__(self):
             # Catch all exceptions if we fail to close the clients
             logging.warning("Exception raised when closing remote clients: {0}".format(e))
 
-    def run_remote_command(self, command, timeout=seconds(5), log_error=True, fail_on_error=True):
+    def run_remote_command(self, command, timeout=DEFAULT_TIMEOUT, log_error=True, fail_on_error=True):
         """
         Execute remote command on the configured host.
 
         :param command: command to execute.
+        :param timeout: timeout for command execution in sec
         :param log_error: log errors.
+        :param fail_on_error: raise Exception on command execution failures
         :return: result of the execution.
         """
         if isinstance(command, list):
             command = " ".join(command)
-        logging.info("Executing remote command command on {0}: {1}".format(self.__user_at_hostname, command))
+        logging.info("Executing remote command on {0}: {1}".format(self.__user_at_hostname, command))
         result = None
         try:
             stdin, stdout, stderr = self.__ssh_client.exec_command(command, get_pty=True)

src/common/schedulers/converters.py

@@ -16,7 +16,7 @@
 
 def from_xml_to_obj(xml, obj_type):
     """
-    Maps a given xml document into a python object.
+    Map a given xml document into a python object.
 
     The python object you want to map the xml into needs to define a MAPPINGS dictionary which declare how
     to map each tag of the xml doc into the object itself.
@@ -55,7 +55,7 @@ def from_xml_to_obj(xml, obj_type):
 
 def from_table_to_obj_list(table, obj_type, separator="|"):
     """
-    Maps a given tabular output into a python object.
+    Map a given tabular output into a python object.
 
     The python object you want to map the table into needs to define a MAPPINGS dictionary which declare how
     to map each row element into the object itself.

src/common/schedulers/sge_commands.py

@@ -56,7 +56,10 @@
 # or some combination thereof.
 # Refer to qstat man page for additional details.
 # o(rphaned) is not considered as busy since we assume a node in orphaned state is not present in ASG anymore
-SGE_BUSY_STATES = ["u", "C", "s", "d", "D", "E", "P"]
+SGE_BUSY_STATES = ["u", "C", "s", "D", "E", "P"]
+
+# This state is set by nodewatcher when the node is locked and is being terminated.
+SGE_DISABLED_STATE = "d"
 
 # If an o(rphaned) state is displayed for a queue instance, it indicates that the queue instance is no longer demanded
 # by the current cluster queue configuration or the host group configuration. The queue instance is kept because jobs
@@ -133,10 +136,11 @@ def remove_hosts_from_queue(hosts):
 def install_sge_on_compute_nodes(hosts, cluster_user):
     """Start sge on compute nodes in parallel."""
     command = (
-        "sudo sh -c 'cd {0} && {0}/inst_sge -noremote -x -auto /opt/parallelcluster/templates/sge/sge_inst.conf'"
+        "sudo sh -c 'ps aux | grep [s]ge_execd || "
+        "(cd {0} && {0}/inst_sge -noremote -x -auto /opt/parallelcluster/templates/sge/sge_inst.conf)'"
     ).format(sge.SGE_ROOT)
     hostnames = [host.hostname for host in hosts]
-    result = RemoteCommandExecutor.run_remote_command_on_multiple_hosts(command, hostnames, cluster_user)
+    result = RemoteCommandExecutor.run_remote_command_on_multiple_hosts(command, hostnames, cluster_user, timeout=20)
 
     succeeded_hosts = []
     for host in hosts:
@@ -206,6 +210,7 @@ def get_jobs_info(hostname_filter=None, job_state_filter=None):
 def get_pending_jobs_info(max_slots_filter=None, skip_if_state=None):
     """
     Retrieve the list of pending jobs.
+
     :param max_slots_filter: discard jobs that require a number of slots bigger than the given value
     :param skip_if_state: discard jobs that are in the given state
     :return: the list of filtered pending jos.