refactor(agent): ServerlessAgent with Span Filter

Introduce a new ServerlessAgent abstract base class to eliminate code
duplication across serverless platforms and add support for the Span
Filtering feature.

This refactoring follows the Template Method pattern, allowing
platform-specific customization through abstract methods while
maintaining a single source of truth for the common serverless agent
behavior.

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

Author: pvital

src/instana/agent/aws_eks_fargate.py

@@ -1,105 +1,39 @@
-# (c) Copyright IBM Corp. 2023
+# (c) Copyright IBM Corp. 2023, 2026
 
 """
 The Instana agent (for AWS EKS Fargate) that manages
 monitoring state and reporting that data.
 """
 
-from instana.agent.base import BaseAgent
+from instana.agent.serverless import ServerlessAgent
 from instana.collector.aws_eks_fargate import EKSFargateCollector
 from instana.collector.helpers.eks.process import get_pod_name
-from instana.log import logger
 from instana.options import EKSFargateOptions
-from instana.util import to_json
-from instana.version import VERSION
 
 
-class EKSFargateAgent(BaseAgent):
-    """In-process agent for AWS Fargate"""
-
-    def __init__(self):
-        super(EKSFargateAgent, self).__init__()
+class EKSFargateAgent(ServerlessAgent):
+    """In-process agent for AWS EKS Fargate"""
 
+    def _initialize_platform(self) -> None:
+        """Initialize EKS Fargate specific options and pod name."""
         self.options = EKSFargateOptions()
-        self.collector = None
-        self.report_headers = None
-        self._can_send = False
         self.podname = get_pod_name()
 
-        # Update log level (if INSTANA_LOG_LEVEL was set)
-        self.update_log_level()
-
-        logger.info(
-            "Stan is on the EKS Pod on AWS Fargate scene.  Starting Instana instrumentation version: %s",
-            VERSION,
-        )
-
-        if self._validate_options():
-            self._can_send = True
-            self.collector = EKSFargateCollector(self)
-            self.collector.start()
-        else:
-            logger.warning(
-                "Required INSTANA_AGENT_KEY and/or INSTANA_ENDPOINT_URL environment variables not set.  "
-                "We will not be able to monitor this Pod."
-            )
-
-    def can_send(self):
-        """
-        Are we in a state where we can send data?
-        @return: Boolean
-        """
-        return self._can_send
-
-    def get_from_structure(self):
-        """
-        Retrieves the From data that is reported alongside monitoring data.
-        @return: dict()
-        """
-
-        return {"hl": True, "cp": "k8s", "e": self.podname}
+    def _create_collector(self) -> EKSFargateCollector:
+        """Create EKS Fargate collector."""
+        return EKSFargateCollector(self)
 
-    def report_data_payload(self, payload):
-        """
-        Used to report metrics and span data to the endpoint URL in self.options.endpoint_url
-        """
-        response = None
-        try:
-            if self.report_headers is None:
-                # Prepare request headers
-                self.report_headers = dict()
-                self.report_headers["Content-Type"] = "application/json"
-                self.report_headers["X-Instana-Host"] = self.podname
-                self.report_headers["X-Instana-Key"] = self.options.agent_key
+    def _get_entity_id(self) -> str:
+        """Get Kubernetes pod name."""
+        return self.podname
 
-            response = self.client.post(
-                self.__data_bundle_url(),
-                data=to_json(payload),
-                headers=self.report_headers,
-                timeout=self.options.timeout,
-                verify=self.options.ssl_verify,
-                proxies=self.options.endpoint_proxy,
-            )
+    def _get_cloud_provider(self) -> str:
+        """Kubernetes cloud provider."""
+        return "k8s"
 
-            if not 200 <= response.status_code < 300:
-                logger.info(
-                    "report_data_payload: Instana responded with status code %s",
-                    response.status_code,
-                )
-        except Exception as exc:
-            logger.debug("report_data_payload: connection error (%s)", type(exc))
-        return response
+    def _get_platform_name(self) -> str:
+        """Platform name for logging."""
+        return "EKS Pod on AWS Fargate"
 
-    def _validate_options(self):
-        """
-        Validate that the options used by this Agent are valid.  e.g. can we report data?
-        """
-        return (
-            self.options.endpoint_url is not None and self.options.agent_key is not None
-        )
 
