Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[job submission] Move HTTP routes to /api/jobs prefix #19995

Merged
merged 6 commits into from
Nov 4, 2021
Merged

[job submission] Move HTTP routes to /api/jobs prefix #19995

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
0ab395d
fix
edoakes Nov 2, 2021
a319902
fix
edoakes Nov 2, 2021
ee4cc11
trailing /
edoakes Nov 3, 2021
225524c
Merge branch 'master' of https://github.com/ray-project/ray into jobs…
edoakes Nov 3, 2021
3d6f362
Merge branch 'master' of https://github.com/ray-project/ray into jobs…
edoakes Nov 4, 2021
8dd07b7
fix
edoakes Nov 4, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions dashboard/modules/job/job_head.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,12 @@

RAY_INTERNAL_JOBS_NAMESPACE = "_ray_internal_jobs_"

JOBS_API_PREFIX = "/api/jobs/"
JOBS_API_ROUTE_LOGS = JOBS_API_PREFIX + "logs"
JOBS_API_ROUTE_SUBMIT = JOBS_API_PREFIX + "submit"
JOBS_API_ROUTE_STATUS = JOBS_API_PREFIX + "status"
JOBS_API_ROUTE_PACKAGE = JOBS_API_PREFIX + "package"


def _ensure_ray_initialized(f: Callable) -> Callable:
@wraps(f)
Expand All @@ -36,7 +42,7 @@ def __init__(self, dashboard_head):

self._job_manager = None

@routes.get("/package")
@routes.get(JOBS_API_ROUTE_PACKAGE)
@_ensure_ray_initialized
async def get_package(self,
req: aiohttp.web.Request) -> aiohttp.web.Response:
Expand All @@ -46,7 +52,7 @@ async def get_package(self,
text=json.dumps(dataclasses.asdict(resp)),
content_type="application/json")

@routes.put("/package")
@routes.put(JOBS_API_ROUTE_PACKAGE)
@_ensure_ray_initialized
async def upload_package(self, req: aiohttp.web.Request):
package_uri = req.query["package_uri"]
Expand All @@ -55,7 +61,7 @@ async def upload_package(self, req: aiohttp.web.Request):

return aiohttp.web.Response()

@routes.post("/submit")
@routes.post(JOBS_API_ROUTE_SUBMIT)
@_ensure_ray_initialized
async def submit(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
# TODO: (jiaodong) Validate if job request is valid without using
Expand All @@ -71,7 +77,7 @@ async def submit(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
text=json.dumps(dataclasses.asdict(resp)),
content_type="application/json")

@routes.get("/status")
@routes.get(JOBS_API_ROUTE_STATUS)
@_ensure_ray_initialized
async def status(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
job_id = req.query["job_id"]
Expand All @@ -82,7 +88,7 @@ async def status(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
text=json.dumps(dataclasses.asdict(resp)),
content_type="application/json")

@routes.get("/logs")
@routes.get(JOBS_API_ROUTE_LOGS)
@_ensure_ray_initialized
async def logs(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
job_id = req.query["job_id"]
Expand Down
19 changes: 12 additions & 7 deletions dashboard/modules/job/sdk.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,10 @@
GetPackageResponse, JobSubmitRequest, JobSubmitResponse, JobStatusResponse,
JobLogsResponse)

from ray.dashboard.modules.job.job_head import (
JOBS_API_ROUTE_LOGS, JOBS_API_ROUTE_SUBMIT, JOBS_API_ROUTE_STATUS,
JOBS_API_ROUTE_PACKAGE)

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

Expand All @@ -37,8 +41,9 @@ def _do_request(self,
json_data: Optional[dict] = None,
params: Optional[dict] = None,
response_type: Optional[type] = None) -> Optional[object]:
url = f"{self._address}/{endpoint}"
logger.info(f"Sending request to {url} with json: {json_data}.")
url = self._address + endpoint
logger.debug(f"Sending request to {url} with "
f"json: {json_data}, params: {params}.")
r = requests.request(
method, url, data=data, json=json_data, params=params)

Expand All @@ -53,7 +58,7 @@ def _do_request(self,
def _package_exists(self, package_uri: str) -> bool:
resp = self._do_request(
"GET",
"package",
JOBS_API_ROUTE_PACKAGE,
params={"package_uri": package_uri},
response_type=GetPackageResponse)
return resp.package_exists
Expand All @@ -73,7 +78,7 @@ def _upload_package(self,
excludes=excludes)
self._do_request(
"PUT",
"package",
JOBS_API_ROUTE_PACKAGE,
data=package_file.read_bytes(),
params={"package_uri": package_uri})
package_file.unlink()
Expand Down Expand Up @@ -120,23 +125,23 @@ def submit_job(self,
entrypoint=entrypoint, runtime_env=runtime_env, metadata=metadata)
resp = self._do_request(
"POST",
"submit",
JOBS_API_ROUTE_SUBMIT,
json_data=dataclasses.asdict(req),
response_type=JobSubmitResponse)
return resp.job_id

def get_job_status(self, job_id: str) -> JobStatus:
resp = self._do_request(
"GET",
"status",
JOBS_API_ROUTE_STATUS,
params={"job_id": job_id},
response_type=JobStatusResponse)
return resp.job_status

def get_job_logs(self, job_id: str) -> Tuple[str, str]:
resp = self._do_request(
"GET",
"logs",
JOBS_API_ROUTE_LOGS,
params={"job_id": job_id},
response_type=JobLogsResponse)
return resp.stdout, resp.stderr