Consolidate logging for HTTP requests, especially in timeout logging

client/AGENTS.md

@@ -37,6 +37,7 @@ client/
     ├── logging.py                # cbl_info / cbl_error / cbl_warning / cbl_log_init, LogLevel
     ├── httpclient.py             # AsyncHTTPClient: wraps ClientSession so timeouts name the request; stream_download()
     ├── httplog.py                # HTTP request/response logging (get_next_writer)
+    ├── shell2http.py             # Shell2HttpClient: sidecar calls with a host-side timeout, logged to http_log/
     ├── greenboarduploader.py     # Upload test results to Greenboard
     ├── websocket_router.py       # WebSocket routing (used by JS server)
     ├── utils.py                  # General utilities

client/src/cbltest/api/couchbaseserver.py

@@ -38,7 +38,8 @@
 from cbltest.api.error import CblTestError
 from cbltest.httpclient import AsyncHTTPClient
 from cbltest.logging import cbl_info, cbl_warning
-from cbltest.utils import SHELL2HTTP_PORT, async_retry_assert, retry_assert
+from cbltest.shell2http import Shell2HttpClient
+from cbltest.utils import async_retry_assert, retry_assert
 from cbltest.version import VERSION
 
 T = TypeVar("T")
@@ -254,7 +255,7 @@ def __init__(
                 headers={"Authorization": encode_basic_auth(username, password, "ascii")},
             )
             # The sidecar is a separate service, so it gets no Couchbase Server credentials
-            self.__shell2http = AsyncHTTPClient(f"http://{self.__hostname}:{SHELL2HTTP_PORT}")
+            self.__shell2http = Shell2HttpClient("Couchbase Server", self.__hostname)
 
             self.__cleanup_mode = cleanup_mode
             # Deleting a bucket between tests leaves nothing to reuse, so nothing needs capping.
@@ -1468,25 +1469,20 @@ async def stop_server(self) -> None:
         """
         Stop the Couchbase Server service via shell2http.
         """
-        async with await self.__shell2http.get("/stop-cbs") as resp:
-            if resp.status != 200:
-                body = await resp.text()
-                raise CblTestError(f"Failed to stop CBS: {resp.status} - {body}")
+        await self.__shell2http.call("get", "/stop-cbs")
 
     async def start_server(self, port: int = 8091) -> None:
         """
         Start the Couchbase Server service via shell2http.
 
         :param port: REST API port to wait for readiness (default 8091)
         """
