fix: Apply PR#861 review suggestions.

- Introduced type hints and code cleanup.
- removed None defaults from .get() calls.
- import reorganization.
- boolean expression formatting.

Signed-off-by: Paulo Vital <paulo.vital@ibm.com>

Author: pvital

src/instana/collector/helpers/fargate/docker.py

@@ -5,15 +5,20 @@
 
 from __future__ import division
 
-from ....log import logger
-from ....util import DictionaryOfStan
-from ..base import BaseHelper
+from typing import TYPE_CHECKING, Any, Type
+
+from instana.collector.helpers.base import BaseHelper
+from instana.log import logger
+from instana.util import DictionaryOfStan
+
+if TYPE_CHECKING:
+    from instana.collector.base import BaseCollector
 
 
 class DockerHelper(BaseHelper):
     """This class acts as a helper to collect Docker snapshot and metric information"""
 
-    def __init__(self, collector):
+    def __init__(self, collector: Type[BaseCollector]) -> None:
         super(DockerHelper, self).__init__(collector)
 
         # The metrics from the previous report cycle
@@ -23,7 +28,7 @@ def __init__(self, collector):
         # Indexed by docker_id:  self.previous_blkio[docker_id][metric]
         self.previous_blkio = DictionaryOfStan()
 
-    def collect_metrics(self, **kwargs):
+    def collect_metrics(self, **kwargs: Any) -> list[dict[str, Any]]:
         """
         Collect and return docker metrics (and optionally snapshot data) for this task
         @return: list - with one or more plugin entities
@@ -33,7 +38,7 @@ def collect_metrics(self, **kwargs):
             if self.collector.task_metadata is not None:
                 containers = self.collector.task_metadata.get("Containers", [])
                 for container in containers:
-                    plugin_data = dict()
+                    plugin_data = {}
                     plugin_data["name"] = "com.instana.plugin.docker"
                     docker_id = container.get("DockerId")
 
@@ -43,7 +48,7 @@ def collect_metrics(self, **kwargs):
 
                     plugin_data["entityId"] = f"{task_arn}::{name}"
                     plugin_data["data"] = DictionaryOfStan()
-                    plugin_data["data"]["Id"] = container.get("DockerId", None)
+                    plugin_data["data"]["Id"] = container.get("DockerId")
 
                     with_snapshot = kwargs.get("with_snapshot", False)
                     # Metrics
@@ -61,25 +66,27 @@ def collect_metrics(self, **kwargs):
             logger.debug("DockerHelper.collect_metrics: ", exc_info=True)
         return plugins
 
-    def _collect_container_snapshot(self, plugin_data, container):
+    def _collect_container_snapshot(
+        self, plugin_data: dict[str, Any], container: dict[str, Any]
+    ) -> None:
         try:
             # Snapshot Data
-            plugin_data["data"]["Created"] = container.get("CreatedAt", None)
-            plugin_data["data"]["Started"] = container.get("StartedAt", None)
-            plugin_data["data"]["Image"] = container.get("Image", None)
-            plugin_data["data"]["Labels"] = container.get("Labels", None)
-            plugin_data["data"]["Ports"] = container.get("Ports", None)
+            plugin_data["data"]["Created"] = container.get("CreatedAt")
+            plugin_data["data"]["Started"] = container.get("StartedAt")
+            plugin_data["data"]["Image"] = container.get("Image")
+            plugin_data["data"]["Labels"] = container.get("Labels")
+            plugin_data["data"]["Ports"] = container.get("Ports")
 
             networks = container.get("Networks", [])
             if len(networks) >= 1:
-                plugin_data["data"]["NetworkMode"] = networks[0].get(
-                    "NetworkMode", None
-                )
+                plugin_data["data"]["NetworkMode"] = networks[0].get("NetworkMode")
         except Exception:
             logger.debug("_collect_container_snapshot: ", exc_info=True)
 
-    def _collect_container_metrics(self, plugin_data, docker_id, with_snapshot):
-        container = self.collector.task_stats_metadata.get(docker_id, None)
+    def _collect_container_metrics(
+        self, plugin_data: dict[str, Any], docker_id: str, with_snapshot: bool
+    ) -> None:
+        container = self.collector.task_stats_metadata.get(docker_id)
         if container is not None:
             self._collect_network_metrics(
                 container, plugin_data, docker_id, with_snapshot
@@ -93,10 +100,14 @@ def _collect_container_metrics(self, plugin_data, docker_id, with_snapshot):
             )
 
     def _collect_network_metrics(
-        self, container, plugin_data, docker_id, with_snapshot
-    ):
+        self,
+        container: dict[str, Any],
+        plugin_data: dict[str, Any],
+        docker_id: str,
+        with_snapshot: bool,
+    ) -> None:
         try:
-            networks = container.get("networks", None)
+            networks = container.get("networks")
             tx_bytes_total = tx_dropped_total = tx_errors_total = tx_packets_total = 0
             rx_bytes_total = rx_dropped_total = rx_errors_total = rx_packets_total = 0
 
@@ -173,11 +184,17 @@ def _collect_network_metrics(
         except Exception:
             logger.debug("_collect_network_metrics: ", exc_info=True)
 
-    def _collect_cpu_metrics(self, container, plugin_data, docker_id, with_snapshot):
+    def _collect_cpu_metrics(
+        self,
+        container: dict[str, Any],
+        plugin_data: dict[str, Any],
+        docker_id: str,
+        with_snapshot: bool,
+    ) -> None:
         try:
             cpu_stats = container.get("cpu_stats", {})
-            cpu_usage = cpu_stats.get("cpu_usage", None)
-            throttling_data = cpu_stats.get("throttling_data", None)
+            cpu_usage = cpu_stats.get("cpu_usage")
+            throttling_data = cpu_stats.get("throttling_data")
 
             if cpu_usage is not None:
                 online_cpus = cpu_stats.get("online_cpus", 1)
@@ -234,10 +251,16 @@ def _collect_cpu_metrics(self, container, plugin_data, docker_id, with_snapshot)
         except Exception:
             logger.debug("_collect_cpu_metrics: ", exc_info=True)
 
-    def _collect_memory_metrics(self, container, plugin_data, docker_id, with_snapshot):
+    def _collect_memory_metrics(
+        self,
+        container: dict[str, Any],
+        plugin_data: dict[str, Any],
+        docker_id: str,
+        with_snapshot: bool,
+    ) -> None:
         try:
             memory = container.get("memory_stats", {})
-            memory_stats = memory.get("stats", None)
+            memory_stats = memory.get("stats")
 
             self.apply_delta(
                 memory,
@@ -307,11 +330,17 @@ def _collect_memory_metrics(self, container, plugin_data, docker_id, with_snapsh
         except Exception:
             logger.debug("_collect_memory_metrics: ", exc_info=True)
 
-    def _collect_blkio_metrics(self, container, plugin_data, docker_id, with_snapshot):
+    def _collect_blkio_metrics(
+        self,
+        container: dict[str, Any],
+        plugin_data: dict[str, Any],
+        docker_id: str,
+        with_snapshot: bool,
+    ) -> None:
         try:
-            blkio_stats = container.get("blkio_stats", None)
+            blkio_stats = container.get("blkio_stats")
             if blkio_stats is not None:
-                service_bytes = blkio_stats.get("io_service_bytes_recursive", None)
+                service_bytes = blkio_stats.get("io_service_bytes_recursive")
                 if service_bytes is not None:
                     for entry in service_bytes:
                         if entry["op"] == "Read":

src/instana/util/config.py

@@ -348,7 +348,7 @@ def parse_span_disabling_dict(items: Dict[str, bool]) -> Tuple[List[str], List[s
 
 def get_disable_trace_configurations_from_env() -> Tuple[List[str], List[str]]:
     # Read INSTANA_TRACING_DISABLE environment variable
-    if tracing_disable := os.environ.get("INSTANA_TRACING_DISABLE", None):
+    if tracing_disable := os.environ.get("INSTANA_TRACING_DISABLE"):
         if is_truthy(tracing_disable):
             # INSTANA_TRACING_DISABLE is True/true/1, then we disable all tracing
             disabled_spans = []
@@ -388,14 +388,14 @@ def get_disable_trace_configurations_from_yaml() -> Tuple[List[str], List[str]]:
     if not root_key:
         return [], []
 
-    if tracing_disable_config := config_reader.data[root_key].get("disable", None):
+    if tracing_disable_config := config_reader.data[root_key].get("disable"):
         return parse_span_disabling(tracing_disable_config)
     return [], []
 
 
 def get_disable_trace_configurations_from_local() -> Tuple[List[str], List[str]]:
     if "tracing" in config and (
-        tracing_disable_config := config["tracing"].get("disable", None)
+        tracing_disable_config := config["tracing"].get("disable")
     ):
         return parse_span_disabling(tracing_disable_config)
     return [], []

src/instana/util/span_utils.py

@@ -57,14 +57,10 @@ def match_key_filter(span_value: str, rule_value: str, match_type: str) -> bool:
 
     return bool(
         rule_value == "*"
-        or match_type == "strict"
-        and span_value == rule_value
-        or match_type == "contains"
-        and rule_value in span_value
-        or match_type == "startswith"
-        and span_value.startswith(rule_value)
-        or match_type == "endswith"
-        and span_value.endswith(rule_value)
+        or (match_type == "strict" and span_value == rule_value)
+        or (match_type == "contains" and rule_value in span_value)
+        or (match_type == "startswith" and span_value.startswith(rule_value))
+        or (match_type == "endswith" and span_value.endswith(rule_value))
     )
 
 

tests/apps/fastapi_app/__init__.py

@@ -2,17 +2,18 @@
 # (c) Copyright Instana Inc. 2020
 
 import uvicorn
-from ...helpers import testenv
-from instana.log import logger as logger
+
+from tests.helpers import testenv
 
 testenv["fastapi_port"] = 10816
 testenv["fastapi_server"] = "http://127.0.0.1:" + str(testenv["fastapi_port"])
 
 
 def launch_fastapi():
-    from .app import fastapi_server
     from instana.singletons import agent
 
+    from .app import fastapi_server
+
     # Hack together a manual custom headers list; We'll use this in tests
     agent.options.extra_http_headers = [
         "X-Capture-This",