Release 2021.3.1

RELEASE_NOTES

@@ -1,3 +1,7 @@
+2021.3.1
+- 2021-03-11 TUN-4051: Add component-tests to test graceful shutdown
+- 2021-03-12 TUN-4052: Add component tests to assert service mode behavior
+
 2021.3.0
 - 2021-03-10 TUN-4075: Dedup test should not compare order of list
 - 2021-03-10 Revert "AUTH-3394: Creates a token per app instead of per path"

cfsetup.yaml

@@ -210,6 +210,9 @@ stretch: &stretch
       - python3.7
       - python3-pip
       - python3-setuptools
+      # procps installs the ps command which is needed in test_sysv_service because the init script
+      # uses ps pid to determine if the agent is running
+      - procps
     pre-cache-copy-paths:
       - component-tests/requirements.txt
     pre-cache:

component-tests/README.md

@@ -31,6 +31,9 @@ If you are using Visual Studio, follow https://code.visualstudio.com/docs/python
 to turn on formatter and https://marketplace.visualstudio.com/items?itemName=cbrevik.toggle-format-on-save
 to turn on format on save.
 
+6. If you have cloudflared running as a service on your machine, you can either stop the service or ignore the service tests
+via `--ignore test_service.py`
+
 # How to run
 Specify path to config file via env var `COMPONENT_TESTS_CONFIG`. This is required.
 ## All tests

component-tests/constants.py

@@ -1,3 +1,3 @@
 METRICS_PORT = 51000
-MAX_RETRIES = 3
-BACKOFF_SECS = 5
+MAX_RETRIES = 5
+BACKOFF_SECS = 7

component-tests/test_logging.py

@@ -5,6 +5,7 @@ from util import start_cloudflared, wait_tunnel_ready, send_requests, LOGGER
 
 
 class TestLogging:
+    # TODO: Test logging when running as a service https://jira.cfops.it/browse/TUN-4082
     # Rolling logger rotate log files after 1 MB
     rotate_after_size = 1000 * 1000
     default_log_file = "cloudflared.log"
@@ -13,7 +14,7 @@ class TestLogging:
     def test_logging_to_terminal(self, tmp_path, component_tests_config):
         config = component_tests_config()
         with start_cloudflared(tmp_path, config, new_process=True) as cloudflared:
-            wait_tunnel_ready()
+            wait_tunnel_ready(tunnel_url=config.get_url())
             self.assert_log_to_terminal(cloudflared)
 
     def test_logging_to_file(self, tmp_path, component_tests_config):
@@ -24,7 +25,7 @@ class TestLogging:
         }
         config = component_tests_config(extra_config)
         with start_cloudflared(tmp_path, config, new_process=True, capture_output=False):
-            wait_tunnel_ready()
+            wait_tunnel_ready(tunnel_url=config.get_url())
             self.assert_log_in_file(log_file)
             self.assert_json_log(log_file)
 
@@ -37,7 +38,7 @@ class TestLogging:
         }
         config = component_tests_config(extra_config)
         with start_cloudflared(tmp_path, config, new_process=True, capture_output=False):
-            wait_tunnel_ready()
+            wait_tunnel_ready(tunnel_url=config.get_url())
             self.assert_log_to_dir(config, log_dir)
 
     def assert_log_to_terminal(self, cloudflared):

component-tests/test_reconnect.py

@@ -4,7 +4,7 @@ import copy
 from retrying import retry
 from time import sleep
 
-from util import start_cloudflared, wait_tunnel_ready, check_tunnel_not_ready, send_requests
+from util import start_cloudflared, wait_tunnel_ready, check_tunnel_not_connected, send_requests
 
 
 class TestReconnect():
@@ -35,16 +35,19 @@ class TestReconnect():
         cloudflared.stdin.flush()
 
     def assert_reconnect(self, config, cloudflared, repeat):
-        wait_tunnel_ready()
+        wait_tunnel_ready(tunnel_url=config.get_url())
         for _ in range(repeat):
             for i in range(self.default_ha_conns):
                 self.send_reconnect(cloudflared, self.default_reconnect_secs)
                 expect_connections = self.default_ha_conns-i-1
                 if expect_connections > 0:
-                    wait_tunnel_ready(expect_connections=expect_connections)
+                    # Don't check if tunnel returns 200 here because there is a race condition between wait_tunnel_ready
+                    # retrying to get 200 response and reconnecting
+                    wait_tunnel_ready(
+                        require_min_connections=expect_connections)
                 else:
-                    check_tunnel_not_ready()
+                    check_tunnel_not_connected()
 
             sleep(self.default_reconnect_secs + 10)
-            wait_tunnel_ready()
-            send_requests(config.get_url(), 1)
+            wait_tunnel_ready(tunnel_url=config.get_url(),
+                              require_min_connections=self.default_ha_conns)

component-tests/test_service.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python
+from contextlib import contextmanager
+import os
+from pathlib import Path
+import platform
+import pytest
+import subprocess
+
+from util import start_cloudflared, cloudflared_cmd, wait_tunnel_ready, LOGGER
+
+
+def select_platform(plat):
+    return pytest.mark.skipif(
+        platform.system() != plat, reason=f"Only runs on {plat}")
+
+
+def default_config_dir():
+    return os.path.join(Path.home(), ".cloudflared")
+
+
+def default_config_file():
+    return os.path.join(default_config_dir(), "config.yml")
+
+
+class TestServiceMode():
+    @select_platform("Darwin")
+    @pytest.mark.skipif(os.path.exists(default_config_file()), reason=f"There is already a config file in default path")
+    def test_launchd_service(self, component_tests_config):
+        # On Darwin cloudflared service defaults to run classic tunnel command
+        additional_config = {
+            "hello-world": True,
+        }
+        config = component_tests_config(
+            additional_config=additional_config, named_tunnel=False)
+        with self.run_service(Path(default_config_dir()), config):
+            self.launchctl_cmd("list")
+            self.launchctl_cmd("start")
+            wait_tunnel_ready(tunnel_url=config.get_url())
+            self.launchctl_cmd("stop")
+
+        os.remove(default_config_file())
+        self.launchctl_cmd("list", success=False)
+
+    @select_platform("Linux")
+    @pytest.mark.skipif(os.path.exists("/etc/cloudflared/config.yml"), reason=f"There is already a config file in default path")
+    def test_sysv_service(self, tmp_path, component_tests_config):
+        config = component_tests_config()
+        with self.run_service(tmp_path, config, root=True):
+            self.sysv_cmd("start")
+            self.sysv_cmd("status")
+            wait_tunnel_ready(tunnel_url=config.get_url())
+            self.sysv_cmd("stop")
+        # Service install copies config file to /etc/cloudflared/config.yml
+        subprocess.run(["sudo", "rm", "/etc/cloudflared/config.yml"])
+        self.sysv_cmd("status", success=False)
+
+    @contextmanager
+    def run_service(self, tmp_path, config, root=False):
+        try:
+            service = start_cloudflared(
+                tmp_path, config, cfd_args=["service", "install"], cfd_pre_args=[], capture_output=False, root=root)
+            yield service
+        finally:
+            start_cloudflared(
+                tmp_path, config, cfd_args=["service", "uninstall"], cfd_pre_args=[], capture_output=False, root=root)
+
+    def launchctl_cmd(self, action, success=True):
+        cmd = subprocess.run(
+            ["launchctl", action, "com.cloudflare.cloudflared"], check=success)
+        if not success:
+            assert cmd.returncode != 0, f"Expect {cmd.args} to fail, but it succeed"
+
+    def sysv_cmd(self, action, success=True):
+        cmd = subprocess.run(
+            ["sudo", "service", "cloudflared", action], check=success)
+        if not success:
+            assert cmd.returncode != 0, f"Expect {cmd.args} to fail, but it succeed"

component-tests/test_termination.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python
+from contextlib import contextmanager
+import requests
+import signal
+import threading
+import time
+
+from util import start_cloudflared, wait_tunnel_ready, check_tunnel_not_connected, LOGGER
+
+
+class TestTermination():
+    grace_period = 5
+    timeout = 10
+    extra_config = {
+        "grace-period": f"{grace_period}s",
+    }
+    signals = [signal.SIGTERM, signal.SIGINT]
+    sse_endpoint = "/sse?freq=1s"
+
+    def test_graceful_shutdown(self, tmp_path, component_tests_config):
+        config = component_tests_config(self.extra_config)
+        for sig in self.signals:
+            with start_cloudflared(
+                    tmp_path, config, new_process=True, capture_output=False) as cloudflared:
+                wait_tunnel_ready(tunnel_url=config.get_url())
+
+                connected = threading.Condition()
+                in_flight_req = threading.Thread(
+                    target=self.stream_request, args=(config, connected,))
+                in_flight_req.start()
+
+                with connected:
+                    connected.wait(self.timeout)
+                # Send signal after the SSE connection is established
+                self.terminate_by_signal(cloudflared, sig)
+                self.wait_eyeball_thread(
+                    in_flight_req, self.grace_period + self.timeout)
+
+    # test cloudflared terminates before grace period expires when all eyeball
+    # connections are drained
+    def test_shutdown_once_no_connection(self, tmp_path, component_tests_config):
+        config = component_tests_config(self.extra_config)
+        for sig in self.signals:
+            with start_cloudflared(
+                    tmp_path, config, new_process=True, capture_output=False) as cloudflared:
+                wait_tunnel_ready(tunnel_url=config.get_url())
+
+                connected = threading.Condition()
+                in_flight_req = threading.Thread(
+                    target=self.stream_request, args=(config, connected, True, ))
+                in_flight_req.start()
+
+                with self.within_grace_period():
+                    with connected:
+                        connected.wait(self.timeout)
+                    # Send signal after the SSE connection is established
+                    self.terminate_by_signal(cloudflared, sig)
+                    self.wait_eyeball_thread(in_flight_req, self.grace_period)
+
+    def test_no_connection_shutdown(self, tmp_path, component_tests_config):
+        config = component_tests_config(self.extra_config)
+        for sig in self.signals:
+            with start_cloudflared(
+                    tmp_path, config, new_process=True, capture_output=False) as cloudflared:
+                wait_tunnel_ready(tunnel_url=config.get_url())
+                with self.within_grace_period():
+                    self.terminate_by_signal(cloudflared, sig)
+
+    def terminate_by_signal(self, cloudflared, sig):
+        cloudflared.send_signal(sig)
+        check_tunnel_not_connected()
+        cloudflared.wait()
+
+    def wait_eyeball_thread(self, thread, timeout):
+        thread.join(timeout)
+        assert thread.is_alive() == False, "eyeball thread is still alive"
+
+    # Using this context asserts logic within the context is executed within grace period
+    @contextmanager
+    def within_grace_period(self):
+        try:
+            start = time.time()
+            yield
+        finally:
+            duration = time.time() - start
+            assert duration < self.grace_period
+
+    def stream_request(self, config, connected, early_terminate):
+        expected_terminate_message = "502 Bad Gateway"
+        url = config.get_url() + self.sse_endpoint
+
+        with requests.get(url, timeout=5, stream=True) as resp:
+            with connected:
+                connected.notifyAll()
+            lines = 0
+            for line in resp.iter_lines():
+                if expected_terminate_message.encode() == line:
+                    break
+                lines += 1
+                if early_terminate and lines == 2:
+                    return
+            # /sse returns count followed by 2 new lines
+            assert lines >= (self.grace_period * 2)

