PYCO-53: Analytics SDK - Logging

Changes
=======
* Add loggers to `acouchbase_analytics` and `couchbase_analytics`
* Remove print/todos related to logging and replace with log messaging
* Allow logging to be configured via `PYCBAC_LOG_LEVEL` environment
  variable
* General clean-up

Author: thejcfactor

acouchbase_analytics/__init__.py

@@ -13,77 +13,8 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-import asyncio
-import selectors
-from asyncio import AbstractEventLoop
-from typing import Optional
 
+from acouchbase_analytics.protocol import get_event_loop as get_event_loop  # noqa: F401
+from couchbase_analytics.common import LOG_DATE_FORMAT as LOG_DATE_FORMAT  # noqa: F401
+from couchbase_analytics.common import LOG_FORMAT as LOG_FORMAT  # noqa: F401
 from couchbase_analytics.common import JSONType as JSONType  # noqa: F401
-
-
-class _LoopValidator:
-    """
-    **INTERNAL**
-    """
-
-    REQUIRED_METHODS = {'add_reader', 'remove_reader', 'add_writer', 'remove_writer'}
-
-    @staticmethod
-    def _get_working_loop() -> AbstractEventLoop:
-        """
-        **INTERNAL**
-        """
-        evloop = asyncio.get_event_loop()
-        gen_new_loop = not _LoopValidator._is_valid_loop(evloop)
-        if gen_new_loop:
-            evloop.close()
-            selector = selectors.SelectSelector()
-            new_loop = asyncio.SelectorEventLoop(selector)
-            asyncio.set_event_loop(new_loop)
-            return new_loop
-
-        return evloop
-
-    @staticmethod
-    def _is_valid_loop(evloop: Optional[AbstractEventLoop] = None) -> bool:
-        """
-        **INTERNAL**
-        """
-        if not evloop:
-            return False
-        for meth in _LoopValidator.REQUIRED_METHODS:
-            abs_meth, actual_meth = (getattr(asyncio.AbstractEventLoop, meth), getattr(evloop.__class__, meth))
-            if abs_meth == actual_meth:
-                return False
-        return True
-
-    @staticmethod
-    def get_event_loop(evloop: Optional[AbstractEventLoop] = None) -> AbstractEventLoop:
-        """
-        **INTERNAL**
-        """
-        if evloop and _LoopValidator._is_valid_loop(evloop):
-            return evloop
-        return _LoopValidator._get_working_loop()
-
-    @staticmethod
-    def close_loop() -> None:
-        """
-        **INTERNAL**
-        """
-        evloop = asyncio.get_event_loop()
-        evloop.close()
-
-
-def get_event_loop(evloop: Optional[AbstractEventLoop] = None) -> AbstractEventLoop:
-    """
-    Get an event loop compatible with acouchbase_analytics.
-    Some Event loops, such as ProactorEventLoop (the default asyncio event
-    loop for Python 3.8 on Windows) are not compatible with acouchbase_analytics as
-    they don't implement all members in the abstract base class.
-
-    :param evloop: preferred event loop
-    :return: The preferred event loop, if compatible, otherwise, a compatible
-    alternative event loop.
-    """
-    return _LoopValidator.get_event_loop(evloop)

acouchbase_analytics/protocol/__init__.py

@@ -12,3 +12,75 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
+
+import asyncio
+import selectors
+from typing import Optional
+
+
+class _LoopValidator:
+    """
+    **INTERNAL**
+    """
+
+    REQUIRED_METHODS = {'add_reader', 'remove_reader', 'add_writer', 'remove_writer'}
+
+    @staticmethod
+    def _get_working_loop() -> asyncio.AbstractEventLoop:
+        """
+        **INTERNAL**
+        """
+        evloop = asyncio.get_event_loop()
+        gen_new_loop = not _LoopValidator._is_valid_loop(evloop)
+        if gen_new_loop:
+            evloop.close()
+            selector = selectors.SelectSelector()
+            new_loop = asyncio.SelectorEventLoop(selector)
+            asyncio.set_event_loop(new_loop)
+            return new_loop
+
+        return evloop
+
+    @staticmethod
+    def _is_valid_loop(evloop: Optional[asyncio.AbstractEventLoop] = None) -> bool:
+        """
+        **INTERNAL**
+        """
+        if not evloop:
+            return False
+        for meth in _LoopValidator.REQUIRED_METHODS:
+            abs_meth, actual_meth = (getattr(asyncio.AbstractEventLoop, meth), getattr(evloop.__class__, meth))
+            if abs_meth == actual_meth:
+                return False
+        return True
+
+    @staticmethod
+    def get_event_loop(evloop: Optional[asyncio.AbstractEventLoop] = None) -> asyncio.AbstractEventLoop:
+        """
+        **INTERNAL**
+        """
+        if evloop and _LoopValidator._is_valid_loop(evloop):
+            return evloop
+        return _LoopValidator._get_working_loop()
+
+    @staticmethod
+    def close_loop() -> None:
+        """
+        **INTERNAL**
+        """
+        evloop = asyncio.get_event_loop()
+        evloop.close()
+
+
+def get_event_loop(evloop: Optional[asyncio.AbstractEventLoop] = None) -> asyncio.AbstractEventLoop:
+    """
+    Get an event loop compatible with acouchbase_analytics.
+    Some Event loops, such as ProactorEventLoop (the default asyncio event
+    loop for Python 3.8 on Windows) are not compatible with acouchbase_analytics as
+    they don't implement all members in the abstract base class.
+
+    :param evloop: preferred event loop
+    :return: The preferred event loop, if compatible, otherwise, a compatible
+    alternative event loop.
+    """
+    return _LoopValidator.get_event_loop(evloop)

