[dashboard] Fixes dashboard issues when environments have set http_pr…

…oxy (ray-project#12598)

* fixes ray start with http_proxy

* format

* fixes

* fixes

* increase timeout

* address comments

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