Source code for multistorageclient.telemetry.attributes.process

 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 multiprocessing
18from collections.abc import Mapping
19
20import opentelemetry.util.types as api_types
21
22from .base import AttributesProvider
23
24

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

[docs]
30    class ProcessAttribute(enum.Enum):
31        """
32        Process attribute.
33
34        Use the enum value in the attributes dictionary values.
35        """
36
37        #: Process ID.
38        PID = "pid"

39
40    #: Attribute key to process attribute map.
41    _attributes: Mapping[str, ProcessAttribute]
42
43    def __init__(self, attributes: Mapping[str, str]):
44        """
45        :param attributes: Map of attribute key to process attribute.
46        """
47
48        self._attributes = {
49            attribute_key: ProcessAttributesProvider.ProcessAttribute(process_attribute)
50            for attribute_key, process_attribute in attributes.items()
51        }
52

[docs]
53    def attributes(self) -> api_types.Attributes:
54        process = multiprocessing.current_process()
55        return {
56            attribute_key: getattr(process, process_attribute.value)
57            for attribute_key, process_attribute in self._attributes.items()
58            if hasattr(process, process_attribute.value)
59        }