Source code for multistorageclient.telemetry.attributes.host

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

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

[docs]
30    class HostAttribute(enum.Enum):
31        """
32        Host attribute.
33
34        Use the enum value in the attributes dictionary values.
35        """
36
37        #: Hostname.
38        NAME = "name"

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

[docs]
53    def attributes(self) -> api_types.Attributes:
54        return {
55            attribute_key: self._host_attribute_value(host_attribute=host_attribute)
56            for attribute_key, host_attribute in self._attributes.items()
57        }

58
59    def _host_attribute_value(self, host_attribute: HostAttribute) -> api_types.AttributeValue:
60        if host_attribute == HostAttributesProvider.HostAttribute.NAME:
61            return socket.gethostname()