-    def __data_bundle_url(self):
-        """
-        URL for posting metrics to the host agent.  Only valid when announced.
-        """
-        return f"{self.options.endpoint_url}/bundle"
+# Made with Bob

src/instana/agent/aws_fargate.py

@@ -1,104 +1,38 @@
-# (c) Copyright IBM Corp. 2021
+# (c) Copyright IBM Corp. 2021, 2026
 # (c) Copyright Instana Inc. 2020
 
 """
 The Instana agent (for AWS Fargate) that manages
 monitoring state and reporting that data.
 """
 
+from instana.agent.serverless import ServerlessAgent
 from instana.collector.aws_fargate import AWSFargateCollector
 from instana.options import AWSFargateOptions
 
-from ..log import logger
-from ..util import to_json
-from ..version import VERSION
-from .base import BaseAgent
 
-
-class AWSFargateAgent(BaseAgent):
+class AWSFargateAgent(ServerlessAgent):
     """In-process agent for AWS Fargate"""
 
-    def __init__(self):
-        super(AWSFargateAgent, self).__init__()
-
+    def _initialize_platform(self) -> None:
+        """Initialize AWS Fargate specific options."""
         self.options = AWSFargateOptions()
-        self.collector = None
-        self.report_headers = None
-        self._can_send = False
-
-        # Update log level (if INSTANA_LOG_LEVEL was set)
-        self.update_log_level()
-
-        logger.info(
-            "Stan is on the AWS Fargate scene.  Starting Instana instrumentation version: %s",
-            VERSION,
-        )
-
-        if self._validate_options():
-            self._can_send = True
-            self.collector = AWSFargateCollector(self)
-            self.collector.start()
-        else:
-            logger.warning(
-                "Required INSTANA_AGENT_KEY and/or INSTANA_ENDPOINT_URL environment variables not set.  "
-                "We will not be able monitor this AWS Fargate cluster."
-            )
-
-    def can_send(self):
-        """
-        Are we in a state where we can send data?
-        @return: Boolean
-        """
-        return self._can_send
 
-    def get_from_structure(self):
-        """
-        Retrieves the From data that is reported alongside monitoring data.
-        @return: dict()
-        """
-        return {"hl": True, "cp": "aws", "e": self.collector.get_fq_arn()}
+    def _create_collector(self) -> AWSFargateCollector:
+        """Create AWS Fargate collector."""
+        return AWSFargateCollector(self)
 
-    def report_data_payload(self, payload):
-        """
-        Used to report metrics and span data to the endpoint URL in self.options.endpoint_url
-        """
-        response = None
-        try:
-            if self.report_headers is None:
-                # Prepare request headers
-                self.report_headers = dict()
-                self.report_headers["Content-Type"] = "application/json"
-                self.report_headers["X-Instana-Host"] = self.collector.get_fq_arn()
-                self.report_headers["X-Instana-Key"] = self.options.agent_key
+    def _get_entity_id(self) -> str:
+        """Get Fargate task ARN."""
+        return self.collector.get_fq_arn()
 
-            response = self.client.post(
-                self.__data_bundle_url(),
-                data=to_json(payload),
-                headers=self.report_headers,
-                timeout=self.options.timeout,
-                verify=self.options.ssl_verify,
-                proxies=self.options.endpoint_proxy,
-            )
+    def _get_cloud_provider(self) -> str:
+        """AWS cloud provider."""
+        return "aws"
 
-            if not 200 <= response.status_code < 300:
-                logger.info(
-                    "report_data_payload: Instana responded with status code %s",
-                    response.status_code,
-                )
-        except Exception as exc:
-            logger.debug("report_data_payload: connection error (%s)", type(exc))
-        return response
+    def _get_platform_name(self) -> str:
+        """Platform name for logging."""
+        return "AWS Fargate"
 
-    def _validate_options(self):
-        """
-        Validate that the options used by this Agent are valid.  e.g. can we report data?
-        """
-        return (
-            self.options.endpoint_url is not None and self.options.agent_key is not None
-        )
 
-    def __data_bundle_url(self):
-        """
-        URL for posting metrics to the host agent.  Only valid when announced.
-        """
-        return f"{self.options.endpoint_url}/bundle"
+# Made with Bob

