Skip to content

Commit

Permalink
Merge branch 'master' of github.com:DataBiosphere/toil into issues/51…
Browse files Browse the repository at this point in the history
…14-filesize-sniffing
  • Loading branch information
stxue1 committed Nov 26, 2024
2 parents 03365dd + f3e98db commit 8b80fc3
Show file tree
Hide file tree
Showing 4 changed files with 258 additions and 12 deletions.
6 changes: 6 additions & 0 deletions docs/appendices/environment_vars.rst
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,12 @@ There are several environment variables that affect the way Toil runs.
| | to S3 (``True`` by default). |
| | Example: ``TOIL_S3_USE_SSL=False`` |
+----------------------------------+----------------------------------------------------+
| TOIL_FTP_USER | The FTP username to override all FTP logins with |
| | Example: ``TOIL_FTP_USER=ftp_user`` |
+----------------------------------+----------------------------------------------------+
| TOIL_FTP_PASSWORD | The FTP password to override all FTP logins with |
| | Example: ``TOIL_FTP_PASSWORD=ftp_password`` |
+----------------------------------+----------------------------------------------------+
| TOIL_WES_BROKER_URL | An optional broker URL to use to communicate |
| | between the WES server and Celery task queue. If |
| | unset, ``amqp://guest:guest@localhost:5672//`` is |
Expand Down
42 changes: 32 additions & 10 deletions src/toil/jobStores/abstractJobStore.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@
)
from urllib.error import HTTPError
from urllib.parse import ParseResult, urlparse
from urllib.request import urlopen
from urllib.request import urlopen, Request
from uuid import uuid4