-        async with await self.__shell2http.post(
+        await self.__shell2http.call(
+            "post",
             "/start-cbs",
             data=json.dumps({"port": port}),
-            headers={"Content-Type": "application/json"},
-        ) as resp:
-            if resp.status != 200:
-                body = await resp.text()
-                raise CblTestError(f"Failed to start CBS: {resp.status} - {body}")
+            content_type="application/json",
+        )
 
     async def get_root_ca_certificate(self) -> bytes:
         """

client/src/cbltest/api/edgeserver.py

@@ -256,20 +256,27 @@ async def _send_request(
         method: str,
         path: str,
         payload: JSONSerializable | None = None,
+        log_failure: bool = True,
     ) -> Any:
+        """
+        :param log_failure: Whether to log a request that gets no response in testserver.log.  Pass
+                            False for a poll that expects failures until the server is up; the HTTP
+                            log still records them.
+        """
         with self.__tracer.start_as_current_span("send_request", attributes={"http.method": method, "http.path": path}):
             headers = {"Content-Type": "application/json"} if payload is not None else None
             data = "" if payload is None else payload.serialize()
             writer = get_next_writer()
             writer.write_begin(f"Edge Server [{self.__hostname}] -> {method.upper()} {path}", data)
-            resp = await self.__session.request(method, path, data=data, headers=headers)
+            with writer.record_failure(f"Edge Server [{self.__hostname}] <- {method.upper()} {path}", log=log_failure):
+                resp = await self.__session.request(method, path, data=data, headers=headers)
+                if resp.content_type.startswith("application/json"):
+                    ret_val = await resp.json()
+                    data = dumps(ret_val, indent=2)
+                else:
+                    data = await resp.text()
+                    ret_val = data
 
-            if resp.content_type.startswith("application/json"):
-                ret_val = await resp.json()
-                data = dumps(ret_val, indent=2)
-            else:
-                data = await resp.text()
-                ret_val = data
             writer.write_end(
                 f"Edge Server [{self.__hostname}] <- {method.upper()} {path} {resp.status}",
                 data,

client/src/cbltest/api/edgeservermanager.py

@@ -22,10 +22,8 @@
 from cbltest.api.jsonserializable import JSONDictionary
 from cbltest.api.syncgateway import get_basic_auth_headers
 from cbltest.configparser import EdgeServerInfo
-from cbltest.httpclient import AsyncHTTPClient
-from cbltest.httplog import get_next_writer
 from cbltest.logging import cbl_warning
-from cbltest.utils import SHELL2HTTP_PORT
+from cbltest.shell2http import DEFAULT_SHELL2HTTP_TIMEOUT, Shell2HttpClient
 from cbltest.version import VERSION
 
 
@@ -42,7 +40,7 @@ def __init__(self, info: EdgeServerInfo) -> None:
         # What the host was provisioned with, which reset_to_initial_state() goes back to.
         self.__info = info
         self.__tracer = get_tracer(__name__, VERSION)
-        self.__shell2http_session = AsyncHTTPClient(f"http://{info.hostname}:{SHELL2HTTP_PORT}")
+        self.__sidecar = Shell2HttpClient("Edge Server", info.hostname, error=CblEdgeServerBadResponseError)
         self.__config_file = info.config_path
         self.__clients: list[EdgeServer] = []
 
@@ -69,32 +67,30 @@ def get_admin_client(self) -> EdgeServer:
 
     async def close(self) -> None:
         """Close the sidecar session and every client handed out."""
-        await self.__shell2http_session.close()
+        await self.__sidecar.close()
         for client in self.__clients:
             await client.close()
         self.__clients.clear()
 
-    async def _call_sidecar(self, method: str, path: str, payload: JSONDictionary | None = None) -> str:
-        """Call a shell2http endpoint on the Edge Server host, raising on anything but a 2xx.
+    async def _call_sidecar(
+        self,
+        method: str,
+        path: str,
+        payload: JSONDictionary | None = None,
+        timeout: float = DEFAULT_SHELL2HTTP_TIMEOUT,
+    ) -> str:
+        """Call a shell2http endpoint on the Edge Server host, raising unless the script succeeded.
 
+        :param timeout: Total HTTP timeout in seconds.  See :meth:`Shell2HttpClient.call`.
         :return: The response body, which every endpoint but this one ignores
         """
-        data = "" if payload is None else payload.serialize()
-        headers = {"Content-Type": "application/json"} if payload is not None else None
-        writer = get_next_writer()
-        writer.write_begin(f"Edge Server host [{self.__info.hostname}] -> {method.upper()} {path}", data)
-        resp = await self.__shell2http_session.request(method, path, data=data, headers=headers)
-        # A sidecar echoes raw files back -- start-edgeserver cats the Edge Server's log
-        # on a failed start -- so one odd byte must not fail the whole call.
-        body = await resp.text(errors="replace")
-        writer.write_end(
-            f"Edge Server host [{self.__info.hostname}] <- {method.upper()} {path} {resp.status}",
-            body,
+        return await self.__sidecar.call(
+            method,
+            path,
+            data="" if payload is None else payload.serialize(),
+            content_type="application/json" if payload is not None else None,
+            timeout=timeout,
         )
-        if not resp.ok:
-            raise CblEdgeServerBadResponseError(resp.status, f"{method} {path} returned {resp.status}", body=body)
-
-        return body
 
     async def kill_server(self) -> None:
         """Stop the Edge Server process."""
@@ -126,7 +122,7 @@ async def __wait_until_serving(self, client: EdgeServer) -> None:
         process exists, which is before it is listening.
         """
         try:
-            await client.get_version()
+            await client._send_request("get", "/", log_failure=False)
         except CblEdgeServerBadResponseError:
             # A rejection is still an answer, so the Edge Server is serving.  A config
             # that declares users answers 401 here.

client/src/cbltest/api/syncgateway.py

@@ -1003,13 +1003,17 @@ async def _send_request(
         payload: JSONSerializable | DatabaseConfig | None = None,
         params: dict[str, str] | None = None,
         log_response: bool = True,
+        log_failure: bool = True,
     ) -> Any:
         """
         :param log_response: Whether to write the response body to the HTTP log.  Pass False for a
                              call whose body is large and uninteresting, such as a changes feed read
                              only to find one document; the request and status are still logged.
+        :param log_failure: Whether to log a request that gets no response in testserver.log.  Pass
+                            False for a poll that expects failures until the node is up; the HTTP
+                            log still records them.
         """
-        body, _ = await self._send_request_with_response(method, path, payload, params, log_response)
+        body, _ = await self._send_request_with_response(method, path, payload, params, log_response, log_failure)
         return body
 
     async def _send_request_with_response(
@@ -1019,6 +1023,7 @@ async def _send_request_with_response(
         payload: JSONSerializable | DatabaseConfig | None = None,
         params: dict[str, str] | None = None,
         log_response: bool = True,
+        log_failure: bool = True,
     ) -> tuple[Any, ClientResponse]:
         """Like :func:`_send_request`, but also returns the response, for its headers and cookies."""
         with self._tracer.start_as_current_span("send_request", attributes={"http.method": method, "http.path": path}):
@@ -1029,13 +1034,17 @@ async def _send_request_with_response(
             logged_path = f"{path}?{urlencode(params)}" if params else path
             writer = get_next_writer()
             writer.write_begin(f"Sync Gateway [{self.__http_url}] -> {method.upper()} {logged_path}", data)
-            resp = await self.__session.request(method, path, data=data, headers=headers, params=params)
-            if resp.content_type.startswith("application/json"):
-                ret_val = await resp.json()
-                data = dumps(ret_val, indent=2)
-            else:
-                data = await resp.text()
-                ret_val = data
+            with writer.record_failure(
+                f"Sync Gateway [{self.__http_url}] <- {method.upper()} {logged_path}", log=log_failure
+            ):
+                resp = await self.__session.request(method, path, data=data, headers=headers, params=params)
+                if resp.content_type.startswith("application/json"):
+                    ret_val = await resp.json()
+                    data = dumps(ret_val, indent=2)
+                else:
+                    data = await resp.text()
+                    ret_val = data
+
             writer.write_end(
                 f"Sync Gateway [{self.__http_url}] <- {method.upper()} {logged_path} {resp.status}",
                 data if log_response or not resp.ok else f"<{len(data)} bytes not logged>",
@@ -2374,7 +2383,7 @@ async def wait_for_rest_api(self) -> None:
 
         async def _wait_for_rest_api_poll() -> None:
             try:
-                await self._send_request("get", "/_ping")
+                await self._send_request("get", "/_ping", log_failure=False)
             # A restart drops in-flight connections, which surfaces as ServerDisconnectedError or
             # ClientOSError as well as ClientConnectorError - all are ClientError/OSError.
             except (CblSyncGatewayBadResponseError, ClientError, OSError) as exc:

client/src/cbltest/httplog.py

@@ -1,7 +1,10 @@
+from collections.abc import Iterator
+from contextlib import contextmanager
 from itertools import count
 from pathlib import Path
 
 from cbltest.globals import CBLPyTestGlobal
+from cbltest.logging import cbl_error
 
 _http_num = count(1)
 
@@ -42,6 +45,25 @@ def write_error(self, msg: str) -> None:
         with open(recv_log_path, "x") as fout:
             fout.write(msg)
 
+    @contextmanager
+    def record_failure(self, request: str, log: bool = True) -> Iterator[None]:
+        """
+        Record an exception raised inside the block, e.g. a timeout or a connection error,
+        in the error file and in testserver.log, then raise it again.
+
+        :param request: What failed, e.g. "Sync Gateway [https://host:4985] <- GET /db/"
+        :param log: Whether to log it in testserver.log too.  Pass False for a poll that
+            expects failures until a server is up.
+        """
+        try:
+            yield
+        except Exception as e:
+            failure = f"{request} failed: {e}"
+            self.write_error(failure)
+            if log:
+                cbl_error(failure, include_stack=False)
+            raise
+
     def write_end(self, header: str, payload: str) -> None:
         send_log_path = self.__get_path("end")
         with open(send_log_path, "x") as fout:

client/src/cbltest/plugins/sgw_cluster_manager.py

@@ -12,13 +12,12 @@
 from typing import Any
 
 import pytest_asyncio
-from aiohttp import ClientTimeout
 from cbltest import CBLPyTest
 from cbltest.api.error import CblTestError
-from cbltest.api.syncgateway import SHELL2HTTP_PORT, SyncGateway
+from cbltest.api.syncgateway import SyncGateway
 from cbltest.api.syncgatewaycluster import SyncGatewayCluster
-from cbltest.httpclient import AsyncHTTPClient
 from cbltest.logging import cbl_info
+from cbltest.shell2http import Shell2HttpClient
 from cbltest.version import VERSION
 from opentelemetry.trace import get_tracer
 
@@ -59,45 +58,41 @@ def __init__(self, node: SyncGateway, token: object = None) -> None:
         _check_token(token, type(self).__name__)
         self.__node = node
         self.__tracer = get_tracer(__name__, VERSION)
-        self.__session = AsyncHTTPClient(f"http://{node.hostname}:{SHELL2HTTP_PORT}")
+        self.__sidecar = Shell2HttpClient("Sync Gateway", node.hostname)
 
     def __str__(self) -> str:
         return str(self.__node)
 
     @property
     def closed(self) -> bool:
         """Whether this node's sidecar session has been closed."""
-        return self.__session.closed
+        return self.__sidecar.closed
 
     async def close(self) -> None:
         """Close this node's sidecar session. The fixture calls this at teardown."""
-        await self.__session.close()
+        await self.__sidecar.close()
 
     @property
     def has_shell2http_sidecar(self) -> bool:
         """Whether this node exposes the shell2http sidecar every operation here goes through."""
         return self.__node.has_shell2http_sidecar
 
-    async def _call_sidecar(self, method: str, path: str, data: str | None = None, timeout: int = 120) -> None:
+    async def _call_sidecar(self, method: str, path: str, data: str | None = None, timeout: float = 120) -> None:
         """
-        Call a sidecar endpoint, raising on anything but a 200.
+        Call a sidecar endpoint, raising unless the script succeeded.
 
         :param method: HTTP method to use
         :param path: Sidecar path, including any query string
         :param data: Request body, for the endpoints that take one
-        :param timeout: Total timeout in seconds
+        :param timeout: Total HTTP timeout in seconds.  See :meth:`Shell2HttpClient.call`.
         """
-        headers = {"Content-Type": "text/plain"} if data is not None else None
-        async with await self.__session.request(
+        await self.__sidecar.call(
             method,
             path,
             data=data,
-            headers=headers,
-            timeout=ClientTimeout(total=timeout),
-        ) as resp:
-            if resp.status != 200:
-                body = await resp.text()
-                raise CblTestError(f"{method.upper()} {path} failed on {self}: {resp.status} - {body}")
+            content_type="text/plain" if data is not None else None,
+            timeout=timeout,
+        )
 
     async def restart_with_config(self, config_name: str = DEFAULT_CONFIG_NAME) -> None:
         """