Task/228 otel wc3traceparent

.github/workflows/CDA-testing.yml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.9', '3.13']
+        python-version: ['3.10', '3.13']
 
     steps:
       - uses: actions/checkout@v7
@@ -50,8 +50,7 @@ jobs:
         id: cache-poetry-venv
         with:
           path: .venv
-          key: ${{ runner.os }}-py${{ matrix.python-version }}-poetry-${{ hashFiles('poetry.lock')
-            }}
+          key: ${{ runner.os }}-py${{ matrix.python-version }}-poetry-${{ hashFiles('poetry.lock')}}
 
       # Install dependencies only if cache is missed
       - name: Install dependencies

.github/workflows/code-check.yml

@@ -1,7 +1,11 @@
 name: Code Check
 
 # Run the workflow on all branches.
-on: [push, pull_request]
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
 
 jobs:
   # Run basic code quality checks.

.github/workflows/testing.yml

@@ -1,7 +1,11 @@
 name: Testing
 
 # Run the workflow on all branches.
-on: [push, pull_request]
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
 
 jobs:
   # Run tests and generate code coverage report.

.gitignore

@@ -60,3 +60,6 @@ bin/
 
 gradle.properties
 .coverage
+
+# Don't commit test run logs.
+tests/run.log

cwms/__init__.py

@@ -18,6 +18,7 @@
 from cwms.projects.project_locks import *
 from cwms.projects.projects import *
 from cwms.projects.water_supply.accounting import *
+from cwms.properties.properties import *
 from cwms.ratings.ratings import *
 from cwms.ratings.ratings_spec import *
 from cwms.ratings.ratings_template import *

cwms/api.py

@@ -34,16 +34,39 @@
 import logging
 from http import HTTPStatus
 from json import JSONDecodeError
-from typing import Any, Optional, cast
+from typing import Any, Final, Optional, cast
 
-from requests import Response, adapters
+from requests import Request, Response, adapters
 from requests.exceptions import RetryError as RequestsRetryError
 from requests_toolbelt import sessions  # type: ignore
 from requests_toolbelt.sessions import BaseUrlSession  # type: ignore
 from urllib3.util.retry import Retry
 
 from cwms.cwms_types import JSON, RequestParams
 
+LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
+
+
+from opentelemetry import propagate, trace
+
+# Setup telemetry
+from opentelemetry.instrumentation.requests import RequestsInstrumentor
+
+
+# Propagate the current W3CTraceContext to the request.
+def request_hook(span: trace.Span, request: Request) -> None:
+    ctx = trace.set_span_in_context(span)
+    propagate.get_global_textmap().inject(request.headers, ctx)
+
+
+tracer = trace.get_tracer(__name__)
+
+RequestsInstrumentor().instrument(
+    tracer=tracer,
+    tracer_provider=trace.get_tracer_provider(),
+    request_hook=request_hook,
+)
+
 # Specify the default API root URL and version.
 API_ROOT = "https://cwms-data.usace.army.mil/cwms-data/"
 API_VERSION = 2
