Completed writing unit tests for functionality implemented so far.

Author: michaelsafyan

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/backend/google/gcs/__init__.py

@@ -1,5 +1,6 @@
 from opentelemetry.instrumentation._blobupload.backend.google.gcs._gcs_impl import GcsBlobUploader
 
+
 __all__ = [
     GcsBlobUploader
 ]

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/backend/google/gcs/_gcs_client_wrapper.py

@@ -0,0 +1,59 @@
+from typing import Any, TypeAlias
+
+import logging
+
+_logger = logging.getLogger(__name__)
+_gcs_initialized = False
+_gcs_client_factory = None
+_gcs_blob_from_uri = None
+
+
+GcsClientType: TypeAlias = Any
+
+
+def set_gcs_client_factory(gcs_client_type, client_factory):
+    global _gcs_initialized
+    global _gcs_client_factory
+    global GcsClientType
+    if _gcs_initialized:
+        _logger.warning('Replacing default GCS client factory')
+    GcsClientType = gcs_client_type
+    _gcs_client_factory = client_factory
+    if _gcs_client_factory and _gcs_blob_from_uri:
+        _gcs_initialized = True
+
+
+def set_gcs_blob_from_uri(blob_from_uri):
+    global _gcs_initialized
+    global _gcs_blob_from_uri
+    if _gcs_initialized:
+        _logger.warning('Replacing default GCS blob_from_uri method')
+    _gcs_blob_from_uri = blob_from_uri
+    if _gcs_client_factory and _gcs_blob_from_uri:
+        _gcs_initialized = True
+
+
+def is_gcs_initialized():
+    return _gcs_initialized
+
+
+def create_gcs_client():
+    if _gcs_client_factory is not None:
+        return _gcs_client_factory()
+    return None
+
+
+def blob_from_uri(uri, client):
+    if _gcs_blob_from_uri is not None:
+        return _gcs_blob_from_uri(uri, client=client)
+    return None
+
+
+try:
+    from google.cloud.storage import Client as _GcsClient
+    from google.cloud.storage.blob import Blob as _GcsBlob
+    set_gcs_client_factory(_GcsClient, _GcsClient)
+    set_gcs_blob_from_uri(getattr(_GcsBlob, 'from_uri', getattr(_GcsBlob, 'from_string')))
+    _logger.debug('Found "google-cloud-storage" optional dependency and successfully registered it.')
+except ImportError:
+    _logger.warning('Missing optional "google-cloud-storage" dependency.')

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/backend/google/gcs/_gcs_impl.py

@@ -1,13 +1,42 @@
 import io
 import uuid
+import logging
 
-from google.cloud.storage import Client as GcsClient
-from google.cloud.storage import Blob as GcsBlob
+from typing import Optional, TypeAlias
 
 from opentelemetry.instrumentation._blobupload.api import Blob
 from opentelemetry.instrumentation._blobupload.api import BlobUploader
 from opentelemetry.instrumentation._blobupload.utils import SimpleBlobUploader
 from opentelemetry.instrumentation._blobupload.utils import blob_uploader_from_simple_blob_uploader
+from opentelemetry.instrumentation._blobupload.backend.google.gcs import _gcs_client_wrapper
+
+_logger = logging.getLogger(__name__)
+
+GcsClient: TypeAlias = _gcs_client_wrapper.GcsClientType
+
+
+def _path_for_span(trace_id, span_id):
+    if not trace_id or not span_id:
+        return ''
+    return 'traces/{}/spans/{}'.format(trace_id, span_id)
+
+
+def _path_for_event(trace_id, span_id, event_name):
+    if not event_name:
+        return ''
+    span_path = _path_for_span(trace_id, span_id)
+    if not span_path:
+        return ''
+    return '{}/events/{}'.format(span_path, event_name)
+
+
+def _path_for_span_event(trace_id, span_id, event_index):
+    if event_index is None:
+        return ''
+    span_path = _path_for_span(trace_id, span_id)
+    if not span_path:
+        return ''
+    return '{}/events/{}'.format(span_path, event_index)
 
 
 def _path_segment_from_labels(labels):
