Repository: ray-project/ray
File: python/ray/autoscaler/v2/instance_manager/common.py (Python)

Code before:
            return False

        # Log when transitioning from TERMINATION_FAILED to TERMINATED to help
        # track and debug potential termination failures.
        if (
            instance.status == Instance.TERMINATION_FAILED
            and new_instance_status == Instance.TERMINATED
        ):
            logger.info(
                f"Instance {instance.instance_id} transitioning from "
                f"TERMINATION_FAILED to TERMINATED. This may indicate the cloud "
                f"instance has already disappeared. Details: {details}"
            )

        # Log when transitioning from ALLOCATION_TIMEOUT to TERMINATED to help
        # track and debug potential allocation or termination issues.
        if (
            instance.status == Instance.ALLOCATION_TIMEOUT
            and new_instance_status == Instance.TERMINATED
        ):
            logger.info(
                f"Instance {instance.instance_id} transitioning from "
                f"ALLOCATION_TIMEOUT to TERMINATED. This may indicate the cloud "
                f"instance has already disappeared. Details: {details}"
            )

        instance.status = new_instance_status
        InstanceUtil._record_status_transition(instance, new_instance_status, details)
        return True

    @staticmethod
    def _record_status_transition(
        instance: Instance, status: Instance.InstanceStatus, details: str
    ):
        """Records the status transition.

        Args:
            instance: The instance to update.
            status: The new status to transition to.
        """
        now_ns = time.time_ns()
        instance.status_history.append(
            Instance.StatusHistory(
                instance_status=status,
                timestamp_ns=now_ns,
                details=details,
            )
        )

    @staticmethod
    def has_timeout(instance: Instance, timeout_s: int) -> bool:

Review comment (suggestion):
but can we remove these logs for now? I think we are planning to allow transition to TERMINATED in general, so the logging won't look like this.

Code after:
        InstanceUtil._record_status_transition(instance, new_instance_status, details)
        return True

    @staticmethod
    def _record_status_transition(
        instance: Instance, status: Instance.InstanceStatus, details: str
    ):
        """Records the status transition.

        Args:
            instance: The instance to update.
            status: The new status to transition to.
        """
        now_ns = time.time_ns()
        instance.status_history.append(
            Instance.StatusHistory(
                instance_status=status,
                timestamp_ns=now_ns,
                details=details,
            )
        )

    @staticmethod
    def has_timeout(instance: Instance, timeout_s: int) -> bool:
        """
        Returns True if the instance has been in the current status for more
        than the timeout_seconds.

        Args:
            instance: The instance to check.
            timeout_seconds: The timeout in seconds.

        Returns:
            True if the instance has been in the current status for more than
            the timeout_s seconds.
        """
        cur_status = instance.status

        status_times_ns = InstanceUtil.get_status_transition_times_ns(
            instance, select_instance_status=cur_status
        )
        assert len(status_times_ns) >= 1, (
            f"instance {instance.instance_id} has {len(status_times_ns)} "
            f"{Instance.InstanceStatus.Name(cur_status)} status"
        )
        status_time_ns = sorted(status_times_ns)[-1]
        if time.time_ns() - status_time_ns <= (timeout_s * 1e9):
            return False

        return True