Fix issue with blank telemetry instance id file (#7433)

python_modules/dagster/dagster/core/telemetry.py

@@ -326,7 +326,7 @@ def _get_instance_telemetry_info(instance):
     dagster_telemetry_enabled = _get_instance_telemetry_enabled(instance)
     instance_id = None
     if dagster_telemetry_enabled:
-        instance_id = _get_or_set_instance_id()
+        instance_id = get_or_set_instance_id()
 
     run_storage_id = None
     if isinstance(instance.run_storage, SqlRunStorage):
@@ -338,7 +338,7 @@ def _get_instance_telemetry_enabled(instance):
     return instance.telemetry_enabled
 
 
-def _get_or_set_instance_id():
+def get_or_set_instance_id():
     instance_id = _get_telemetry_instance_id()
     if instance_id == None:
         instance_id = _set_telemetry_instance_id()
@@ -353,8 +353,10 @@ def _get_telemetry_instance_id():
 
     with open(telemetry_id_path, "r") as telemetry_id_file:
         telemetry_id_yaml = yaml.safe_load(telemetry_id_file)
-        if INSTANCE_ID_STR in telemetry_id_yaml and isinstance(
-            telemetry_id_yaml[INSTANCE_ID_STR], str
+        if (
+            telemetry_id_yaml
+            and INSTANCE_ID_STR in telemetry_id_yaml
+            and isinstance(telemetry_id_yaml[INSTANCE_ID_STR], str)
         ):
             return telemetry_id_yaml[INSTANCE_ID_STR]
     return None
@@ -389,7 +391,7 @@ def log_external_repo_stats(instance, source, external_repo, external_pipeline=N
     check.opt_inst_param(external_pipeline, "external_pipeline", ExternalPipeline)
 
     if _get_instance_telemetry_enabled(instance):
-        instance_id = _get_or_set_instance_id()
+        instance_id = get_or_set_instance_id()
 
         pipeline_name_hash = hash_name(external_pipeline.name) if external_pipeline else ""
         repo_hash = hash_name(external_repo.name)
@@ -422,7 +424,7 @@ def log_repo_stats(instance, source, pipeline=None, repo=None):
     check.opt_inst_param(repo, "repo", ReconstructableRepository)
 
     if _get_instance_telemetry_enabled(instance):
-        instance_id = _get_or_set_instance_id()
+        instance_id = get_or_set_instance_id()
 
         if isinstance(pipeline, ReconstructablePipeline):
             pipeline_name_hash = hash_name(pipeline.get_definition().name)

python_modules/dagster/dagster_tests/cli_tests/command_tests/test_telemetry.py

@@ -8,9 +8,11 @@
 from dagster.cli.pipeline import pipeline_execute_command
 from dagster.core.definitions.reconstruct import get_ephemeral_repository_name
 from dagster.core.telemetry import (
+    TELEMETRY_STR,
     UPDATE_REPO_STATS,
     cleanup_telemetry_logger,
     get_or_create_dir_from_dagster_home,
+    get_or_set_instance_id,
     hash_name,
     log_workspace_stats,
     write_telemetry_log_line,
@@ -221,3 +223,13 @@ def test_write_telemetry_log_line_writes_to_dagster_home():
 
         # Needed to avoid file contention issues on windows with the telemetry log file
         cleanup_telemetry_logger()
+
+
+def test_set_instance_id_from_empty_file():
+    with tempfile.TemporaryDirectory() as temp_dir:
+        with environ({"DAGSTER_HOME": temp_dir}):
+            # Write an empty file to the path
+            open(
+                os.path.join(get_or_create_dir_from_dagster_home(TELEMETRY_STR), "id.yaml"), "w"
+            ).close()
+            assert get_or_set_instance_id()