from toil.common import Config, getNodeID, safeUnpickleFromStream
Expand All @@ -46,6 +46,7 @@
JobException,
ServiceJobDescription,
)
from toil.lib.ftp_utils import FtpFsAccess
from toil.lib.compatibility import deprecated
from toil.lib.exceptions import UnimplementedURLException
from toil.lib.io import WriteWatchingStream
Expand Down Expand Up @@ -1856,18 +1857,31 @@ class JobStoreSupport(AbstractJobStore, metaclass=ABCMeta):
stores.
"""

@classmethod
def _setup_ftp(cls) -> FtpFsAccess:
# FTP connections are not reused. Ideally, a thread should watch any reused FTP connections
# and close them when necessary
return FtpFsAccess()

@classmethod
def _supports_url(cls, url: ParseResult, export: bool = False) -> bool:
return url.scheme.lower() in ("http", "https", "ftp") and not export

@classmethod
def _url_exists(cls, url: ParseResult) -> bool:
# Deal with FTP first to support user/password auth
if url.scheme.lower() == "ftp":
ftp = cls._setup_ftp()
return ftp.exists(url.geturl())

try:
# TODO: Figure out how to HEAD instead of this.
with cls._open_url(url):
with closing(urlopen(Request(url.geturl(), method="HEAD"))):
return True
except FileNotFoundError:
return False
except HTTPError as e:
if e.code in (404, 410):
return False
else:
raise
# Any other errors we should pass through because something really went
# wrong (e.g. server is broken today but file may usually exist)

Expand All @@ -1880,11 +1894,13 @@ def _url_exists(cls, url: ParseResult) -> bool:
)
def _get_size(cls, url: ParseResult) -> Optional[int]:
if url.scheme.lower() == "ftp":
return None
with closing(urlopen(url.geturl())) as readable:
# just read the header for content length
size = readable.info().get("content-length")
return int(size) if size is not None else None
ftp = cls._setup_ftp()
return ftp.size(url.geturl())

# just read the header for content length
resp = urlopen(Request(url.geturl(), method="HEAD"))
size = resp.info().get("content-length")
return int(size) if size is not None else None

@classmethod
def _read_from_url(
Expand Down Expand Up @@ -1917,6 +1933,12 @@ def count(l: int) -> None:
]
)
def _open_url(cls, url: ParseResult) -> IO[bytes]:
# Deal with FTP first so we support user/password auth
if url.scheme.lower() == "ftp":
ftp = cls._setup_ftp()
# we open in read mode as write mode is not supported
return ftp.open(url.geturl(), mode="r")

try:
return cast(IO[bytes], closing(urlopen(url.geturl())))
except HTTPError as e:
Expand Down
5 changes: 3 additions & 2 deletions src/toil/lib/aws/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from urllib.parse import ParseResult

from toil.lib.aws import AWSRegionName, AWSServerErrors, session
from toil.lib.conversions import strtobool
from toil.lib.misc import printq
from toil.lib.retry import (
DEFAULT_DELAYS,
Expand Down Expand Up @@ -363,7 +364,7 @@ def get_object_for_url(url: ParseResult, existing: Optional[bool] = None) -> "S3
host = os.environ.get("TOIL_S3_HOST", None)
port = os.environ.get("TOIL_S3_PORT", None)
protocol = "https"
if os.environ.get("TOIL_S3_USE_SSL", True) == "False":
if strtobool(os.environ.get("TOIL_S3_USE_SSL", 'True')) is False:
protocol = "http"
if host:
endpoint_url = f"{protocol}://{host}" + f":{port}" if port else ""
Expand Down Expand Up @@ -425,7 +426,7 @@ def list_objects_for_url(url: ParseResult) -> list[str]:
host = os.environ.get("TOIL_S3_HOST", None)
port = os.environ.get("TOIL_S3_PORT", None)
protocol = "https"
if os.environ.get("TOIL_S3_USE_SSL", True) == "False":
if strtobool(os.environ.get("TOIL_S3_USE_SSL", 'True')) is False:
protocol = "http"
if host:
endpoint_url = f"{protocol}://{host}" + f":{port}" if port else ""
Expand Down
217 changes: 217 additions & 0 deletions src/toil/lib/ftp_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,217 @@
# Copyright 2017 Oregon Health and Science University
#
# Copyright (C) 2015-2021 Regents of the University of California
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import ftplib
import logging
import netrc
import os
from contextlib import closing
from typing import Optional, Any, cast, IO
from urllib.parse import urlparse
from urllib.request import urlopen

logger = logging.getLogger(__name__)


class FtpFsAccess:
"""
FTP access with upload.
Taken and modified from https://github.com/ohsu-comp-bio/cwl-tes/blob/03f0096f9fae8acd527687d3460a726e09190c3a/cwl_tes/ftp.py#L37-L251
"""
# TODO: Properly support FTP over SSL

def __init__(
self, cache: Optional[dict[Any, ftplib.FTP]] = None
):
"""
FTP object to handle FTP connections. By default, connect over FTP with TLS.
:param cache: cache of generated FTP objects
"""
self.cache = cache or {}
self.netrc = None
try:
if "HOME" in os.environ:
if os.path.exists(os.path.join(os.environ["HOME"], ".netrc")):
self.netrc = netrc.netrc(os.path.join(os.environ["HOME"], ".netrc"))
elif os.path.exists(os.path.join(os.curdir, ".netrc")):
self.netrc = netrc.netrc(os.path.join(os.curdir, ".netrc"))
except netrc.NetrcParseError as err:
logger.debug(err)

def exists(self, fn: str) -> bool:
"""
Check if a file/directory exists over an FTP server
:param fn: FTP url
:return: True or false depending on whether the object exists on the server
"""
return self.isfile(fn) or self.isdir(fn)

def isfile(self, fn: str) -> bool:
"""
Check if the FTP url points to a file
:param fn: FTP url
:return: True if url is file, else false
"""
ftp = self._connect(fn)
if ftp:
try:
if not self.size(fn) is None:
return True
else:
return False
except ftplib.all_errors:
return False
return False

def isdir(self, fn: str) -> bool:
"""
Check if the FTP url points to a directory
:param fn: FTP url
:return: True if url is directory, else false
"""
ftp = self._connect(fn)
if ftp:
try:
cwd = ftp.pwd()
ftp.cwd(urlparse(fn).path)
ftp.cwd(cwd)
return True
except ftplib.all_errors:
return False
return False

def open(self, fn: str, mode: str) -> IO[bytes]:
"""
Open an FTP url.
Only supports reading, no write support.
:param fn: FTP url
:param mode: Mode to open FTP url in
:return:
"""
if "r" in mode:
host, port, user, passwd, path = self._parse_url(fn)
handle = urlopen("ftp://{}:{}@{}:{}/{}".format(user, passwd, host, port, path))
return cast(IO[bytes], closing(handle))
# TODO: support write mode
raise Exception("Write mode FTP not implemented")

def _parse_url(
self, url: str
) -> tuple[str, int, Optional[str], Optional[str], str]:
"""
Parse an FTP url into hostname, username, password, and path
:param url:
:return: hostname, username, password, path
"""
parse = urlparse(url)
user = parse.username
passwd = parse.password
host = parse.hostname
port = parse.port
path = parse.path
if host is None:
# The URL we connect to must have a host
raise RuntimeError(f"FTP URL does not contain a host: {url}")
# default port is 21
if port is None:
port = 21
if parse.scheme == "ftp":
if not user and self.netrc:
if host is not None:
creds = self.netrc.authenticators(host)
if creds:
user, _, passwd = creds
if not user:
if host is not None:
user, passwd = self._recall_credentials(host)
if passwd is None:
passwd = "anonymous@"
if user is None:
user = "anonymous"
return host, port, user, passwd, path

def _connect(self, url: str) -> Optional[ftplib.FTP]:
"""
Connect to an FTP server. Handles authentication.
:param url: FTP url
:return: FTP object
"""
parse = urlparse(url)
if parse.scheme == "ftp":
host, port, user, passwd, _ = self._parse_url(url)
if host is None:
# there has to be a host
return None
if (host, user, passwd) in self.cache:
if self.cache[(host, user, passwd)].pwd():
return self.cache[(host, user, passwd)]
ftp = ftplib.FTP_TLS()
# Note: the FTP lib logger handles logging itself and doesn't go through our logging implementation
ftp.set_debuglevel(1 if logger.isEnabledFor(logging.DEBUG) else 0)
ftp.connect(host, port)
env_user = os.getenv("TOIL_FTP_USER")
env_passwd = os.getenv("TOIL_FTP_PASSWORD")
if env_user:
user = env_user
if env_passwd:
passwd = env_passwd
ftp.login(user or "", passwd or "", secure=False)
self.cache[(host, user, passwd)] = ftp
return ftp
return None

def _recall_credentials(
self, desired_host: str
) -> tuple[Optional[str], Optional[str]]:
"""
Grab the cached credentials
:param desired_host: FTP hostname
:return: username, password
"""
for host, user, passwd in self.cache:
if desired_host == host:
return user, passwd
return None, None

def size(self, fn: str) -> Optional[int]:
"""
Get the size of an FTP object
:param fn: FTP url
:return: Size of object
"""
ftp = self._connect(fn)
if ftp:
host, port, user, passwd, path = self._parse_url(fn)
try:
return ftp.size(path)
except ftplib.all_errors as e:
if str(e) == "550 SIZE not allowed in ASCII mode":
# some servers don't allow grabbing size in ascii mode
# https://stackoverflow.com/questions/22090001/get-folder-size-using-ftplib/22093848#22093848
ftp.voidcmd("TYPE I")
return ftp.size(path)
handle = urlopen("ftp://{}:{}@{}:{}/{}".format(user, passwd, host, port, path))
info = handle.info()
handle.close()
if "Content-length" in info:
return int(info["Content-length"])
return None

return None

0 comments on commit 8b80fc3

Please sign in to comment.