Merge pull request #153 from aws/develop

Merge Release 2.4.0

Author: lukeseawalker

.flake8

@@ -0,0 +1,31 @@
+[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
+exclude =
+    .tox,
+    .git,
+    .pytest_cache,
+    docs/source/conf.py,
+    build,
+    dist,
+    tests/fixtures/*,
+    *.pyc,
+    *.egg-info,
+    .cache,
+    .eggs
+max-complexity = 10
+max-line-length = 120
+import-order-style = google
+application-import-names = flake8
+format = ${cyan}%(path)s${reset}:${yellow_bold}%(row)d${reset}:${green_bold}%(col)d${reset}: ${red_bold}%(code)s${reset} %(text)s

.gitignore

@@ -8,3 +8,4 @@ build/
 *.egg-info/
 .idea/
 .DS_Store
+.tox/

.isort.cfg

@@ -0,0 +1,13 @@
+[settings]
+line_length=120
+known_future_library=future
+known_third_party=boto3,botocore,awscli,tabulate,argparse,configparser,pytest,pytest,pytest-datadir,pytest-html,pytest-rerunfailures,pytest-xdist,argparse,retrying,junitparser,Jinja2
+# 3 - Vertical Hanging Indent
+# from third_party import (
+#     lib1,
+#     lib2,
+#     lib3,
+#     lib4,
+# )
+multi_line_output=3
+include_trailing_comma=true

.travis.yml

@@ -8,12 +8,15 @@ python:
   - "3.6"
 
 install:
-  - if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then xargs -L 1 pip install < requirements26.txt; fi
-  - if [[ $TRAVIS_PYTHON_VERSION != '2.6' ]]; then pip install -r requirements.txt; fi
-  - pip install -e .
+  - pip install tox-travis
 
 sudo: false
 
-script:
-  - sh tests/test.sh
-  - python jobwatcher/plugins/unittests.py
+matrix:
+  include:
+    - name: Code Checks
+      python: 3.6
+      stage: linters
+      env: TOXENV=code-linters
+
+script: tox

CHANGELOG.md

@@ -3,6 +3,39 @@ aws-parallelcluster-node CHANGELOG
 
 This file is used to list changes made in each version of the aws-parallelcluster-node package.
 
+2.4.0
+-----
+
+**ENHANCEMENTS**
+- Dynamically fetch compute instance type and cluster size in order to support updates
+- SGE:
+  - process nodes added to or removed from the cluster in batches in order to speed up cluster scaling.
+  - scale up only if required slots/nodes can be satisfied
+  - scale down if pending jobs have unsatisfiable CPU/nodes requirements
+  - add support for jobs in hold/suspended state (this includes job dependencies)
+  - automatically terminate and replace faulty or unresponsive compute nodes
+  - add retries in case of failures when adding or removing nodes
+- Slurm:
+  - scale up only if required slots/nodes can be satisfied
+  - scale down if pending jobs have unsatisfiable CPU/nodes requirements
+  - automatically terminate and replace faulty or unresponsive compute nodes
+- Dump logs of replaced failing compute nodes to shared home directory
+
+**CHANGES**
+- SQS messages that fail to be processed are re-queued only 3 times and not forever
+- Reset idletime to 0 when the host becomes essential for the cluster (because of min size of ASG or because there are
+  pending jobs in the scheduler queue)
+- SGE: a node is considered as busy when in one of the following states "u", "C", "s", "d", "D", "E", "P", "o".
+  This allows a quick replacement of the node without waiting for the `nodewatcher` to terminate it.
+
+**BUG FIXES**
+- Slurm: add "BeginTime", "NodeDown", "Priority" and "ReqNodeNotAvail" to the pending reasons that trigger
+  a cluster scaling
+- Add a timeout on remote commands execution so that the daemons are not stuck if the compute node is unresponsive
+- Fix an edge case that was causing the `nodewatcher` to hang forever in case the node had become essential to the
+  cluster during a call to `self_terminate`.
+
+
 2.3.1
 -----
 

common/remote_command_executor.py

@@ -0,0 +1,130 @@
+# Copyright 2019 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
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "LICENSE.txt" file accompanying this file.
+# This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, express or implied.
+# See the License for the specific language governing permissions and limitations under the License.
+import collections
+import logging
+import os
+import time
+from math import ceil
+from multiprocessing import Pool
+
+from retrying import retry
+
+from common.time_utils import seconds
+from paramiko import AutoAddPolicy, SSHClient
+
+RemoteCommandResult = collections.namedtuple("RemoteCommandResult", ["return_code", "stdout", "stderr"])
+
+
+class RemoteCommandExecutionError(Exception):
+    """Signal a failure in remote command execution."""
+
+    pass
+
+
+class RemoteCommandExecutor:
+    """Execute remote commands."""
+
+    def __init__(self, hostname, user, ssh_key_file=None):
+        try:
+            if not ssh_key_file:
+                ssh_key_file = os.path.expanduser("~" + user) + "/.ssh/id_rsa"
+            self.__ssh_client = SSHClient()
+            self.__ssh_client.load_system_host_keys()
+            self.__ssh_client.set_missing_host_key_policy(AutoAddPolicy())
+            self.__ssh_client.connect(hostname=hostname, username=user, key_filename=ssh_key_file)
+            self.__user_at_hostname = "{0}@{1}".format(user, hostname)
+        except Exception as e:
+            logging.error("Failed when connecting to host %s with error: %s", hostname, e)
+            raise
+
+    def __del__(self):
+        try:
+            self.__ssh_client.close()
+        except Exception as e:
+            # 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):
+        """
+        Execute remote command on the configured host.
+
+        :param command: command to execute.
+        :param log_error: log errors.
+        :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))
+        result = None
+        try:
+            stdin, stdout, stderr = self.__ssh_client.exec_command(command, get_pty=True)
+            self._wait_for_command_execution(timeout, stdout)
+            result = RemoteCommandResult(
+                return_code=stdout.channel.recv_exit_status(),
+                stdout="\n".join(stdout.read().decode().splitlines()),
+                stderr="\n".join(stderr.read().decode().splitlines()),
+            )
+            if result.return_code != 0 and fail_on_error:
+                raise RemoteCommandExecutionError(result)
+            return result
+        except Exception:
+            if log_error and result:
+                logging.error(
+                    "Command {0} failed with error:\n{1}\nand output:\n{2}".format(
+                        command, result.stderr, result.stdout
+                    )
+                )
+            raise
+
+    @staticmethod
+    def _wait_for_command_execution(timeout, stdout):
+        # Using the non-blocking exit_status_ready to avoid being stuck forever on recv_exit_status
+        # especially when a compute node is terminated during this operation
+        while timeout > 0 and not stdout.channel.exit_status_ready():
+            timeout = timeout - 1
+            time.sleep(1)
+        if not stdout.channel.exit_status_ready():
+            raise RemoteCommandExecutionError("Timeout occurred when executing remote command")
+
+    @staticmethod
+    def run_remote_command_on_multiple_hosts(
+        command, hostnames, user, ssh_key_file=None, parallelism=10, timeout=10, fail_on_error=True
+    ):
+        if not hostnames:
+            return {}
+
+        pool = Pool(parallelism)
+        try:
+            r = pool.map_async(
+                _pickable_run_remote_command,
+                [(hostname, command, user, ssh_key_file, timeout, fail_on_error) for hostname in hostnames],
+            )
+            # The pool timeout is computed by adding 2 times the command timeout for each batch of hosts that is
+            # processed in sequence. Where the size of a batch is given by the degree of parallelism.
+            results = r.get(timeout=int(ceil(len(hostnames) / float(parallelism)) * (2 * timeout)))
+        finally:
+            pool.terminate()
+
+        return dict(results)
+
+
+@retry(stop_max_attempt_number=2)
+def _pickable_run_remote_command(args):
+    """Pickable version of the run_command method that can be used by a pool."""
+    (hostname, command, user, ssh_key_file, timeout, fail_on_error) = args
+    try:
+        remote_command_executor = RemoteCommandExecutor(hostname, user, ssh_key_file)
+        remote_command_executor.run_remote_command(command, timeout, fail_on_error=fail_on_error)
+        return hostname, True
+    except Exception as e:
+        logging.error("Failed when executing remote command on node %s with error %s", hostname, e)
+        return hostname, False

common/schedulers/__init__.py

@@ -0,0 +1,10 @@
+# Copyright 2019 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
+#
+# http://aws.amazon.com/apache2.0/
+#
+# or in the "LICENSE.txt" file accompanying this file. This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, express or implied. See the License for the specific language governing permissions and
+# limitations under the License.