Release 1.12.12 (tracks 5ec85d9e05cd66e09277f165583468cea0338f3d)

Author: elementl-devtools

dagster-cloud-examples/dagster_cloud_examples/version.py

@@ -1 +1 @@
-__version__ = "1.12.11"
+__version__ = "1.12.12"

dagster-cloud-examples/pyproject.toml

@@ -19,7 +19,7 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = [
-    "dagster-cloud==1.12.11",
+    "dagster-cloud==1.12.12",
 ]
 
 [project.license]

dagster-cloud/dagster_cloud/agent/dagster_cloud_agent.py

@@ -32,10 +32,6 @@
 from dagster_cloud_cli.core.workspace import CodeLocationDeployData
 from dagster_shared.record import replace
 
-from dagster_cloud.agent.instrumentation.constants import DAGSTER_CLOUD_AGENT_METRIC_PREFIX
-from dagster_cloud.agent.instrumentation.run_launch import extract_run_attributes
-from dagster_cloud.agent.instrumentation.schedule import inspect_schedule_result
-from dagster_cloud.agent.instrumentation.sensor import inspect_sensor_result
 from dagster_cloud.agent.queries import (
     ADD_AGENT_HEARTBEATS_MUTATION,
     DEPLOYMENTS_QUERY,
@@ -59,7 +55,6 @@
 )
 from dagster_cloud.batching import Batcher
 from dagster_cloud.instance import DagsterCloudAgentInstance