acouchbase_analytics/protocol/_core/async_json_stream.py

@@ -15,14 +15,16 @@
 
 from __future__ import annotations
 
-from typing import AsyncIterator, Optional
+from typing import AsyncIterator, Callable, Optional
 
 import ijson
 from anyio import EndOfStream, Event, create_memory_object_stream
 
 from acouchbase_analytics.protocol._core.async_json_token_parser import AsyncJsonTokenParser
 from couchbase_analytics.common._core.json_parsing import JsonStreamConfig, ParsedResult, ParsedResultType
+from couchbase_analytics.common._core.json_token_parser_base import JsonTokenParsingError
 from couchbase_analytics.common.errors import AnalyticsError
+from couchbase_analytics.common.logging import LogLevel
 
 
 class AsyncJsonStream:
@@ -31,6 +33,7 @@ def __init__(
         http_stream_iter: AsyncIterator[bytes],
         *,
         stream_config: Optional[JsonStreamConfig] = None,
+        logger_handler: Optional[Callable[[str, LogLevel], None]] = None,
     ) -> None:
         # HTTP stream handling
         if stream_config is None:
@@ -40,10 +43,13 @@ def __init__(
         self._http_response_buffer = bytearray()
         self._http_stream_exhausted = False
 
+        # logging
+        self._log_handler = logger_handler
+
         # results handling
         self._send_stream, self._receive_stream = create_memory_object_stream[ParsedResult](
             max_buffer_size=stream_config.buffered_row_max
-        )  # noqa: E501
+        )
         self._json_stream_parser = None
         self._buffer_entire_result = stream_config.buffer_entire_result
         handler = None if self._buffer_entire_result is True else self._handle_json_result
@@ -87,6 +93,10 @@ def _continue_processing(self) -> bool:
             return False
         return True
 
+    def _log_message(self, message: str, level: LogLevel) -> None:
+        if self._log_handler is not None:
+            self._log_handler(message, level)
+
     async def _send_to_stream(self, result: ParsedResult, close: Optional[bool] = False) -> None:
         """
         **INTERNAL**
@@ -130,10 +140,18 @@ async def _process_token_stream(self) -> None:
                 await self._json_token_parser.parse_token(event, value)
             except StopAsyncIteration:
                 self._token_stream_exhausted = True
+            except JsonTokenParsingError as ex:
+                ex_str = str(ex)
+                self._log_message(f'JSON token parsing error encountered: {ex_str}', LogLevel.ERROR)
+                self._token_stream_exhausted = True
+                await self._send_to_stream(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR), close=True)
+                self._handle_notification(ParsedResultType.ERROR)
+                return
             except ijson.common.IncompleteJSONError as ex:
-                # TODO: log this error
+                ex_str = str(ex)
+                self._log_message(f'Incomplete JSON error encountered: {ex_str}', LogLevel.ERROR)
                 self._token_stream_exhausted = True
-                await self._send_to_stream(ParsedResult(str(ex).encode('utf-8'), ParsedResultType.ERROR), close=True)
+                await self._send_to_stream(ParsedResult(ex_str.encode('utf-8'), ParsedResultType.ERROR), close=True)
                 self._handle_notification(ParsedResultType.ERROR)
                 return
 
@@ -176,10 +194,9 @@ async def get_result(self) -> ParsedResult:
 
     async def start_parsing(self) -> None:
         if self._json_stream_parser is not None:
-            # TODO: logging; I don't think this is an error...
+            self._log_message('JSON stream parser already exists', LogLevel.WARNING)
             return
         await self._process_token_stream()
 
     async def continue_parsing(self) -> None:
-        # TODO: error is _json_stream_parser is None?
         await self._process_token_stream()

acouchbase_analytics/protocol/_core/async_json_token_parser.py

@@ -22,6 +22,7 @@
     START_EVENTS,
     VALUE_TOKENS,
     JsonTokenParserBase,
+    JsonTokenParsingError,
     ParsingState,
     TokenType,
 )
@@ -49,7 +50,9 @@ async def _handle_pop_event(self, token_type: TokenType) -> None:
             next_token = self._pop()
             if next_token.type == matching_token.type:
                 should_emit = self._handle_pop_transition(next_token.state)
-                # I think obj_pairs.reverse() is O(n); while reversed is O(1)
+                # NOTE: obj_pairs.reverse() vs. reversed(obj_pairs) are essentially the same _because_ we convert
+                #       the obj_pairs to a string (e.g. ",".join(...)); using reversed() in this case is slightly
+                #       more convenient as it returns an iterator
                 if matching_token.type == TokenType.START_ARRAY:
                     obj = f'[{",".join(reversed(obj_pairs))}]'
                 else:
@@ -88,5 +91,4 @@ async def parse_token(self, token: str, value: str) -> None:
         elif token_type in POP_EVENTS:
             await self._handle_pop_event(token_type)
         else:
-            # TODO: custom exception
-            raise ValueError(f'Invalid token type: {token_type}; {value=}')
+            raise JsonTokenParsingError(f'Invalid token type: {token_type}; {value=}')

acouchbase_analytics/protocol/_core/client_adapter.py

@@ -15,14 +15,15 @@
 
 from __future__ import annotations
 
-import socket
-from typing import TYPE_CHECKING, Optional
+import logging
+from typing import TYPE_CHECKING, Optional, cast
 from uuid import uuid4
 
 from httpx import URL, AsyncClient, BasicAuth, Response
 
 from couchbase_analytics.common.credential import Credential
 from couchbase_analytics.common.deserializer import Deserializer
+from couchbase_analytics.common.logging import LogLevel, log_message
 from couchbase_analytics.protocol.connection import _ConnectionDetails
 from couchbase_analytics.protocol.options import OptionsBuilder
 
@@ -35,13 +36,17 @@ class _AsyncClientAdapter:
     **INTERNAL**
     """
 
