Observability/Ray_observability_part_1.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ray Observability Part 1\n",
    "\n",
    "<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Generic/ray_logo.png\" width=\"20%\" loading=\"lazy\">\n",
    "\n",
    "## About this notebook\n",
    "\n",
    "### Is this module right for you?\n",
    "\n",
    "This module provides a general purpose introduction to the most common observability tools to effectively debug, optimize, and monitor Ray applications. It is for data scientists, ML  practitioners, ML engineers, and Python developers looking for ways to understand the behavior of their Ray systems.\n",
    "\n",
    "### Prerequisites\n",
    "\n",
    "For this notebook, you should satisfy the following minimum requirements:\n",
    "\n",
    "-   Practical Python experience\n",
    "-   Familiarity with Ray equivalent to completing these training modules:\n",
    "    -   [Overview of Ray](https://github.com/ray-project/ray-educational-materials/blob/main/Introductory_modules/Overview_of_Ray.ipynb)\n",
    "    -   [Ray Core](https://github.com/ray-project/ray-educational-materials/tree/main/Ray_Core)\n",
    "\n",
    "### Learning objectives\n",
    "\n",
    "-   Understand the major tools available for observability with Ray, namely the State API and Dashboard UI.\n",
    "-   Debug a sample application and surface errors through multiple different observability points.\n",
    "-   Optimize an application with a known anti-pattern and identify the bottleneck using Ray Dashboard, and implement a common design pattern to address it.\n",
    "\n",
    "### What will you do?\n",
    "\n",
    "-   Introduction to the Ray observability toolbox\n",
    "    -   Learn about what observability is and why it can be so difficult in distributed settings.\n",
    "    -   Read about the State API and Dashboard UI and leverage them in common development workflows.\n",
    "-   Ray observability workflows\n",
    "    -   Debugging\n",
    "        -   Reproduce an out of memory error and retrieve metrics and logs related to the failure.\n",
    "        -   Reproduce a hanging bug and observe its behavior.\n",
    "    -   Optimizing\n",
    "        -   Run some `ray.get()` anti-patterns and observe performance bottlenecks and implement the corresponding design pattern to optimize it.\n",
    "-   Summarize the most common observability tools and find resources for further advanced exploration.\n",
    "\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction to observability in distributed systems\n",
    "\n",
    "---\n",
    "\n",
    "Effective debugging and optimizing is crucial in any system, but it becomes even more vital in [distributed settings](https://en.wikipedia.org/wiki/Distributed_computing). With a high number of inter-connected, heterogeneous components, updates can cause opaque failures, making it difficult to identify and fix issues. The goal of observability is to allow for the monitoring and understanding of the internal state of the system in real-time to ensure its stability and performance.\n",
    "\n",
    "### What is observability?\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "  <strong>Observability:</a></strong> the ability to understand the behavior of the internal state of a system inferred from its external outputs.\n",
    "</div>\n",
    "\n",
    "Within the context of Ray, [observability](https://docs.ray.io/en/latest/ray-observability/index.html) refers to the extent of visibility into distributed applications along with the available tools for inspecting and aggregating performance data.\n",
    "\n",
    "### Why is observability challenging in distributed systems?\n",
    "\n",
    "A [Ray cluster](https://docs.ray.io/en/latest/cluster/key-concepts.html) consists of a head node that manages a large number of worker nodes which execute the code of an application. As the scale of the cluster increases, so does the number of [tasks](https://docs.ray.io/en/latest/ray-core/tasks.html#ray-remote-functions), [actors](https://docs.ray.io/en/latest/ray-core/actors.html#actor-guide), and [objects](https://docs.ray.io/en/latest/ray-core/objects.html#objects-in-ray) that are concurrently executed and [stored across heterogeneous machines](https://docs.ray.io/en/latest/ray-core/scheduling/memory-management.html#memory).\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Observability_part_1/actor_failure.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|Actor failure in a Ray cluster with millions of concurrent tasks among [thousands of worker nodes](https://github.com/ray-project/ray/blob/master/release/benchmarks/README.md).|\n",
    "\n",
    "Debugging issues in this environment can be challenging. Among [thousands of nodes and tens of thousands of actors](https://github.com/ray-project/ray/blob/master/release/benchmarks/README.md), performance bottlenecks, failures, and unpredictable behavior are inevitable. For example, diagnosing a cluster with thousands of actors and millions of processes could contain any number of non-trivial pitfalls:\n",
    "\n",
    "-   How do I know if an actor has failed, especially if the actor failed to initialize in the first place?\n",
    "    -   Some processes may become stuck indefinitely due to incomplete scheduling, known as \"hang.\"\n",
    "-   When I become aware of actor failure(s), how do I know which one(s) caused the issue?\n",
    "    -   In a set-up of tens of thousands of actors, how do I begin to identify the culprit?\n",
    "-   Once I know which actor(s) to inspect, how can I find the log and fix the bug?\n",
    "    -   Filtering through logs and tracing failures is impossible without robust tooling.\n",
    "\n",
    "The [Ray runtime](https://docs.ray.io/en/latest/ray-core/starting-ray.html#what-is-the-ray-runtime) manages much of the low-level system behavior in a Ray application which poses a unique opportunity to offer built-in performance data. By providing the right tools to successfully debug, optimize, and monitor Ray applications, developers can troubleshoot issues to improve a system's overall reliability and efficiency."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ray observability toolbox\n",
    "\n",
    "---\n",
    "\n",
    "### Observability tooling by layer\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Observability_part_1/observability_stack.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|Ray offers observability tooling and third-party integrations at each layer of development that allow you to understand the Ray cluster, Ray application, and ML application.|\n",
    "\n",
    "| Layer | Tooling | Purpose | Scenarios |\n",
    "|---|---|---|---|\n",
    "| **Ray Cluster** | <ul><li>State API</li><li>Ray Dashboard</li></ul> | Infrastructure observability, like [`htop`](https://htop.dev/) for a Ray cluster. Monitor the status and utilization including hardware (CPUs, GPUs, TPUs), network, and memory. | Which nodes in my Ray cluster are experiencing high CPU or memory usage so that I can optimize consumption and reduce costs? |\n",
    "| **Ray Application (Core and AIR)** | <ul><li>State API</li><li>Ray Dashboard</li><li>Profiler</li><li>Debugger</li></ul> | Debug, optimize, and monitor Ray applications including the status of tasks, actors, objects, placement groups, jobs, and more. | Are my thousands of in-flight tasks and actors progressing normally, or are some failing or hanging in unintended ways? If so, which ones, and how can I access the logs easily for any given node? |\n",
    "| **ML Application** | Interactive Development <ul><li>[Weights & Biases](https://www.anyscale.com/events/2023/01/19/simplify-building-scaling-tracking-and-monitoring-your-ai-ml-models)</li><li>[MLflow](https://docs.ray.io/en/latest/tune/examples/tune-mlflow.html)</li><li>[Comet](https://docs.ray.io/en/latest/tune/examples/tune-comet.html)</li></ul> Production <ul><li>[Ray Dashboard](https://docs.ray.io/en/latest/ray-observability/ray-metrics.html)</li><li>[Arize](https://www.anyscale.com/events/2023/02/07/productionizing-machine-learning-with-observability-quality-and-flexibility)</li><li>[WhyLabs](https://docs.whylabs.ai/docs/ray-integration/)</li></ul> | Monitor ML models in interactive development and production through third-party integrations. | Which hyperparameters are optimal for my model? How does model performance vary over time or across different input data? Are there any anomalies in the model’s behavior in production?  |\n",
    "\n",
    "This section provides a solid introduction to the two main observability tools in Ray: the [State API](https://docs.ray.io/en/master/ray-observability/state/state-api.html) and the [Dashboard UI](https://docs.ray.io/en/master/ray-core/ray-dashboard.html). To demonstrate their functionality, consider the following simple example to practice on:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import ray\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def task():\n",
    "    time.sleep(60)\n",
    "\n",
    "\n",
    "@ray.remote\n",
    "class Actor:\n",
    "    def call(self):\n",
    "        print(\"Actor called.\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### State API\n",
    "\n",
    "[State APIs](https://docs.ray.io/en/latest/ray-observability/state/state-api.html#monitoring-ray-states) allow users to access the current state of resources of Ray through [CLI](https://docs.ray.io/en/latest/ray-observability/state/cli.html#state-api-cli-ref) or [Python SDK](https://docs.ray.io/en/latest/ray-observability/state/ray-state-api-reference.html#state-api-ref).\n",
    "\n",
    "-   **Resources** include Ray tasks, actors, objects, placement groups, and more.\n",
    "-   **States** refer to the immutable metadata (e.g. actor's name) and mutable states (e.g. actor's scheduling state or pid) of Ray resources.\n",
    "\n",
    "There are three main APIs that allow you to inspect cluster resources with varying levels of granularity:\n",
    "\n",
    "-   **`summary`** returns a summarized view of a given resource (i.e. tasks, actors, objects).\n",
    "-   **`list`** returns a list of resources filterable by type and state.\n",
    "-   **`get`** returns information about a specific resource in detail.\n",
    "\n",
    "In addition, you can also easily retrieve and filter through [ray logs](https://docs.ray.io/en/latest/ray-observability/state/cli.html#ray-logs-api-cli-ref):\n",
    "\n",
    "-   **`logs`** returns the logs of tasks, actors, workers, or system log files.\n",
    "\n",
    "#### Example: Inspect cluster resources."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task.remote()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!ray summary tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "actor = Actor.remote()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "actor.call.remote()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!ray list actors"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Coding exercise\n",
    "\n",
    "Access more information about the actor you just created by using `ray get <actor_id>` using [CLI](https://docs.ray.io/en/latest/ray-observability/state/cli.html#ray-get) or [Python SDK](https://docs.ray.io/en/latest/ray-observability/state/ray-state-api-reference.html#get-apis) ."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### YOUR CODE HERE ###"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### SAMPLE IMPLEMENTATION ###\n",
    "!ray get actors ### FILL IN HERE WITH YOUR <actor_id> ###"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dashboard UI\n",
    "\n",
    "The [Ray Dashboard](https://docs.ray.io/en/master/ray-core/ray-dashboard.html) offers a built-in mechanism for viewing the state of cluster resources, [time series](https://en.wikipedia.org/wiki/Time_series) metrics, and other features at a glance. Information made available via the terminal with the State API is also available to view in the Dashboard UI. You can access the dashboard in a few ways:\n",
    "\n",
    "-   Through the URL printed when Ray is initialized (the default is <https://localhost:8265>).\n",
    "-   When using the [cluster launcher](https://docs.ray.io/en/master/cluster/vms/references/ray-cluster-cli.html#monitor-cluster).\n",
    "-   Under the \"Tools\" tab in the [Anyscale Console](https://www.anyscale.com/platform) (all built-in; no setup).\\\n",
    "\n",
    "Note: In the Anyscale console, time series metrics are built-in, so you don't have to set anything up. Otherwise, download and configuration instructions for integration with Prometheus and Grafana can be found [here](https://docs.ray.io/en/latest/ray-observability/ray-metrics.html#ray-metrics).\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Observability_part_1/dashboard_ui.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|Ray dashboard is a web-based UI to help users monitor their cluster that acts as a central hub for the best observability tools available for Ray.|\n",
    "\n",
    "#### Dashboard Navigation\n",
    "\n",
    "The overview page of the dashboard offers live monitoring and quick links to common views such as metrics, nodes, jobs, and events related to [Ray job submission APIs](https://docs.ray.io/en/master/cluster/running-applications/job-submission/quickstart.html#jobs-quickstart) and the [Ray autoscaler](https://docs.ray.io/en/master/cluster/key-concepts.html#cluster-autoscaler). Along the top navigation bar, you will find five viewing displays:\n",
    "\n",
    "1.  **Jobs** - View the status and logs of [Ray jobs](https://docs.ray.io/en/latest/cluster/running-applications/job-submission/index.html#jobs-overview).\n",
    "2.  **Cluster** - View state, resource utilization, and logs for each node and worker.\n",
    "3.  **Actors** - View information about the actors that have existed on the Ray cluster.\n",
    "4.  **Metrics** - View time series metrics that automatically refresh every 15 seconds; requires [Prometheus](https://prometheus.io/) and [Grafana](https://grafana.com/oss/grafana/) running for your cluster.\n",
    "5.  **Logs** - View all logs, organized by node and file name; supports filter and search.\n",
    "\n",
    "Each view offers plenty of categories to monitor, and you can refer to the [dashboard references](https://docs.ray.io/en/latest/ray-core/ray-dashboard.html#references) for a more complete description. A more detailed investigation will come in the coming sections on Ray observability workflows.\n",
    "\n",
    "#### Coding exercise\n",
    "\n",
    "Open up the Ray Dashboard from the URL provided when you called `ray.init()`.\n",
    "\n",
    "Try running a task or creating a new actor. You can use the basic task or actor provided in the beginning of this section or set up custom ones. Monitor the updating displays to see states and access logs.\\\n",
    "Remember: In the Anyscale console, time series metrics are built-in, so you don't have to set anything up. If you're following along locally, download and configuration instructions for integration with Prometheus and Grafana can be found [here](https://docs.ray.io/en/latest/ray-observability/ray-metrics.html#ray-metrics)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### YOUR CODE HERE ###"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "task.remote()\n",
    "sample_actor = Actor.remote()\n",
    "sample_actor.call.remote()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Summary\n",
    "\n",
    "#### Key concepts\n",
    "\n",
    "-   **State API**\n",
    "    -   A way to inspect the state of resources through CLI and Python SDK.\n",
    "-   **Dashboard UI** \n",
    "    -   A central way to view resources, state, and time series metrics at a glance; a great entry point to using the best support observability tools in Ray.\n",
    "-   **Other tools**\n",
    "    -   Profiling\n",
    "    -   Debugger"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ray observability workflows\n",
    "\n",
    "---\n",
    "\n",
    "The previous two sections discussed the significance of visibility into distributed systems and introduced the State API and Dashboard UI with a simple Ray application. Having laid the groundwork, you can now walk through some common observability stories organized by typical developer workflows: debugging failures and optimizing performance bottlenecks.\n",
    "\n",
    "### Debugging\n",
    "\n",
    "Within the context of Ray, debugging an application refers to failures that emerge during remote processes. This involves the interplay of two main APIs (which have an error handling model [very similar](https://docs.ray.io/en/latest/ray-core/actors/async_api.html#objectrefs-as-concurrent-futures-futures) to standard Python future APIs):\n",
    "\n",
    "-   **`.remote`** - Creates a task or actor and starts a remote process; will return an exception if the remote process fails.\n",
    "-   **`.get`** - Retrieves the result from an object reference; raises an exception if the remote process failed.\n",
    "\n",
    "The [exceptions APIs](https://docs.ray.io/en/master/ray-core/package-ref.html#exceptions) can be grouped into a framework of three primary failure modes:\n",
    "\n",
    "1.  **Application failures** - \n",
    "This happens when a remote task or actor fails resulting from errors in user-generated code. Exceptions thrown include `RayTaskError` and `RayActorError`.\n",
    "\n",
    "2.  **Intentional system failures** - \n",
    "These indicate that while Ray has failed, this failure is a deliberate action. Common examples include using cancellation APIs such as `ray.cancel` for tasks or `ray.kill` for actors.\n",
    "\n",
    "3.  **Unintended system failures** - \n",
    "These arise when a remote process has failed due to unforeseen system failures, such as process crashes or node failures. The following are typical cases:\n",
    "    -  The out of memory killer randomly terminates processes.\n",
    "    -  The machine is being terminated, particularly in the case of spot instances.\n",
    "    -  The system being highly overloaded or stressed leading to failure.\n",
    "    -  Bugs within Ray Core (relatively infrequent).\n",
    "\n",
    "Each of these failures necessitates a specialized approach. In the following section, you will focus on an out of memory (OOM) example, and feel free to refer to the [debugging user guides](https://docs.ray.io/en/latest/ray-observability/monitoring-debugging/troubleshoot-failures.html) for further information on the other types of failures.\n",
    "\n",
    "#### Example: Out of memory errors.\n",
    "\n",
    "In the context of distributed computing, an out of memory (OOM) error occurs when a node in a cluster tries to use more memory than the amount available, leading to a failure of the entire system. Ray applications [use memory](https://docs.ray.io/en/latest/ray-core/scheduling/memory-management.html#concepts) in two main ways:\n",
    "\n",
    "1.  **System memory:** Memory used internally by Ray to manage resources and processes.\n",
    "2.  **Application memory:** Memory used by your application including creating objects in the object store with `ray.put` and reading them with `ray.get`; objects will [spill to disk](https://docs.ray.io/en/latest/ray-core/objects/object-spilling.html#id1) if the store fills up.\n",
    "\n",
    "Consider the following example which continuously leaks memory by appending gigabyte arrays of zeros until the node runs out of memory:\n",
    "\n",
    "<div class=\"alert alert-info\">\n",
    "  <strong>Warning:</a></strong> This script is designed to cause a memory leak and lead to an out of memory (OOM) error. Use caution when executing this code and only run it in a controlled environment.\n",
    "</div>\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "running = False  # Set to True to run the memory leaker.\n",
    "\n",
    "\n",
    "@ray.remote(max_retries=0)\n",
    "def memory_leaker():\n",
    "    chunks = []\n",
    "    bytes_per_chunk = 1024 * 1024 * 1024  # 1 gigabyte.\n",
    "    while running:\n",
    "        chunks.append([0] * bytes_per_chunk)\n",
    "        time.sleep(5)  # Delay to observe the leak.\n",
    "\n",
    "\n",
    "ray.get(memory_leaker.remote())"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this script, you define a Ray Task that leaks memory by continuously appending one-gigabyte arrays of zeros to a list. Depending on your resource configurations, this will eventually cause the memory usage to continuously increase until the OOM killer throws the following error message:\n",
    "\n",
    "```\n",
    "ray.exceptions.OutOfMemoryError: Task was killed due to the node running low on memory.\n",
    "```\n",
    "\n",
    "Note: By default, if a worker dies unexpectedly, Ray will [rerun the process](https://docs.ray.io/en/latest/ray-core/tasks/fault-tolerance.html#retries) up to 3 times. You can specify the number of `max_retries` in the `ray.remote` decorator (e.g. 0 to disable retries, -1 for infinite retries).\n",
    "\n",
    "Effective use of observability tools is essential for [preventing OOM incidents](https://docs.ray.io/en/latest/ray-core/scheduling/ray-oom-prevention.html). While you may be familiar with using [`htop`](https://htop.dev/) or [`free -h`](https://linuxhint.com/linux-free-command-examples/) to gather snapshots of system's running processes, these provide limited utility in distributed systems. Ideally, you want a tool that summarizes resource utilization and usage per node and worker without having to specify each one individually. Through the [Ray Dashboard](https://docs.ray.io/en/master/ray-core/ray-dashboard.html#node-view), you can view a live feed of vital information about your cluster and application during OOM events:\n",
    "\n",
    "-   **Nodes:** Offers a snapshot of memory usage per node and worker.\n",
    "-   **Metrics:** Shows historical usage on an active cluster via Prometheus and Grafana; comes built-in with the Anyscale console.\n",
    "-   **Logs:** Accessible, searchable, and filterable via Node and Logs view.\n",
    "\n",
    "In this section, set up the Ray Dashboard as the focal point for collecting observability data and insights.\n",
    "\n",
    "#### OOM on Ray Dashboard\n",
    "\n",
    "Access the Ray Dashboard from the URL provided when calling `ray.init()`. If you haven't already, download and configuration instructions for integration with Prometheus and Grafana can be found [here](https://docs.ray.io/en/latest/ray-observability/ray-metrics.html#ray-metrics).\n",
    "\n",
    "If you are using the Anyscale console, you can find a quick link to the Dashboard in the \"Tools\" menu, and all monitoring tools will be built-in for you.\n",
    "\n",
    "Try running the memory leaker task once more, and this time pay attention to the memory usage per node as well as the historical usage graphs.\n",
    "\n",
    "#### Coding exercise\n",
    "\n",
    "Try monitoring another OOM script, this time with a memory leaking actor instead of a task. The workflow should be very similar, except with different components to monitor. Try checking out the \"Actors\" view in the Ray Dashboard to inspect the leaking actor."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### SAMPLE STARTER SCRIPT ###\n",
    "import math\n",
    "\n",
    "@ray.remote\n",
    "class Leaker:\n",
    "    def __init__(self):\n",
    "        self.leaks = []\n",
    "\n",
    "    def allocate(self, num_bytes: int, sleep_time_s: int):\n",
    "        # Each element in the array occupies 8 bytes.\n",
    "        new_list = [0] * math.ceil(num_bytes / 8)\n",
    "        self.leaks.append(new_list)\n",
    "\n",
    "        time.sleep(sleep_time_s)\n",
    "\n",
    "\n",
    "### YOUR CODE HERE ###"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Example: Hanging errors.\n",
    "\n",
    "Another common error that Ray users encounter is hang, which refers to the situation where an object reference created by `.remote()` cannot be retrieved by `.get()`. Hanging can cause delays or failure. You can detect these issues in the following ways:\n",
    "\n",
    "1.  **State API**\n",
    "- `ray status` - Pending tasks and actors will surface in the \"Demands\" summary.\n",
    "2.  **Ray Dashboard**\n",
    "- Progress Bar - Check the [status](https://github.com/ray-project/ray/blob/de2f8da435359bed6c704c1cac288ab06fcaaeca/src/ray/protobuf/common.proto#L648) of tasks and actors.\n",
    "- Metrics - Visualize the state of tasks and actors over time.\n",
    "\n",
    "Consider the following example that reproduces a lightweight hang (not an indefinite, failing hang error) due to the nature of the dependencies.\n",
    "\n",
    "#### Coding Exercise"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def long_running_task():\n",
    "    time.sleep(random.randint(10, 60))\n",
    "\n",
    "@ray.remote\n",
    "def dependent_task(dependencies: list[ray._raylet.ObjectRef]):\n",
    "    ray.get(dependencies)\n",
    "\n",
    "dependencies = [long_running_task.remote() for _ in range(100)]\n",
    "dependent_task.remote(dependencies)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here, the `dependent_task` must wait until the dependencies (100 `long_running_task`s) return in order to resolve. Using the State API and Ray Dashboard, you can track the progress of each task and subtask.\n",
    "\n",
    "While this example represents an engineered hanging, typically you will find a handful of common causes for encountering hang:\n",
    "\n",
    "1.  **Application bugs**\n",
    "- User-generated bugs or anti-patterns in a Ray application. [Stacktrace](https://docs.ray.io/en/master/ray-observability/monitoring-debugging/profiling.html#python-cpu-profiling-in-the-dashboard) via the Ray Dashboard will offer visibility.\n",
    "2.  **Resource constraints**\n",
    "- Waiting for available resources, which can be complicated by placement groups.\n",
    "3.  **Object store memory insufficient**\n",
    "- There's not enough memory to pull objects efficiently to the local node.\n",
    "4.  **Pending upstream dependencies**\n",
    "- Dependencies may not be scheduled yet or they are still running.\n",
    "\n",
    "#### Summary\n",
    "\n",
    "Debugging Ray applications involves detecting errors, accessing snapshots of current usage, and analyzing historical data to hone in on the issue. These workflows can be facilitated by using observability tools, especially the Ray Dashboard which acts as a central repository for collecting metrics and logs about a Ray cluster."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Optimizing\n",
    "\n",
    "Within the context of Ray, optimization refers to the process of improving the speed and efficiency of an application. Distributed systems can be costly to run and maintain due to multiple interconnected components needing to work together consistently and reliably. By optimizing performance, organizations can achieve better results and reduce costs.\n",
    "\n",
    "Ray has two common types of optimization issues: resource constraints and design anti-patterns.\n",
    "\n",
    "1.  **Resource constraints**\n",
    "- **Number of cores:** Ray will not schedule more tasks than the number of CPUs that it automatically detects.\n",
    "- **Physical vs. logical CPUs:** Speedup is proportional to [physical CPUs, not logical](https://www.linkedin.com/pulse/understanding-physical-logical-cpus-akshay-deshpande).\n",
    "\n",
    "2.  **Design anti-patterns**\n",
    "- **Small tasks:** Ray introduces overhead for managing each task, so if the task takes less than that overhead (around 10 milliseconds), you are likely to see worse performance.\n",
    "- **Variable durations:** Calling `ray.get` on a batch of tasks with varying durations limits performance to the slowest task.\n",
    "- **Multi-threaded libraries:** When all tasks compete for all resources, you experience contention that prevents runtime improvements.\n",
    "- [**And much more**](https://docs.ray.io/en/latest/ray-core/patterns/index.html).\n",
    "\n",
    "Each use case comes with its own unique set of challenges to [troubleshoot](https://docs.ray.io/en/latest/ray-observability/monitoring-debugging/troubleshoot-performance.html). In the following section, you will walk through a known design anti-pattern and practice using the relevant observability tools to profile the bottleneck.\n",
    "\n",
    "#### Example: Anti-pattern using `ray.get`\n",
    "\n",
    "Consider a scenario where you have a batch of independent tasks that are submitted at the same time. Each task can take a variable amount of time to complete. That is, one task may be completed quickly while another may take a long time.\n",
    "\n",
    "In [this anti-pattern](https://docs.ray.io/en/latest/ray-core/patterns/ray-get-submission-order.html), if you were to call `ray.get()` on the entire batch, your performance would be limited by the longest running task.\n",
    "\n",
    "|<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Observability_part_1/perfetto_timeline.png\" width=\"70%\" loading=\"lazy\">|\n",
    "|:--|\n",
    "|The [Ray timeline](https://docs.ray.io/en/latest/ray-observability/monitoring-debugging/profiling.html#visualizing-tasks-in-the-ray-timeline) provides a high-level view of the tasks that are currently running in your application, how long they take to run, and how well the workload is distributed across all the workers in your cluster. Using `ray.get` to retrieve results from a batch of remote functions is limited by the longest running task in the batch.|\n",
    "\n",
    "Open the Ray Dashboard, and try running the corresponding anti-pattern code below. Pay special attention to the progress bar for tasks as well as the timeline view."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def sleep_task(i: int) -> int:\n",
    "    time.sleep(i)\n",
    "    return i\n",
    "\n",
    "\n",
    "def post_processing_step(new_val: int):\n",
    "    time.sleep(0.5)\n",
    "\n",
    "\n",
    "big_sleep_times = [25]\n",
    "small_sleep_times = [random.random() for _ in range(20)]\n",
    "SLEEP_TIMES = big_sleep_times + small_sleep_times\n",
    "\n",
    "# Launch remote tasks\n",
    "refs = [sleep_task.remote(i) for i in SLEEP_TIMES]\n",
    "for ref in refs:\n",
    "    # Blocks until this ObjectRef is ready.\n",
    "    result = ray.get(ref)  # Retrieve result in submission order.\n",
    "    post_processing_step(result)  # Process the result."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Example: Design pattern using `ray.wait()`\n",
    "\n",
    "Instead of calling `ray.get()` on a batch with variable tasks, [use `ray.wait()`](https://docs.ray.io/en/latest/ray-core/package-ref.html#ray-wait) to get the first object reference ready to return and then use `ray.get()` to retrieve the result. In this way, you can apply the post-processing function as soon as a result becomes available instead of having the submission order potentially slow down the pipeline.\n",
    "\n",
    "#### Coding exercise\n",
    "\n",
    "By running the anti-pattern, you may have noticed that one long-running task will block the entire process when using `ray.get()` on the entire batch.\n",
    "\n",
    "Implement the design pattern illustrated in the diagram above which uses `ray.wait()` to process results as soon as one becomes available. With the Ray Dashboard open (especially the timeline view), observe the differences between using `ray.get()` and `ray.wait()` for pipelining data submission."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### YOUR CODE HERE ###"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### SAMPLE IMPLEMENTATION ###\n",
    "\n",
    "# Launch remote tasks.\n",
    "refs = [sleep_task.remote(i) for i in SLEEP_TIMES]\n",
    "unfinished = refs\n",
    "while unfinished:\n",
    "    # Returns the first ObjectRef that is ready.\n",
    "    finished, unfinished = ray.wait(unfinished, num_returns=1)\n",
    "    # Retrieve the first ready result.\n",
    "    result = ray.get(finished[0])\n",
    "    # Process the result.\n",
    "    post_processing_step(result)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Coding exercise\n",
    "\n",
    "Another [common anti-pattern](https://docs.ray.io/en/latest/ray-core/tips-for-first-time.html#tip-1-delay-ray-get) involves using `ray.get()` unnecessarily, which leads to performance issues since it's a blocking call.\n",
    "\n",
    "Run the following anti-pattern and pattern code examples with the Ray Dashboard open. Observe the differences between the two approaches using the time series metrics views.\n",
    "\n",
    "Can you identify further improvements to this base code? Verify your results with the Ray Dashboard."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def f(i: int) -> int:\n",
    "    return i\n",
    "\n",
    "# Anti-pattern: no parallelism due to calling ray.get inside of the loop.\n",
    "sequential_returns = []\n",
    "for i in range(100):\n",
    "    sequential_returns.append(ray.get(f.remote(i)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### YOUR CODE HERE ###"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if ray.is_initialized():\n",
    "    ray.shutdown()\n",
    "\n",
    "ray.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### SAMPLE IMPLEMENTATION ###\n",
    "refs = []\n",
    "for i in range(100):\n",
    "    refs.append(f.remote(i))\n",
    "\n",
    "parallel_returns = ray.get(refs)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Summary\n",
    "\n",
    "The Ray Dashboard offers a range of tools to help you optimize performance and identify performance bottlenecks. It provides a timeline view, progress bar for tasks, and other profiling tools to assist you in improving the speed and efficiency of your application. With these tools, you can analyze different aspects of your application, identify areas for improvement, and resolve any performance issues, ultimately reducing costs."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusion\n",
    "\n",
    "---\n",
    "\n",
    "Congratulations! You have now practiced using the two main observability tools (the [State API](https://docs.ray.io/en/master/ray-observability/state/state-api.html) and the [Dashboard UI](https://docs.ray.io/en/master/ray-core/ray-dashboard.html)) for debugging and optimizing Ray applications. In the next module, you will explore how to take observability beyond interactive development and into production.\n",
    "\n",
    "### Summary\n",
    "\n",
    "-   Introduction to the Ray observability toolbox\n",
    "    -   Observability in distributed systems is essential and challenging given the scale of coordination among [thousands of heterogeneous resources](https://github.com/ray-project/ray/blob/master/release/benchmarks/README.md).\n",
    "    -   The [State API](https://docs.ray.io/en/master/ray-observability/state/state-api.html) and the [Dashboard UI](https://docs.ray.io/en/master/ray-core/ray-dashboard.html) offer visibility into Ray applications.\n",
    "-   Ray observability workflows\n",
    "    -   Debugging\n",
    "        -   Monitor and debug errors through the Ray Dashboard where you can access logs, time series, metrics, and more.\n",
    "    -   Optimizing\n",
    "        -   Identify performance bottlenecks by inspecting the progress bar, timeline, and node view available in the Ray Dashboard.\n",
    "-   Advanced exploration\n",
    "    -   Ray Observability Part 2\n",
    "    -   [Ray Observability Documentation](https://docs.ray.io/en/latest/ray-observability/index.html)\n",
    "    -   [Ray Observability Roadmap (upcoming releases)](https://github.com/ray-project/ray/issues/30097)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Connect with the Ray community\n",
    "\n",
    "You can learn and get more involved with the Ray community of developers and researchers:\n",
    "\n",
    "* [**Ray documentation**](https://docs.ray.io/en/latest)\n",
    "\n",
    "* [**Official Ray site**](https://www.ray.io/)  \n",
    "Browse the ecosystem and use this site as a hub to get the information that you need to get going and building with Ray.\n",
    "\n",
    "* [**Join the community on Slack**](https://forms.gle/9TSdDYUgxYs8SA9e8)  \n",
    "Find friends to discuss your new learnings in our Slack space.\n",
    "\n",
    "* [**Use the discussion board**](https://discuss.ray.io/)  \n",
    "Ask questions, follow topics, and view announcements on this community forum.\n",
    "\n",
    "* [**Join a meetup group**](https://www.meetup.com/Bay-Area-Ray-Meetup/)  \n",
    "Tune in on meet-ups to listen to compelling talks, get to know other users, and meet the team behind Ray.\n",
    "\n",
    "* [**Open an issue**](https://github.com/ray-project/ray/issues/new/choose)  \n",
    "Ray is constantly evolving to improve developer experience. Submit feature requests, bug-reports, and get help via GitHub issues.\n",
    "\n",
    "* [**Become a Ray contributor**](https://docs.ray.io/en/latest/ray-contribute/getting-involved.html)  \n",
    "We welcome community contributions to improve our documentation and Ray framework."
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "<img src=\"https://technical-training-assets.s3.us-west-2.amazonaws.com/Generic/ray_logo.png\" width=\"20%\" loading=\"lazy\">"
   ],
   "metadata": {
    "collapsed": false
   }
  }
 ],
 "metadata": {
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  },
  "vscode": {
   "interpreter": {
    "hash": "31f2aee4e71d21fbe5cf8b01ff0e069b9275f58929596ceb00d14d90e3e16cd6"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}