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

Code before:
            end_date=ti.end_date,
            rendered_map_index=ti.rendered_map_index,
        )
        state = TaskInstanceState.FAILED
        error = e
    except SystemExit as e:
        # SystemExit needs to be retried if they are eligible.
        log.error("Task exited", exit_code=e.code)
        msg, state = _handle_current_task_failed(ti)
        error = e
    except BaseException as e:
        log.exception("Task failed with exception")
        msg, state = _handle_current_task_failed(ti)
        error = e
    finally:
        if msg:
            SUPERVISOR_COMMS.send(msg=msg)

    # Return the message to make unit tests easier too
    ti.state = state
    return state, msg, error


def _handle_current_task_success(
    context: Context,
    ti: RuntimeTaskInstance,
) -> tuple[SucceedTask, TaskInstanceState]:
    end_date = datetime.now(tz=timezone.utc)
    ti.end_date = end_date

    # Record operator and task instance success metrics
    operator = ti.task.__class__.__name__
    stats_tags = {"dag_id": ti.dag_id, "task_id": ti.task_id}

    Stats.incr(f"operator_successes_{operator}", tags=stats_tags)
    # Same metric with tagging
    Stats.incr("operator_successes", tags={**stats_tags, "operator": operator})
    Stats.incr("ti_successes", tags=stats_tags)

    task_outlets = list(_build_asset_profiles(ti.task.outlets))
    outlet_events = list(_serialize_outlet_events(context["outlet_events"]))
    msg = SucceedTask(
        end_date=end_date,
        task_outlets=task_outlets,
        outlet_events=outlet_events,
        rendered_map_index=ti.rendered_map_index,
    )
    return msg, TaskInstanceState.SUCCESS


def _handle_current_task_failed(

Review comment (suggestion):
Similarly here

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

`tags` are used for both new and legacy metrics. `extra_tags` are referring to the tags on the legacy metric name.

Code after:
        # If these are thrown, we should mark the TI state as failed.
        log.exception("Task failed with exception")
        ti.end_date = datetime.now(tz=timezone.utc)
        msg = TaskState(
            state=TaskInstanceState.FAILED,
            end_date=ti.end_date,
            rendered_map_index=ti.rendered_map_index,
        )
        state = TaskInstanceState.FAILED
        error = e
    except SystemExit as e:
        # SystemExit needs to be retried if they are eligible.
        log.error("Task exited", exit_code=e.code)
        msg, state = _handle_current_task_failed(ti)
        error = e
    except BaseException as e:
        log.exception("Task failed with exception")
        msg, state = _handle_current_task_failed(ti)
        error = e
    finally:
        Stats.incr(
            f"ti.finish.{ti.dag_id}.{ti.task_id}.{state.value}",
            tags=stats_tags,
        )
        # Same metric with tagging
        Stats.incr("ti.finish", tags={**stats_tags, "state": state.value})

        if msg:
            SUPERVISOR_COMMS.send(msg=msg)

    # Return the message to make unit tests easier too
    ti.state = state
    return state, msg, error


def _handle_current_task_success(
    context: Context,
    ti: RuntimeTaskInstance,
) -> tuple[SucceedTask, TaskInstanceState]:
    end_date = datetime.now(tz=timezone.utc)
    ti.end_date = end_date

    # Record operator and task instance success metrics
    operator = ti.task.__class__.__name__
    stats_tags = {"dag_id": ti.dag_id, "task_id": ti.task_id}

    Stats.incr(f"operator_successes_{operator}", tags=stats_tags)
    # Same metric with tagging
    Stats.incr("operator_successes", tags={**stats_tags, "operator": operator})
    Stats.incr("ti_successes", tags=stats_tags)