src/instana/agent/aws_lambda.py

@@ -1,107 +1,38 @@
-# (c) Copyright IBM Corp. 2021
+# (c) Copyright IBM Corp. 2021, 2026
 # (c) Copyright Instana Inc. 2020
 
 """
 The Instana Agent for AWS Lambda functions that manages
 monitoring state and reporting that data.
 """
 
-from typing import Any, Dict
-from instana.agent.base import BaseAgent
+from instana.agent.serverless import ServerlessAgent
 from instana.collector.aws_lambda import AWSLambdaCollector
-from instana.log import logger
 from instana.options import AWSLambdaOptions
-from instana.util import to_json
-from instana.version import VERSION
 
 
-class AWSLambdaAgent(BaseAgent):
+class AWSLambdaAgent(ServerlessAgent):
     """In-process Agent for AWS Lambda"""
 
-    def __init__(self) -> None:
-        super(AWSLambdaAgent, self).__init__()
-
-        self.collector = None
+    def _initialize_platform(self) -> None:
+        """Initialize AWS Lambda specific options."""
         self.options = AWSLambdaOptions()
-        self.report_headers = None
-        self._can_send = False
-
-        # Update log level from what Options detected
-        self.update_log_level()
-
-        logger.info(
-            f"Stan is on the AWS Lambda scene.  Starting Instana instrumentation version: {VERSION}",
-        )
-
-        if self._validate_options():
-            self._can_send = True
-            self.collector = AWSLambdaCollector(self)
-            self.collector.start()
-        else:
-            logger.warning(
-                "Required INSTANA_AGENT_KEY and/or INSTANA_ENDPOINT_URL environment variables not set.  "
-                "We will not be able monitor this function."
-            )
-
-    def can_send(self) -> bool:
-        """
-        Are we in a state where we can send data?
-        @return: Boolean
-        """
-        return self._can_send
-
-    def get_from_structure(self) -> Dict[str, Any]:
-        """
-        Retrieves the From data that is reported alongside monitoring data.
-        @return: dict()
-        """
-        return {"hl": True, "cp": "aws", "e": self.collector.get_fq_arn()}
 
-    def report_data_payload(self, payload):
-        """
-        Used to report metrics and span data to the endpoint URL in self.options.endpoint_url
-        """
-        response = None
-        try:
-            if self.report_headers is None:
-                # Prepare request headers
-                self.report_headers = dict()
-                self.report_headers["Content-Type"] = "application/json"
-                self.report_headers["X-Instana-Host"] = self.collector.get_fq_arn()
-                self.report_headers["X-Instana-Key"] = self.options.agent_key
+    def _create_collector(self) -> AWSLambdaCollector:
+        """Create AWS Lambda collector."""
+        return AWSLambdaCollector(self)
 
-            response = self.client.post(
-                self.__data_bundle_url(),
-                data=to_json(payload),
-                headers=self.report_headers,
-                timeout=self.options.timeout,
-                verify=self.options.ssl_verify,
-                proxies=self.options.endpoint_proxy,
-            )
+    def _get_entity_id(self) -> str:
+        """Get Lambda function ARN."""
+        return self.collector.get_fq_arn()
 
-            if 200 <= response.status_code < 300:
-                logger.debug(
-                    "report_data_payload: Instana responded with status code %s",
-                    response.status_code,
-                )
-            else:
-                logger.info(
-                    "report_data_payload: Instana responded with status code %s",
-                    response.status_code,
-                )
-        except Exception as exc:
-            logger.debug("report_data_payload: connection error (%s)", type(exc))
+    def _get_cloud_provider(self) -> str:
+        """AWS cloud provider."""
+        return "aws"
 
-        return response
+    def _get_platform_name(self) -> str:
+        """Platform name for logging."""
+        return "AWS Lambda"
 
-    def _validate_options(self) -> bool:
-        """
-        Validate that the options used by this Agent are valid.  e.g. can we report data?
-        """
-        return self.options.endpoint_url and self.options.agent_key
 
-    def __data_bundle_url(self) -> str:
-        """
-        URL for posting metrics to the host agent.  Only valid when announced.
-        """
-        return f"{self.options.endpoint_url}/bundle"
+# Made with Bob