Skip to content

Serverless Span Filtering #872

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pvital wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Serverless Span Filtering #872

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
87b4477
refactor(agent): ServerlessAgent with Span Filter
pvital May 11, 2026
cf2c136
tests(agent): Update tests for agent refactoring.
pvital May 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 19 additions & 85 deletions src/instana/agent/aws_eks_fargate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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):
Copy link
Copy Markdown
Contributor

@CagriYonca CagriYonca May 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Response type is not specified.

Suggested change
def _create_collector(self):
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
102 changes: 18 additions & 84 deletions src/instana/agent/aws_fargate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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):
Copy link
Copy Markdown
Contributor

@CagriYonca CagriYonca May 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Response type is not specified.

Suggested change
def _create_collector(self):
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
105 changes: 18 additions & 87 deletions src/instana/agent/aws_lambda.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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):
Copy link
Copy Markdown
Contributor

@CagriYonca CagriYonca May 13, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Response type is not specified.

Suggested change
def _create_collector(self):
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
Loading
Loading