odps/models/instance.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 1999-2025 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import print_function

import base64
import functools
import json
import logging
import sys
import threading
import time
import warnings
from collections import OrderedDict
from datetime import datetime

import requests

from .. import compat, errors, options, readers, serializers, utils
from ..accounts import BearerTokenAccount
from ..compat import Enum, six
from ..lib.monotonic import monotonic
from ..lib.tblib import pickling_support
from ..utils import to_str
from .core import JSONRemoteModel, LazyLoad, XMLRemoteModel
from .job import Job
from .readers import TunnelArrowReader, TunnelRecordReader
from .tasks import SQLTask
from .worker import LOG_TYPES_MAPPING, WorkerDetail2

logger = logging.getLogger(__name__)
pickling_support.install()


_RESULT_LIMIT_HELPER_MSG = (
    "See https://pyodps.readthedocs.io/zh_CN/latest/base-sql.html#read-sql-exec-result "
    "for more information about limits on instance results."
)
_STATUS_QUERY_TIMEOUT = 5 * 60  # timeout when getting status


def _with_status_api_lock(func):
    @six.wraps(func)
    def wrapped(self, *args, **kw):
        with self._status_api_lock:
            return func(self, *args, **kw)

    return wrapped


class SpawnedInstanceReaderMixin(object):
    @property
    def schema(self):
        return self._schema

    @staticmethod
    def _read_instance_split(
        conn,
        download_id,
        start,
        count,
        idx,
        rest_client=None,
        project=None,
        instance_id=None,
        tunnel_endpoint=None,
        columns=None,
        arrow=False,
    ):
        # read part data
        from ..tunnel import InstanceTunnel

        try:
            instance_tunnel = InstanceTunnel(
                client=rest_client, project=project, endpoint=tunnel_endpoint
            )
            session = utils.call_with_retry(
                instance_tunnel.create_download_session,
                instance=instance_id,
                download_id=download_id,
            )

            def _data_to_pandas():
                if not arrow:
                    with session.open_record_reader(
                        start, count, columns=columns
                    ) as reader:
                        return reader.to_pandas()
                else:
                    with session.open_arrow_reader(
                        start, count, columns=columns
                    ) as reader:
                        return reader.to_pandas()

            data = utils.call_with_retry(_data_to_pandas)
            conn.send((idx, data, True))
        except:
            try:
                conn.send((idx, sys.exc_info(), False))
            except:
                logger.exception("Failed to write in process %d", idx)
                raise

    def _get_process_split_reader(self, columns=None, append_partitions=None):  # noqa
        rest_client = self._parent._client
        project = self._parent.project.name
        tunnel_endpoint = self._parent.project._tunnel_endpoint
        instance_id = self._parent.id

        return functools.partial(
            self._read_instance_split,
            rest_client=rest_client,
            project=project,
            instance_id=instance_id,
            arrow=isinstance(self, TunnelArrowReader),
            tunnel_endpoint=tunnel_endpoint,
            columns=columns or self._column_names,
        )


class InstanceRecordReader(SpawnedInstanceReaderMixin, TunnelRecordReader):
    def __init__(self, instance, download_session, columns=None):
        super(InstanceRecordReader, self).__init__(
            instance, download_session, columns=columns
        )
        self._schema = download_session.schema


class InstanceArrowReader(SpawnedInstanceReaderMixin, TunnelArrowReader):
    def __init__(self, instance, download_session, columns=None):
        super(InstanceArrowReader, self).__init__(
            instance, download_session, columns=columns
        )
        self._schema = download_session.schema


