Skip to content

Add worker resource and queue metrics to OTel instrumentation #613

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

Merged
s3rius merged 3 commits into from
May 4, 2026
Merged

Add worker resource and queue metrics to OTel instrumentation #613

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
65223b7
feat: add OTel metrics to OpenTelemetryMiddleware
Mohammed0tarek May 2, 2026
0670feb
Added cpu and memory utilization guages
Mohammed0tarek May 2, 2026
4188421
feat: add worker queue and active task OTel metrics
Mohammed0tarek May 2, 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
Prev Previous commit
Next Next commit
Added cpu and memory utilization guages
  • Loading branch information
@Mohammed0tarek @s3rius
Mohammed0tarek authored and s3rius committed May 4, 2026
commit 0670febf91e3f6cd10f9e626334c282c1e94bff9
29 changes: 27 additions & 2 deletions taskiq/middlewares/opentelemetry_middleware.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,8 @@
from contextlib import AbstractContextManager
from datetime import datetime, timezone
from importlib.metadata import version
from typing import Any, TypeVar
from typing import Any, Generator, TypeVar
import psutil

from packaging.version import Version, parse

Expand All @@ -17,7 +18,7 @@

from opentelemetry import context as context_api
from opentelemetry import trace
from opentelemetry.metrics import Meter, MeterProvider, get_meter
from opentelemetry.metrics import Meter, MeterProvider, Observation, get_meter
from opentelemetry.propagate import extract, inject
from opentelemetry.semconv.trace import SpanAttributes
from opentelemetry.trace import Span, Tracer, TracerProvider
Expand Down Expand Up @@ -205,6 +206,30 @@ def __init__(
unit="s",
description="Time the tasks waited before executing",
)
# current metrics to watch for in workers: CPU and memory utilization
self._process = psutil.Process()
# 6- CPU utilization
self.worker_cpu_utilization = self._meter.create_observable_gauge(
"worker_cpu_utilization",
callbacks=[self._observe_cpu],
unit="%",
description="Worker CPU utilization percentage. Only for worker processes",
)
# 7- Memory utilization
self.worker_memory_utilization = self._meter.create_observable_gauge(
"worker_memory_utilization",
callbacks=[self._observe_memory],
unit="By",
description="Worker memory utilization in bytes. Only for worker processes",
)

def _observe_memory(self, _options: Any) -> Generator[Observation, None, None]:
if self.broker and self.broker.is_worker_process:
yield Observation(self._process.memory_info().rss)

def _observe_cpu(self, _options: Any) -> Generator[Observation, None, None]:
if self.broker and self.broker.is_worker_process:
yield Observation(self._process.cpu_percent())

def pre_send(self, message: TaskiqMessage) -> TaskiqMessage:
"""
Expand Down