Merge pull request #4176 from oliver-sanders/mod-etc-hosts

actions: fudge /etc/hosts

.github/workflows/bash.yml

@@ -10,6 +10,7 @@ on:
 jobs:
   bash-docker:
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     strategy:
       matrix:
         bash-version:

.github/workflows/test_fast.yml

@@ -19,6 +19,18 @@ jobs:
             python-version: '3.7'
 
     steps:
+      - name: Fudge /etc/hosts
+        run: |
+          # Add the current IP address, long hostname and short hostname
+          # record to /etc/hosts file
+          # https://github.com/actions/virtual-environments/issues/3185
+          echo -e \
+            "$(ip addr show eth0 \
+            | grep "inet\b" \
+            | awk '{print $2}' \
+            | cut -d/ -f1)\t$(hostname -f) $(hostname -s)" \
+            | sudo tee -a /etc/hosts
+
       - name: Checkout
         uses: actions/checkout@v2
         with:

.github/workflows/test_functional.yml

@@ -67,6 +67,18 @@ jobs:
       REMOTE_PLATFORM: ${{ contains(matrix.platform, '_remote') }}
 
     steps:
+      - name: Fudge /etc/hosts
+        run: |
+          # Add the current IP address, long hostname and short hostname
+          # record to /etc/hosts file
+          # https://github.com/actions/virtual-environments/issues/3185
+          echo -e \
+            "$(ip addr show eth0 \
+            | grep "inet\b" \
+            | awk '{print $2}' \
+            | cut -d/ -f1)\t$(hostname -f) $(hostname -s)" \
+            | sudo tee -a /etc/hosts
+
       - name: Checkout
         uses: actions/checkout@v2
         with:

cylc/flow/cfgspec/globalcfg.py

@@ -22,6 +22,7 @@
 from cylc.flow import LOG
 from cylc.flow import __version__ as CYLC_VERSION
 from cylc.flow.hostuserutil import get_user_home
+from cylc.flow.network.client_factory import CommsMeth
 from cylc.flow.parsec.config import ParsecConfig, ConfigNode as Conf
 from cylc.flow.parsec.exceptions import ParsecError
 from cylc.flow.parsec.upgrade import upgrader