class Instance(LazyLoad):
    """
    Instance means that a ODPS task will sometimes run as an instance.

    ``status`` can reflect the current situation of a instance.
    ``is_terminated`` method indicates if the instance has finished.
    ``is_successful`` method indicates if the instance runs successfully.
    ``wait_for_success`` method will block the main process until the instance has finished.

    For a SQL instance, we can use open_reader to read the results.

    :Example:

    >>> instance = odps.execute_sql('select * from dual')  # this sql return the structured data
    >>> with instance.open_reader() as reader:
    >>>     # handle the record
    >>>
    >>> instance = odps.execute_sql('desc dual')  # this sql do not return structured data
    >>> with instance.open_reader() as reader:
    >>>    print(reader.raw)  # just return the raw result
    """

    __slots__ = (
        "_task_results",
        "_is_sync",
        "_id_thread_local",
        "_status_api_lock",
        "_logview_address",
        "_logview_address_time",
        "_last_progress_value",
        "_last_progress_time",
        "_logview_logged",
        "_job_source",
    )

    _download_id = utils.thread_local_attribute("_id_thread_local", lambda: None)

    def __init__(self, **kwargs):
        if "task_results" in kwargs:
            kwargs["_task_results"] = kwargs.pop("task_results")
        super(Instance, self).__init__(**kwargs)

        try:
            del self._id_thread_local
        except AttributeError:
            pass

        if self._task_results is not None and len(self._task_results) > 0:
            self._is_sync = True
            self._status = Instance.Status.TERMINATED
        else:
            self._is_sync = False

        self._status_api_lock = threading.RLock()

        self._logview_address = None
        self._logview_address_time = None
        self._last_progress_value = None
        self._last_progress_time = None
        self._logview_logged = False
        self._job_source = None

    @property
    def id(self):
        return self.name

    class Status(Enum):
        RUNNING = "Running"
        SUSPENDED = "Suspended"
        TERMINATED = "Terminated"

    class InstanceStatus(XMLRemoteModel):
        _root = "Instance"

        status = serializers.XMLNodeField("Status")

    class InstanceResult(XMLRemoteModel):
        class TaskResult(XMLRemoteModel):
            class Result(XMLRemoteModel):
                transform = serializers.XMLNodeAttributeField(attr="Transform")
                format = serializers.XMLNodeAttributeField(attr="Format")
                text = serializers.XMLNodeField(".", default="")

                def __str__(self):
                    if six.PY2:
                        text = utils.to_binary(self.text)
                    else:
                        text = self.text
                    if self.transform is not None and self.transform == "Base64":
                        try:
                            return utils.to_str(base64.b64decode(text))
                        except TypeError:
                            return text
                    return text

                def __bytes__(self):
                    text = utils.to_binary(self.text)
                    if self.transform is not None and self.transform == "Base64":
                        try:
                            return utils.to_binary(base64.b64decode(text))
                        except TypeError:
                            return text
                    return text

            type = serializers.XMLNodeAttributeField(attr="Type")
            name = serializers.XMLNodeField("Name")
            result = serializers.XMLNodeReferenceField(Result, "Result")

        task_results = serializers.XMLNodesReferencesField(TaskResult, "Tasks", "Task")

    class Task(XMLRemoteModel):
        """
        Task stands for each task inside an instance.

        It has a name, a task type, the start to end time, and a running status.
        """

        name = serializers.XMLNodeField("Name")
        type = serializers.XMLNodeAttributeField(attr="Type")
        start_time = serializers.XMLNodeField(
            "StartTime", parse_callback=utils.parse_rfc822
        )
        end_time = serializers.XMLNodeField(
            "EndTime", parse_callback=utils.parse_rfc822
        )
        status = serializers.XMLNodeField(
            "Status", parse_callback=lambda s: Instance.Task.TaskStatus(s.upper())
        )
        histories = serializers.XMLNodesReferencesField(
            "Instance.Task", "Histories", "History"
        )

        class TaskStatus(Enum):
            WAITING = "WAITING"
            RUNNING = "RUNNING"
            SUCCESS = "SUCCESS"
            FAILED = "FAILED"
            SUSPENDED = "SUSPENDED"
            CANCELLED = "CANCELLED"

        class TaskProgress(XMLRemoteModel):
            """
            TaskProgress reprents for the progress of a task.

            A single TaskProgress may consist of several stages.

            :Example:

            >>> progress = instance.get_task_progress('task_name')
            >>> progress.get_stage_progress_formatted_string()
            2015-11-19 16:39:07 M1_Stg1_job0:0/0/1[0%]	R2_1_Stg1_job0:0/0/1[0%]
            """

            class StageProgress(XMLRemoteModel):
                name = serializers.XMLNodeAttributeField(attr="ID")
                backup_workers = serializers.XMLNodeField(
                    "BackupWorkers", parse_callback=int
                )
                terminated_workers = serializers.XMLNodeField(
                    "TerminatedWorkers", parse_callback=int
                )
                running_workers = serializers.XMLNodeField(
                    "RunningWorkers", parse_callback=int
                )
                total_workers = serializers.XMLNodeField(
                    "TotalWorkers", parse_callback=int
                )
                input_records = serializers.XMLNodeField(
                    "InputRecords", parse_callback=int
                )
                output_records = serializers.XMLNodeField(
                    "OutputRecords", parse_callback=int
                )
                finished_percentage = serializers.XMLNodeField(
                    "FinishedPercentage", parse_callback=int
                )

            stages = serializers.XMLNodesReferencesField(StageProgress, "Stage")

            def get_stage_progress_formatted_string(self):
                buf = six.StringIO()

                buf.write(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
                buf.write(" ")

                for stage in self.stages:
                    buf.write(
                        "{0}:{1}/{2}/{3}{4}[{5}%]\t".format(
                            stage.name,
                            stage.running_workers,
                            stage.terminated_workers,
                            stage.total_workers,
                            "(+%s backups)" % stage.backup_workers
                            if stage.backup_workers > 0
                            else "",
                            stage.finished_percentage,
                        )
                    )

                return buf.getvalue()

    class TaskInfo(serializers.XMLSerializableModel):
        _root = "Instance"
        __slots__ = "key", "value"

        key = serializers.XMLNodeField("Key")
        value = serializers.XMLNodeField("Value")

    class TaskCost(object):
        __slots__ = "cpu_cost", "memory_cost", "input_size"

        def __init__(self, cpu_cost=None, memory_cost=None, input_size=None):
            self.cpu_cost = cpu_cost
            self.memory_cost = memory_cost
            self.input_size = input_size

    class SQLCost(object):
        __slots__ = "udf_num", "complexity", "input_size"

        def __init__(self, udf_num=None, complexity=None, input_size=None):
            self.udf_num = udf_num
            self.complexity = complexity
            self.input_size = input_size

    class DownloadSessionCreationError(errors.InternalServerError):
        pass

    class TaskSummary(dict):
        def __init__(self, *args, **kwargs):
            super(Instance.TaskSummary, self).__init__(*args, **kwargs)
            self.summary_text, self.json_summary = None, None

    class AnonymousSubmitInstance(XMLRemoteModel):
        _root = "Instance"
        job = serializers.XMLNodeReferenceField(Job, "Job")

    class InstanceQueueingInfo(JSONRemoteModel):
        __slots__ = ("_instance",)

        class Status(Enum):
            RUNNING = "Running"
            SUSPENDED = "Suspended"
            TERMINATED = "Terminated"
            UNKNOWN = "Unknown"

        _properties = serializers.JSONRawField()  # hold the raw dict
        instance_id = serializers.JSONNodeField("instanceId")
        priority = serializers.JSONNodeField("instancePriority")
        progress = serializers.JSONNodeField("instanceProcess")
        job_name = serializers.JSONNodeField("jobName")
        project = serializers.JSONNodeField("projectName")
        skynet_id = serializers.JSONNodeField("skynetId")
        start_time = serializers.JSONNodeField(
            "startTime", parse_callback=utils.strptime_with_tz
        )
        task_type = serializers.JSONNodeField("taskType")
        task_name = serializers.JSONNodeField("taskName")
        user_account = serializers.JSONNodeField("userAccount")
        status = serializers.JSONNodeField("status", parse_callback=Status)

        @property
        def instance(self):
            if hasattr(self, "_instance") and self._instance:
                return self._instance

            from .projects import Projects

            self._instance = Projects(client=self._client)[self.project].instances[
                self.instance_id
            ]
            return self._instance

        def __getattr__(self, item):
            item = utils.underline_to_camel(item)

            if item in self._properties:
                return self._properties[item]

            return super(Instance.InstanceQueueingInfo, self).__getattr__(item)

    name = serializers.XMLNodeField("Name")
    owner = serializers.XMLNodeField("Owner")
    start_time = serializers.XMLNodeField(
        "StartTime", parse_callback=utils.parse_rfc822
    )
    end_time = serializers.XMLNodeField("EndTime", parse_callback=utils.parse_rfc822)
    _status = serializers.XMLNodeField(
        "Status", parse_callback=lambda s: Instance.Status(s)
    )
    _tasks = serializers.XMLNodesReferencesField(Task, "Tasks", "Task")

    def reload(self, blocking=False):
        actions = []
        if blocking:
            actions.append("instancestatus")
        resp = self._client.get(self.resource(), actions=actions)

        self.owner = resp.headers.get("x-odps-owner")
        self.start_time = utils.parse_rfc822(resp.headers.get("x-odps-start-time"))
        end_time_header = "x-odps-end-time"
        if (
            end_time_header in resp.headers
            and len(resp.headers[end_time_header].strip()) > 0
        ):
            self.end_time = utils.parse_rfc822(resp.headers.get(end_time_header))

        self.parse(self._client, resp, obj=self)
        # remember not to set `_loaded = True`

    def stop(self):
        """
        Stop this instance.

        :return: None
        """

        instance_status = Instance.InstanceStatus(status="Terminated")
        xml_content = instance_status.serialize()

        headers = {"Content-Type": "application/xml"}
        self._client.put(self.resource(), xml_content, headers=headers)

    @staticmethod
    def _call_with_retry(func, retry=False, retry_timeout=None):
        retry_kw = {
            "retry_times": options.retry_times if retry else 0,
            "exc_type": (errors.InternalServerError, errors.RequestTimeTooSkewed),
        }
        if retry and retry_timeout is not None:
            # use retry timeout instead of retry count
            retry_kw.update({"retry_times": None, "retry_timeout": retry_timeout})
        return utils.call_with_retry(func, **retry_kw)

    @_with_status_api_lock
    def get_task_results_without_format(self, timeout=None, retry=True):
        if self._is_sync:
            return self._task_results

        def _get_resp():
            return self._client.get(self.resource(), action="result", timeout=timeout)

        resp = self._call_with_retry(_get_resp, retry=retry, retry_timeout=timeout)
        instance_result = Instance.InstanceResult.parse(self._client, resp)
        return OrderedDict([(r.name, r.result) for r in instance_result.task_results])

    @_with_status_api_lock
    def get_task_results(self, timeout=None, retry=True):
        """
        Get all the task results.

        :return: a dict which key is task name, and value is the task result as string
        :rtype: dict
        """

        results = self.get_task_results_without_format(timeout=timeout, retry=retry)
        if options.tunnel.string_as_binary:
            return OrderedDict(
                [(k, bytes(result)) for k, result in six.iteritems(results)]
            )
        else:
            return OrderedDict(
                [(k, str(result)) for k, result in six.iteritems(results)]
            )

    def _get_default_task_name(self):
        job = self._get_job()
        if len(job.tasks) != 1:
            msg = "No tasks" if len(job.tasks) == 0 else "Multiple tasks"
            raise errors.ODPSError("%s in instance." % msg)
        return job.tasks[0].name

    @_with_status_api_lock
    def get_task_result(self, task_name=None, timeout=None, retry=True):
        """
        Get a single task result.

        :param task_name: task name
        :return: task result
        :rtype: str
        """
        task_name = task_name or self._get_default_task_name()
        return self.get_task_results(timeout=timeout, retry=retry).get(task_name)

    @_with_status_api_lock
    def get_task_summary(self, task_name=None):
        """
        Get a task's summary, mostly used for MapReduce.

        :param task_name: task name
        :return: summary as a dict parsed from JSON
        :rtype: dict
        """
        task_name = task_name or self._get_default_task_name()
        params = {"taskname": task_name}
        resp = self._client.get(
            self.resource(), action="instancesummary", params=params
        )

        map_reduce = resp.json().get("Instance")
        if map_reduce:
            json_summary = map_reduce.get("JsonSummary")
            if json_summary:
                summary = Instance.TaskSummary(json.loads(json_summary))
                summary.summary_text = map_reduce.get("Summary")
                summary.json_summary = json_summary

                return summary

    @_with_status_api_lock
    def get_task_statuses(self, retry=True, timeout=None):
        """
        Get all tasks' statuses

        :return: a dict which key is the task name and value is the :class:`odps.models.Instance.Task` object
        :rtype: dict
        """

        def _get_resp():
            return self._client.get(self.resource(), action="taskstatus")

        resp = self._call_with_retry(_get_resp, retry=retry, retry_timeout=timeout)
        self.parse(self._client, resp, obj=self)
        return dict([(task.name, task) for task in self._tasks])

    @_with_status_api_lock
    def get_task_names(self, retry=True, timeout=None):
        """
        Get names of all tasks

        :return: task names
        :rtype: list
        """

        return compat.lkeys(self.get_task_statuses(retry=retry, timeout=timeout))

    def get_task_cost(self, task_name=None):
        """
        Get task cost

        :param task_name: name of the task
        :return: task cost
        :rtype: Instance.TaskCost

        :Example:

        >>> cost = instance.get_task_cost(instance.get_task_names()[0])
        >>> cost.cpu_cost
        200
        >>> cost.memory_cost
        4096
        >>> cost.input_size
        0
        """
        task_name = task_name or self._get_default_task_name()
        summary = self.get_task_summary(task_name)
        if summary is None:
            return None

        if "Cost" in summary:
            task_cost = summary["Cost"]

            cpu_cost = task_cost.get("CPU")
            memory = task_cost.get("Memory")
            input_size = task_cost.get("Input")

            return Instance.TaskCost(cpu_cost, memory, input_size)

    def _raise_empty_task_info(self, resp):
        raise errors.EmptyTaskInfoError(
            "Empty response. Task server maybe dead.",
            code=resp.status_code,
            instance_id=self.id,
            endpoint=self._client.endpoint,
            request_id=resp.headers.get("x-odps-request-id"),
            tag="ODPS",
        )

    def get_task_info(self, task_name, key, raise_empty=False):
        """
        Get task related information.

        :param task_name: name of the task
        :param key: key of the information item
        :param raise_empty: if True, will raise error when response is empty
        :return: a string of the task information
        """
        actions = ["info"]
        params = OrderedDict([("taskname", task_name), ("key", key)])

        resp = self._client.get(self.resource(), actions=actions, params=params)
        resp_data = resp.content.decode()
        if raise_empty and not resp_data:
            self._raise_empty_task_info(resp)
        return resp_data

    def put_task_info(
        self, task_name, key, value, check_location=False, raise_empty=False
    ):
        """
        Put information into a task.

        :param task_name: name of the task
        :param key: key of the information item
        :param value: value of the information item
        :param check_location: raises if Location header is missing
        :param raise_empty: if True, will raise error when response is empty
        """
        actions = ["info"]
        params = {"taskname": task_name}
        headers = {"Content-Type": "application/xml"}
        body = self.TaskInfo(key=key, value=value).serialize()

        resp = self._client.put(
            self.resource(), actions=actions, params=params, headers=headers, data=body
        )

        location = resp.headers.get("Location")
        if check_location and (location is None or len(location) == 0):
            raise errors.ODPSError("Invalid response, Location header required.")
        resp_data = resp.content.decode()
        if raise_empty and not resp_data:
            self._raise_empty_task_info(resp)
        return resp_data

    def get_task_quota(self, task_name=None):
        """
        Get queueing info of the task.
        Note that time between two calls should larger than 30 seconds, otherwise empty dict is returned.

        :param task_name: name of the task
        :return: quota info in dict format
        """
        task_name = task_name or self._get_default_task_name()
        actions = ["instancequota"]
        params = {"taskname": task_name}
        resp = self._client.get(self.resource(), actions=actions, params=params)
        return json.loads(resp.text)

    def get_sql_task_cost(self):
        """
        Get cost information of the sql cost task, including input data size,
        number of UDF, Complexity of the sql task.

        NOTE that DO NOT use this function directly as it cannot be applied to
        instances returned from SQL. Use ``o.execute_sql_cost`` instead.

        :return: cost info in dict format
        """
        resp = self.get_task_result(self.get_task_names()[0])
        cost = json.loads(resp)
        sql_cost = cost["Cost"]["SQL"]

        udf_num = sql_cost.get("UDF")
        complexity = sql_cost.get("Complexity")
        input_size = sql_cost.get("Input")
        return Instance.SQLCost(udf_num, complexity, input_size)

    def _get_status(self, blocking=False):
        if self._status != Instance.Status.TERMINATED:
            self.reload(blocking)
        return self._status

    @property
    @_with_status_api_lock
    def status(self):
        return self._get_status()

    def is_terminated(self, retry=True, blocking=False, retry_timeout=None):
        """
        If this instance has finished or not.

        :return: True if finished else False
        :rtype: bool
        """
        return self._call_with_retry(
            lambda: self._get_status(blocking) == Instance.Status.TERMINATED,
            retry=retry,
            retry_timeout=retry_timeout,
        )

    def is_running(self, retry=True, blocking=False, retry_timeout=None):
        """
        If this instance is still running.

        :return: True if still running else False
        :rtype: bool
        """
        return self._call_with_retry(
            lambda: self._get_status(blocking) == Instance.Status.RUNNING,
            retry=retry,
            retry_timeout=retry_timeout,
        )

    def is_successful(self, retry=True, retry_timeout=None):
        """
        If the instance runs successfully.

        :return: True if successful else False
        :rtype: bool
        """

        if not self.is_terminated(retry=retry):
            return False

        def _get_successful():
            statuses = self.get_task_statuses()
            return all(
                task.status == Instance.Task.TaskStatus.SUCCESS
                for task in statuses.values()
            )

        return self._call_with_retry(
            _get_successful, retry=retry, retry_timeout=retry_timeout
        )

    @property
    def is_sync(self):
        return self._is_sync

    def get_all_task_progresses(self):
        return {
            task_name: self.get_task_progress(task_name)
            for task_name in self.get_task_names()
        }

    def _dump_instance_progress(self, start_time, check_time, final=False):
        if logger.getEffectiveLevel() > logging.INFO:
            return

        prog_time_interval = options.progress_time_interval
        prog_percentage_gap = options.progress_percentage_gap
        logview_latency = min(options.logview_latency, prog_time_interval)
        try:
            task_progresses = self.get_all_task_progresses()
            total_progress = sum(
                stage.finished_percentage
                for progress in task_progresses.values()
                for stage in progress.stages
            )

            if not self._logview_logged and check_time - start_time >= logview_latency:
                self._logview_logged = True
                logger.info(
                    "Instance ID: %s\n  Log view: %s",
                    self.id,
                    self.get_logview_address(),
                )

            # final log need to be outputed once the progress is updated and logview
            #  address is printed
            need_final_log = (
                final
                and self._logview_logged
                and self._last_progress_value < total_progress
            )
            # intermediate log need to be outputed once current progress exceeds certain
            #  gap or certain time elapsed
            need_intermediate_log = check_time - start_time >= prog_time_interval and (
                total_progress - self._last_progress_value >= prog_percentage_gap
                or check_time - self._last_progress_time >= prog_time_interval
            )
            if need_final_log or need_intermediate_log:
                output_parts = [str(self.id)] + [
                    progress.get_stage_progress_formatted_string()
                    for progress in task_progresses.values()
                ]
                if len(output_parts) > 1:
                    logger.info(" ".join(output_parts))
                self._last_progress_value = total_progress
                self._last_progress_time = check_time
        except:  # pragma: no cover
            # make sure progress display does not affect execution
            pass

    def wait_for_completion(
        self, interval=1, timeout=None, max_interval=None, blocking=True
    ):
        """
        Wait for the instance to complete, and neglect the consequence.

        :param interval: time interval to check
        :param max_interval: if specified, next check interval will be
            multiplied by 2 till max_interval is reached.
        :param timeout: time
        :param blocking: whether to block waiting at server side. Note that this option does
            not affect client behavior.
        :return: None
        """

        start_time = check_time = self._last_progress_time = monotonic()
        self._last_progress_value = 0
        while not self.is_terminated(
            retry=True, blocking=blocking, retry_timeout=_STATUS_QUERY_TIMEOUT
        ):
            try:
                sleep_interval_left = interval - (monotonic() - check_time)
                if sleep_interval_left > 0:
                    time.sleep(sleep_interval_left)
                check_time = monotonic()
                if max_interval is not None:
                    interval = min(interval * 2, max_interval)
                if timeout is not None and check_time - start_time > timeout:
                    raise errors.WaitTimeoutError(
                        "Wait completion of instance %s timed out" % self.id,
                        instance_id=self.id,
                    )

                self._dump_instance_progress(start_time, check_time)
            except KeyboardInterrupt:
                break
        # dump final progress
        self._dump_instance_progress(start_time, check_time, final=True)

    def wait_for_success(
        self, interval=1, timeout=None, max_interval=None, blocking=True
    ):
        """
        Wait for instance to complete, and check if the instance is successful.

        :param interval: time interval to check
        :param max_interval: if specified, next check interval will be
            multiplied by 2 till max_interval is reached.
        :param timeout: time
        :param blocking: whether to block waiting at server side. Note that this option does
            not affect client behavior.
        :return: None
        :raise: :class:`odps.errors.ODPSError` if the instance failed
        """

        self.wait_for_completion(
            interval=interval,
            max_interval=max_interval,
            timeout=timeout,
            blocking=blocking,
        )

        if not self.is_successful(retry=True):
            for task_name, task in six.iteritems(self.get_task_statuses()):
                exc = None
                if task.status == Instance.Task.TaskStatus.FAILED:
                    exc = errors.parse_instance_error(self.get_task_result(task_name))
                elif task.status != Instance.Task.TaskStatus.SUCCESS:
                    exc = errors.ODPSError(
                        "%s, status=%s" % (task_name, task.status.value)
                    )
                if exc:
                    exc.instance_id = self.id
                    raise exc

    @_with_status_api_lock
    def get_task_progress(self, task_name=None):
        """
        Get task's current progress

        :param task_name: task_name
        :return: the task's progress
        :rtype: :class:`odps.models.Instance.Task.TaskProgress`
        """
        task_name = task_name or self._get_default_task_name()
        params = {"instanceprogress": task_name, "taskname": task_name}

        resp = self._client.get(self.resource(), params=params)
        return Instance.Task.TaskProgress.parse(self._client, resp)

    @_with_status_api_lock
    def get_task_detail(self, task_name=None):
        """
        Get task's detail

        :param task_name: task name
        :return: the task's detail
        :rtype: list or dict according to the JSON
        """

        def _get_detail():
            from ..compat import json  # fix object_pairs_hook parameter for Py2.6

            params = {"taskname": task_name}
            resp = self._client.get(
                self.resource(), action="instancedetail", params=params
            )
            res = resp.content.decode() if six.PY3 else resp.content
            try:
                return json.loads(res, object_pairs_hook=OrderedDict)
            except ValueError:
                return res

        task_name = task_name or self._get_default_task_name()
        result = _get_detail()
        if not result:
            # todo: this is a workaround for the bug that get_task_detail returns nothing.
            return self.get_task_detail2(task_name)
        else:
            return result

    @_with_status_api_lock
    def get_task_detail2(self, task_name=None, **kw):
        """
        Get task's detail v2

        :param task_name: task name
        :return: the task's detail
        :rtype: list or dict according to the JSON
        """
        task_name = task_name or self._get_default_task_name()
        params = {"taskname": task_name}
        if "subquery_id" in kw:
            params["subquery_id"] = str(kw.pop("subquery_id"))

        resp = self._client.get(self.resource(), action="detail", params=params)
        res = resp.content.decode() if six.PY3 else resp.content
        try:
            return json.loads(res, object_pairs_hook=OrderedDict)
        except ValueError:
            return res

    @_with_status_api_lock
    def get_task_workers(self, task_name=None, json_obj=None):
        """
        Get workers from task
        :param task_name: task name
        :param json_obj: json object parsed from get_task_detail2
        :return: list of workers

        .. seealso:: :class:`odps.models.Worker`
        """
        if json_obj is None:
            task_name = task_name or self._get_default_task_name()

        if json_obj is None:
            json_obj = self.get_task_detail2(task_name)
        return WorkerDetail2.extract_from_json(
            json_obj, client=self._client, parent=self
        )

    @_with_status_api_lock
    def get_worker_log(self, log_id, log_type, size=0):
        """
        Get logs from worker.

        :param log_id: id of log, can be retrieved from details.
        :param log_type: type of logs. Possible log types contains {log_types}
        :param size: length of the log to retrieve
        :return: log content
        """
        params = OrderedDict([("log", ""), ("id", log_id)])
        if log_type is not None:
            log_type = log_type.lower()
            if log_type not in LOG_TYPES_MAPPING:
                raise ValueError(
                    "log_type should choose a value in "
                    + " ".join(six.iterkeys(LOG_TYPES_MAPPING))
                )
            params["logtype"] = LOG_TYPES_MAPPING[log_type]
        if size > 0:
            params["size"] = str(size)
        resp = self._client.get(self.resource(), params=params)
        return resp.text

    get_worker_log.__doc__ = get_worker_log.__doc__.format(
        log_types=", ".join(sorted(six.iterkeys(LOG_TYPES_MAPPING)))
    )

    @_with_status_api_lock
    def get_logview_address(self, hours=None, use_legacy=None):
        """
        Get logview address of the instance object by hours.

        :param hours:
        :return: logview address
        :rtype: str
        """
        if use_legacy is None:
            use_legacy = options.use_legacy_logview
        if use_legacy is None and self.project.odps.job_insight_host is not None:
            use_legacy = False
        if (
            self.project.odps.job_insight_host is None
            or self.project.odps.region_name is None
        ):
            use_legacy = True
        if use_legacy is False:
            return self._get_job_insight_address()
        return self._get_legacy_logview_address(hours=hours)

    def _get_job_insight_address(self):
        return (
            "%(job_insight_host)s/%(region_id)s/job-insights?h=%(endpoint)s"
            "&p=%(project_name)s&i=%(instance_id)s"
        ) % dict(
            job_insight_host=self.project.odps.job_insight_host,
            region_id=self.project.odps.region_name,
            endpoint=self._client.endpoint,
            project_name=self.project.name,
            instance_id=self.id,
        )

    def _get_legacy_logview_address(self, hours=None):
        if (
            self._logview_address is not None
            and monotonic() - self._logview_address_time < 600
        ):
            return self._logview_address

        project = self.project
        if isinstance(project.odps.account, BearerTokenAccount):
            token = to_str(project.odps.account.token)
        else:
            hours = hours or options.logview_hours
            policy = {
                "Statement": [
                    {
                        "Action": ["odps:Read"],
                        "Effect": "Allow",
                        "Resource": "acs:odps:*:projects/%s/instances/%s"
                        % (project.name, self.id),
                    }
                ],
                "Version": "1",
            }
            token = self.project.generate_auth_token(policy, "bearer", hours)

        link = (
            "%(logview_host)s/logview/?h=%(endpoint)s&p=%(project_name)s"
            "&i=%(instance_id)s&token=%(token)s"
        ) % dict(
            logview_host=self.project.odps.logview_host,
            endpoint=self._client.endpoint,
            project_name=project.name,
            instance_id=self.id,
            token=token,
        )
        self._logview_address = link
        self._logview_address_time = monotonic()
        return link

    def __str__(self):
        return self.id

    def _get_job(self):
        if not self._job_source:
            url = self.resource()
            resp = self._client.get(url, action="source")

            self._job_source = Job.parse(self._client, resp, parent=self)
        return self._job_source

    def get_tasks(self):
        return self.tasks

    @property
    def tasks(self):
        job = self._get_job()
        return job.tasks

    @property
    def priority(self):
        job = self._get_job()
        return job.priority

    def _get_queueing_info(self, **kw):
        params = {}
        if "subquery_id" in kw:
            params["subquery_id"] = str(kw.pop("subquery_id"))

        url = self.resource()
        resp = self._client.get(url, action="cached", params=params)
        return (
            Instance.InstanceQueueingInfo.parse(
                self._client, resp, parent=self.project.instance_queueing_infos
            ),
            resp,
        )

    def get_queueing_info(self):
        info, _ = self._get_queueing_info()
        return info

    def get_sql_query(self):
        task = [t for t in self.tasks if isinstance(t, SQLTask)]
        if not task:
            raise errors.ODPSError("Instance %s does not contain a SQLTask.", self.id)
        if len(task) > 1:  # pragma: no cover
            raise errors.ODPSError("Multiple SQLTasks exist in instance %s.", self.id)
        return task[0].query

    def get_unique_identifier_id(self):
        job = self._get_job()
        return job.unique_identifier_id

    def _check_get_task_name(self, task_type, task_name=None, err_head=None):
        if not self.is_successful(retry=True):
            raise errors.ODPSError(
                "%s, instance(%s) may fail or has not finished yet"
                % (err_head, self.id)
            )

        task_type = task_type.lower()
        filtered_tasks = {
            name: task
            for name, task in six.iteritems(self.get_task_statuses())
            if task.type.lower() == task_type
        }
        if len(filtered_tasks) > 1:
            if task_name is None:
                raise errors.ODPSError(
                    "%s, job has more than one %s tasks, please specify one"
                    % (err_head, task_type)
                )
            elif task_name not in filtered_tasks:
                raise errors.ODPSError(
                    "%s, unknown task name: %s" % (err_head, task_name)
                )
            return task_name
        elif len(filtered_tasks) == 1:
            return list(filtered_tasks)[0]
        else:
            raise errors.ODPSError("%s, job has no %s task" % (err_head, task_type))

    def _create_instance_tunnel(self, endpoint=None, quota_name=None):
        from ..tunnel import InstanceTunnel

        return InstanceTunnel(
            client=self._client,
            project=self.project,
            quota_name=quota_name,
            endpoint=endpoint or self.project._tunnel_endpoint,
        )

    @utils.survey
    def _open_result_reader(self, schema=None, task_name=None, timeout=None, **kw):
        task_name = self._check_get_task_name(
            "sql", task_name=task_name, err_head="Cannot open reader"
        )
        result = self.get_task_result(task_name, timeout=timeout)
        reader = readers.CsvRecordReader(schema, result, **kw)
        if options.result_reader_create_callback:
            options.result_reader_create_callback(reader)
        return reader

    def _open_tunnel_reader(self, **kw):
        from ..tunnel.instancetunnel import InstanceDownloadSession

        reopen = kw.pop("reopen", False)
        endpoint = kw.pop("endpoint", None)
        quota_name = kw.pop("quota_name", None)
        arrow = kw.pop("arrow", False)
        columns = kw.pop("columns", None)

        tunnel = self._create_instance_tunnel(endpoint=endpoint, quota_name=quota_name)
        download_id = self._download_id if not reopen else None

        try:
            download_session = tunnel.create_download_session(
                instance=self, download_id=download_id, **kw
            )
            if (
                download_id
                and download_session.status != InstanceDownloadSession.Status.Normal
            ):
                download_session = tunnel.create_download_session(instance=self, **kw)
        except errors.InternalServerError:
            e, tb = sys.exc_info()[1:]
            e.__class__ = Instance.DownloadSessionCreationError
            six.reraise(Instance.DownloadSessionCreationError, e, tb)

        self._download_id = download_session.id

        if arrow:
            return InstanceArrowReader(self, download_session, columns=columns)
        else:
            return InstanceRecordReader(self, download_session, columns=columns)

    def open_reader(self, *args, **kwargs):
        """
        Open the reader to read records from the result of the instance. If `tunnel` is `True`,
        instance tunnel will be used. Otherwise conventional routine will be used. If instance tunnel
        is not available and `tunnel` is not specified, the method will fall back to the
        conventional routine.
        Note that the number of records returned is limited unless `options.limited_instance_tunnel`
        is set to `True` or `limit=True` is configured under instance tunnel mode. Otherwise
        the number of records returned is always limited.

        :param tunnel: if true, use instance tunnel to read from the instance.
                       if false, use conventional routine.
                       if absent, `options.tunnel.use_instance_tunnel` will be used and automatic fallback
                       is enabled.
        :param bool limit: if True, enable the limitation
        :param bool reopen: the reader will reuse last one, reopen is true means open a new reader.
        :param endpoint: the tunnel service URL
        :param compress_option: compression algorithm, level and strategy
        :type compress_option: :class:`odps.tunnel.CompressOption`
        :param compress_algo: compression algorithm, work when ``compress_option`` is not provided,
                              can be ``zlib``, ``snappy``
        :param compress_level: used for ``zlib``, work when ``compress_option`` is not provided
        :param compress_strategy: used for ``zlib``, work when ``compress_option`` is not provided
        :return: reader, ``count`` means the full size, ``status`` means the tunnel status

        :Example:

        >>> with instance.open_reader() as reader:
        >>>     count = reader.count  # How many records of a table or its partition
        >>>     for record in reader[0: count]:
        >>>         # read all data, actually better to split into reading for many times
        """
        use_tunnel = kwargs.get("use_tunnel", kwargs.get("tunnel"))
        auto_fallback_result = use_tunnel is None

        timeout = kwargs.pop("timeout", None)
        if use_tunnel is None:
            use_tunnel = options.tunnel.use_instance_tunnel
            if use_tunnel:
                timeout = (
                    timeout
                    if timeout is not None
                    else options.tunnel.legacy_fallback_timeout
                )
        kwargs["timeout"] = timeout

        result_fallback_errors = (
            errors.InvalidProjectTable,
            errors.InvalidArgument,
            errors.NoSuchProject,
        )
        if use_tunnel:
            # for compatibility
            if "limit_enabled" in kwargs:
                kwargs["limit"] = kwargs["limit_enabled"]
                del kwargs["limit_enabled"]

            if "limit" not in kwargs:
                kwargs["limit"] = options.tunnel.limit_instance_tunnel

            auto_fallback_protection = False
            if kwargs["limit"] is None:
                kwargs["limit"] = False
                auto_fallback_protection = True

            try:
                return self._open_tunnel_reader(**kwargs)
            except result_fallback_errors:
                # service version too low to support instance tunnel.
                if not auto_fallback_result:
                    raise
                if not kwargs.get("limit"):
                    warnings.warn(
                        "Instance tunnel not supported, will fallback to "
                        "restricted approach. 10000 records will be limited. "
                        + _RESULT_LIMIT_HELPER_MSG
                    )
            except requests.Timeout:
                # tunnel creation timed out, which might be caused by too many files
                # on the service.
                if not auto_fallback_result:
                    raise
                if not kwargs.get("limit"):
                    warnings.warn(
                        "Instance tunnel timed out, will fallback to restricted approach. "
                        "10000 records will be limited. You may try merging small files "
                        "on your source table. " + _RESULT_LIMIT_HELPER_MSG
                    )
            except (
                Instance.DownloadSessionCreationError,
                errors.InstanceTypeNotSupported,
            ):
                # this is for DDL sql instances such as `show partitions` which raises
                # InternalServerError when creating download sessions.
                if not auto_fallback_result:
                    raise
            except errors.NoPermission as exc:
                # project is protected or data permission is configured
                if not auto_fallback_protection:
                    raise
                if not kwargs.get("limit"):
                    warnings.warn(
                        "Project or data under protection, 10000 records will be limited. "
                        "Raw error message:\n"
                        + str(exc)
                        + "\n"
                        + _RESULT_LIMIT_HELPER_MSG
                    )
                    kwargs["limit"] = True
                    return self.open_reader(*args, **kwargs)

        return self._open_result_reader(*args, **kwargs)

    def _iter_reader_with_pandas(self, iter_func, **kw):
        try:
            with self.open_reader(**kw) as reader:
                for batch in iter_func(reader):
                    yield batch
        except (errors.ChecksumError, errors.MethodNotAllowed):
            # arrow tunnel not implemented or not supported
            kw.pop("arrow", None)
            with self.open_reader(**kw) as reader:
                for batch in iter_func(reader):
                    yield batch

    def to_pandas(
        self,
        columns=None,
        limit=None,
        start=None,
        count=None,
        n_process=1,
        quota_name=None,
        tags=None,
        **kwargs
    ):
        """
        Read instance data into pandas DataFrame. The limit argument follows definition
        of `open_reader` API.

        :param list columns: columns to read
        :param bool limit: if True, enable the limitation
        :param int start: start row index from 0
        :param int count: data count to read
        :param int n_process: number of processes to accelerate reading
        :param str quota_name: name of tunnel quota to use
        """
        try:
            import pyarrow as pa
        except ImportError:
            pa = None

        kw = dict(
            limit=limit,
            columns=columns,
            arrow=pa is not None,
            quota_name=quota_name,
            tags=tags,
            **kwargs
        )
        if limit is None:
            kw.pop("limit")

        def _it(reader):
            yield reader.to_pandas(start=start, count=count, n_process=n_process)

        return next(self._iter_reader_with_pandas(_it, **kw))

    def iter_pandas(
        self,
        columns=None,
        limit=None,
        batch_size=None,
        start=None,
        count=None,
        quota_name=None,
        tags=None,
        **kwargs
    ):
        """
        Iterate table data in blocks as pandas DataFrame. The limit argument
        follows definition of `open_reader` API.

        :param list columns: columns to read
        :param bool limit: if True, enable the limitation
        :param int batch_size: size of DataFrame batch to read
        :param int start: start row index from 0
        :param int count: data count to read
        :param str quota_name: name of tunnel quota to use
        """
        try:
            import pyarrow as pa
        except ImportError:
            pa = None

        batch_size = batch_size or options.tunnel.read_row_batch_size
        kw = dict(
            limit=limit,
            columns=columns,
            arrow=pa is not None,
            quota_name=quota_name,
            tags=tags,
            **kwargs
        )
        if limit is None:
            kw.pop("limit")

        def _it(reader):
            for batch in reader.iter_pandas(
                batch_size, start=start, count=count, columns=columns
            ):
                yield batch

        for batch in self._iter_reader_with_pandas(_it, **kw):
            yield batch