-from dagster_cloud.opentelemetry.observers.execution_observer import observe_execution
 from dagster_cloud.util import SERVER_HANDLE_TAG, compressed_namedtuple_upload_file, is_isolated_run
 from dagster_cloud.version import __version__
 from dagster_cloud.workspace.user_code_launcher import (
@@ -981,29 +976,12 @@ def _handle_api_request(
                 instance_ref=self._get_user_code_instance_ref(deployment_name),
             )
 
-            schedule_attributes = {
-                "schedule": args.schedule_name,
-                "repository": args.repository_origin.repository_name,
-                "location": args.repository_origin.code_location_origin.location_name,
-                "deployment": deployment_name,
-            }
-
-            with observe_execution(
-                opentelemetry=self._instance.opentelemetry,
-                event_key=f"{DAGSTER_CLOUD_AGENT_METRIC_PREFIX}.schedule.evaluation",
-                short_description="schedule evaluation requests",
-                attributes=schedule_attributes,
-                result_evaluator_callback=inspect_schedule_result,
-            ) as observer:
-                serialized_schedule_data_or_error = client.external_schedule_execution(
-                    external_schedule_execution_args=args,
-                )
-                observer.evaluate_result(
-                    serialized_data_or_error=serialized_schedule_data_or_error,
-                )
-                return DagsterCloudApiGrpcResponse(
-                    serialized_response_or_error=serialized_schedule_data_or_error
-                )
+            serialized_schedule_data_or_error = client.external_schedule_execution(
+                external_schedule_execution_args=args,
+            )
+            return DagsterCloudApiGrpcResponse(
+                serialized_response_or_error=serialized_schedule_data_or_error
+            )
 
         elif api_name == DagsterCloudApi.GET_EXTERNAL_SENSOR_EXECUTION_DATA:
             client = self._get_grpc_client(
@@ -1015,27 +993,12 @@ def _handle_api_request(
                 instance_ref=self._get_user_code_instance_ref(deployment_name),
             )
 
-            sensor_attributes = {
-                "sensor": args.sensor_name,
-                "repository": args.repository_origin.repository_name,
-                "location": args.repository_origin.code_location_origin.location_name,
-                "deployment": deployment_name,
-            }
-
-            with observe_execution(
-                opentelemetry=self._instance.opentelemetry,
-                event_key=f"{DAGSTER_CLOUD_AGENT_METRIC_PREFIX}.sensor.evaluation",
-                short_description="sensor evaluation requests",
-                attributes=sensor_attributes,
-                result_evaluator_callback=inspect_sensor_result,
-            ) as observer:
-                serialized_sensor_data_or_error = client.external_sensor_execution(
-                    sensor_execution_args=args,
-                )
-                observer.evaluate_result(serialized_sensor_data_or_error)
-                return DagsterCloudApiGrpcResponse(
-                    serialized_response_or_error=serialized_sensor_data_or_error
-                )
+            serialized_sensor_data_or_error = client.external_sensor_execution(
+                sensor_execution_args=args,
+            )
+            return DagsterCloudApiGrpcResponse(
+                serialized_response_or_error=serialized_sensor_data_or_error
+            )
         elif api_name == DagsterCloudApi.GET_EXTERNAL_NOTEBOOK_DATA:
             client = self._get_grpc_client(
                 user_code_launcher, deployment_name, cast("str", location_name)
@@ -1066,42 +1029,32 @@ def _handle_api_request(
                     ),
                 )
 
-                run_attributes = extract_run_attributes(deployment_name, run)
-                with observe_execution(
-                    opentelemetry=self._instance.opentelemetry,
-                    event_key=f"{DAGSTER_CLOUD_AGENT_METRIC_PREFIX}.run.launches",
-                    short_description="run execution requests",
-                    attributes=run_attributes,
-                ) as observer:
-                    launcher = scoped_instance.get_run_launcher_for_run(run)  # type: ignore  # (instance subclass)
+                launcher = scoped_instance.get_run_launcher_for_run(run)  # type: ignore  # (instance subclass)
 
-                    if is_isolated_run(run):
-                        launcher.launch_run(LaunchRunContext(dagster_run=run, workspace=None))
-                    else:
-                        scoped_instance.report_engine_event(
-                            f"Launching {run.run_id} without an isolated run environment.",
-                            run,
-                            cls=self.__class__,
-                        )
+                if is_isolated_run(run):
+                    launcher.launch_run(LaunchRunContext(dagster_run=run, workspace=None))
+                else:
+                    scoped_instance.report_engine_event(
+                        f"Launching {run.run_id} without an isolated run environment.",
+                        run,
+                        cls=self.__class__,
+                    )
 
-                        run_location_name = cast(
-                            "str",
-                            run.remote_job_origin.repository_origin.code_location_origin.location_name,
-                        )
+                    run_location_name = cast(
+                        "str",
+                        run.remote_job_origin.repository_origin.code_location_origin.location_name,
+                    )
 
-                        server = user_code_launcher.get_grpc_server(
-                            deployment_name, run_location_name
-                        )
+                    server = user_code_launcher.get_grpc_server(deployment_name, run_location_name)
 
-                        # Record the server handle that we launched it on to for run monitoring
-                        scoped_instance.add_run_tags(
-                            run.run_id, new_tags={SERVER_HANDLE_TAG: str(server.server_handle)}
-                        )
+                    # Record the server handle that we launched it on to for run monitoring
+                    scoped_instance.add_run_tags(
+                        run.run_id, new_tags={SERVER_HANDLE_TAG: str(server.server_handle)}
+                    )
 
-                        launcher.launch_run_from_grpc_client(
-                            scoped_instance, run, server.server_endpoint.create_client()
-                        )
-                    observer.evaluate_result(run=run)
+                    launcher.launch_run_from_grpc_client(
+                        scoped_instance, run, server.server_endpoint.create_client()
+                    )
 
                 return DagsterCloudApiSuccess()
         elif api_name == DagsterCloudApi.TERMINATE_RUN:

dagster-cloud/dagster_cloud/agent/instrumentation/__init__.py

@@ -1,7 +1,7 @@
 import copy
 import socket
 import uuid
-from collections.abc import Mapping, Sequence
+from collections.abc import Sequence
 from contextlib import ExitStack
 from functools import lru_cache
 from typing import TYPE_CHECKING, Any, Optional, Union
@@ -34,11 +34,8 @@
 
 from dagster_cloud.agent import AgentQueuesConfig
 from dagster_cloud.auth.constants import decode_agent_token
-from dagster_cloud.opentelemetry.config import opentelemetry_config_schema
-from dagster_cloud.opentelemetry.controller import OpenTelemetryController
 from dagster_cloud.storage.client import dagster_cloud_api_config
 from dagster_cloud.util import get_env_names_from_config, is_isolated_run
-from dagster_cloud.version import __version__
 
 if TYPE_CHECKING:
     from requests import Session
@@ -83,7 +80,6 @@ def __init__(
         allowed_full_deployment_locations=None,
         allowed_branch_deployment_locations=None,
         agent_metrics=None,
-        opentelemetry=None,
         **kwargs,
     ):
         super().__init__(*args, **kwargs)
@@ -155,12 +151,6 @@ def __init__(
             allowed_branch_deployment_locations
         )
 
-        self._opentelemetry_config: Optional[Mapping[str, Any]] = self._get_processed_config(
-            "opentelemetry", opentelemetry, opentelemetry_config_schema()
-        )
-
-        self._opentelemetry_controller: Optional[OpenTelemetryController] = None
-
         self._instance_uuid = str(uuid.uuid4())
 
     @property
@@ -582,9 +572,6 @@ def config_schema(cls):
                 is_required=False,
                 description="List of allowed location names for branch deployments",
             ),
-            "opentelemetry": Field(
-                opentelemetry_config_schema(), is_required=False, default_value={"enabled": False}
-            ),
         }
 
     @classmethod
@@ -667,9 +654,6 @@ def config_defaults(base_dir):
 
     def dispose(self) -> None:
         super().dispose()
-        if self._opentelemetry_controller:
-            self._opentelemetry_controller.dispose()
-            self._opentelemetry_controller = None
         self._exit_stack.close()
 
     @property
@@ -690,17 +674,6 @@ def dagster_cloud_run_worker_monitoring_interval_seconds(self) -> int:
         # potentially overridden interval in the serverless user code launcher
         return 30
 
-    @property
-    def opentelemetry(self) -> OpenTelemetryController:
-        if not self._opentelemetry_controller:
-            self._opentelemetry_controller = OpenTelemetryController(
-                instance_id=self.instance_uuid,
-                version=__version__,
-                config=self._opentelemetry_config,
-            )
-
-        return self._opentelemetry_controller
-
 
 @lru_cache(maxsize=100)  # Scales on order of active branch deployments
 def _cached_inject_deployment(