feat: graphql add Run.hasRunMetricsEnabled field (#24662)

## Summary & Motivation

This change introduces a new field `hasRunMetricsEnabled` to the GrapheneRun type in the GraphQL schema. It allows clients to query whether run metrics are enabled for a specific run. The implementation checks for the presence of run metric tags and their boolean values.

This enables hiding the run metrics tag to declutter the frontend of this implementation detail.

## How I Tested These Changes

- BK w/ added test

## Changelog

NOCHANGELOG

python_modules/dagster-graphql/dagster_graphql/schema/pipelines/pipeline.py

@@ -19,8 +19,9 @@
     RunRecord,
     RunsFilter,
 )
-from dagster._core.storage.tags import REPOSITORY_LABEL_TAG, TagType, get_tag_type
+from dagster._core.storage.tags import REPOSITORY_LABEL_TAG, RUN_METRIC_TAGS, TagType, get_tag_type
 from dagster._core.workspace.permissions import Permissions
+from dagster._utils.tags import get_boolean_tag_value
 from dagster._utils.yaml_utils import dump_run_config_yaml
 
 from dagster_graphql.implementation.events import from_event_record, iterate_metadata_entries
@@ -82,6 +83,11 @@
     from dagster_graphql.schema.asset_graph import GrapheneAssetNode
     from dagster_graphql.schema.partition_sets import GrapheneJobSelectionPartition
 
+UNSTARTED_STATUSES = [
+    DagsterRunStatus.QUEUED,
+    DagsterRunStatus.NOT_STARTED,
+    DagsterRunStatus.STARTING,
+]
 
 STARTED_STATUSES = {
     DagsterRunStatus.STARTED,
@@ -384,6 +390,7 @@ class GrapheneRun(graphene.ObjectType):
     hasConcurrencyKeySlots = graphene.NonNull(graphene.Boolean)
     rootConcurrencyKeys = graphene.List(graphene.NonNull(graphene.String))
     hasUnconstrainedRootNodes = graphene.NonNull(graphene.Boolean)
+    hasRunMetricsEnabled = graphene.NonNull(graphene.Boolean)
 
     class Meta:
         interfaces = (GraphenePipelineRun, GrapheneRunsFeedEntry)
@@ -624,6 +631,13 @@ def resolve_rootConcurrencyKeys(self, graphene_info: ResolveInfo):
             root_concurrency_keys.extend([concurrency_key] * count)
         return root_concurrency_keys
 
+    def resolve_hasRunMetricsEnabled(self, graphene_info: ResolveInfo):
+        if self.dagster_run.status in UNSTARTED_STATUSES:
+            return False
+
+        run_tags = self.dagster_run.tags
+        return any(get_boolean_tag_value(run_tags.get(tag)) for tag in RUN_METRIC_TAGS)
+
 
 class GrapheneIPipelineSnapshotMixin:
     # Mixin this class to implement IPipelineSnapshot

python_modules/dagster-graphql/dagster_graphql_tests/graphql/test_runs.py

@@ -1,6 +1,7 @@
 import copy
 import tempfile
 
+import pytest
 import yaml
 from dagster import AssetMaterialization, Output, define_asset_job, job, op, repository
 from dagster._core.definitions.decorators.asset_decorator import asset
@@ -292,6 +293,22 @@
 }
 """
 
+RUN_METRICS_QUERY = """
+query RunQuery($runId: ID!) {
+  pipelineRunOrError(runId: $runId) {
+    __typename
+    ... on Run {
+        runId
+        tags {
+          key
+          value
+        }
+        hasRunMetricsEnabled
+    }
+  }
+}
+"""
+
 
 def _get_runs_data(result, run_id):
     for run_data in result.data["pipelineOrError"]["runs"]:
@@ -981,3 +998,47 @@ def test_run_has_concurrency_slots():
                 assert result.data["pipelineRunsOrError"]["results"][0]["runId"] == run_id
                 assert result.data["pipelineRunsOrError"]["results"][0]["hasConcurrencyKeySlots"]
                 assert result.data["pipelineRunsOrError"]["results"][0]["rootConcurrencyKeys"]
+
+
+@pytest.mark.parametrize(
+    "run_tag, run_tag_value, run_metrics_enabled, failure_message",
+    [
+        (None, None, False, "run_metrics tag not present should result to false"),
+        (".dagster/run_metrics", "true", True, "run_metrics tag set to true should result to true"),
+        (
+            ".dagster/run_metrics",
+            "false",
+            False,
+            "run_metrics tag set to falsy value should result to false",
+        ),
+        (
+            "dagster/run_metrics",
+            "true",
+            True,
+            "public run_metrics tag set to true should result to true",
+        ),
+    ],
+)
+def test_run_has_run_metrics_enabled(run_tag, run_tag_value, run_metrics_enabled, failure_message):
+    with instance_for_test() as instance:
+        repo = get_repo_at_time_1()
+        tags = (
+            {
+                run_tag: run_tag_value,
+            }
+            if run_tag
+            else {}
+        )
+        run = instance.create_run_for_job(
+            repo.get_job("foo_job"), status=DagsterRunStatus.STARTED, tags=tags
+        )
+
+        with define_out_of_process_context(__file__, "get_repo_at_time_1", instance) as context:
+            result = execute_dagster_graphql(
+                context,
+                RUN_METRICS_QUERY,
+                variables={"runId": run.run_id},
+            )
+            assert result.data
+            has_run_metrics_enabled = result.data["pipelineRunOrError"]["hasRunMetricsEnabled"]
+            assert has_run_metrics_enabled == run_metrics_enabled, failure_message

python_modules/dagster/dagster/_core/storage/tags.py

@@ -90,6 +90,12 @@
     RETRY_ON_ASSET_OR_OP_FAILURE_TAG,
 ]
 
+# Supports for the public tag is deprecated
+RUN_METRIC_TAGS = [
+    f"{HIDDEN_TAG_PREFIX}run_metrics",
+    f"{SYSTEM_TAG_PREFIX}run_metrics",
+]
+
 
 class TagType(Enum):
     # Custom tag provided by a user