Feature/sfm logs per endpoint (#120)

* Basic sfm sending
* EndpointStatuses reported explicitly
* Sending SFM log per endpoint
* SFM logs blocked for custom extensions
* Unit tests for SFM statuses per endpoint
* Refactored building of overall status, fixed enrichment, linter fixes.
* Removed `clear_endpoint_error` (redundant and potentially misleading)
* Status implementation extracted to a separate file
* Additional log if SFMs not allowed for custom extensions.

Author: DtDamianGrzes

dynatrace_extension/__about__.py

@@ -3,4 +3,4 @@
 # SPDX-License-Identifier: MIT
 
 
-__version__ = "1.6.3"
+__version__ = "1.7.0"

dynatrace_extension/__init__.py

@@ -6,7 +6,6 @@
 # ruff: noqa: F401
 
 from .sdk.activation import ActivationConfig, ActivationType
-from .sdk.communication import EndpointStatus, EndpointStatuses, IgnoreStatus, MultiStatus, Status, StatusValue
 from .sdk.event import Severity
 from .sdk.extension import DtEventType, Extension
 from .sdk.helper import (
@@ -25,3 +24,4 @@
     schedule_function,
 )
 from .sdk.metric import Metric, MetricType, SummaryStat
+from .sdk.status import EndpointStatus, EndpointStatuses, IgnoreStatus, MultiStatus, Status, StatusValue

dynatrace_extension/sdk/callback.py

@@ -9,7 +9,7 @@
 from timeit import default_timer as timer
 
 from .activation import ActivationType
-from .communication import EndpointStatuses, IgnoreStatus, MultiStatus, Status, StatusValue
+from .status import EndpointStatuses, IgnoreStatus, MultiStatus, Status, StatusValue
 
 
 class WrappedCallback:

dynatrace_extension/sdk/communication.py

@@ -10,11 +10,10 @@
 from abc import ABC, abstractmethod
 from collections.abc import Generator, Sequence
 from dataclasses import dataclass
-from enum import Enum
 from pathlib import Path
-from threading import RLock
 from typing import Any, TypeVar
 
+from .status import Status
 from .vendor.mureq.mureq import HTTPException, Response, request
 
 CONTENT_TYPE_JSON = "application/json;charset=utf-8"
@@ -30,173 +29,6 @@
 HTTP_BAD_REQUEST = 400
 
 
-class StatusValue(Enum):
-    EMPTY = ""
-    OK = "OK"
-    GENERIC_ERROR = "GENERIC_ERROR"
-    INVALID_ARGS_ERROR = "INVALID_ARGS_ERROR"
-    EEC_CONNECTION_ERROR = "EEC_CONNECTION_ERROR"
-    INVALID_CONFIG_ERROR = "INVALID_CONFIG_ERROR"
-    AUTHENTICATION_ERROR = "AUTHENTICATION_ERROR"
-    DEVICE_CONNECTION_ERROR = "DEVICE_CONNECTION_ERROR"
-    WARNING = "WARNING"
-    UNKNOWN_ERROR = "UNKNOWN_ERROR"
-
-
-class IgnoreStatus:
-    pass
-
-
-class Status:
-    def __init__(self, status: StatusValue = StatusValue.EMPTY, message: str = "", timestamp: int | None = None):
-        self.status = status
-        self.message = message
-        self.timestamp = timestamp
-
-    def to_json(self) -> dict:
-        status = {"status": self.status.value, "message": self.message}
-        if self.timestamp:
-            status["timestamp"] = self.timestamp  # type: ignore
-        return status
-
-    def __repr__(self):
-        return json.dumps(self.to_json())
-
-    def is_error(self) -> bool:
-        # WARNING is treated as an error
-        return self.status not in (StatusValue.OK, StatusValue.EMPTY)
-
-    def is_warning(self) -> bool:
-        return self.status == StatusValue.WARNING
-
-
-class MultiStatus:
-    def __init__(self):
-        self.statuses: list[Status] = []
-
-    def add_status(self, status: StatusValue, message):
-        self.statuses.append(Status(status, message))
-
-    def build(self) -> Status:
-        ret = Status(StatusValue.OK)
-        if len(self.statuses) == 0:
-            return ret
-
-        messages = []
-        all_ok = True
-        all_err = True
-        any_warning = False
-
-        for stored_status in self.statuses:
-            if stored_status.message != "":
-                messages.append(stored_status.message)
-
-            if stored_status.is_warning():
-                any_warning = True
-
-            if stored_status.is_error():
-                all_ok = False
-            else:
-                all_err = False
-
-        ret.message = ", ".join(messages)
-
-        if any_warning:
-            ret.status = StatusValue.WARNING
-        elif all_ok:
-            ret.status = StatusValue.OK
-        elif all_err:
-            ret.status = StatusValue.GENERIC_ERROR
-        else:
-            ret.status = StatusValue.WARNING
-
-        return ret
-
-
-class EndpointStatus:
-    def __init__(self, endpoint_hint: str, short_status: StatusValue, message: str):
-        self.endpoint = endpoint_hint
-        self.status: StatusValue = short_status
-        self.message = message
-
-    def __str__(self):
-        return str(self.__dict__)
-
-
-class EndpointStatuses:
-    class TooManyEndpointStatusesError(Exception):
-        pass
-
-    class MergeConflictError(Exception):
-        def __init__(self, first: EndpointStatus, second: EndpointStatus):
-            super().__init__(f"Endpoint Statuses conflict while merging - first: {first}; second: {second}")
-
-    def __init__(self, total_endpoints_number: int):
-        self._lock = RLock()
-        self._faulty_endpoints: dict[str, EndpointStatus] = {}
-        self._num_endpoints = total_endpoints_number
-
-    def add_endpoint_status(self, status: EndpointStatus):
-        with self._lock:
-            if status.status == StatusValue.OK:
-                self.clear_endpoint_error(status.endpoint)
-            else:
-                if len(self._faulty_endpoints) == self._num_endpoints:
-                    message = "Cannot add another endpoint status. \
-                    The number of reported statuses already has reached preconfigured maximum of {self._num_endpoints} endpoints."
-                    raise EndpointStatuses.TooManyEndpointStatusesError(message)
-
-                self._faulty_endpoints[status.endpoint] = status
-
-    def clear_endpoint_error(self, endpoint_hint: str):
-        with self._lock:
-            try:
-                del self._faulty_endpoints[endpoint_hint]
-            except KeyError:
-                pass
-
-    def merge(self, other: EndpointStatuses):
-        with self._lock:
-            with other._lock:
-                self._num_endpoints += other._num_endpoints
-
-                for endpoint, status in other._faulty_endpoints.items():
-                    if endpoint not in self._faulty_endpoints.keys():
-                        self._faulty_endpoints[endpoint] = status
-                    else:
-                        self._num_endpoints -= 1
-                        raise EndpointStatuses.MergeConflictError(
-                            self._faulty_endpoints[endpoint], other._faulty_endpoints[endpoint]
-                        )
-
-    def build_common_status(self) -> Status:
-        with self._lock:
-            ok_count = self._num_endpoints - len(self._faulty_endpoints)
-            nok_count = len(self._faulty_endpoints)
-
-            if nok_count == 0:
-                return Status(StatusValue.OK, f"Endpoints OK: {self._num_endpoints} NOK: 0")
-
-            error_messages = []
-            for ep_status in self._faulty_endpoints.values():
-                error_messages.append(f"{ep_status.endpoint} - {ep_status.status.value} {ep_status.message}")
-            common_msg = ", ".join(error_messages)
-
-            # Determine status value
-            all_endpoints_faulty = nok_count == self._num_endpoints
-            has_warning_status = StatusValue.WARNING in [
-                ep_status.status for ep_status in self._faulty_endpoints.values()
-            ]
-
-            if all_endpoints_faulty and not has_warning_status:
-                status_value = StatusValue.GENERIC_ERROR
-            else:
-                status_value = StatusValue.WARNING
-
-            message = f"Endpoints OK: {ok_count} NOK: {nok_count} NOK_reported_errors: {common_msg}"
-            return Status(status=status_value, message=message)
-
-
 class CommunicationClient(ABC):
     """
     Abstract class for extension communication
@@ -250,6 +82,10 @@ def get_cluster_time_diff(self) -> int:
     def send_dt_event(self, event: dict) -> None:
         pass
 
+    @abstractmethod
+    def send_sfm_logs(self, sfm_logs: dict | list[dict]) -> list[dict | None]:
+        pass
+
 
 class HttpClient(CommunicationClient):
     """
@@ -261,6 +97,7 @@ def __init__(self, base_url: str, datasource_id: str, id_token_file_path: str, l
         self._extension_config_url = f"{base_url}/extconfig/{datasource_id}"
         self._metric_url = f"{base_url}/mint/{datasource_id}"
         self._sfm_url = f"{base_url}/sfm/{datasource_id}"
+        self._sfm_logs_url = f"{base_url}/sfmlogs/{datasource_id}"
         self._keep_alive_url = f"{base_url}/alive/{datasource_id}"
         self._timediff_url = f"{base_url}/timediffms"
         self._events_url = f"{base_url}/logs/{datasource_id}"
@@ -417,7 +254,13 @@ def send_metrics(self, mint_lines: list[str]) -> list[MintResponse]:
 
     def send_events(self, events: dict | list[dict], eec_enrichment: bool = True) -> list[dict | None]:
         self.logger.debug(f"Sending log events: {events}")
+        return self._send_events(self._events_url, events, eec_enrichment)
+
+    def send_sfm_logs(self, sfm_logs: dict | list[dict]):
+        self.logger.debug(f"Sending SFM logs: {sfm_logs}")
+        return self._send_events(self._sfm_logs_url, sfm_logs)
 
+    def _send_events(self, url, events: dict | list[dict], eec_enrichment: bool = True) -> list[dict | None]:
         responses = []
         if isinstance(events, dict):
             events = [events]
@@ -426,7 +269,7 @@ def send_events(self, events: dict | list[dict], eec_enrichment: bool = True) ->
         for batch in batches:
             try:
                 eec_response = self._make_request(
-                    self._events_url,
+                    url,
                     "POST",
                     batch,
                     extra_headers={"Content-Type": CONTENT_TYPE_JSON, "eec-enrichment": str(eec_enrichment).lower()},
@@ -583,6 +426,17 @@ def replace_secrets_in_activation_config(self, secrets: dict, activation_config_
 
         return activation_config_string
 
+    def send_sfm_logs(self, sfm_logs: dict | list[dict]) -> list[dict | None]:
+        if isinstance(sfm_logs, dict):
+            sfm_logs = [sfm_logs]
+
+        self.logger.info(f"send_sfm_logs: {len(sfm_logs)} logs")
+
+        if self.print_metrics:
+            for log in sfm_logs:
+                self.logger.info(f"send_sfm_log: {log}")
+        return []
+
 
 def divide_into_batches(
     items: Sequence[dict | str], max_size_bytes: int, join_with: str | None = None

dynatrace_extension/sdk/extension.py

@@ -20,19 +20,12 @@
 
 from .activation import ActivationConfig, ActivationType
 from .callback import WrappedCallback
-from .communication import (
-    CommunicationClient,
-    DebugClient,
-    EndpointStatuses,
-    HttpClient,
-    IgnoreStatus,
-    Status,
-    StatusValue,
-)
+from .communication import CommunicationClient, DebugClient, HttpClient
 from .event import Severity
 from .metric import Metric, MetricType, SfmMetric, SummaryStat
 from .runtime import RuntimeProperties
 from .snapshot import Snapshot
+from .status import EndpointStatuses, EndpointStatusesMap, IgnoreStatus, Status, StatusValue
 from .throttled_logger import StrictThrottledHandler, ThrottledHandler
 
 HEARTBEAT_INTERVAL = timedelta(seconds=50)
@@ -285,6 +278,13 @@ def __init__(self, name: str = "") -> None:
         # Error message from caught exception in self.initialize()
         self._initialization_error: str = ""
 
+        # Map of all Endpoint Statuses
+        self._sfm_logs_allowed = not self.extension_name.startswith("custom:")
+        if not self._sfm_logs_allowed:
+            self.logger.warning("SFM logs not allowed for custom extensions.")
+
+        self._ep_statuses = EndpointStatusesMap(send_sfm_logs_function=self._send_sfm_logs)
+
         self._parse_args()
 
         for function, interval, args, activation_type in Extension.schedule_decorators:
@@ -1037,8 +1037,7 @@ def _build_current_status(self):
             if internal_callback_error:
                 return Status(overall_status_value, "\n".join(messages))
 
-        # Handle regular statuses, merge all EndpointStatuses
-        ep_status_merged = EndpointStatuses(0)
+        # Handle regular statuses, report all EndpointStatuses
         all_ok = True
         all_err = True
         any_warning = False
@@ -1048,10 +1047,7 @@ def _build_current_status(self):
                 continue
 
             if isinstance(callback.status, EndpointStatuses):
-                try:
-                    ep_status_merged.merge(callback.status)
-                except EndpointStatuses.MergeConflictError as e:
-                    self.logger.exception(e)
+                self._ep_statuses.update_ep_statuses(callback.status)
                 continue
 
             if callback.status.is_warning():
@@ -1066,8 +1062,9 @@ def _build_current_status(self):
                 messages.append(f"{callback.name()}: {callback.status.status.value} - {callback.status.message}")
 
         # Handle merged EndpointStatuses
-        if ep_status_merged._num_endpoints > 0:
-            ep_status_merged = ep_status_merged.build_common_status()
+        if self._ep_statuses.contains_any_status():
+            self._ep_statuses.send_ep_logs()
+            ep_status_merged = self._ep_statuses.build_common_status()
             messages.insert(0, ep_status_merged.message)
 
             if ep_status_merged.is_warning():
@@ -1221,3 +1218,32 @@ def get_snapshot(self, snapshot_file: Path | str | None = None) -> Snapshot:
                 raise FileNotFoundError(msg)
 
         return Snapshot.parse_from_file(snapshot_file)
+
+    def _send_sfm_logs_internal(self, logs: dict | list[dict]):
+        try:
+            responses = self._client.send_sfm_logs(logs)
+
+            for response in responses:
+                with self._internal_callbacks_results_lock:
+                    self._internal_callbacks_results[self._send_sfm_logs_internal.__name__] = Status(StatusValue.OK)
+                    if not response or "error" not in response or "message" not in response["error"]:
+                        return
+                    self._internal_callbacks_results[self._send_sfm_logs_internal.__name__] = Status(
+                        StatusValue.GENERIC_ERROR, response["error"]["message"]
+                    )
+        except Exception as e:
+            api_logger.error(f"Error sending SFM logs: {e!r}", exc_info=True)
+            with self._internal_callbacks_results_lock:
+                self._internal_callbacks_results[self._send_sfm_logs_internal.__name__] = Status(StatusValue.GENERIC_ERROR, str(e))
+
+
+    def _send_sfm_logs(self, logs: dict | list[dict]):
+        if not self._sfm_logs_allowed or not logs:
+            return
+
+        for log in logs:
+            log.update(self._metadata)
+            log["dt.extension.config.label"] = self.monitoring_config_name
+            log.pop("monitoring.configuration", None)
+
+        self._internal_executor.submit(self._send_sfm_logs_internal, logs)