-    _ANALYTICS_PATH = '/api/v1/request'
+    ANALYTICS_PATH = '/api/v1/request'
+    LOGGER_NAME = 'acouchbase_analytics'
 
     def __init__(
         self, http_endpoint: str, credential: Credential, options: Optional[object] = None, **kwargs: object
     ) -> None:
         self._client_id = str(uuid4())
+        self._prefix = ''
+        self._cluster_id = cast(str, kwargs.pop('cluster_id', ''))
         self._opts_builder = OptionsBuilder()
+        kwargs['logger_name'] = self.logger_name
         self._conn_details = _ConnectionDetails.create(self._opts_builder, http_endpoint, credential, options, **kwargs)
         # TODO:  do we want to support custom HTTP transports for the async client?
         self._http_transport_cls = None
@@ -51,7 +56,7 @@ def analytics_path(self) -> str:
         """
         **INTERNAL**
         """
-        return self._ANALYTICS_PATH
+        return self.ANALYTICS_PATH
 
     @property
     def client(self) -> AsyncClient:
@@ -88,6 +93,30 @@ def has_client(self) -> bool:
         """
         return hasattr(self, '_client')
 
+    @property
+    def log_prefix(self) -> str:
+        """
+        **INTERNAL**
+        """
+        if self._prefix:
+            return self._prefix
+        self._prefix = f'[{self._cluster_id}'
+        if self.has_client:
+            self._prefix += f'/{self._client_id}'
+            if self.connection_details.is_secure():
+                self._prefix += '/https]'
+            else:
+                self._prefix += '/http]'
+
+        return self._prefix
+
+    @property
+    def logger_name(self) -> str:
+        """
+        **INTERNAL**
+        """
+        return self.LOGGER_NAME
+
     @property
     def options_builder(self) -> OptionsBuilder:
         """
@@ -101,6 +130,7 @@ async def close_client(self) -> None:
         """
         if hasattr(self, '_client'):
             await self._client.aclose()
+            self.log_message('Cluster HTTP client closed', LogLevel.INFO)
 
     async def create_client(self) -> None:
         """
@@ -123,7 +153,15 @@ async def create_client(self) -> None:
                 if self._http_transport_cls is not None:
                     transport = self._http_transport_cls()
                 self._client = AsyncClient(auth=BasicAuth(*self._conn_details.credential), transport=transport)
-            # TODO: log message
+            self.log_message(
+                (f'Cluster HTTP client created: connection_details={self._conn_details.get_init_details()}'),
+                LogLevel.INFO,
+            )
+        else:
+            self.log_message('Cluster HTTP client already exists, skipping creation.', LogLevel.INFO)
+
+    def log_message(self, message: str, log_level: LogLevel) -> None:
+        log_message(logger, f'{self.log_prefix} {message}', log_level)
 
     async def send_request(self, request: QueryRequest) -> Response:
         """
@@ -139,15 +177,14 @@ async def send_request(self, request: QueryRequest) -> Response:
             path=request.url.path,
         )
         req = self._client.build_request(request.method, url, json=request.body, extensions=request.extensions)
-        try:
-            return await self._client.send(req, stream=True)
-        except socket.gaierror as err:
-            req_url = self._conn_details.url.get_formatted_url()
-            raise RuntimeError(f'Unable to connect to {req_url}') from err
+        return await self._client.send(req, stream=True)
 
     def reset_client(self) -> None:
         """
         **INTERNAL**
         """
         if hasattr(self, '_client'):
             del self._client
+
+
+logger = logging.getLogger(_AsyncClientAdapter.LOGGER_NAME)