component-tests/util.py

@@ -12,26 +12,34 @@ from constants import METRICS_PORT, MAX_RETRIES, BACKOFF_SECS
 LOGGER = logging.getLogger(__name__)
 
 
-def write_config(path, config):
-    config_path = path / "config.yaml"
+def write_config(directory, config):
+    config_path = directory / "config.yml"
     with open(config_path, 'w') as outfile:
         yaml.dump(config, outfile)
     return config_path
 
 
-def start_cloudflared(path, config, cfd_args=["run"], cfd_pre_args=["tunnel"], new_process=False, allow_input=False, capture_output=True):
-    config_path = write_config(path, config.full_config)
-    cmd = [config.cloudflared_binary]
-    cmd += cfd_pre_args
-    cmd += ["--config", config_path]
-    cmd += cfd_args
-    LOGGER.info(f"Run cmd {cmd} with config {config}")
+def start_cloudflared(directory, config, cfd_args=["run"], cfd_pre_args=["tunnel"], new_process=False, allow_input=False, capture_output=True, root=False):
+    config_path = write_config(directory, config.full_config)
+    cmd = cloudflared_cmd(config, config_path, cfd_args, cfd_pre_args, root)
     if new_process:
         return run_cloudflared_background(cmd, allow_input, capture_output)
     # By setting check=True, it will raise an exception if the process exits with non-zero exit code
     return subprocess.run(cmd, check=True, capture_output=capture_output)
 
 
+def cloudflared_cmd(config, config_path, cfd_args, cfd_pre_args, root):
+    cmd = []
+    if root:
+        cmd += ["sudo"]
+    cmd += [config.cloudflared_binary]
+    cmd += cfd_pre_args
+    cmd += ["--config", config_path]
+    cmd += cfd_args
+    LOGGER.info(f"Run cmd {cmd} with config {config}")
+    return cmd
+
+
 @contextmanager
 def run_cloudflared_background(cmd, allow_input, capture_output):
     output = subprocess.PIPE if capture_output else subprocess.DEVNULL
@@ -44,25 +52,26 @@ def run_cloudflared_background(cmd, allow_input, capture_output):
 
 
 @retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
-def wait_tunnel_ready(expect_connections=4):
-    url = f'http://localhost:{METRICS_PORT}/ready'
+def wait_tunnel_ready(tunnel_url=None, require_min_connections=1):
+    metrics_url = f'http://localhost:{METRICS_PORT}/ready'
 
     with requests.Session() as s:
-        resp = send_request(s, url, True)
+        resp = send_request(s, metrics_url, True)
         assert resp.json()[
-            "readyConnections"] == expect_connections, f"Ready endpoint returned {resp.json()} but we expect {expect_connections} ready connections"
+            "readyConnections"] >= require_min_connections, f"Ready endpoint returned {resp.json()} but we expect at least {require_min_connections} connections"
+        if tunnel_url is not None:
+            send_request(s, tunnel_url, True)
 
 
 @retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
-def check_tunnel_not_ready():
+def check_tunnel_not_connected():
     url = f'http://localhost:{METRICS_PORT}/ready'
 
     resp = requests.get(url, timeout=1)
     assert resp.status_code == 503, f"Expect {url} returns 503, got {resp.status_code}"
 
+
 # In some cases we don't need to check response status, such as when sending batch requests to generate logs
-
-
 def send_requests(url, count, require_ok=True):
     errors = 0
     with requests.Session() as s: