Add files using upload-large-folder tool

.venv/lib/python3.11/site-packages/ray/dashboard/modules/event/event_utils.py

@@ -0,0 +1,208 @@
+import asyncio
+import collections
+import fnmatch
+import itertools
+import json
+import logging.handlers
+import mmap
+import os
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+from ray._private.utils import get_or_create_event_loop, run_background_task
+from ray.dashboard.modules.event import event_consts
+from ray.dashboard.utils import async_loop_forever
+
+logger = logging.getLogger(__name__)
+
+
+def _get_source_files(event_dir, source_types=None, event_file_filter=None):
+    event_log_names = os.listdir(event_dir)
+    source_files = {}
+    all_source_types = set(event_consts.EVENT_SOURCE_ALL)
+    for source_type in source_types or event_consts.EVENT_SOURCE_ALL:
+        assert source_type in all_source_types, f"Invalid source type: {source_type}"
+        files = []
+        for n in event_log_names:
+            if fnmatch.fnmatch(n, f"*{source_type}*"):
+                f = os.path.join(event_dir, n)
+                if event_file_filter is not None and not event_file_filter(f):
+                    continue
+                files.append(f)
+        if files:
+            source_files[source_type] = files
+    return source_files
+
+
+def _restore_newline(event_dict):
+    try:
+        event_dict["message"] = (
+            event_dict["message"].replace("\\n", "\n").replace("\\r", "\n")
+        )
+    except Exception:
+        logger.exception("Restore newline for event failed: %s", event_dict)
+    return event_dict
+
+
+def _parse_line(event_str):
+    return _restore_newline(json.loads(event_str))
+
+
+def parse_event_strings(event_string_list):
+    events = []
+    for data in event_string_list:
+        if not data:
+            continue
+        try:
+            event = _parse_line(data)
+            events.append(event)
+        except Exception:
+            logger.exception("Parse event line failed: %s", repr(data))
+    return events
+
+
+ReadFileResult = collections.namedtuple(
+    "ReadFileResult", ["fid", "size", "mtime", "position", "lines"]
+)
+
+
+def _read_file(
+    file, pos, n_lines=event_consts.EVENT_READ_LINE_COUNT_LIMIT, closefd=True
+):
+    with open(file, "rb", closefd=closefd) as f:
+        # The ino may be 0 on Windows.
+        stat = os.stat(f.fileno())
+        fid = stat.st_ino or file
+        lines = []
+        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
+            start = pos
+            for _ in range(n_lines):
+                sep = mm.find(b"\n", start)
+                if sep == -1:
+                    break
+                if sep - start <= event_consts.EVENT_READ_LINE_LENGTH_LIMIT:
+                    lines.append(mm[start:sep].decode("utf-8"))
+                else:
+                    truncated_size = min(100, event_consts.EVENT_READ_LINE_LENGTH_LIMIT)
+                    logger.warning(
+                        "Ignored long string: %s...(%s chars)",
+                        mm[start : start + truncated_size].decode("utf-8"),
+                        sep - start,
+                    )
+                start = sep + 1
+        return ReadFileResult(fid, stat.st_size, stat.st_mtime, start, lines)
+
+
+def monitor_events(
+    event_dir,
+    callback,
+    monitor_thread_pool_executor: ThreadPoolExecutor,
+    scan_interval_seconds=event_consts.SCAN_EVENT_DIR_INTERVAL_SECONDS,
+    start_mtime=time.time() + event_consts.SCAN_EVENT_START_OFFSET_SECONDS,
+    monitor_files=None,
+    source_types=None,
+):
+    """Monitor events in directory. New events will be read and passed to the
+    callback.
+
+    Args:
+        event_dir: The event log directory.
+        callback (def callback(List[str]): pass): A callback accepts a list of
+            event strings.
+        monitor_thread_pool_executor: A thread pool exector to monitor/update
+            events. None means it will use the default execturo which uses
+            num_cpus of the machine * 5 threads (before python 3.8) or
+            min(32, num_cpus + 5) (from Python 3.8).
+        scan_interval_seconds: An interval seconds between two scans.
+        start_mtime: Only the event log files whose last modification
+            time is greater than start_mtime are monitored.
+        monitor_files (Dict[int, MonitorFile]): The map from event log file id
+            to MonitorFile object. Monitor all files start from the beginning
+            if the value is None.
+        source_types (List[str]): A list of source type name from
+            event_pb2.Event.SourceType.keys(). Monitor all source types if the
+            value is None.
+    """
+    loop = get_or_create_event_loop()
+    if monitor_files is None:
+        monitor_files = {}
+
+    logger.info(
+        "Monitor events logs modified after %s on %s, " "the source types are %s.",
+        start_mtime,
+        event_dir,
+        "all" if source_types is None else source_types,
+    )
+
+    MonitorFile = collections.namedtuple("MonitorFile", ["size", "mtime", "position"])
+
+    def _source_file_filter(source_file):
+        stat = os.stat(source_file)
+        return stat.st_mtime > start_mtime
+
+    def _read_monitor_file(file, pos):
+        assert isinstance(
+            file, str
+        ), f"File should be a str, but a {type(file)}({file}) found"
+        fd = os.open(file, os.O_RDONLY)
+        try:
+            stat = os.stat(fd)
+            # Check the file size to avoid raising the exception
+            # ValueError: cannot mmap an empty file
+            if stat.st_size <= 0:
+                return []
+            fid = stat.st_ino or file
+            monitor_file = monitor_files.get(fid)
+            if monitor_file:
+                if (
+                    monitor_file.position == monitor_file.size
+                    and monitor_file.size == stat.st_size
+                    and monitor_file.mtime == stat.st_mtime
+                ):
+                    logger.debug(
+                        "Skip reading the file because " "there is no change: %s", file
+                    )
+                    return []
+                position = monitor_file.position
+            else:
+                logger.info("Found new event log file: %s", file)
+                position = pos
+            # Close the fd in finally.
+            r = _read_file(fd, position, closefd=False)
+            # It should be fine to update the dict in executor thread.
+            monitor_files[r.fid] = MonitorFile(r.size, r.mtime, r.position)
+            loop.call_soon_threadsafe(callback, r.lines)
+        except Exception as e:
+            raise Exception(f"Read event file failed: {file}") from e
+        finally:
+            os.close(fd)
+
+    @async_loop_forever(scan_interval_seconds, cancellable=True)
+    async def _scan_event_log_files():
+        # Scan event files.
+        source_files = await loop.run_in_executor(
+            monitor_thread_pool_executor,
+            _get_source_files,
+            event_dir,
+            source_types,
+            _source_file_filter,
+        )
+
+        # Limit concurrent read to avoid fd exhaustion.
+        semaphore = asyncio.Semaphore(event_consts.CONCURRENT_READ_LIMIT)
+
+        async def _concurrent_coro(filename):
+            async with semaphore:
+                return await loop.run_in_executor(
+                    monitor_thread_pool_executor, _read_monitor_file, filename, 0
+                )
+
+        # Read files.
+        await asyncio.gather(
+            *[
+                _concurrent_coro(filename)
+                for filename in list(itertools.chain(*source_files.values()))
+            ]
+        )
+
+    return run_background_task(_scan_event_log_files())

.venv/lib/python3.11/site-packages/ray/dashboard/modules/healthz/healthz_agent.py

@@ -0,0 +1,54 @@
+from aiohttp.web import Request, Response
+
+import ray.dashboard.optional_utils as optional_utils
+import ray.dashboard.utils as dashboard_utils
+import ray.exceptions
+from ray.dashboard.modules.healthz.utils import HealthChecker
+
+routes = optional_utils.DashboardAgentRouteTable
+
+
+class HealthzAgent(dashboard_utils.DashboardAgentModule):
+    """Health check in the agent.
+
+    This module adds health check related endpoint to the agent to check
+    local components' health.
+    """
+
+    def __init__(self, dashboard_agent):
+        super().__init__(dashboard_agent)
+        self._health_checker = HealthChecker(
+            dashboard_agent.gcs_aio_client,
+            f"{dashboard_agent.ip}:{dashboard_agent.node_manager_port}",
+        )
+
+    @routes.get("/api/local_raylet_healthz")
+    async def health_check(self, req: Request) -> Response:
+        try:
+            alive = await self._health_checker.check_local_raylet_liveness()
+            if alive is False:
+                return Response(status=503, text="Local Raylet failed")
+        except ray.exceptions.RpcError as e:
+            # We only consider the error other than GCS unreachable as raylet failure
+            # to avoid false positive.
+            # In case of GCS failed, Raylet will crash eventually if GCS is not back
+            # within a given time and the check will fail since agent can't live
+            # without a local raylet.
+            if e.rpc_code not in (
+                ray._raylet.GRPC_STATUS_CODE_UNAVAILABLE,
+                ray._raylet.GRPC_STATUS_CODE_UNKNOWN,
+                ray._raylet.GRPC_STATUS_CODE_DEADLINE_EXCEEDED,
+            ):
+                return Response(status=503, text=f"Health check failed due to: {e}")
+
+        return Response(
+            text="success",
+            content_type="application/text",
+        )
+
+    async def run(self, server):
+        pass
+
+    @staticmethod
+    def is_minimal_module():
+        return False

.venv/lib/python3.11/site-packages/ray/dashboard/modules/healthz/healthz_head.py

@@ -0,0 +1,41 @@
+from aiohttp.web import HTTPServiceUnavailable, Request, Response
+
+import ray.dashboard.optional_utils as optional_utils
+import ray.dashboard.utils as dashboard_utils
+from ray.dashboard.modules.healthz.utils import HealthChecker
+
+routes = optional_utils.DashboardHeadRouteTable
+
+
+class HealthzHead(dashboard_utils.DashboardHeadModule):
+    """Health check in the head.
+
+    This module adds health check related endpoint to the head to check
+    GCS's heath.
+    """
+
+    def __init__(self, config: dashboard_utils.DashboardHeadModuleConfig):
+        super().__init__(config)
+        self._health_checker = HealthChecker(self.gcs_aio_client)
+
+    @routes.get("/api/gcs_healthz")
+    async def health_check(self, req: Request) -> Response:
+        alive = False
+        try:
+            alive = await self._health_checker.check_gcs_liveness()
+            if alive is True:
+                return Response(
+                    text="success",
+                    content_type="application/text",
+                )
+        except Exception as e:
+            return HTTPServiceUnavailable(reason=f"Health check failed: {e}")
+
+        return HTTPServiceUnavailable(reason="Health check failed")
+
+    async def run(self, server):
+        pass
+
+    @staticmethod
+    def is_minimal_module():
+        return True

.venv/lib/python3.11/site-packages/ray/dashboard/modules/healthz/utils.py

@@ -0,0 +1,24 @@
+from typing import Optional
+
+from ray._private.gcs_utils import GcsAioClient
+
+
+class HealthChecker:
+    def __init__(
+        self, gcs_aio_client: GcsAioClient, local_node_address: Optional[str] = None
+    ):
+        self._gcs_aio_client = gcs_aio_client
+        self._local_node_address = local_node_address
+
+    async def check_local_raylet_liveness(self) -> bool:
+        if self._local_node_address is None:
+            return False
+
+        liveness = await self._gcs_aio_client.check_alive(
+            [self._local_node_address.encode()], 0.1
+        )
+        return liveness[0]
+
+    async def check_gcs_liveness(self) -> bool:
+        await self._gcs_aio_client.check_alive([], 0.1)
+        return True

.venv/lib/python3.11/site-packages/ray/dashboard/modules/snapshot/snapshot_head.py

@@ -0,0 +1,235 @@
+import concurrent.futures
+import enum
+import json
+import logging
+import os
+from datetime import datetime
+from typing import Optional
+
+import aiohttp.web
+
+import ray.dashboard.optional_utils as dashboard_optional_utils
+import ray.dashboard.utils as dashboard_utils
+from ray import ActorID
+from ray._private.pydantic_compat import BaseModel, Extra, Field, validator
+from ray._private.storage import _load_class
+from ray.dashboard.consts import RAY_CLUSTER_ACTIVITY_HOOK
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+routes = dashboard_optional_utils.DashboardHeadRouteTable
+
+SNAPSHOT_API_TIMEOUT_SECONDS = 30
+
+
+class RayActivityStatus(str, enum.Enum):
+    ACTIVE = "ACTIVE"
+    INACTIVE = "INACTIVE"
+    ERROR = "ERROR"
+
+
+class RayActivityResponse(BaseModel, extra=Extra.allow):
+    """
+    Pydantic model used to inform if a particular Ray component can be considered
+    active, and metadata about observation.
+    """
+
+    is_active: RayActivityStatus = Field(
+        ...,
+        description=(
+            "Whether the corresponding Ray component is considered active or inactive, "
+            "or if there was an error while collecting this observation."
+        ),
+    )
+    reason: Optional[str] = Field(
+        None, description="Reason if Ray component is considered active or errored."
+    )
+    timestamp: float = Field(
+        ...,
+        description=(
+            "Timestamp of when this observation about the Ray component was made. "
+            "This is in the format of seconds since unix epoch."
+        ),
+    )
+    last_activity_at: Optional[float] = Field(
+        None,
+        description=(
+            "Timestamp when last actvity of this Ray component finished in format of "
+            "seconds since unix epoch. This field does not need to be populated "
+            "for Ray components where it is not meaningful."
+        ),
+    )
+
+    @validator("reason", always=True)
+    def reason_required(cls, v, values, **kwargs):
+        if "is_active" in values and values["is_active"] != RayActivityStatus.INACTIVE:
+            if v is None:
+                raise ValueError(
+                    'Reason is required if is_active is "active" or "error"'
+                )
+        return v
+
+
+class APIHead(dashboard_utils.DashboardHeadModule):
+    def __init__(self, config: dashboard_utils.DashboardHeadModuleConfig):
+        super().__init__(config)
+        # For offloading CPU intensive work.
+        self._thread_pool = concurrent.futures.ThreadPoolExecutor(
+            max_workers=2, thread_name_prefix="api_head"
+        )
+
+    @routes.get("/api/actors/kill")
+    async def kill_actor_gcs(self, req) -> aiohttp.web.Response:
+        actor_id = req.query.get("actor_id")
+        force_kill = req.query.get("force_kill", False) in ("true", "True")
+        no_restart = req.query.get("no_restart", False) in ("true", "True")
+        if not actor_id:
+            return dashboard_optional_utils.rest_response(
+                success=False, message="actor_id is required."
+            )
+
+        await self.gcs_aio_client.kill_actor(
+            ActorID.from_hex(actor_id),
+            force_kill,
+            no_restart,
+            timeout=SNAPSHOT_API_TIMEOUT_SECONDS,
+        )
+
+        message = (
+            f"Force killed actor with id {actor_id}"
+            if force_kill
+            else f"Requested actor with id {actor_id} to terminate. "
+            + "It will exit once running tasks complete"
+        )
+
+        return dashboard_optional_utils.rest_response(success=True, message=message)
+
+    @routes.get("/api/component_activities")
+    async def get_component_activities(self, req) -> aiohttp.web.Response:
+        timeout = req.query.get("timeout", None)
+        if timeout and timeout.isdigit():
+            timeout = int(timeout)
+        else:
+            timeout = SNAPSHOT_API_TIMEOUT_SECONDS
+
+        # Get activity information for driver
+        driver_activity_info = await self._get_job_activity_info(timeout=timeout)
+        resp = {"driver": dict(driver_activity_info)}
+
+        if RAY_CLUSTER_ACTIVITY_HOOK in os.environ:
+            try:
+                cluster_activity_callable = _load_class(
+                    os.environ[RAY_CLUSTER_ACTIVITY_HOOK]
+                )
+                external_activity_output = cluster_activity_callable()
+                assert isinstance(external_activity_output, dict), (
+                    f"Output of hook {os.environ[RAY_CLUSTER_ACTIVITY_HOOK]} "
+                    "should be Dict[str, RayActivityResponse]. Got "
+                    f"output: {external_activity_output}"
+                )
+                for component_type in external_activity_output:
+                    try:
+                        component_activity_output = external_activity_output[
+                            component_type
+                        ]
+                        # Parse and validate output to type RayActivityResponse
+                        component_activity_output = RayActivityResponse(
+                            **dict(component_activity_output)
+                        )
+                        resp[component_type] = dict(component_activity_output)
+                    except Exception as e:
+                        logger.exception(
+                            f"Failed to get activity status of {component_type} "
+                            f"from user hook {os.environ[RAY_CLUSTER_ACTIVITY_HOOK]}."
+                        )
+                        resp[component_type] = {
+                            "is_active": RayActivityStatus.ERROR,
+                            "reason": repr(e),
+                            "timestamp": datetime.now().timestamp(),
+                        }
+            except Exception as e:
+                logger.exception(
+                    "Failed to get activity status from user "
+                    f"hook {os.environ[RAY_CLUSTER_ACTIVITY_HOOK]}."
+                )
+                resp["external_component"] = {
+                    "is_active": RayActivityStatus.ERROR,
+                    "reason": repr(e),
+                    "timestamp": datetime.now().timestamp(),
+                }
+
+        return aiohttp.web.Response(
+            text=json.dumps(resp),
+            content_type="application/json",
+            status=aiohttp.web.HTTPOk.status_code,
+        )
+
+    async def _get_job_activity_info(self, timeout: int) -> RayActivityResponse:
+        # Returns if there is Ray activity from drivers (job).
+        # Drivers in namespaces that start with _ray_internal_ are not
+        # considered activity.
+        # This includes the _ray_internal_dashboard job that gets automatically
+        # created with every cluster
+        try:
+            reply = await self.gcs_aio_client.get_all_job_info(
+                skip_submission_job_info_field=True,
+                skip_is_running_tasks_field=True,
+                timeout=timeout,
+            )
+
+            num_active_drivers = 0
+            latest_job_end_time = 0
+            for job_table_entry in reply.values():
+                is_dead = bool(job_table_entry.is_dead)
+                in_internal_namespace = job_table_entry.config.ray_namespace.startswith(
+                    "_ray_internal_"
+                )
+                latest_job_end_time = (
+                    max(latest_job_end_time, job_table_entry.end_time)
+                    if job_table_entry.end_time
+                    else latest_job_end_time
+                )
+                if not is_dead and not in_internal_namespace:
+                    num_active_drivers += 1
+
+            current_timestamp = datetime.now().timestamp()
+            # Latest job end time must be before or equal to the current timestamp.
+            # Job end times may be provided in epoch milliseconds. Check if this
+            # is true, and convert to seconds
+            if latest_job_end_time > current_timestamp:
+                latest_job_end_time = latest_job_end_time / 1000
+                assert current_timestamp >= latest_job_end_time, (
+                    f"Most recent job end time {latest_job_end_time} must be "
+                    f"before or equal to the current timestamp {current_timestamp}"
+                )
+
+            is_active = (
+                RayActivityStatus.ACTIVE
+                if num_active_drivers > 0
+                else RayActivityStatus.INACTIVE
+            )
+            return RayActivityResponse(
+                is_active=is_active,
+                reason=f"Number of active drivers: {num_active_drivers}"
+                if num_active_drivers
+                else None,
+                timestamp=current_timestamp,
+                # If latest_job_end_time == 0, no jobs have finished yet so don't
+                # populate last_activity_at
+                last_activity_at=latest_job_end_time if latest_job_end_time else None,
+            )
+        except Exception as e:
+            logger.exception("Failed to get activity status of Ray drivers.")
+            return RayActivityResponse(
+                is_active=RayActivityStatus.ERROR,
+                reason=repr(e),
+                timestamp=datetime.now().timestamp(),
+            )
+
+    async def run(self, server):
+        pass
+
+    @staticmethod
+    def is_minimal_module():
+        return False

.venv/lib/python3.11/site-packages/ray/data/_internal/arrow_ops/transform_polars.py

@@ -0,0 +1,40 @@
+from typing import TYPE_CHECKING, List
+
+try:
+    import pyarrow
+except ImportError:
+    pyarrow = None
+
+
+if TYPE_CHECKING:
+    from ray.data._internal.planner.exchange.sort_task_spec import SortKey
+
+pl = None
+
+
+def check_polars_installed():
+    try:
+        global pl
+        import polars as pl
+    except ImportError:
+        raise ImportError(
+            "polars not installed. Install with `pip install polars` or set "
+            "`DataContext.use_polars = False` to fall back to pyarrow"
+        )
+
+
+def sort(table: "pyarrow.Table", sort_key: "SortKey") -> "pyarrow.Table":
+    check_polars_installed()
+    df = pl.from_arrow(table)
+    return df.sort(sort_key.get_columns(), reverse=sort_key.get_descending()).to_arrow()
+
+
+def concat_and_sort(
+    blocks: List["pyarrow.Table"], sort_key: "SortKey"
+) -> "pyarrow.Table":
+    check_polars_installed()
+    blocks = [pl.from_arrow(block) for block in blocks]
+    df = pl.concat(blocks).sort(
+        sort_key.get_columns(), reverse=sort_key.get_descending()
+    )
+    return df.to_arrow()