@@ -22,24 +51,19 @@ def _path_segment_from_labels(labels):
     ...depending on the particular type of signal source.
 
     """
-    segments = []
-    target_segments = [
-        ('traces', 'trace_id', 'unknown'),
-        ('spans', 'span_id', 'unknown'),
-        ('events', 'event_index', None),
-    ]
-    for segment_prefix, label_key, default_val in target_segments:
-        label_value = labels.get(label_key) or default_val
-        if label_value:
-            segments.append(segment_prefix)
-            segments.append(label_value)
-    if ((labels.get('otel_type') in ['event', 'span_event']) and 
-        ('events' not in segments)):
-        event_name = labels.get('event_name') or 'unknown'
-        segments.append('events')
-        segments.append(event_name)
-    return '/'.join(segments)
-
+    signal_type = labels.get('otel_type')
+    if not signal_type or signal_type not in ['span', 'event', 'span_event']:
+        return ''
+    trace_id = labels.get('trace_id')
+    span_id = labels.get('span_id')
+    event_name = labels.get('event_name')
+    event_index = labels.get('event_index')
+    if signal_type == 'span':
+        return _path_for_span(trace_id, span_id)
+    elif signal_type == 'event':
+        return _path_for_event(trace_id, span_id, event_name)
+    elif signal_type == 'span_event':
+        return _path_for_span_event(trace_id, span_id, event_index)
 
 
 class _SimpleGcsBlobUploader(SimpleBlobUploader):
@@ -52,15 +76,18 @@ def __init__(self, prefix: str, client:Optional[GcsClient]=None):
         if not prefix.endswith('/'):
             prefix = '{}/'.format(prefix)
         self._prefix = prefix
-        self._client = client or GcsClient()
+        self._client = client or _gcs_client_wrapper.create_gcs_client()
 
     def generate_destination_uri(self, blob: Blob) -> str:
         origin_path = _path_segment_from_labels(blob.labels)
+        if origin_path and not origin_path.endswith('/'):
+            origin_path = '{}/'.format(origin_path)
         upload_id = uuid.uuid4().hex
-        return '{}{}/uploads/{}'.format(self._prefix, origin_path, upload_id)
+        return '{}{}uploads/{}'.format(self._prefix, origin_path, upload_id)
 
     def upload_sync(self, uri: str, blob: Blob):
-        gcs_blob = GcsBlob.from_string(uri, client=self._client)
+        _logger.debug('Uploading blob: size: {} -> "{}"'.format(len(blob.raw_bytes), uri))
+        gcs_blob = _gcs_client_wrapper.blob_from_uri(uri, client=self._client)
         gcs_blob.upload_from_file(
             io.BytesIO(blob.raw_bytes),
             content_type=blob.content_type)
@@ -69,9 +96,12 @@ def upload_sync(self, uri: str, blob: Blob):
         gcs_blob.metadata = metadata
 
 
+
 class GcsBlobUploader(BlobUploader):
 
     def __init__(self, prefix: str, client:Optional[GcsClient]=None):
+        if not _gcs_client_wrapper.is_gcs_initialized():
+            raise NotImplementedError("GcsBlobUploader implementation unavailable without 'google-cloud-storage' optional dependency.")
         simple_uploader = _SimpleGcsBlobUploader(prefix, client)
         self._delegate = blob_uploader_from_simple_blob_uploader(simple_uploader)
 

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/utils/simple_blob_uploader_adaptor.py

@@ -35,7 +35,7 @@ def __call__(self):
         try:
             self._simple_uploader.upload_sync(self._uri, self._blob)
         except:
-            _logger.error('Failed to upload blob to "{}".'.format(self._uri))
+            _logger.exception('Failed to upload blob to "{}".'.format(self._uri))
 
 
 def _create_default_executor_no_cleanup():

opentelemetry-instrumentation/tests/_blobupload/api/test_blob.py

@@ -5,6 +5,7 @@
     sys.path.append("../../../src")
 
 import base64
+import logging
 import unittest
 
 
@@ -105,4 +106,5 @@ def assert_labels_equal(self, a, b):
 
 
 if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
     unittest.main()

opentelemetry-instrumentation/tests/_blobupload/api/test_content_type.py

@@ -5,6 +5,7 @@
     sys.path.append("../../../src")
 
 import io
+import logging
 import unittest
 
 from PIL import Image
@@ -56,4 +57,5 @@ def test_detects_png(self):
 
 
 if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
     unittest.main()

opentelemetry-instrumentation/tests/_blobupload/api/test_labels.py

@@ -4,6 +4,7 @@
     import sys
     sys.path.append("../../../src")
 
+import logging
 import unittest
 
 from opentelemetry.instrumentation._blobupload.api import (
@@ -68,4 +69,5 @@ def test_generate_labels_for_span_event(self):
 
 
 if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
     unittest.main()

opentelemetry-instrumentation/tests/_blobupload/api/test_provider.py

@@ -4,6 +4,7 @@
     import sys
     sys.path.append("../../../src")
 
+import logging
 import unittest
 
 from opentelemetry.instrumentation._blobupload.api import (
@@ -61,4 +62,5 @@ def get_blob_uploader(self, use_case):
 
 
 if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
     unittest.main()