Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

feat: add opentelemetry on top of newly refactored pyinfra

Browse Source
This commit is contained in:
Isaac Riley 2024-01-24 08:09:42 +01:00
parent 725d6dce45
commit 936bb4fe80
6 changed files with 1084 additions and 451 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.pre-commit-config.yaml
View File

@ -2,10 +2,10 @@
# See https://pre-commit.com/hooks.html for more hooks
exclude: ^(docs/|notebooks/|data/|src/secrets/|src/static/|src/templates/|tests)
default_language_version:
python: python3.8
python: python3.10
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.4.0
rev: v4.5.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
@ -26,7 +26,7 @@ repos:
args: ["--profile", "black"]
- repo: https://github.com/psf/black
rev: 23.1.0
rev: 23.12.1
hooks:
- id: black
# exclude: ^(docs/|notebooks/|data/|src/secrets/)

4
README.md
View File

@ -58,6 +58,8 @@ the [complete example](pyinfra/examples.py).
| STORAGE__AZURE__CONNECTION_STRING | storage.azure.connection_string | Connection string for the Azure server |
| STORAGE__TENANT_SERVER__PUBLIC_KEY | storage.tenant_server.public_key | Public key of the tenant server |
| STORAGE__TENANT_SERVER__ENDPOINT | storage.tenant_server.endpoint | Endpoint of the tenant server |
| TRACING__ENDPOINT | tracing.endpoint | Endpoint to which OpenTelemetry traces are exported
| TRACING__SERVER_NAME | tracing.server_name | Name of the service as displayed in the traces collected
## Queue Manager
@ -139,7 +141,7 @@ $ cd tests && docker compose up
**Shell 2**: Start pyinfra with callback mock
```bash
$ python scripts/start_pyinfra.py
$ python scripts/start_pyinfra.py
```
**Shell 3**: Upload dummy content on storage and publish message

1458
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

5
pyinfra/config/validators.py
View File

@ -44,3 +44,8 @@ webserver_validators = [
Validator("webserver.host", must_exist=True, is_type_of=str),
Validator("webserver.port", must_exist=True, is_type_of=int),
]
opentelemetry_validators = [
Validator("tracing.endpoint", must_exist=True, is_type_of=str),
Validator("tracing.service.name", must_exist=True, is_type_of=str),
]

24
pyinfra/queue/manager.py
View File

@ -4,17 +4,23 @@ import json
import logging
import signal
import sys
from typing import Union, Callable
from typing import Callable, Union
import pika
import pika.exceptions
from dynaconf import Dynaconf
from kn_utils.logging import logger
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.pika import PikaInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from pika.adapters.blocking_connection import BlockingChannel, BlockingConnection
from retry import retry
from pyinfra.config.validators import queue_manager_validators
from pyinfra.config.loader import validate_settings
from pyinfra.config.validators import queue_manager_validators
pika_logger = logging.getLogger("pika")
pika_logger.setLevel(logging.WARNING) # disables non-informative pika log clutter
@ -36,6 +42,9 @@ class QueueManager:
self.channel: Union[BlockingChannel, None] = None
self.connection_sleep = settings.rabbitmq.connection_sleep
self.tracing_endpoint = settings.tracing.endpoint
self.service_name = settings.tracing.service_name
atexit.register(self.stop_consuming)
signal.signal(signal.SIGTERM, self._handle_stop_signal)
signal.signal(signal.SIGINT, self._handle_stop_signal)
@ -76,6 +85,17 @@ class QueueManager:
logger.info("Connection to RabbitMQ established, channel open.")
resource = Resource(attributes={"service.name": self.service_name})
trace.set_tracer_provider(TracerProvider(resource=resource))
tracer = trace.get_tracer(__name__)
otlp_exporter = OTLPSpanExporter(endpoint=self.tracing_endpoint)
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)
pika_instrumentation = PikaInstrumentor()
pika_instrumentation.instrument_channel(channel=self.channel)
def is_ready(self):
self.establish_connection()
return self.channel.is_open

38
pyproject.toml
View File

@ -4,7 +4,6 @@ version = "1.10.0"
description = ""
authors = ["Team Research <research@knecon.com>"]
license = "All rights reseverd"
readme = "README.md"
[tool.poetry.dependencies]
python = ">=3.10,<3.11"
@ -31,6 +30,18 @@ black = "^23.10"
pylint = "^3"
coverage = "^7.3"
requests = "^2.31"
pre-commit = "^3.6.0"
[tool.poetry.group.telemetry.dependencies]
opentelemetry-instrumentation-pika = "^0.43b0"
opentelemetry-exporter-otlp = "^1.22.0"
opentelemetry-instrumentation = "^0.43b0"
opentelemetry-api = "^1.22.0"
opentelemetry-sdk = "^1.22.0"
opentelemetry-exporter-otlp-proto-http = "^1.22.0"
opentelemetry-instrumentation-flask = "^0.43b0"
opentelemetry-instrumentation-requests = "^0.43b0"
[tool.pytest.ini_options]
minversion = "6.0"
@ -39,6 +50,31 @@ testpaths = ["tests", "integration"]
log_cli = 1
log_cli_level = "DEBUG"
[tool.mypy]
exclude = ['.venv']
[tool.black]
line-length = 120
target-version = ["py310"]
[tool.isort]
profile = "black"
[tool.pylint.format]
max-line-length = 120
disable = [
"C0114",
"C0325",
"R0801",
"R0902",
"R0903",
"R0904",
"R0913",
"R0914",
"W0511"
]
docstring-min-length = 3
[[tool.poetry.source]]
name = "PyPI"
priority = "primary"