Source code for multistorageclient.telemetry.attributes.thread

 1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 2# SPDX-License-Identifier: Apache-2.0
 3#
 4# Licensed under the Apache License, Version 2.0 (the "License");
 5# you may not use this file except in compliance with the License.
 6# You may obtain a copy of the License at
 7#
 8# http://www.apache.org/licenses/LICENSE-2.0
 9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import enum
17import threading
18from collections.abc import Mapping
19
20import opentelemetry.util.types as api_types
21
22from .base import AttributesProvider
23
24

[docs]
25class ThreadAttributesProvider(AttributesProvider):
26    """
27    Provides :py:type:`opentelemetry.util.types.Attributes` from current thread information.
28    """
29

[docs]
30    class ThreadAttribute(enum.Enum):
31        """
32        Thread attribute.
33
34        Use the enum value in the attributes dictionary values.
35        """
36
37        #: Python thread ID.
38        IDENT = "ident"
39        #: OS thread ID.
40        NATIVE_ID = "native_id"

41
42    #: Attribute key to thread attribute key map.
43    _attributes: Mapping[str, ThreadAttribute]
44
45    def __init__(self, attributes: Mapping[str, str]):
46        """
47        :param attributes: Map of attribute key to thread attribute.
48        """
49
50        self._attributes = {
51            attribute_key: ThreadAttributesProvider.ThreadAttribute(thread_attribute)
52            for attribute_key, thread_attribute in attributes.items()
53        }
54

[docs]
55    def attributes(self) -> api_types.Attributes:
56        thread = threading.current_thread()
57        return {
58            attribute_key: getattr(thread, thread_attribute.value)
59            for attribute_key, thread_attribute in self._attributes.items()
60            if hasattr(thread, thread_attribute.value)
61        }