1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
"""
Code used for the Queries module in Sentry
"""
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations.redis.utils import _get_safe_command
from sentry_sdk.utils import capture_internal_exceptions
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from redis import Redis
from sentry_sdk.integrations.redis import RedisIntegration
from sentry_sdk.tracing import Span
from typing import Any
def _compile_db_span_properties(integration, redis_command, args):
# type: (RedisIntegration, str, tuple[Any, ...]) -> dict[str, Any]
description = _get_db_span_description(integration, redis_command, args)
properties = {
"op": OP.DB_REDIS,
"description": description,
}
return properties
def _get_db_span_description(integration, command_name, args):
# type: (RedisIntegration, str, tuple[Any, ...]) -> str
description = command_name
with capture_internal_exceptions():
description = _get_safe_command(command_name, args)
data_should_be_truncated = (
integration.max_data_size and len(description) > integration.max_data_size
)
if data_should_be_truncated:
description = description[: integration.max_data_size - len("...")] + "..."
return description
def _set_db_data_on_span(span, connection_params):
# type: (Span, dict[str, Any]) -> None
span.set_data(SPANDATA.DB_SYSTEM, "redis")
db = connection_params.get("db")
if db is not None:
span.set_data(SPANDATA.DB_NAME, str(db))
host = connection_params.get("host")
if host is not None:
span.set_data(SPANDATA.SERVER_ADDRESS, host)
port = connection_params.get("port")
if port is not None:
span.set_data(SPANDATA.SERVER_PORT, port)
def _set_db_data(span, redis_instance):
# type: (Span, Redis[Any]) -> None
try:
_set_db_data_on_span(span, redis_instance.connection_pool.connection_kwargs)
except AttributeError:
pass # connections_kwargs may be missing in some cases
|