@@ -190,7 +213,7 @@ def init_session(
     if api_root:
         # Ensure the API_ROOT ends with a single slash
         api_root = api_root.rstrip("/") + "/"
-        logging.debug(f"Initializing root URL: api_root={api_root}")
+        LOGGER.debug(f"Initializing root URL: api_root={api_root}")
         SESSION = sessions.BaseUrlSession(base_url=api_root)
         adapter = adapters.HTTPAdapter(
             pool_connections=pool_connections,
@@ -200,7 +223,7 @@ def init_session(
         SESSION.mount("https://", adapter)
     if token:
         if api_key:
-            logging.warning(
+            LOGGER.warning(
                 "Both token and api_key were provided to init_session(); using token for Authorization."
             )
         # Ensure we don't provide the bearer text twice
@@ -300,12 +323,13 @@ def _process_response(response: Response) -> Any:
         # Fallback for remaining content types
         return response.content.decode("utf-8")
     except JSONDecodeError as error:
-        logging.error(
+        LOGGER.error(
             f"Error decoding CDA response as JSON: {error} on line {error.lineno}\n\tFalling back to text"
         )
         return response.text
 
 
+@tracer.start_as_current_span("get")
 def get(
     endpoint: str,
     params: Optional[RequestParams] = None,
@@ -328,18 +352,24 @@ def get(
     Raises:
         ApiError: If an error response is return by the API.
     """
-
+    span = trace.get_current_span()
     headers = {"Accept": api_version_text(api_version)}
     try:
         with SESSION.get(endpoint, params=params, headers=headers) as response:
             if not response.ok:
                 logging.error(f"CDA Error: response={response}")
-                raise ApiError(response)
+                error = ApiError(response)
+                span.set_status(status=trace.StatusCode.ERROR)
+                span.record_exception(error)
+                raise error
             return _process_response(response)
     except RequestsRetryError as error:
+        span.set_status(status=trace.StatusCode.ERROR)
+        span.record_exception(error)
         raise _unwrap_retry_error(error) from None
 
 
+@tracer.start_as_current_span("get-paging")
 def get_with_paging(
     selector: str,
     endpoint: str,
@@ -366,6 +396,8 @@ def get_with_paging(
     """
 
     first_pass = True
+    current_span = trace.get_current_span()
+    current_span.set_attribute("page", "first")
     while (params["page"] is not None) or first_pass:
         temp = get(endpoint, params, api_version=api_version)
         if first_pass:
@@ -374,20 +406,23 @@ def get_with_paging(
             response[selector] = response[selector] + temp[selector]
         if "next-page" in temp.keys():
             params["page"] = temp["next-page"]
+            current_span.set_attribute("page", params["page"])
         else:
             params["page"] = None
+            current_span.set_attribute("page", "last")
         first_pass = False
     return response
 
 
+@tracer.start_as_current_span("post")
 def _post_function(
     endpoint: str,
     data: Any,
     params: Optional[RequestParams] = None,
     *,
     api_version: int = API_VERSION,
 ) -> Any:
-
+    span = trace.get_current_span()
     # post requires different headers than get for
     headers = {"accept": "*/*", "Content-Type": api_version_text(api_version)}
     if isinstance(data, dict) or isinstance(data, list):
@@ -398,9 +433,14 @@ def _post_function(
         ) as response:
             if not response.ok:
                 logging.error(f"CDA Error: response={response}")
-                raise ApiError(response)
+                error = ApiError(response)
+                span.set_status(status=trace.StatusCode.ERROR)
+                span.record_exception(error)
+                raise error
             return response
     except RequestsRetryError as error:
+        span.set_status(status=trace.StatusCode.ERROR)
+        span.record_exception(error)
         raise _unwrap_retry_error(error) from None
 
 
@@ -431,6 +471,7 @@ def post(
     _post_function(endpoint=endpoint, data=data, params=params, api_version=api_version)
 
 
+@tracer.start_as_current_span("post_with_return")
 def post_with_returned_data(
     endpoint: str,
     data: Any,
@@ -462,6 +503,7 @@ def post_with_returned_data(
     return _process_response(response)
 
 
+@tracer.start_as_current_span("patch")
 def patch(
     endpoint: str,
     data: Optional[Any] = None,
@@ -486,7 +528,7 @@ def patch(
     Raises:
         ApiError: If an error response is return by the API.
     """
-
+    span = trace.get_current_span()
     headers = {"accept": "*/*", "Content-Type": api_version_text(api_version)}
 
     if data and isinstance(data, dict) or isinstance(data, list):
@@ -497,11 +539,17 @@ def patch(
         ) as response:
             if not response.ok:
                 logging.error(f"CDA Error: response={response}")
-                raise ApiError(response)
+                error = ApiError(response)
+                span.set_status(status=trace.StatusCode.ERROR)
+                span.record_exception(error)
+                raise error
     except RequestsRetryError as error:
+        span.set_status(status=trace.StatusCode.ERROR)
+        span.record_exception(error)
         raise _unwrap_retry_error(error) from None
 
 
+@tracer.start_as_current_span("delete")
 def delete(
     endpoint: str,
     params: Optional[RequestParams] = None,
@@ -521,12 +569,17 @@ def delete(
     Raises:
         ApiError: If an error response is return by the API.
     """
-
+    span = trace.get_current_span()
     headers = {"Accept": api_version_text(api_version)}
     try:
         with SESSION.delete(endpoint, params=params, headers=headers) as response:
             if not response.ok:
                 logging.error(f"CDA Error: response={response}")
-                raise ApiError(response)
+                error = ApiError(response)
+                span.set_status(status=trace.StatusCode.ERROR)
+                span.record_exception(error)
+                raise error
     except RequestsRetryError as error:
+        span.set_status(status=trace.StatusCode.ERROR)
+        span.record_exception(error)
         raise _unwrap_retry_error(error) from None