Try to open file in UploadCompletionHook initializer with given path to ensure file upload works

util/opentelemetry-util-genai/CHANGELOG.md

@@ -18,6 +18,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4320](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4320))
 - Add workflow invocation type to genAI utils
   ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4310](#4310))
+- Check if upload works at startup in initializer of the `UploadCompletionHook`, instead
+of repeatedly failing on every upload ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4390](#4390)).
 
 ## Version 0.3b0 (2026-02-20)
 

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_upload/completion_hook.py

@@ -156,6 +156,24 @@ def __init__(
                 f"Invalid {upload_format=}. Must be one of {_FORMATS}"
             )
         self._format = upload_format
+        self._content_type = (
+            "application/json"
+            if self._format == "json"
+            else "application/jsonl"
+        )
+        test_path = posixpath.join(
+            self._base_path,
+            f"one_off_test_to_see_if_upload_works.{self._format}",
+        )
+        try:
+            with self._fs.open(
+                test_path, "w", content_type=self._content_type
+            ) as file:
+                file.write("\n")
+        except Exception as e:
+            raise ValueError(
+                f"Failed to write file to the following path to test if upload working: {test_path}.\n Got error: {e}"
+            )
 
         # Use a ThreadPoolExecutor for its queueing and thread management. The semaphore
         # limits the number of queued tasks. If the queue is full, data will be dropped.
@@ -271,13 +289,7 @@ def _do_upload(
             for message_idx, line in enumerate(message_lines):
                 line[_MESSAGE_INDEX_KEY] = message_idx
 
-        content_type = (
-            "application/json"
-            if self._format == "json"
-            else "application/jsonl"
-        )
-
-        with self._fs.open(path, "w", content_type=content_type) as file:
+        with self._fs.open(path, "w", content_type=self._content_type) as file:
             for message in message_lines:
                 gen_ai_json_dump(message, file)
                 file.write("\n")

util/opentelemetry-util-genai/tests/test_upload.py

@@ -102,6 +102,8 @@ def setUp(self):
         self.hook = UploadCompletionHook(
             base_path=BASE_PATH, max_queue_size=MAXSIZE, lru_cache_max_size=5
         )
+        # 1 upload is done when creating the UploadHook to ensure upload works. Reset mock.
+        self.mock_fs.reset_mock()
 
     def tearDown(self) -> None:
         self.hook.shutdown()
@@ -145,6 +147,18 @@ def test_upload_then_shutdown(self):
             "should have uploaded 4 files",
         )
 
+    def test_failed_upload_causes_initializer_to_throw(self):
+        self.mock_fs.open.side_effect = ValueError("Failed for some reason!")
+        with self.assertRaisesRegex(
+            ValueError,
+            "Failed to write file to the following path to test if upload working",
+        ):
+            UploadCompletionHook(
+                base_path=BASE_PATH,
+                max_queue_size=MAXSIZE,
+                lru_cache_max_size=5,
+            )
+
     def test_lru_cache_works(self):
         record = LogRecord()
         self.hook.on_completion(