@@ -440,7 +441,8 @@
             ''')
             Conf('suite definition directory', VDR.V_STRING)
             Conf('communication method',
-                 VDR.V_STRING, 'zmq', options=['zmq', 'poll', 'ssh'], desc='''
+                 VDR.V_STRING, 'zmq',
+                 options=[meth.value for meth in CommsMeth], desc='''
                 The means by which task progress messages are reported back to
                 the running suite.
 

cylc/flow/network/client.py

@@ -231,15 +231,7 @@ def get_header(self) -> dict:
             dict: dictionary with the header information, such as
                 program and hostname.
         """
-
         host = socket.gethostname()
-        # Identify communication method
-        comms_method = os.getenv("CLIENT_COMMS_METH", default=CommsMeth.ZMQ)
-        if (self.host and
-            (comms_method == CommsMeth.ZMQ) and
-            (socket.gethostbyname(
-                self.host) == socket.gethostbyname(socket.gethostname()))):
-            comms_method = CommsMeth.LOCAL
         if len(sys.argv) > 1:
             cmd = sys.argv[1]
         else:
@@ -259,7 +251,11 @@ def get_header(self) -> dict:
             'meta': {
                 'prog': cmd,
                 'host': host,
-                'comms_method': comms_method,
+                'comms_method':
+                    os.getenv(
+                        "CLIENT_COMMS_METH",
+                        default=CommsMeth.ZMQ.value
+                    )
             }
         }
 

cylc/flow/network/client_factory.py

@@ -14,40 +14,39 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from enum import Enum
 import os
 
 
-class CommsMeth():
+class CommsMeth(Enum):
     """String literals used for identifying communication methods"""
 
+    POLL = 'poll'
     SSH = 'ssh'
     ZMQ = 'zmq'
-    LOCAL = 'local'  # used for local commands
 
 
-def get_comms_method():
+def get_comms_method() -> CommsMeth:
     """"Return Communication Method from environment variable, default zmq"""
+    return CommsMeth(
+        os.getenv('CYLC_TASK_COMMS_METHOD', CommsMeth.ZMQ.value)
+    )
 
-    return os.getenv('CYLC_TASK_COMMS_METHOD', CommsMeth.ZMQ)
 
-
-def get_runtime_client(comms_method, workflow, timeout=None):
+def get_runtime_client(comms_method: CommsMeth, workflow, timeout=None):
     """Return client for the provided communication method.
 
         Args:
             comm_method: communication method
             workflow: workflow name
     """
-
     if comms_method == CommsMeth.SSH:
         from cylc.flow.network.ssh_client import SuiteRuntimeClient
     else:
-        from cylc.flow.network.client import SuiteRuntimeClient
+        from cylc.flow.network.client import SuiteRuntimeClient  # type: ignore
     return SuiteRuntimeClient(workflow, timeout=timeout)
 
 
 def get_client(workflow, timeout=None):
     """Get communication method and return correct SuiteRuntimeClient"""
-
-    comms_method = get_comms_method()
-    return get_runtime_client(comms_method, workflow, timeout=timeout)
+    return get_runtime_client(get_comms_method(), workflow, timeout=timeout)

cylc/flow/network/ssh_client.py

@@ -71,7 +71,7 @@ def send_request(self, command, args=None, timeout=None):
         """
         # Set environment variable to determine the communication for use on
         # the scheduler
-        os.environ["CLIENT_COMMS_METH"] = CommsMeth.SSH
+        os.environ["CLIENT_COMMS_METH"] = CommsMeth.SSH.value
         cmd = ["client"]
         if timeout:
             cmd += [f'comms_timeout={timeout}']

cylc/flow/task_message.py

@@ -25,10 +25,13 @@
 import os
 import sys
 
-
 from cylc.flow.exceptions import SuiteStopped
 import cylc.flow.flags
-from cylc.flow.network.client_factory import get_client
+from cylc.flow.network.client_factory import (
+    CommsMeth,
+    get_client,
+    get_comms_method
+)
 from cylc.flow.pathutil import get_suite_run_job_dir
 from cylc.flow.task_outputs import TASK_OUTPUT_STARTED, TASK_OUTPUT_SUCCEEDED
 from cylc.flow.wallclock import get_current_time_string
@@ -80,6 +83,12 @@ def record_messages(suite, task_job, messages):
     # Record the event time, in case the message is delayed in some way.
     event_time = get_current_time_string(
         override_use_utc=(os.getenv('CYLC_UTC') == 'True'))
+    write_messages(suite, task_job, messages, event_time)
+    if get_comms_method() != CommsMeth.POLL:
+        send_messages(suite, task_job, messages, event_time)
+
+
+def write_messages(suite, task_job, messages, event_time):
     # Print to stdout/stderr
     for severity, message in messages:
         if severity in STDERR_LEVELS:
@@ -91,6 +100,9 @@ def record_messages(suite, task_job, messages):
     # Write to job.status
     _append_job_status_file(suite, task_job, event_time, messages)
     # Send messages
+
+
+def send_messages(suite, task_job, messages, event_time):
     suite = os.path.normpath(suite)
     try:
         pclient = get_client(suite)

cylc/flow/task_remote_mgr.py

@@ -182,8 +182,8 @@ def remote_init(
         self.remote_init_map[install_target] = REMOTE_INIT_IN_PROGRESS
 
         # Determine what items to install
-        comm_meth = platform['communication method']
-        items = self._remote_init_items(comm_meth)
+        comms_meth: CommsMeth = CommsMeth(platform['communication method'])
+        items = self._remote_init_items(comms_meth)
 
         # Create a TAR archive with the service files,
         # so they can be sent later via SSH's STDIN to the task remote.
@@ -390,7 +390,7 @@ def _file_install_callback(self, ctx, install_target):
             LOG.error(ctx)
             self.ready = True
 
-    def _remote_init_items(self, comm_meth):
+    def _remote_init_items(self, comms_meth: CommsMeth):
         """Return list of items to install based on communication method.
 
         Return (list):
@@ -401,7 +401,7 @@ def _remote_init_items(self, comm_meth):
         """
         items = []
 
-        if comm_meth in [CommsMeth.SSH, CommsMeth.ZMQ]:
+        if comms_meth in [CommsMeth.SSH, CommsMeth.ZMQ]:
             # Contact file
             items.append((
                 get_contact_file(self.suite),

etc/bin/shellchecker

@@ -77,7 +77,9 @@ default () {
     main '.' \
         --exclude 'etc/bin/live-graph-movie.sh' \
         --exclude '.git' \
-        -- -e SC1090
+        -- \
+        -e SC1090 \
+        -e SC1091
 }
 
 if [[ $# -gt 0 ]]; then

setup.py

@@ -63,7 +63,8 @@ def find_version(*file_paths):
     'flake8>=3.0.0',
     'mypy>=0.800',
     'pycodestyle>=2.5.0',
-    'pytest-asyncio>=0.14.0',
+    # TODO: https://github.com/pytest-dev/pytest-asyncio/issues/ 209
+    'pytest-asyncio==0.14.0',
     'pytest-cov>=2.8.0',
     'pytest-xdist>=2',
     'pytest-env>=0.6.2',

tests/flakyfunctional/cylc-poll/16-execution-time-limit.t

@@ -16,16 +16,15 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #-------------------------------------------------------------------------------
 # Test execution time limit polling.
+export REQUIRE_PLATFORM='loc:* comms:poll runner:background'
 . "$(dirname "$0")/test_header"
 #-------------------------------------------------------------------------------
 set_test_number 4
 create_test_global_config '' "
 [platforms]
    [[$CYLC_TEST_PLATFORM]]
-        communication method = poll
         submission polling intervals = PT2S
         execution polling intervals = PT1M
-        job runner = background
         execution time limit polling intervals = PT5S
 "
 install_suite "${TEST_NAME_BASE}" "${TEST_NAME_BASE}"

tests/flakyfunctional/cylc-poll/16-execution-time-limit/flow.cylc

@@ -14,6 +14,12 @@
     [[foo]]
         platform = {{ environ['CYLC_TEST_PLATFORM'] }}
         init-script = cylc__job__disable_fail_signals ERR EXIT
-        script = cylc__job__wait_cylc_message_started; exit 1
+        script = """
+            cylc__job__wait_cylc_message_started
+            # give it a while for the started message to get picked up by
+            # the scheduler
+            sleep 10
+            exit 1
+        """
         [[[job]]]
             execution time limit = PT5S

tests/functional/cylc-poll/13-comm-method.t

@@ -22,12 +22,13 @@ set_test_number 6
 install_suite "${TEST_NAME_BASE}" "${TEST_NAME_BASE}"
 #-------------------------------------------------------------------------------
 run_ok "${TEST_NAME_BASE}-validate" cylc validate "${SUITE_NAME}"
-create_test_global_config '
+create_test_global_config "
 [platforms]
-   [[localhost]]
+   [[$CYLC_TEST_PLATFORM]]
         communication method = poll
         execution polling intervals = 10*PT6S
-        submission polling intervals = 10*PT6S'
+        submission polling intervals = 10*PT6S
+"
 
 suite_run_ok "${TEST_NAME_BASE}-run" \
     cylc play --reference-test --debug --no-detach "${SUITE_NAME}"

tests/functional/cylc-poll/13-comm-method/flow.cylc

@@ -1,10 +1,19 @@
+#!Jinja2
+
 [scheduling]
     [[graph]]
         R1 = t1 & t2
 
 [runtime]
     [[root]]
-        script = true
+        script = """
+            wait
+            # sleep for twice the polling interval to make sure
+            # the started message gets picked up before the succeeded
+            # message is issued
+            sleep 12
+        """
+        platform = {{ environ['CYLC_TEST_PLATFORM'] }}
     [[t1]]
     [[t2]]
         [[[job]]]

tests/functional/platforms/03-platform-db.t

@@ -16,7 +16,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #-------------------------------------------------------------------------------
 # Check db stores correct platform
-export REQUIRE_PLATFORM='loc:remote'
+export REQUIRE_PLATFORM='loc:remote comms:?(tcp|ssh)'
 . "$(dirname "$0")/test_header"
 set_test_number 3
 

tests/functional/remote/06-poll.t

@@ -0,0 +1,59 @@
+#!/usr/bin/env bash
+# THIS FILE IS PART OF THE CYLC SUITE ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#-------------------------------------------------------------------------------
+# Test remote host settings.
+export REQUIRE_PLATFORM='loc:remote comms:poll'
+. "$(dirname "$0")/test_header"
+#-------------------------------------------------------------------------------
+set_test_number 4
+create_test_global_config "" "
+[platforms]
+    [[$CYLC_TEST_PLATFORM]]
+        retrieve job logs = True
+"
+#-------------------------------------------------------------------------------
+init_suite "${TEST_NAME_BASE}" <<__HERE__
+[scheduling]
+    [[graph]]
+        R1 = foo
+[runtime]
+    [[foo]]
+        script = cylc message -- foo
+        platform = $CYLC_TEST_PLATFORM
+        [[[outputs]]]
+            foo = foo
+__HERE__
+#-------------------------------------------------------------------------------
+TEST_NAME="${TEST_NAME_BASE}-validate"
+run_ok "${TEST_NAME}" cylc validate "${SUITE_NAME}"
+
+TEST_NAME="${TEST_NAME_BASE}-run"
+suite_run_ok "${TEST_NAME}" \
+    cylc play "${SUITE_NAME}" \
+    --debug \
+    --no-detach
+
+log_scan \
+    "${TEST_NAME_BASE}-poll" \
+    "$(cylc cat-log -m p "$SUITE_NAME")" \
+    10 \
+    1 \
+    '\[foo.1\] status=submitted: (polled)foo' \
+    '\[foo.1\] status=succeeded: (polled)succeeded'
+
+purge
+exit

tests/unit/test_task_remote_mgr.py

@@ -18,6 +18,7 @@
 from typing import Any
 from unittest.mock import MagicMock, Mock
 
+from cylc.flow.network.client_factory import CommsMeth
 from cylc.flow.task_remote_mgr import (
     REMOTE_FILE_INSTALL_DONE, REMOTE_INIT_IN_PROGRESS, TaskRemoteMgr)
 
@@ -42,7 +43,7 @@ def test_remote_init_skip(
     """
     platform = {
         'install target': install_target,
-        'communication method': 'whatever'
+        'communication method': CommsMeth.POLL
     }
     mock_task_remote_mgr = MagicMock(remote_init_map={})
     mock_construct_ssh_cmd = Mock()