fix(sentry): count triggers deployed through deploy_task (#1436)

Signed-off-by: Kevin Su <pingsutw@apache.org>

Author: pingsutw

src/flyte/_deploy.py

@@ -20,7 +20,7 @@
 from ._image import Image
 from ._initialize import ensure_client, get_client, get_init_config, requires_initialization
 from ._logging import logger
-from ._sentry import track_operation
+from ._sentry import count, track_operation
 from ._status import status
 from ._task import TaskTemplate
 from ._task_environment import TaskEnvironment
@@ -296,6 +296,12 @@ async def _deploy_task(
                     )
                 )
                 status.success(f"Deployed task {task.name} (version {task_id.version})")
+                if deployable_triggers:
+                    count(
+                        "flyte.operation",
+                        len(deployable_triggers),
+                        tags={"operation": "deploy_trigger", "status": "success"},
+                    )
             except ConnectError as e:
                 if e.code == Code.ALREADY_EXISTS:
                     status.info(f"Task {task.name} already exists, skipping")

tests/flyte/test_deploy.py

@@ -605,3 +605,63 @@ def __init__(self):
         deployment = await apply(plan, copy_style="loaded_modules", dryrun=True)
 
     assert "e" in deployment.envs
+
+
+@pytest.mark.asyncio
+async def test_deploy_task_counts_deployed_triggers(monkeypatch):
+    """Triggers ship inside DeployTaskRequest, so deploy_task is the only place they can be counted."""
+    from flyteidl2.task import task_definition_pb2
+
+    root_dir = pathlib.Path(__file__).parents[2].resolve()
+    monkeypatch.setattr(sys, "path", [str(root_dir / "src"), *sys.path])
+
+    env = flyte.TaskEnvironment(name="test_env", image="python:3.10")
+
+    @env.task()
+    async def task() -> None:
+        pass
+
+    task = replace(task, triggers=(Mock(), Mock()))
+    context = SerializationContext(version="v1", root_dir=root_dir)
+    config = Mock(sync_local_sys_paths=False)
+
+    with (
+        patch("flyte._deploy.ensure_client"),
+        patch("flyte._deploy.get_init_config", return_value=config),
+        patch("flyte._deploy.get_client", return_value=Mock(task_service=Mock(deploy_task=AsyncMock()))),
+        patch("flyte._internal.runtime.convert.convert_upload_default_inputs", AsyncMock(return_value=[])),
+        patch(
+            "flyte._internal.runtime.trigger_serde.to_task_trigger",
+            AsyncMock(return_value=task_definition_pb2.TaskTrigger(name="t")),
+        ),
+        patch("flyte._deploy.count") as count_mock,
+    ):
+        await _deploy_task(task, context, dryrun=False)
+
+    count_mock.assert_called_once_with("flyte.operation", 2, tags={"operation": "deploy_trigger", "status": "success"})
+
+
+@pytest.mark.asyncio
+async def test_deploy_task_without_triggers_emits_no_trigger_count(monkeypatch):
+    root_dir = pathlib.Path(__file__).parents[2].resolve()
+    monkeypatch.setattr(sys, "path", [str(root_dir / "src"), *sys.path])
+
+    env = flyte.TaskEnvironment(name="test_env", image="python:3.10")
+
+    @env.task()
+    async def task() -> None:
+        pass
+
+    context = SerializationContext(version="v1", root_dir=root_dir)
+    config = Mock(sync_local_sys_paths=False)
+
+    with (
+        patch("flyte._deploy.ensure_client"),
+        patch("flyte._deploy.get_init_config", return_value=config),
+        patch("flyte._deploy.get_client", return_value=Mock(task_service=Mock(deploy_task=AsyncMock()))),
+        patch("flyte._internal.runtime.convert.convert_upload_default_inputs", AsyncMock(return_value=[])),
+        patch("flyte._deploy.count") as count_mock,
+    ):
+        await _deploy_task(task, context, dryrun=False)
+
+    count_mock.assert_not_called()