diff --git a/.changesets/support-logging.md b/.changesets/support-logging.md new file mode 100644 index 00000000..c3389da6 --- /dev/null +++ b/.changesets/support-logging.md @@ -0,0 +1,8 @@ +--- +bump: minor +type: add +--- + +Support logging through the external collector experimental feature. When the `collector_endpoint` configuration option is provided, the OpenTelemetry stack will be automatically configured to instrument logs. + +The `logging` module will be automatically instrumented, such that log lines emitted through loggers that propagate to the root logger will be automatically sent to AppSignal. To disable this behaviour, add `"logging"` to the `disable_default_instrumentations` configuration option list. \ No newline at end of file diff --git a/.changesets/support-usage-with-external-collector.md b/.changesets/support-usage-with-external-collector.md new file mode 100644 index 00000000..9bb51e57 --- /dev/null +++ b/.changesets/support-usage-with-external-collector.md @@ -0,0 +1,14 @@ +--- +bump: minor +type: add +--- + +Support usage with external collector. When the `collector_endpoint` configuration option is provided, instead of booting up the AppSignal agent bundled with the application, the OpenTelemetry stack will be configured to send data to the given collector. + +This is an **experimental** feature. The following functionality is not currently supported when using the collector: + +- NGINX metrics +- StatsD metrics +- Host metrics + +Some configuration options are only supported when using the agent or when using the collector. A warning will be emitted if a configuration option that is only supported by one is set while using the other. \ No newline at end of file diff --git a/conftest.py b/conftest.py index 48cb9860..6f3cd5a7 100644 --- a/conftest.py +++ b/conftest.py @@ -16,7 +16,6 @@ from opentelemetry.trace import set_tracer_provider from appsignal import probes -from appsignal.agent import agent from appsignal.check_in.heartbeat import ( _kill_continuous_heartbeats, _reset_heartbeat_continuous_interval_seconds, @@ -105,11 +104,6 @@ def stop_and_clear_probes_after_tests() -> Any: probes.clear() -@pytest.fixture(scope="function", autouse=True) -def reset_agent_active_state() -> Any: - agent.active = False - - @pytest.fixture(scope="function", autouse=True) def reset_global_client() -> Any: _reset_client() diff --git a/src/appsignal/agent.py b/src/appsignal/agent.py index 71240f8f..1d3c41ea 100644 --- a/src/appsignal/agent.py +++ b/src/appsignal/agent.py @@ -1,18 +1,27 @@ from __future__ import annotations +import os +import signal import subprocess +import time from dataclasses import dataclass from pathlib import Path +from . import internal_logger as logger +from .binary import Binary from .config import Config @dataclass -class Agent: +class Agent(Binary): package_path: Path = Path(__file__).parent agent_path: Path = package_path / "appsignal-agent" platform_path: Path = package_path / "_appsignal_platform" - active: bool = False + _active: bool = False + + @property + def active(self) -> bool: + return self._active def start(self, config: Config) -> None: config.set_private_environ() @@ -33,12 +42,24 @@ def start(self, config: Config) -> None: p.wait(timeout=1) returncode = p.returncode if returncode == 0: - self.active = True + self._active = True else: output, _ = p.communicate() out = output.decode("utf-8") print(f"AppSignal agent is unable to start ({returncode}): ", out) + def stop(self, config: Config) -> None: + working_dir = config.option("working_directory_path") or "/tmp/appsignal" + lock_path = os.path.join(working_dir, "agent.lock") + try: + with open(lock_path) as file: + line = file.readline() + pid = int(line.split(";")[2]) + os.kill(pid, signal.SIGTERM) + time.sleep(2) + except FileNotFoundError: + logger.info("Agent lock file not found; not stopping the agent") + def diagnose(self, config: Config) -> bytes: config.set_private_environ() return subprocess.run( @@ -56,6 +77,3 @@ def architecture_and_platform(self) -> list[str]: return file.read().split("-", 1) except FileNotFoundError: return ["", ""] - - -agent = Agent() diff --git a/src/appsignal/binary.py b/src/appsignal/binary.py new file mode 100644 index 00000000..c2f831e6 --- /dev/null +++ b/src/appsignal/binary.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod + +from .config import Config + + +class Binary(ABC): + @property + @abstractmethod + def active(self) -> bool: ... + + @abstractmethod + def start(self, config: Config) -> None: ... + + @abstractmethod + def stop(self, config: Config) -> None: ... + + +class NoopBinary(Binary): + def __init__(self, active: bool = False) -> None: + self._active = active + + @property + def active(self) -> bool: + return self._active + + def start(self, config: Config) -> None: + pass + + def stop(self, config: Config) -> None: + pass diff --git a/src/appsignal/client.py b/src/appsignal/client.py index cf46abad..bff064b7 100644 --- a/src/appsignal/client.py +++ b/src/appsignal/client.py @@ -1,12 +1,9 @@ from __future__ import annotations -import os -import signal -import time from typing import TYPE_CHECKING from . import internal_logger as logger -from .agent import agent +from .binary import NoopBinary from .config import Config, Options from .opentelemetry import start as start_opentelemetry from .probes import start as start_probes @@ -15,6 +12,8 @@ if TYPE_CHECKING: from typing_extensions import Unpack + from .binary import Binary + _client: Client | None = None @@ -26,11 +25,13 @@ def _reset_client() -> None: class Client: _config: Config + _binary: Binary def __init__(self, **options: Unpack[Options]) -> None: global _client self._config = Config(options) + self._binary = NoopBinary() _client = self @classmethod @@ -41,10 +42,13 @@ def config(cls) -> Config | None: return _client._config def start(self) -> None: + self._set_binary() + if self._config.is_active(): logger.info("Starting AppSignal") - agent.start(self._config) - if not agent.active: + self._config.warn() + self._binary.start(self._config) + if not self._binary.active: return start_opentelemetry(self._config) self._start_probes() @@ -56,17 +60,23 @@ def stop(self) -> None: logger.info("Stopping AppSignal") scheduler().stop() - working_dir = self._config.option("working_directory_path") or "/tmp/appsignal" - lock_path = os.path.join(working_dir, "agent.lock") - try: - with open(lock_path) as file: - line = file.readline() - pid = int(line.split(";")[2]) - os.kill(pid, signal.SIGTERM) - time.sleep(2) - except FileNotFoundError: - logger.info("Agent lock file not found") + self._binary.stop(self._config) def _start_probes(self) -> None: if self._config.option("enable_minutely_probes"): start_probes() + + def _set_binary(self) -> None: + if self._config.should_use_external_collector(): + # When a custom collector endpoint is set, use a `NoopBinary` + # set to active, so that OpenTelemetry and probes are started, + # but the agent is not started. + logger.info( + "Not starting the AppSignal agent: using collector endpoint instead" + ) + self._binary = NoopBinary(active=True) + else: + # Use the agent when a custom collector endpoint is not set. + from .agent import Agent + + self._binary = Agent() diff --git a/src/appsignal/config.py b/src/appsignal/config.py index 6419e92b..39fa33fc 100644 --- a/src/appsignal/config.py +++ b/src/appsignal/config.py @@ -2,9 +2,11 @@ import os import platform +import socket import tempfile from typing import Any, ClassVar, List, Literal, TypedDict, cast, get_args +from . import internal_logger as logger from .__about__ import __version__ @@ -13,6 +15,7 @@ class Options(TypedDict, total=False): app_path: str | None bind_address: str | None ca_file_path: str | None + collector_endpoint: str | None cpu_count: float | None diagnose_endpoint: str | None disable_default_instrumentations: None | ( @@ -26,7 +29,11 @@ class Options(TypedDict, total=False): endpoint: str | None environment: str | None files_world_accessible: bool | None + filter_attributes: list[str] | None + filter_function_parameters: list[str] | None filter_parameters: list[str] | None + filter_request_payload: list[str] | None + filter_request_query_parameters: list[str] | None filter_session_data: list[str] | None hostname: str | None host_role: str | None @@ -44,10 +51,15 @@ class Options(TypedDict, total=False): push_api_key: str | None revision: str | None request_headers: list[str] | None + response_headers: list[str] | None running_in_container: bool | None send_environment_metadata: bool | None + send_function_parameters: bool | None send_params: bool | None + send_request_payload: bool | None + send_request_query_parameters: bool | None send_session_data: bool | None + service_name: str | None statsd_port: str | int | None working_directory_path: str | None @@ -97,22 +109,23 @@ class Config: ) DefaultInstrumentation = Literal[ - "opentelemetry.instrumentation.aiopg", - "opentelemetry.instrumentation.asyncpg", - "opentelemetry.instrumentation.celery", - "opentelemetry.instrumentation.django", - "opentelemetry.instrumentation.flask", - "opentelemetry.instrumentation.jinja2", - "opentelemetry.instrumentation.mysql", - "opentelemetry.instrumentation.mysqlclient", - "opentelemetry.instrumentation.pika", - "opentelemetry.instrumentation.psycopg", - "opentelemetry.instrumentation.psycopg2", - "opentelemetry.instrumentation.pymysql", - "opentelemetry.instrumentation.redis", - "opentelemetry.instrumentation.requests", - "opentelemetry.instrumentation.sqlite3", - "opentelemetry.instrumentation.sqlalchemy", + "aiopg", + "asyncpg", + "celery", + "django", + "flask", + "jinja2", + "mysql", + "mysqlclient", + "pika", + "psycopg", + "psycopg2", + "pymysql", + "redis", + "requests", + "sqlite3", + "sqlalchemy", + "logging", ] DEFAULT_INSTRUMENTATIONS = cast( List[DefaultInstrumentation], list(get_args(DefaultInstrumentation)) @@ -140,9 +153,29 @@ def is_active(self) -> bool: def option(self, option: str) -> Any: return self.options.get(option) + # Whether it should instrument logging. + def should_instrument_logging(self) -> bool: + # The agent does not support logging, so this is equivalent + # to whether it should use a collector. + return self.should_use_collector() + + # Whether it should use a collector to send data to AppSignal. + def should_use_collector(self) -> bool: + # This is currently equivalent to whether it should use an + # external collector. + return self.should_use_external_collector() + + # Whether it should use a collector to send data to AppSignal + # which is booted externally to this integration. + def should_use_external_collector(self) -> bool: + return self.option("collector_endpoint") is not None + @staticmethod def load_from_system() -> Options: - return Options(app_path=os.getcwd()) + return Options( + app_path=os.getcwd(), + hostname=os.environ.get("HOSTNAME") or socket.gethostname(), + ) @staticmethod def load_from_environment() -> Options: @@ -150,6 +183,7 @@ def load_from_environment() -> Options: active=parse_bool(os.environ.get("APPSIGNAL_ACTIVE")), bind_address=os.environ.get("APPSIGNAL_BIND_ADDRESS"), ca_file_path=os.environ.get("APPSIGNAL_CA_FILE_PATH"), + collector_endpoint=os.environ.get("APPSIGNAL_COLLECTOR_ENDPOINT"), cpu_count=parse_float(os.environ.get("APPSIGNAL_CPU_COUNT")), diagnose_endpoint=os.environ.get("APPSIGNAL_DIAGNOSE_ENDPOINT"), disable_default_instrumentations=parse_disable_default_instrumentations( @@ -171,7 +205,17 @@ def load_from_environment() -> Options: files_world_accessible=parse_bool( os.environ.get("APPSIGNAL_FILES_WORLD_ACCESSIBLE") ), + filter_attributes=parse_list(os.environ.get("APPSIGNAL_FILTER_ATTRIBUTES")), + filter_function_parameters=parse_list( + os.environ.get("APPSIGNAL_FILTER_FUNCTION_PARAMETERS") + ), filter_parameters=parse_list(os.environ.get("APPSIGNAL_FILTER_PARAMETERS")), + filter_request_payload=parse_list( + os.environ.get("APPSIGNAL_FILTER_REQUEST_PAYLOAD") + ), + filter_request_query_parameters=parse_list( + os.environ.get("APPSIGNAL_FILTER_REQUEST_QUERY_PARAMETERS") + ), filter_session_data=parse_list( os.environ.get("APPSIGNAL_FILTER_SESSION_DATA") ), @@ -191,14 +235,25 @@ def load_from_environment() -> Options: push_api_key=os.environ.get("APPSIGNAL_PUSH_API_KEY"), revision=os.environ.get("APP_REVISION"), request_headers=parse_list(os.environ.get("APPSIGNAL_REQUEST_HEADERS")), + response_headers=parse_list(os.environ.get("APPSIGNAL_RESPONSE_HEADERS")), running_in_container=parse_bool( os.environ.get("APPSIGNAL_RUNNING_IN_CONTAINER") ), send_environment_metadata=parse_bool( os.environ.get("APPSIGNAL_SEND_ENVIRONMENT_METADATA") ), + send_function_parameters=parse_bool( + os.environ.get("APPSIGNAL_SEND_FUNCTION_PARAMETERS") + ), send_params=parse_bool(os.environ.get("APPSIGNAL_SEND_PARAMS")), + send_request_payload=parse_bool( + os.environ.get("APPSIGNAL_SEND_REQUEST_PAYLOAD") + ), + send_request_query_parameters=parse_bool( + os.environ.get("APPSIGNAL_SEND_REQUEST_QUERY_PARAMETERS") + ), send_session_data=parse_bool(os.environ.get("APPSIGNAL_SEND_SESSION_DATA")), + service_name=os.environ.get("APPSIGNAL_SERVICE_NAME"), statsd_port=os.environ.get("APPSIGNAL_STATSD_PORT"), working_directory_path=os.environ.get("APPSIGNAL_WORKING_DIRECTORY_PATH"), ) @@ -313,6 +368,122 @@ def _validate(self) -> None: if len(push_api_key.strip()) > 0: self.valid = True + def warn(self) -> None: + if self.should_use_collector(): + self._warn_agent_exclusive_options() + else: + self._warn_collector_exclusive_options() + + # Emit a warning if agent-exclusive configuration options are used. + def _warn_agent_exclusive_options(self) -> None: + exclusive_options = [ + "bind_address", + "cpu_count", + "dns_servers", + "enable_host_metrics", + "enable_nginx_metrics", + "enable_statsd", + "files_world_accessible", + "filter_parameters", + "host_role", + "nginx_port", + "opentelemetry_port", + "running_in_container", + "send_environment_metadata", + "send_params", + "working_directory_path", + "statsd_port", + ] + + option_specific_warnings = { + "filter_parameters": ( + "Use the 'filter_attributes', 'filter_function_parameters'," + " 'filter_request_payload' and 'filter_request_query_parameters'" + " configuration options instead." + ), + "send_params": ( + "Use the 'send_function_parameters', 'send_request_payload'" + " and 'send_request_query_parameters' configuration options instead." + ), + "opentelemetry_port": ( + "Set the collector's OpenTelemetry port as part of the" + " 'collector_endpoint' configuration option." + ), + } + + user_modified_options = self._filter_user_modified_options(exclusive_options) + + for option in user_modified_options: + logger.warning( + f"The collector is in use. The '{option}' configuration option" + " is only used by the agent and will be ignored." + ) + if option in option_specific_warnings: + logger.warning(option_specific_warnings[option]) + + if user_modified_options: + logger.info( + "To use the agent, unset the 'collector_endpoint' configuration option." + ) + + # Emit a warning if collector-exclusive configuration options are used. + def _warn_collector_exclusive_options(self) -> None: + exclusive_options = [ + "filter_attributes", + "filter_function_parameters", + "filter_request_payload", + "filter_request_query_parameters", + "response_headers", + "send_function_parameters", + "send_request_payload", + "send_request_query_parameters", + "service_name", + ] + + filter_warning = "Use the 'filter_parameters' option instead." + send_warning = "Use the 'send_params' option instead." + + option_specific_warnings = { + "filter_attributes": filter_warning, + "filter_function_parameters": filter_warning, + "filter_request_payload": filter_warning, + "filter_request_query_parameters": filter_warning, + "send_function_parameters": send_warning, + "send_request_payload": send_warning, + "send_request_query_parameters": send_warning, + } + + user_modified_options = self._filter_user_modified_options(exclusive_options) + + for option in user_modified_options: + logger.warning( + f"The agent is in use. The '{option}' configuration option" + " is only used by the collector and will be ignored." + ) + if option in option_specific_warnings: + logger.warning(option_specific_warnings[option]) + + if user_modified_options: + logger.info( + "To use the collector, set the 'collector_endpoint' configuration option." + ) + + # Filter a list of options, returning a list of those options for which + # a value has been set by the user (through the initialiser or in the + # environment) which differs from that of the default configuration. + def _filter_user_modified_options(self, options: list[str]) -> list[str]: + return [ + option + for option in options + if ( + ( + option in self.sources["initial"] + or option in self.sources["environment"] + ) + and self.option(option) != self.sources["default"].get(option) + ) + ] + def parse_bool(value: str | None) -> bool | None: if value is None: diff --git a/src/appsignal/opentelemetry.py b/src/appsignal/opentelemetry.py index b8522d27..0f41b893 100644 --- a/src/appsignal/opentelemetry.py +++ b/src/appsignal/opentelemetry.py @@ -1,11 +1,15 @@ from __future__ import annotations import os -from typing import TYPE_CHECKING, Callable, Mapping +from typing import TYPE_CHECKING, Callable, List, Mapping, Union, cast +from opentelemetry import _logs as logs from opentelemetry import metrics, trace +from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor from opentelemetry.sdk.metrics import ( Counter, Histogram, @@ -35,25 +39,25 @@ from opentelemetry.trace.span import Span -def add_aiopg_instrumentation() -> None: +def add_aiopg_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.aiopg import AiopgInstrumentor AiopgInstrumentor().instrument() -def add_asyncpg_instrumentation() -> None: +def add_asyncpg_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor AsyncPGInstrumentor().instrument() -def add_celery_instrumentation() -> None: +def add_celery_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.celery import CeleryInstrumentor CeleryInstrumentor().instrument() -def add_django_instrumentation() -> None: +def add_django_instrumentation(_config: Config) -> None: import json from django.http.request import HttpRequest @@ -69,7 +73,7 @@ def response_hook(span: Span, request: HttpRequest, response: HttpResponse) -> N DjangoInstrumentor().instrument(response_hook=response_hook) -def add_flask_instrumentation() -> None: +def add_flask_instrumentation(_config: Config) -> None: import json from urllib.parse import parse_qs @@ -85,93 +89,118 @@ def request_hook(span: Span, environ: dict[str, str]) -> None: FlaskInstrumentor().instrument(request_hook=request_hook) -def add_jinja2_instrumentation() -> None: +def add_jinja2_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.jinja2 import Jinja2Instrumentor Jinja2Instrumentor().instrument() -def add_mysql_instrumentation() -> None: +def add_mysql_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.mysql import MySQLInstrumentor MySQLInstrumentor().instrument() -def add_mysqlclient_instrumentation() -> None: +def add_mysqlclient_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.mysqlclient import MySQLClientInstrumentor MySQLClientInstrumentor().instrument() -def add_pika_instrumentation() -> None: +def add_pika_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.pika import PikaInstrumentor PikaInstrumentor().instrument() -def add_psycopg2_instrumentation() -> None: +def add_psycopg2_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor Psycopg2Instrumentor().instrument() -def add_psycopg_instrumentation() -> None: +def add_psycopg_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.psycopg import PsycopgInstrumentor PsycopgInstrumentor().instrument() -def add_pymysql_instrumentation() -> None: +def add_pymysql_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.pymysql import PyMySQLInstrumentor PyMySQLInstrumentor().instrument() -def add_redis_instrumentation() -> None: +def add_redis_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.redis import RedisInstrumentor RedisInstrumentor().instrument(sanitize_query=True) -def add_requests_instrumentation() -> None: +def add_requests_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.requests import RequestsInstrumentor RequestsInstrumentor().instrument() -def add_sqlalchemy_instrumentation() -> None: +def add_sqlalchemy_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor SQLAlchemyInstrumentor().instrument() -def add_sqlite3_instrumentation() -> None: +def add_sqlite3_instrumentation(_config: Config) -> None: from opentelemetry.instrumentation.sqlite3 import SQLite3Instrumentor SQLite3Instrumentor().instrument() -DefaultInstrumentationAdder = Callable[[], None] +def add_logging_instrumentation(config: Config) -> None: + # Do not add a root logging handler if we should not support + # instrumenting logging. + if not config.should_instrument_logging(): + return + + import logging + + from opentelemetry.sdk._logs import LoggingHandler + + # Attach OTel LoggingHandler to the root logger + handler = LoggingHandler(level=logging.NOTSET) + logger = logging.getLogger() + logger.addHandler(handler) + + # Configure the OpenTelemetry loggers and the AppSignal internal + # logger to not propagate to the root logger. This prevents + # internal logs from being sent to AppSignal. + opentelemetry_logger = logging.getLogger("opentelemetry") + opentelemetry_logger.propagate = False + appsignal_logger = logging.getLogger("appsignal") + appsignal_logger.propagate = False + + +DefaultInstrumentationAdder = Callable[[Config], None] DEFAULT_INSTRUMENTATION_ADDERS: Mapping[ Config.DefaultInstrumentation, DefaultInstrumentationAdder ] = { - "opentelemetry.instrumentation.aiopg": add_aiopg_instrumentation, - "opentelemetry.instrumentation.asyncpg": add_asyncpg_instrumentation, - "opentelemetry.instrumentation.celery": add_celery_instrumentation, - "opentelemetry.instrumentation.django": add_django_instrumentation, - "opentelemetry.instrumentation.flask": add_flask_instrumentation, - "opentelemetry.instrumentation.jinja2": add_jinja2_instrumentation, - "opentelemetry.instrumentation.mysql": add_mysql_instrumentation, - "opentelemetry.instrumentation.mysqlclient": add_mysqlclient_instrumentation, - "opentelemetry.instrumentation.pika": add_pika_instrumentation, - "opentelemetry.instrumentation.psycopg2": add_psycopg2_instrumentation, - "opentelemetry.instrumentation.psycopg": add_psycopg_instrumentation, - "opentelemetry.instrumentation.pymysql": add_pymysql_instrumentation, - "opentelemetry.instrumentation.redis": add_redis_instrumentation, - "opentelemetry.instrumentation.requests": add_requests_instrumentation, - "opentelemetry.instrumentation.sqlalchemy": add_sqlalchemy_instrumentation, - "opentelemetry.instrumentation.sqlite3": add_sqlite3_instrumentation, + "aiopg": add_aiopg_instrumentation, + "asyncpg": add_asyncpg_instrumentation, + "celery": add_celery_instrumentation, + "django": add_django_instrumentation, + "flask": add_flask_instrumentation, + "jinja2": add_jinja2_instrumentation, + "mysql": add_mysql_instrumentation, + "mysqlclient": add_mysqlclient_instrumentation, + "pika": add_pika_instrumentation, + "psycopg2": add_psycopg2_instrumentation, + "psycopg": add_psycopg_instrumentation, + "pymysql": add_pymysql_instrumentation, + "redis": add_redis_instrumentation, + "requests": add_requests_instrumentation, + "sqlalchemy": add_sqlalchemy_instrumentation, + "sqlite3": add_sqlite3_instrumentation, + "logging": add_logging_instrumentation, } @@ -183,16 +212,20 @@ def start(config: Config) -> None: request_headers ) - opentelemetry_port = config.option("opentelemetry_port") - _start_tracer(opentelemetry_port, config.option("log_level") == "trace") - _start_metrics(opentelemetry_port) + _start_tracer(config) + _start_metrics(config) + + # Configure OpenTelemetry logging only if a collector is used + # (it is not currently supported by the agent) + if config.should_instrument_logging(): + _start_logging(config) add_instrumentations(config) -def _otlp_span_processor(opentelemetry_port: str | int) -> BatchSpanProcessor: +def _otlp_span_processor(config: Config) -> BatchSpanProcessor: otlp_exporter = OTLPSpanExporter( - endpoint=f"http://localhost:{opentelemetry_port}/v1/traces" + endpoint=f"{_opentelemetry_endpoint(config)}/v1/traces", ) return BatchSpanProcessor(otlp_exporter) @@ -202,9 +235,11 @@ def _console_span_processor() -> SimpleSpanProcessor: return SimpleSpanProcessor(console_exporter) -def _start_tracer(opentelemetry_port: str | int, should_trace: bool = False) -> None: - otlp_span_processor = _otlp_span_processor(opentelemetry_port) - provider = TracerProvider() +def _start_tracer(config: Config) -> None: + otlp_span_processor = _otlp_span_processor(config) + provider = TracerProvider(resource=_resource(config)) + + should_trace = config.option("log_level") == "trace" if should_trace: multi_span_processor = ConcurrentMultiSpanProcessor() @@ -227,20 +262,93 @@ def _start_tracer(opentelemetry_port: str | int, should_trace: bool = False) -> } -def _start_metrics(opentelemetry_port: str | int) -> None: +def _start_metrics(config: Config) -> None: metric_exporter = OTLPMetricExporter( - endpoint=f"http://localhost:{opentelemetry_port}/v1/metrics", + endpoint=f"{_opentelemetry_endpoint(config)}/v1/metrics", preferred_temporality=METRICS_PREFERRED_TEMPORALITY, ) metric_reader = PeriodicExportingMetricReader( metric_exporter, export_interval_millis=10000 ) - resource = Resource(attributes={"appsignal.service.process_id": os.getpid()}) - provider = MeterProvider(resource=resource, metric_readers=[metric_reader]) + provider = MeterProvider(resource=_resource(config), metric_readers=[metric_reader]) metrics.set_meter_provider(provider) +def _start_logging(config: Config) -> None: + log_exporter = OTLPLogExporter( + endpoint=f"{_opentelemetry_endpoint(config)}/v1/logs", + ) + provider = LoggerProvider(resource=_resource(config)) + provider.add_log_record_processor(BatchLogRecordProcessor(log_exporter)) + + logs.set_logger_provider(provider) + + +def _resource(config: Config) -> Resource: + attributes = { + key: value + for key, value in { + "appsignal.config.name": config.options.get("name"), + "appsignal.config.environment": config.options.get("environment"), + "appsignal.config.push_api_key": config.options.get("push_api_key"), + "appsignal.config.revision": config.options.get("revision", "unknown"), + "appsignal.config.language_integration": "python", + "service.name": config.options.get("service_name", "unknown"), + "host.name": config.options.get("hostname", "unknown"), + "appsignal.service.process_id": os.getpid(), + "appsignal.config.filter_attributes": config.options.get( + "filter_attributes" + ), + "appsignal.config.filter_function_parameters": config.options.get( + "filter_function_parameters" + ), + "appsignal.config.filter_request_query_parameters": config.options.get( + "filter_request_query_parameters" + ), + "appsignal.config.filter_request_payload": config.options.get( + "filter_request_payload" + ), + "appsignal.config.filter_request_session_data": config.options.get( + "filter_session_data" + ), + "appsignal.config.ignore_actions": config.options.get("ignore_actions"), + "appsignal.config.ignore_errors": config.options.get("ignore_errors"), + "appsignal.config.ignore_namespaces": config.options.get( + "ignore_namespaces" + ), + "appsignal.config.response_headers": config.options.get("response_headers"), + "appsignal.config.request_headers": config.options.get("request_headers"), + "appsignal.config.send_function_parameters": config.options.get( + "send_function_parameters" + ), + "appsignal.config.send_request_query_parameters": config.options.get( + "send_request_query_parameters" + ), + "appsignal.config.send_request_payload": config.options.get( + "send_request_payload" + ), + "appsignal.config.send_request_session_data": config.options.get( + "send_session_data" + ), + }.items() + if value is not None + } + + return Resource(attributes=cast(Mapping[str, Union[str, List[str]]], attributes)) + + +def _opentelemetry_endpoint(config: Config) -> str: + collector_endpoint = config.options.get("collector_endpoint") + if collector_endpoint: + # Remove trailing slashes (it will be concatenated + # with /v1/{traces,metrics,logs} later) + return collector_endpoint.rstrip("/") + + opentelemetry_port = config.option("opentelemetry_port") + return f"http://localhost:{opentelemetry_port}" + + def add_instrumentations( config: Config, _adders: Mapping[ @@ -253,9 +361,12 @@ def add_instrumentations( return for name, adder in _adders.items(): - if name not in disable_list: + if ( + name not in disable_list + and f"opentelemetry.instrumentation.{name}" not in disable_list + ): try: - adder() + adder(config) logger.info(f"Instrumented {name}") except ModuleNotFoundError: pass diff --git a/tests/diagnose b/tests/diagnose index 69570f56..30f1c121 160000 --- a/tests/diagnose +++ b/tests/diagnose @@ -1 +1 @@ -Subproject commit 69570f567f23f3e5a94e527e27866fc722082849 +Subproject commit 30f1c121a4960999fccf02d18317e99edeb0d320 diff --git a/tests/test_client.py b/tests/test_client.py index 033873f1..bf8c069a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -4,7 +4,8 @@ import signal from unittest.mock import call, mock_open, patch -from appsignal.agent import agent +from appsignal.agent import Agent +from appsignal.binary import NoopBinary from appsignal.client import Client @@ -23,7 +24,8 @@ def test_client_agent_inactive(): client.start() assert os.environ.get("_APPSIGNAL_ACTIVE") is None - assert agent.active is False + assert type(client._binary) is Agent + assert client._binary.active is False def test_client_agent_active(): @@ -33,7 +35,8 @@ def test_client_agent_active(): client.start() assert os.environ.get("_APPSIGNAL_ACTIVE") == "true" - assert agent.active is True + assert type(client._binary) is Agent + assert client._binary.active is True def test_client_agent_active_invalid(): @@ -43,7 +46,34 @@ def test_client_agent_active_invalid(): client.start() assert os.environ.get("_APPSIGNAL_ACTIVE") is None - assert agent.active is False + assert type(client._binary) is Agent + assert client._binary.active is False + + +def test_client_active_noopbinary_when_collector_endpoint_set(): + client = Client( + active=True, + name="MyApp", + push_api_key="0000-0000-0000-0000", + request_headers=["accept", "x-custom-header"], + collector_endpoint="https://custom-endpoint.appsignal.com", + ) + + client.start() + + assert type(client._binary) is NoopBinary + assert client._binary.active + + # Does not set the private config environment variables + assert os.environ.get("_APPSIGNAL_ACTIVE") is None + assert os.environ.get("_APPSIGNAL_APP_NAME") is None + assert os.environ.get("_APPSIGNAL_PUSH_API_KEY") is None + + # Sets the OpenTelemetry config environment variables + assert ( + os.environ.get("OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST") + == "accept,x-custom-header" + ) def test_client_active(): @@ -64,11 +94,15 @@ def test_client_active(): assert os.environ.get("_APPSIGNAL_ACTIVE") == "true" assert os.environ.get("_APPSIGNAL_APP_NAME") == "MyApp" assert os.environ.get("_APPSIGNAL_PUSH_API_KEY") == "0000-0000-0000-0000" + + # Sets the OpenTelemetry config environment variables assert ( os.environ.get("OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST") == "accept,x-custom-header" ) - assert agent.active + + assert type(client._binary) is Agent + assert client._binary.active def test_client_active_without_request_headers(): diff --git a/tests/test_config.py b/tests/test_config.py index c41c5ba1..adfc4694 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -38,8 +38,9 @@ def test_source_order(): def test_system_source(): config = Config() - assert list(config.sources["system"].keys()) == ["app_path"] + assert list(config.sources["system"].keys()) == ["app_path", "hostname"] assert "app_path" in list(config.options.keys()) + assert "hostname" in list(config.options.keys()) def test_environ_source(): @@ -149,17 +150,15 @@ def test_environ_source_bool_is_invalid(): def test_environ_source_disable_default_instrumentations_list(): os.environ["APPSIGNAL_DISABLE_DEFAULT_INSTRUMENTATIONS"] = ",".join( - ["opentelemetry.instrumentation.celery", "something.else"] + ["celery", "not a real default instrumentation name"] ) config = Config() assert config.sources["environment"]["disable_default_instrumentations"] == [ - "opentelemetry.instrumentation.celery" - ] - assert config.options["disable_default_instrumentations"] == [ - "opentelemetry.instrumentation.celery" + "celery" ] + assert config.options["disable_default_instrumentations"] == ["celery"] def test_environ_source_disable_default_instrumentations_bool(): @@ -254,6 +253,102 @@ def test_set_private_environ(): assert os.environ["_APP_REVISION"] == "abc123" +def test_opentelemetry_resource(): + import os + + from appsignal.opentelemetry import _resource + + config = Config( + Options( + name="TestApp", + environment="test", + push_api_key="test-key", + revision="abc123", + service_name="test-service", + hostname="test-host", + filter_attributes=["password", "secret"], + filter_function_parameters=["param1"], + filter_request_query_parameters=["query1"], + filter_request_payload=["payload1"], + filter_session_data=["session1"], + ignore_actions=["action1", "action2"], + ignore_errors=["error1"], + ignore_namespaces=["namespace1"], + response_headers=["x-response"], + request_headers=["x-request"], + send_function_parameters=True, + send_request_query_parameters=True, + send_request_payload=True, + send_session_data=True, + ) + ) + + resource = _resource(config) + + # Test required attributes + assert resource.attributes["appsignal.config.name"] == "TestApp" + assert resource.attributes["appsignal.config.environment"] == "test" + assert resource.attributes["appsignal.config.push_api_key"] == "test-key" + assert resource.attributes["appsignal.config.revision"] == "abc123" + assert resource.attributes["appsignal.config.language_integration"] == "python" + assert resource.attributes["service.name"] == "test-service" + assert resource.attributes["host.name"] == "test-host" + assert resource.attributes["appsignal.service.process_id"] == os.getpid() + + # Test filter attributes + assert resource.attributes["appsignal.config.filter_attributes"] == ( + "password", + "secret", + ) + assert resource.attributes["appsignal.config.filter_function_parameters"] == ( + "param1", + ) + assert resource.attributes["appsignal.config.filter_request_query_parameters"] == ( + "query1", + ) + assert resource.attributes["appsignal.config.filter_request_payload"] == ( + "payload1", + ) + assert resource.attributes["appsignal.config.filter_request_session_data"] == ( + "session1", + ) + + # Test ignore attributes + assert resource.attributes["appsignal.config.ignore_actions"] == ( + "action1", + "action2", + ) + assert resource.attributes["appsignal.config.ignore_errors"] == ("error1",) + assert resource.attributes["appsignal.config.ignore_namespaces"] == ("namespace1",) + + # Test header attributes + assert resource.attributes["appsignal.config.response_headers"] == ("x-response",) + assert resource.attributes["appsignal.config.request_headers"] == ("x-request",) + + # Test send attributes + assert resource.attributes["appsignal.config.send_function_parameters"] is True + assert resource.attributes["appsignal.config.send_request_query_parameters"] is True + assert resource.attributes["appsignal.config.send_request_payload"] is True + assert resource.attributes["appsignal.config.send_request_session_data"] is True + + +def test_opentelemetry_resource_with_defaults(): + from appsignal.opentelemetry import _resource + + # Test with minimal config to check default values + config = Config(Options()) + resource = _resource(config) + + # Test default values + assert resource.attributes["appsignal.config.revision"] == "unknown" + assert resource.attributes["service.name"] == "unknown" + assert resource.attributes["appsignal.config.language_integration"] == "python" + + # Test that None values are excluded + assert "appsignal.config.name" not in resource.attributes + assert "appsignal.config.push_api_key" not in resource.attributes + + def test_set_private_environ_valid_log_path(): cwdir = os.getcwd() config = Config(Options(log_path=cwdir)) @@ -320,3 +415,251 @@ def test_is_active_invalid_push_api_key(): config = Config(Options(active=True, push_api_key=" ")) assert config.is_active() is False + + +def test_warn_no_warnings_when_using_default_values(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + + # Using collector with default values should not emit warnings + config = Config(Options(collector_endpoint="http://localhost:4318")) + config.warn() + + assert mock_warning.call_count == 0 + + # Using agent with default values should not emit warnings + config = Config(Options()) + config.warn() + + assert mock_warning.call_count == 0 + + +def test_warn_all_agent_exclusive_options(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + mock_info = mocker.patch("appsignal.internal_logger.info") + + def config_builder() -> Config: + return Config( + Options( + collector_endpoint="http://localhost:4318", + bind_address="0.0.0.0", + cpu_count=2.0, + dns_servers=["8.8.8.8"], + enable_host_metrics=False, + enable_nginx_metrics=True, + enable_statsd=True, + files_world_accessible=False, + filter_parameters=["password"], + host_role="web", + nginx_port="8080", + opentelemetry_port="9999", + running_in_container=True, + send_environment_metadata=False, + send_params=False, + working_directory_path="/app", + statsd_port="8125", + ) + ) + + initial_config = config_builder() + # Build a config object where all the given options + # come from the environment + environment_config = config_builder() + environment_config.sources["environment"] = environment_config.sources["initial"] + environment_config.sources["initial"] = {} + + for config in [initial_config, environment_config]: + mock_warning.reset_mock() + mock_info.reset_mock() + + config.warn() + + warning_messages = [call.args[0] for call in mock_warning.call_args_list] + + agent_exclusive_options = [ + "bind_address", + "cpu_count", + "dns_servers", + "enable_host_metrics", + "enable_nginx_metrics", + "enable_statsd", + "files_world_accessible", + "filter_parameters", + "host_role", + "nginx_port", + "opentelemetry_port", + "running_in_container", + "send_environment_metadata", + "send_params", + "working_directory_path", + "statsd_port", + ] + + for option in agent_exclusive_options: + assert any( + f"'{option}' configuration option is only used by the agent" in msg + for msg in warning_messages + ), f"Expected warning for '{option}' not found" + + # Info log about using the agent should only be emitted once + mock_info.assert_called_once_with( + "To use the agent, unset the 'collector_endpoint' configuration option." + ) + + +def test_warn_all_collector_exclusive_options(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + mock_info = mocker.patch("appsignal.internal_logger.info") + + def config_builder() -> Config: + return Config( + Options( + filter_attributes=["attr1"], + filter_function_parameters=["param1"], + filter_request_payload=["payload1"], + filter_request_query_parameters=["query1"], + response_headers=["x-response"], + send_function_parameters=True, + send_request_payload=True, + send_request_query_parameters=True, + service_name="my-service", + ) + ) + + initial_config = config_builder() + # Build a config object where all the given options + # come from the environment + environment_config = config_builder() + environment_config.sources["environment"] = environment_config.sources["initial"] + environment_config.sources["initial"] = {} + + for config in [initial_config, environment_config]: + mock_warning.reset_mock() + mock_info.reset_mock() + + config.warn() + + warning_messages = [call.args[0] for call in mock_warning.call_args_list] + + collector_exclusive_options = [ + "filter_attributes", + "filter_function_parameters", + "filter_request_payload", + "filter_request_query_parameters", + "response_headers", + "send_function_parameters", + "send_request_payload", + "send_request_query_parameters", + "service_name", + ] + + for option in collector_exclusive_options: + assert any( + f"'{option}' configuration option is only used by the collector" in msg + for msg in warning_messages + ), f"Expected warning for '{option}' not found" + + # Info log about using the collector should only be emitted once + mock_info.assert_called_once_with( + "To use the collector, set the 'collector_endpoint' configuration option." + ) + + +def test_warn_filter_parameters_emits_specific_advice(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + + config = Config( + Options( + collector_endpoint="http://localhost:4318", + filter_parameters=["password"], + ) + ) + + config.warn() + + warning_messages = [call.args[0] for call in mock_warning.call_args_list] + + assert any( + "Use the 'filter_attributes', 'filter_function_parameters'," + " 'filter_request_payload' and 'filter_request_query_parameters'" + " configuration options instead." in msg + for msg in warning_messages + ), "Expected specific advice for 'filter_parameters' not found" + + +def test_warn_send_params_emits_specific_advice(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + + config = Config( + Options( + collector_endpoint="http://localhost:4318", + send_params=False, + ) + ) + + config.warn() + + warning_messages = [call.args[0] for call in mock_warning.call_args_list] + + assert any( + "Use the 'send_function_parameters', 'send_request_payload'" + " and 'send_request_query_parameters' configuration options instead." in msg + for msg in warning_messages + ), "Expected specific advice for 'send_params' not found" + + +def test_warn_opentelemetry_port_emits_specific_advice(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + + config = Config( + Options( + collector_endpoint="http://localhost:4318", + opentelemetry_port="9999", + ) + ) + + config.warn() + + warning_messages = [call.args[0] for call in mock_warning.call_args_list] + + assert any( + "Set the collector's OpenTelemetry port as part of the" + " 'collector_endpoint' configuration option." in msg + for msg in warning_messages + ), "Expected specific advice for 'opentelemetry_port' not found" + + +def test_warn_collector_filter_options_emit_use_filter_parameters_advice(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + + config = Config( + Options( + filter_attributes=["attr1"], + filter_function_parameters=["param1"], + filter_request_payload=["payload1"], + filter_request_query_parameters=["query1"], + ) + ) + + config.warn() + + warning_messages = [call.args[0] for call in mock_warning.call_args_list] + + assert warning_messages.count("Use the 'filter_parameters' option instead.") == 4 + + +def test_warn_collector_send_options_emit_use_send_params_advice(mocker): + mock_warning = mocker.patch("appsignal.internal_logger.warning") + + config = Config( + Options( + send_function_parameters=True, + send_request_payload=True, + send_request_query_parameters=True, + ) + ) + + config.warn() + + warning_messages = [call.args[0] for call in mock_warning.call_args_list] + + assert warning_messages.count("Use the 'send_params' option instead.") == 3 diff --git a/tests/test_opentelemetry.py b/tests/test_opentelemetry.py index 2c2fefe4..29d0884a 100644 --- a/tests/test_opentelemetry.py +++ b/tests/test_opentelemetry.py @@ -1,21 +1,20 @@ from __future__ import annotations +from typing import List, cast from unittest.mock import Mock from appsignal.config import Config, Options from appsignal.opentelemetry import add_instrumentations -def raise_module_not_found_error(): +def raise_module_not_found_error(_config: Config) -> None: raise ModuleNotFoundError def mock_adders() -> dict[Config.DefaultInstrumentation, Mock]: return { - "opentelemetry.instrumentation.celery": Mock(), - "opentelemetry.instrumentation.jinja2": Mock( - side_effect=raise_module_not_found_error - ), + "celery": Mock(), + "jinja2": Mock(side_effect=raise_module_not_found_error), } @@ -30,17 +29,30 @@ def test_add_instrumentations(): def test_add_instrumentations_disable_some_default_instrumentations(): + adders = mock_adders() + config = Config(Options(disable_default_instrumentations=["celery"])) + + add_instrumentations(config, _adders=adders) + + adders["celery"].assert_not_called() + adders["jinja2"].assert_called_once() + + +def test_disable_default_instrumentations_backwards_compatibility_prefix(): adders = mock_adders() config = Config( Options( - disable_default_instrumentations=["opentelemetry.instrumentation.celery"] + disable_default_instrumentations=cast( + List[Config.DefaultInstrumentation], + ["opentelemetry.instrumentation.celery"], + ) ) ) add_instrumentations(config, _adders=adders) - adders["opentelemetry.instrumentation.celery"].assert_not_called() - adders["opentelemetry.instrumentation.jinja2"].assert_called_once() + adders["celery"].assert_not_called() + adders["jinja2"].assert_called_once() def test_add_instrumentations_disable_all_default_instrumentations():