Repository: apache/airflow
File: task-sdk/src/airflow/sdk/execution_time/task_runner.py (Python)

Code before:
        TaskDeferred,
    )

    if TYPE_CHECKING:
        assert ti.task is not None
        assert isinstance(ti.task, BaseOperator)

    parent_pid = os.getpid()

    def _on_term(signum, frame):
        pid = os.getpid()
        if pid != parent_pid:
            return

        ti.task.on_kill()

    signal.signal(signal.SIGTERM, _on_term)

    msg: ToSupervisor | None = None
    state: TaskInstanceState
    error: BaseException | None = None

    try:
        # First, clear the xcom data sent from server
        if ti._ti_context_from_server and (keys_to_delete := ti._ti_context_from_server.xcom_keys_to_clear):
            for x in keys_to_delete:
                log.debug("Clearing XCom with key", key=x)
                XCom.delete(
                    key=x,
                    dag_id=ti.dag_id,
                    task_id=ti.task_id,
                    run_id=ti.run_id,
                    map_index=ti.map_index,
                )

        with set_current_context(context):
            # This is the earliest that we can render templates -- as if it excepts for any reason we need to
            # catch it and handle it like a normal task failure
            if early_exit := _prepare(ti, log, context):
                msg = early_exit
                ti.state = state = TaskInstanceState.FAILED
                return state, msg, error

            try:
                result = _execute_task(context=context, ti=ti, log=log)
            except Exception:
                import jinja2

                # If the task failed, swallow rendering error so it doesn't mask the main error.
                with contextlib.suppress(jinja2.TemplateSyntaxError, jinja2.UndefinedError):
                    previous_rendered_map_index = ti.rendered_map_index

Review comment (suggestion):
You should use the `DualStatsManager` instead of `Stats` because of the legacy metric.

`ti.start` is the regular metric name and the `ti.start.{ti.dag_id}.{ti.task_id}` is the legacy name.


```diff
- Stats.incr(f"ti.start.{ti.dag_id}.{ti.task_id}", tags=stats_tags)
- # Same metric with tagging
- Stats.incr("ti.start", tags=stats_tags)
+ DualStatsManager.incr("ti.start", extra_tags=stats_tags)
```

There is a new config option to disable legacy metrics. The `DualStatsManager` will check the option and skip the legacy name if enabled. Also it will read the `metrics_tamplate.yaml` and get the legacy name and the name tags from there

https://github.com/astronomer/airflow/blob/ff0c4bd334d215a1f2d717f1f0f2e1c25985c257/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml#L167-L179


Code after:
        TaskDeferred,
    )

    if TYPE_CHECKING:
        assert ti.task is not None
        assert isinstance(ti.task, BaseOperator)

    parent_pid = os.getpid()

    def _on_term(signum, frame):
        pid = os.getpid()
        if pid != parent_pid:
            return

        ti.task.on_kill()

    signal.signal(signal.SIGTERM, _on_term)

    msg: ToSupervisor | None = None
    state: TaskInstanceState
    error: BaseException | None = None

    stats_tags = {"dag_id": ti.dag_id, "task_id": ti.task_id}
    Stats.incr(f"ti.start.{ti.dag_id}.{ti.task_id}", tags=stats_tags)
    # Same metric with tagging
    Stats.incr("ti.start", tags=stats_tags)

    try:
        # First, clear the xcom data sent from server
        if ti._ti_context_from_server and (keys_to_delete := ti._ti_context_from_server.xcom_keys_to_clear):
            for x in keys_to_delete:
                log.debug("Clearing XCom with key", key=x)
                XCom.delete(
                    key=x,
                    dag_id=ti.dag_id,
                    task_id=ti.task_id,
                    run_id=ti.run_id,
                    map_index=ti.map_index,
                )

        with set_current_context(context):
            # This is the earliest that we can render templates -- as if it excepts for any reason we need to
            # catch it and handle it like a normal task failure
            if early_exit := _prepare(ti, log, context):
                msg = early_exit
                ti.state = state = TaskInstanceState.FAILED
                return state, msg, error

            try:
                result = _execute_task(context=context, ti=ti, log=log)
            except Exception: