[dashboard] Add `RAY_CLUSTER_ACTIVITY_HOOK` to `/api/component_activities`
diff --git a/dashboard/consts.py b/dashboard/consts.py
index a81d3ebc73c23..0380dbbc079e8 100644
--- a/dashboard/consts.py
+++ b/dashboard/consts.py
@@ -36,3 +36,10 @@
 BAD_RUNTIME_ENV_CACHE_TTL_SECONDS = env_integer(
     "BAD_RUNTIME_ENV_CACHE_TTL_SECONDS", 60 * 10
 )
+# Hook that is invoked on the dashboard `/api/component_activities` endpoint.
+# Environment variable stored here should be a callable that does not
+# take any arguments and should return a dictionary mapping
+# activity component type (str) to
+# ray.dashboard.modules.snapshot.snapshot_head.RayActivityResponse.
+# Example: "your.module.ray_cluster_activity_hook".
+RAY_CLUSTER_ACTIVITY_HOOK = "RAY_CLUSTER_ACTIVITY_HOOK"
diff --git a/dashboard/modules/snapshot/component_activities_schema.json b/dashboard/modules/snapshot/component_activities_schema.json
index 8eb93c8816bc8..57fda426987ad 100644
--- a/dashboard/modules/snapshot/component_activities_schema.json
+++ b/dashboard/modules/snapshot/component_activities_schema.json
@@ -7,7 +7,8 @@
       "type": "object",
       "properties":{
         "is_active": {
-          "type": "boolean"
+          "type": "string",
+          "enum": ["ACTIVE", "INACTIVE", "ERROR"]
         },
         "reason": {
           "type": [
diff --git a/dashboard/modules/snapshot/snapshot_head.py b/dashboard/modules/snapshot/snapshot_head.py
index 5b24c2a0fee51..7f6de4ed57cf2 100644
--- a/dashboard/modules/snapshot/snapshot_head.py
+++ b/dashboard/modules/snapshot/snapshot_head.py
@@ -2,16 +2,21 @@
 import concurrent.futures
 import dataclasses
 from datetime import datetime
+import enum
+import logging
 import hashlib
 import json
+import os
 from typing import Any, Dict, List, Optional
 
 import aiohttp.web
 
 import ray
+from ray.dashboard.consts import RAY_CLUSTER_ACTIVITY_HOOK
 import ray.dashboard.optional_utils as dashboard_optional_utils
 import ray.dashboard.utils as dashboard_utils
 from ray._private import ray_constants
+from ray._private.storage import _load_class
 from ray.core.generated import gcs_pb2, gcs_service_pb2, gcs_service_pb2_grpc
 from ray.dashboard.modules.job.common import JOB_ID_METADATA_KEY, JobInfoStorageClient
 from ray.experimental.internal_kv import (
@@ -22,9 +27,18 @@
 from ray.job_submission import JobInfo
 from ray.runtime_env import RuntimeEnv
 
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
 routes = dashboard_optional_utils.ClassMethodRouteTable
 
 
+class RayActivityStatus(str, enum.Enum):
+    ACTIVE = "ACTIVE"
+    INACTIVE = "INACTIVE"
+    ERROR = "ERROR"
+
+
 @dataclasses.dataclass
 class RayActivityResponse:
     """
@@ -32,11 +46,12 @@ class RayActivityResponse:
     active, and metadata about observation.
     """
 
-    # Whether the corresponding Ray component is considered active
-    is_active: bool
-    # Reason if Ray component is considered active
+    # Whether the corresponding Ray component is considered active or inactive,
+    # or if there was an error while collecting this observation.
+    is_active: RayActivityStatus
+    # Reason if Ray component is considered active or errored.
     reason: Optional[str] = None
-    # Timestamp of when this observation about the Ray component was made
+    # Timestamp of when this observation about the Ray component was made.
     timestamp: Optional[float] = None
 
 
@@ -108,16 +123,64 @@ async def snapshot(self, req):
 
     @routes.get("/api/component_activities")
     async def get_component_activities(self, req) -> aiohttp.web.Response:
-        # Get activity information for driver
         timeout = req.query.get("timeout", None)
         if timeout and timeout.isdigit():
             timeout = int(timeout)
         else:
             timeout = 5
 
+        # Get activity information for driver
         driver_activity_info = await self._get_job_activity_info(timeout=timeout)
-
         resp = {"driver": dataclasses.asdict(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
+                        ]
+                        # Cast output to type RayActivityResponse
+                        component_activity_output = RayActivityResponse(
+                            **dataclasses.asdict(component_activity_output)
+                        )
+                        # Validate is_active field is of type RayActivityStatus
+                        component_activity_output.is_active = RayActivityStatus[
+                            component_activity_output.is_active
+                        ]
+                        resp[component_type] = dataclasses.asdict(
+                            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",
@@ -128,25 +191,40 @@ 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_job_info_ are not
         # considered activity.
-        request = gcs_service_pb2.GetAllJobInfoRequest()
-        reply = await self._gcs_job_info_stub.GetAllJobInfo(request, timeout=timeout)
+        try:
+            request = gcs_service_pb2.GetAllJobInfoRequest()
+            reply = await self._gcs_job_info_stub.GetAllJobInfo(
+                request, timeout=timeout
+            )
 
-        num_active_drivers = 0
-        for job_table_entry in reply.job_info_list:
-            is_dead = bool(job_table_entry.is_dead)
-            in_internal_namespace = job_table_entry.config.ray_namespace.startswith(
-                JobInfoStorageClient.JOB_DATA_KEY_PREFIX
+            num_active_drivers = 0
+            for job_table_entry in reply.job_info_list:
+                is_dead = bool(job_table_entry.is_dead)
+                in_internal_namespace = job_table_entry.config.ray_namespace.startswith(
+                    JobInfoStorageClient.JOB_DATA_KEY_PREFIX
+                )
+                if not is_dead and not in_internal_namespace:
+                    num_active_drivers += 1
+
+            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=datetime.now().timestamp(),
+            )
+        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(),
             )
-            if not is_dead and not in_internal_namespace:
-                num_active_drivers += 1
-
-        return RayActivityResponse(
-            is_active=num_active_drivers > 0,
-            reason=f"Number of active drivers: {num_active_drivers}"
-            if num_active_drivers
-            else None,
-            timestamp=datetime.now().timestamp(),
-        )
 
     def _get_job_info(self, metadata: Dict[str, str]) -> Optional[JobInfo]:
         # If a job submission ID has been added to a job, the status is
diff --git a/dashboard/modules/snapshot/tests/test_snapshot.py b/dashboard/modules/snapshot/tests/test_snapshot.py
index 0789f5c601c39..d024a929c9a1e 100644
--- a/dashboard/modules/snapshot/tests/test_snapshot.py
+++ b/dashboard/modules/snapshot/tests/test_snapshot.py
@@ -18,12 +18,51 @@
     run_string_as_driver_nonblocking,
 )
 from ray.dashboard import dashboard
+from ray.dashboard.consts import RAY_CLUSTER_ACTIVITY_HOOK
 from ray.dashboard.modules.snapshot.snapshot_head import RayActivityResponse
 from ray.dashboard.tests.conftest import *  # noqa
 
 
-def test_inactive_component_activities(call_ray_start):
-    # Verify no activity in response if no active drivers
+@pytest.fixture
+def set_ray_cluster_activity_hook(request):
+    """
+    Fixture that sets RAY_CLUSTER_ACTIVITY_HOOK environment variable
+    for test_e2e_component_activities_hook.
+    """
+    external_hook = getattr(request, "param")
+    assert (
+        external_hook
+    ), "Please pass value of RAY_CLUSTER_ACTIVITY_HOOK env var to this fixture"
+    old_hook = os.environ.get(RAY_CLUSTER_ACTIVITY_HOOK)
+    os.environ[RAY_CLUSTER_ACTIVITY_HOOK] = external_hook
+
+    yield external_hook
+
+    if old_hook is not None:
+        os.environ[RAY_CLUSTER_ACTIVITY_HOOK] = old_hook
+    else:
+        del os.environ[RAY_CLUSTER_ACTIVITY_HOOK]
+
+
+@pytest.mark.parametrize(
+    "set_ray_cluster_activity_hook",
+    [
+        "ray._private.test_utils.external_ray_cluster_activity_hook1",
+        "ray._private.test_utils.external_ray_cluster_activity_hook2",
+        "ray._private.test_utils.external_ray_cluster_activity_hook3",
+        "ray._private.test_utils.external_ray_cluster_activity_hook4",
+    ],
+    indirect=True,
+)
+async def test_component_activities_hook(set_ray_cluster_activity_hook, call_ray_start):
+    """
+    Tests /api/component_activities returns correctly for various
+    responses of RAY_CLUSTER_ACTIVITY_HOOK defined in ray._private.test_utils.
+
+    Verify no active drivers are correctly reflected in response.
+    """
+    external_hook = set_ray_cluster_activity_hook
+
     response = requests.get("http://127.0.0.1:8265/api/component_activities")
     response.raise_for_status()
 
@@ -36,15 +75,46 @@ def test_inactive_component_activities(call_ray_start):
     pprint.pprint(data)
     jsonschema.validate(instance=data, schema=json.load(open(schema_path)))
 
-    # Validate ray_activity_response field can be cast to RayActivityResponse object
+    # Validate driver response can be cast to RayActivityResponse object
+    # and that there are no active drivers.
     driver_ray_activity_response = RayActivityResponse(**data["driver"])
-    assert not driver_ray_activity_response.is_active
+    assert driver_ray_activity_response.is_active == "INACTIVE"
     assert driver_ray_activity_response.reason is None
 
+    # Validate external component response can be cast to RayActivityResponse object
+    if external_hook[-1] == "4":
+        external_activity_response = RayActivityResponse(**data["external_component"])
+        assert external_activity_response.is_active == "ERROR"
+        assert (
+            external_activity_response.reason
+            == "Exception('Error in external cluster activity hook')"
+        )
+    elif external_hook[-1] == "3":
+        external_activity_response = RayActivityResponse(**data["external_component"])
+        assert external_activity_response.is_active == "ERROR"
+    elif external_hook[-1] == "2":
+        external_activity_response = RayActivityResponse(**data["test_component2"])
+        assert external_activity_response.is_active == "ERROR"
+    elif external_hook[-1] == "1":
+        external_activity_response = RayActivityResponse(**data["test_component1"])
+        assert external_activity_response.is_active == "ACTIVE"
+        assert external_activity_response.reason == "Counter: 1"
+
+        # Call endpoint again to validate different response
+        response = requests.get("http://127.0.0.1:8265/api/component_activities")
+        response.raise_for_status()
+        data = response.json()
+        jsonschema.validate(instance=data, schema=json.load(open(schema_path)))
+
+        external_activity_response = RayActivityResponse(**data["test_component1"])
+        assert external_activity_response.is_active == "ACTIVE"
+        assert external_activity_response.reason == "Counter: 2"
+
 
 def test_active_component_activities(ray_start_with_dashboard):
     # Verify drivers which don't have namespace starting with _ray_internal_job_info_
     # are considered active.
+
     driver_template = """
 import ray
 
@@ -77,7 +147,7 @@ def test_active_component_activities(ray_start_with_dashboard):
     # Validate ray_activity_response field can be cast to RayActivityResponse object
     driver_ray_activity_response = RayActivityResponse(**data["driver"])
 
-    assert driver_ray_activity_response.is_active
+    assert driver_ray_activity_response.is_active == "ACTIVE"
     # Drivers with namespace starting with "_ray_internal_job_info_" are not
     # considered active drivers. Three active drivers are the two
     # run with namespace "my_namespace" and the one started
diff --git a/python/ray/_private/test_utils.py b/python/ray/_private/test_utils.py
index 52b00672679b0..a780165eaf9f3 100644
--- a/python/ray/_private/test_utils.py
+++ b/python/ray/_private/test_utils.py
@@ -1,4 +1,5 @@
 import asyncio
+import dataclasses
 import fnmatch
 import functools
 import io
@@ -1366,3 +1367,67 @@ def job_hook(**kwargs):
     cmd = " ".join(kwargs["entrypoint"])
     print(f"hook intercepted: {cmd}")
     sys.exit(0)
+
+
+@dataclasses.dataclass
+class TestRayActivityResponse:
+    """
+    Redefinition of dashboard.modules.snapshot.snapshot_head.RayActivityResponse
+    used in test_component_activities_hook to mimic typical
+    usage of redefining or extending response type.
+    """
+
+    is_active: str
+    reason: Optional[str] = None
+    timestamp: Optional[float] = None
+
+
+# Global counter to test different return values
+# for external_ray_cluster_activity_hook1.
+ray_cluster_activity_hook_counter = 0
+
+
+def external_ray_cluster_activity_hook1():
+    """
+    Example external hook for test_component_activities_hook.
+
+    Returns valid response and increments counter in `reason`
+    field on each call.
+    """
+    global ray_cluster_activity_hook_counter
+    ray_cluster_activity_hook_counter += 1
+    return {
+        "test_component1": TestRayActivityResponse(
+            is_active="ACTIVE",
+            reason=f"Counter: {ray_cluster_activity_hook_counter}",
+        )
+    }
+
+
+def external_ray_cluster_activity_hook2():
+    """
+    Example external hook for test_component_activities_hook.
+
+    Returns invalid output because the value of `test_component2`
+    should be of type RayActivityResponse.
+    """
+    return {"test_component2": "bad_output"}
+
+
+def external_ray_cluster_activity_hook3():
+    """
+    Example external hook for test_component_activities_hook.
+
+    Returns invalid output because return type is not
+    Dict[str, RayActivityResponse]
+    """
+    return "bad_output"
+
+
+def external_ray_cluster_activity_hook4():
+    """
+    Example external hook for test_component_activities_hook.
+
+    Errors during execution.
+    """
+    raise Exception("Error in external cluster activity hook")

<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

<!-- Please give a short summary of the change and the problem this solves. -->
- Add external hook to `/api/component_activities` endpoint in dashboard snapshot router
- Change `is_active` field of `RayActivityResponse` to take an enum `RayActivityStatus` instead of `bool`. This is a backward incompatible change, but should be ok because https://github.com/ray-project/ray/pull/25996 wasn't included in any branch cuts. `RayActivityResponse` now supports informing when there was an error getting the activity observation and the reason.

## Related issue number

<!-- For example: "Closes #1234" -->

## Checks

- [x] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [x] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(
