Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/storage/_opentelemetry_tracing.py: 38%

1# Copyright 2024 Google LLC 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Manages OpenTelemetry tracing span creation and handling. This is a PREVIEW FEATURE: Coverage and functionality may change.""" 

16 

17import logging 

18import os 

19 

20from contextlib import contextmanager 

21 

22from google.api_core import exceptions as api_exceptions 

23from google.api_core import retry as api_retry 

24from google.cloud.storage import __version__ 

25from google.cloud.storage.retry import ConditionalRetryPolicy 

26 

27 

28ENABLE_OTEL_TRACES_ENV_VAR = "ENABLE_GCS_PYTHON_CLIENT_OTEL_TRACES" 

29_DEFAULT_ENABLE_OTEL_TRACES_VALUE = False 

30 

31enable_otel_traces = os.environ.get( 

32 ENABLE_OTEL_TRACES_ENV_VAR, _DEFAULT_ENABLE_OTEL_TRACES_VALUE 

33) 

34logger = logging.getLogger(__name__) 

35 

36try: 

37 from opentelemetry import trace 

38 

39 HAS_OPENTELEMETRY = True 

40 

41except ImportError: 

42 logger.debug( 

43 "This service is instrumented using OpenTelemetry. " 

44 "OpenTelemetry or one of its components could not be imported; " 

45 "please add compatible versions of opentelemetry-api and " 

46 "opentelemetry-instrumentation packages in order to get Storage " 

47 "Tracing data." 

48 ) 

49 HAS_OPENTELEMETRY = False 

50 

51_default_attributes = { 

52 "rpc.service": "CloudStorage", 

53 "rpc.system": "http", 

54 "user_agent.original": f"gcloud-python/{__version__}", 

55} 

56 

57_cloud_trace_adoption_attrs = { 

58 "gcp.client.service": "storage", 

59 "gcp.client.version": __version__, 

60 "gcp.client.repo": "googleapis/python-storage", 

61} 

62 

63 

64@contextmanager 

65def create_trace_span(name, attributes=None, client=None, api_request=None, retry=None): 

66 """Creates a context manager for a new span and set it as the current span 

67 in the configured tracer. If no configuration exists yields None.""" 

68 if not HAS_OPENTELEMETRY or not enable_otel_traces: 

69 yield None 

70 return 

71 

72 tracer = trace.get_tracer(__name__) 

73 final_attributes = _get_final_attributes(attributes, client, api_request, retry) 

74 # Yield new span. 

75 with tracer.start_as_current_span( 

76 name=name, kind=trace.SpanKind.CLIENT, attributes=final_attributes 

77 ) as span: 

78 try: 

79 yield span 

80 except api_exceptions.GoogleAPICallError as error: 

81 span.set_status(trace.Status(trace.StatusCode.ERROR)) 

82 span.record_exception(error) 

83 raise 

84 

85 

86def _get_final_attributes(attributes=None, client=None, api_request=None, retry=None): 

87 collected_attr = _default_attributes.copy() 

88 collected_attr.update(_cloud_trace_adoption_attrs) 

89 if api_request: 

90 collected_attr.update(_set_api_request_attr(api_request, client)) 

91 if isinstance(retry, api_retry.Retry): 

92 collected_attr.update(_set_retry_attr(retry)) 

93 if isinstance(retry, ConditionalRetryPolicy): 

94 collected_attr.update( 

95 _set_retry_attr(retry.retry_policy, retry.conditional_predicate) 

96 ) 

97 if attributes: 

98 collected_attr.update(attributes) 

99 final_attributes = {k: v for k, v in collected_attr.items() if v is not None} 

100 return final_attributes 

101 

102 

103def _set_api_request_attr(request, client): 

104 attr = {} 

105 if request.get("method"): 

106 attr["http.request.method"] = request.get("method") 

107 if request.get("path"): 

108 path = request.get("path") 

109 full_path = f"{client._connection.API_BASE_URL}{path}" 

110 attr["url.full"] = full_path 

111 if request.get("timeout"): 

112 attr["connect_timeout,read_timeout"] = request.get("timeout") 

113 return attr 

114 

115 

116def _set_retry_attr(retry, conditional_predicate=None): 

117 predicate = conditional_predicate if conditional_predicate else retry._predicate 

118 retry_info = f"multiplier{retry._multiplier}/deadline{retry._deadline}/max{retry._maximum}/initial{retry._initial}/predicate{predicate}" 

119 return {"retry": retry_info}