[dashboard] Fixes dashboard issues when environments have set http_proxy (ray-project#12598)

* fixes ray start with http_proxy

* format

* fixes

* fixes

* increase timeout

* address comments

Author: ConeyLiu

dashboard/agent.py

@@ -75,8 +75,9 @@ def __init__(self,
         logger.info("Dashboard agent grpc address: %s:%s", self.ip,
                     self.grpc_port)
         self.aioredis_client = None
+        options = (("grpc.enable_http_proxy", 0), )
         self.aiogrpc_raylet_channel = aiogrpc.insecure_channel(
-            f"{self.ip}:{self.node_manager_port}")
+            f"{self.ip}:{self.node_manager_port}", options=options)
         self.http_session = None
 
     def _load_modules(self):

dashboard/head.py

@@ -159,7 +159,9 @@ async def run(self):
                 if not gcs_address:
                     raise Exception("GCS address not found.")
                 logger.info("Connect to GCS at %s", gcs_address)
-                channel = aiogrpc.insecure_channel(gcs_address)
+                options = (("grpc.enable_http_proxy", 0), )
+                channel = aiogrpc.insecure_channel(
+                    gcs_address, options=options)
             except Exception as ex:
                 logger.error("Connect to GCS failed: %s, retry...", ex)
                 await asyncio.sleep(

dashboard/modules/logical_view/logical_view_head.py

@@ -46,7 +46,9 @@ async def kill_actor(self, req) -> aiohttp.web.Response:
         except KeyError:
             return rest_response(success=False, message="Bad Request")
         try:
-            channel = aiogrpc.insecure_channel(f"{ip_address}:{port}")
+            options = (("grpc.enable_http_proxy", 0), )
+            channel = aiogrpc.insecure_channel(
+                f"{ip_address}:{port}", options=options)
             stub = core_worker_pb2_grpc.CoreWorkerServiceStub(channel)
 
             await stub.KillActor(

dashboard/modules/reporter/reporter_head.py

@@ -38,7 +38,9 @@ async def _update_stubs(self, change):
         if change.new:
             node_id, ports = change.new
             ip = DataSource.node_id_to_ip[node_id]
-            channel = aiogrpc.insecure_channel(f"{ip}:{ports[1]}")
+            options = (("grpc.enable_http_proxy", 0), )
+            channel = aiogrpc.insecure_channel(
+                f"{ip}:{ports[1]}", options=options)
             stub = reporter_pb2_grpc.ReporterServiceStub(channel)
             self._stubs[ip] = stub
 

dashboard/modules/stats_collector/stats_collector_head.py

@@ -71,7 +71,8 @@ async def _update_stubs(self, change):
             node_id, node_info = change.new
             address = "{}:{}".format(node_info["nodeManagerAddress"],
                                      int(node_info["nodeManagerPort"]))
-            channel = aiogrpc.insecure_channel(address)
+            options = (("grpc.enable_http_proxy", 0), )
+            channel = aiogrpc.insecure_channel(address, options=options)
             stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)
             self._stubs[node_id] = stub
 

dashboard/tests/conftest.py

@@ -1,17 +1,40 @@
-import os
-import pytest
-from ray.tests.conftest import *  # noqa
-
-
-@pytest.fixture
-def enable_test_module():
-    os.environ["RAY_DASHBOARD_MODULE_TEST"] = "true"
-    yield
-    os.environ.pop("RAY_DASHBOARD_MODULE_TEST", None)
-
-
-@pytest.fixture
-def disable_aiohttp_cache():
-    os.environ["RAY_DASHBOARD_NO_CACHE"] = "true"
-    yield
-    os.environ.pop("RAY_DASHBOARD_NO_CACHE", None)
+import os
+import pytest
+from ray.tests.conftest import *  # noqa
+
+
+@pytest.fixture
+def enable_test_module():
+    os.environ["RAY_DASHBOARD_MODULE_TEST"] = "true"
+    yield
+    os.environ.pop("RAY_DASHBOARD_MODULE_TEST", None)
+
+
+@pytest.fixture
+def disable_aiohttp_cache():
+    os.environ["RAY_DASHBOARD_NO_CACHE"] = "true"
+    yield
+    os.environ.pop("RAY_DASHBOARD_NO_CACHE", None)
+
+
+@pytest.fixture
+def set_http_proxy():
+    http_proxy = os.environ.get("http_proxy", None)
+    https_proxy = os.environ.get("https_proxy", None)
+
+    # set http proxy
+    os.environ["http_proxy"] = "www.example.com:990"
+    os.environ["https_proxy"] = "www.example.com:990"
+
+    yield
+
+    # reset http proxy
+    if http_proxy:
+        os.environ["http_proxy"] = http_proxy
+    else:
+        del os.environ["http_proxy"]
+
+    if https_proxy:
+        os.environ["https_proxy"] = https_proxy
+    else:
+        del os.environ["https_proxy"]

dashboard/tests/test_dashboard.py

@@ -571,5 +571,38 @@ def test_immutable_types():
         print(d3[1])
 
 
+def test_http_proxy(enable_test_module, set_http_proxy, shutdown_only):
+    address_info = ray.init(num_cpus=1, include_dashboard=True)
+    assert (wait_until_server_available(address_info["webui_url"]) is True)
+
+    webui_url = address_info["webui_url"]
+    webui_url = format_web_url(webui_url)
+
+    timeout_seconds = 10
+    start_time = time.time()
+    while True:
+        time.sleep(1)
+        try:
+            response = requests.get(
+                webui_url + "/test/dump",
+                proxies={
+                    "http": None,
+                    "https": None
+                })
+            response.raise_for_status()
+            try:
+                response.json()
+                assert response.ok
+            except Exception as ex:
+                logger.info("failed response: %s", response.text)
+                raise ex
+            break
+        except (AssertionError, requests.exceptions.ConnectionError) as e:
+            logger.info("Retry because of %s", e)
+        finally:
+            if time.time() > start_time + timeout_seconds:
+                raise Exception("Timed out while testing.")
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", __file__]))

dashboard/utils.py

@@ -1,34 +1,35 @@
 import abc
-import os
-import socket
-import time
 import asyncio
 import collections
-import json
 import datetime
 import functools
 import importlib
 import inspect
+import json
 import logging
+import os
 import pkgutil
+import socket
 import traceback
-from base64 import b64decode
 from abc import ABCMeta, abstractmethod
-from collections.abc import MutableMapping, Mapping, Sequence
+from base64 import b64decode
 from collections import namedtuple
+from collections.abc import MutableMapping, Mapping, Sequence
 from typing import Any
 
-import aioredis
+import aiohttp.signals
 import aiohttp.web
-import ray.new_dashboard.consts as dashboard_consts
+import aioredis
+import time
 from aiohttp import hdrs
 from aiohttp.frozenlist import FrozenList
 from aiohttp.typedefs import PathLike
 from aiohttp.web import RouteDef
-import aiohttp.signals
 from google.protobuf.json_format import MessageToDict
-from ray.utils import binary_to_hex
+
+import ray.new_dashboard.consts as dashboard_consts
 from ray.ray_constants import env_bool
+from ray.utils import binary_to_hex
 
 try:
     create_task = asyncio.create_task