Merge pull request #27 from ENCODE-DCC/dev

v0.3.0

.circleci/config.yml

@@ -10,19 +10,21 @@ defaults:
 
 machine_defaults: &machine_defaults
   machine:
-    image: ubuntu-1604:202007-01
+    image: ubuntu-2004:202010-01
   working_directory: ~/autouri
 
 
+update_apt:
+  name: Update apt
+  command: |
+    sudo apt-get update -y
+    sudo DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends tzdata
+
+
 install_python3: &install_python3
   name: Install python3, pip3
   command: |
-    sudo apt-get update && sudo apt-get install software-properties-common git wget curl -y
-    sudo add-apt-repository ppa:deadsnakes/ppa -y
-    sudo apt-get update && sudo apt-get install python3.6 -y
-    sudo wget https://bootstrap.pypa.io/get-pip.py
-    sudo python3.6 get-pip.py
-    sudo ln -s /usr/bin/python3.6 /usr/local/bin/python3
+    sudo apt-get install -y software-properties-common git wget curl python3 python3-pip
 
 
 install_shellcheck: &install_shellcheck
@@ -31,18 +33,12 @@ install_shellcheck: &install_shellcheck
     curl -Ls https://github.com/koalaman/shellcheck/releases/download/stable/shellcheck-stable.linux.x86_64.tar.xz | tar xJ && sudo mv shellcheck-stable/shellcheck /usr/local/bin/
 
 
-install_precommit: &install_precommit
-  name: Install Python pre-commit
-  command: |
-    sudo pip3 install PyYAML --ignore-installed
-    sudo pip3 install pre-commit
-
-
 install_py3_packages: &install_py3_packages
   name: Install Python packages
   command: |
-    sudo pip3 install pytest requests dateparser filelock "six>=1.13.0"
-    sudo pip3 install --upgrade pyasn1-modules
+    python3 -m pip install --upgrade pip
+    pip3 install PyYAML --ignore-installed
+    pip3 install pre-commit pytest requests dateparser filelock six ntplib
 
 
 install_gcs_lib: &install_gcs_lib
@@ -51,13 +47,13 @@ install_gcs_lib: &install_gcs_lib
     echo "deb [signed-by=/usr/share/keyrings/cloud.google.gpg] http://packages.cloud.google.com/apt cloud-sdk main" | sudo tee -a /etc/apt/sources.list.d/google-cloud-sdk.list
     curl https://packages.cloud.google.com/apt/doc/apt-key.gpg | sudo apt-key --keyring /usr/share/keyrings/cloud.google.gpg add -
     sudo apt-get update && sudo apt-get install google-cloud-sdk -y
-    sudo pip3 install google-cloud-storage
+    pip3 install google-cloud-storage
 
 
 install_aws_lib: &install_aws_lib
   name: Install AWS Python API (boto3) and CLI (awscli)
   command: |
-    sudo pip3 install boto3 awscli
+    pip3 install awscli boto3
 
 
 make_root_only_dir: &make_root_only_dir
@@ -73,7 +69,7 @@ jobs:
     steps:
       - checkout
       - run: *install_python3
-      - run: *install_precommit
+      - run: *install_py3_packages
       - run: *install_shellcheck
       - run:
           no_output_timeout: 10m
@@ -91,7 +87,6 @@ jobs:
           no_output_timeout: 60m
           command: |
             cd tests/
-            # sign in
             echo ${GCLOUD_SERVICE_ACCOUNT_SECRET_JSON} > tmp_key.json
             gcloud auth activate-service-account --project=${GOOGLE_PROJECT_ID} --key-file=tmp_key.json
             gcloud config set project ${GOOGLE_PROJECT_ID}

.isort.cfg

@@ -4,7 +4,7 @@ include_trailing_comma = True
 force_grid_wrap = 0
 use_parentheses = True
 line_length = 88
-known_third_party = boto3,botocore,dateparser,dateutil,filelock,google,pytest,requests,setuptools
+known_third_party = boto3,botocore,dateparser,dateutil,filelock,google,ntplib,pytest,requests,setuptools
 
 [mypy-bin]
 ignore_errors = True

.pre-commit-config.yaml

@@ -3,7 +3,7 @@
     rev: 19.3b0
     hooks:
       - id: black
-        language_version: python3.6
+        language_version: python3
 
   - repo: https://github.com/asottile/seed-isort-config
     rev: v1.9.2
@@ -14,7 +14,7 @@
     rev: v4.3.21
     hooks:
       - id: isort
-        language_version: python3.6
+        language_version: python3
 
   - repo: https://github.com/detailyang/pre-commit-shell
     rev: v1.0.6

autouri/__init__.py

@@ -5,4 +5,4 @@
 from .s3uri import S3URI
 
 __all__ = ["AbsPath", "AutoURI", "URIBase", "GCSURI", "HTTPURL", "S3URI"]
-__version__ = "0.2.6"
+__version__ = "0.3.0"

autouri/gcsuri.py

@@ -29,6 +29,7 @@
 
 from .autouri import AutoURI, URIBase
 from .metadata import URIMetadata, get_seconds_from_epoch, parse_md5_str
+from .ntp_now import now_utc
 
 logger = logging.getLogger(__name__)
 
@@ -75,10 +76,15 @@ class GCSURILock(BaseFileLock):
             Retry if release (deletion) of a lock file fails.
         DEFAULT_RETRY_RELEASE_INTERVAL:
             Interval for retrial in seconds.
+        DEFAULT_LOCK_FILE_EXPIRATION_SEC:
+            Expiration time of a lock file in seconds.
+            If a lock file is older than this then it is released from hold
+            (based on GCS's temporary hold) and then deleted.
     """
 
     DEFAULT_RETRY_RELEASE = 3
     DEFAULT_RETRY_RELEASE_INTERVAL = 3
+    DEFAULT_LOCK_FILE_EXPIRATION_SEC = 3600
 
     def __init__(
         self,
@@ -88,13 +94,15 @@ def __init__(
         poll_interval=10.0,
         retry_release=DEFAULT_RETRY_RELEASE,
         retry_release_interval=DEFAULT_RETRY_RELEASE_INTERVAL,
+        lockfile_expiration_sec=DEFAULT_LOCK_FILE_EXPIRATION_SEC,
         no_lock=False,
     ):
         super().__init__(lock_file, timeout=timeout)
         self._poll_interval = poll_interval
         self._thread_id = thread_id
         self._retry_release = retry_release
         self._retry_release_interval = retry_release_interval
+        self._lockfile_expiration_sec = lockfile_expiration_sec
 
     def acquire(self, timeout=None, poll_intervall=5.0):
         """Use self._poll_interval instead of poll_intervall in args
@@ -125,6 +133,22 @@ def _acquire(self):
         """
         u = GCSURI(self._lock_file, thread_id=self._thread_id)
         try:
+            metadata = u.get_metadata()
+
+            if metadata.exists:
+                # if lock file already exists then check if it's expired
+                if (
+                    now_utc().timestamp()
+                    > metadata.mtime + self._lockfile_expiration_sec
+                ):
+                    logger.debug("Found expired lock file, will release/delete it.")
+                    blob, _ = u.get_blob()
+                    if blob.temporary_hold:
+                        blob.temporary_hold = False
+                        blob.patch()
+                    blob.delete()
+                else:
+                    return
             blob, _ = u.get_blob(new=True)
             blob.upload_from_string("")
             blob.temporary_hold = True
@@ -213,7 +237,7 @@ class GCSURI(URIBase):
     DURATION_PRESIGNED_URL: int = 4233600
 
     RETRY_BUCKET: int = 3
-    RETRY_BUCKET_DELAY: int = 1
+    RETRY_BUCKET_DELAY: int = 3
     USE_GSUTIL_FOR_S3: bool = False
 
     _CACHED_GCS_CLIENTS = {}

autouri/ntp_now.py

@@ -0,0 +1,61 @@
+"""Accurate now() based on cached offset between NTP server
+and local system time.
+
+Make sure not to change system time once the offset is cached.
+"""
+
+import logging
+from datetime import datetime, timezone
+
+import ntplib
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_NTP_SERVER = "pool.ntp.org"
+cached_offset_between_ntp_server_and_local_system = None
+
+
+def now_utc(ntp_server=DEFAULT_NTP_SERVER):
+    global cached_offset_between_ntp_server_and_local_system
+
+    if cached_offset_between_ntp_server_and_local_system is not None:
+        return (
+            datetime.now(timezone.utc)
+            + cached_offset_between_ntp_server_and_local_system
+        )
+
+    try:
+        resp = ntplib.NTPClient().request(ntp_server)
+        adjusted_timestamp = resp.tx_time + resp.delay * 0.5
+        ntp_server_time = datetime.fromtimestamp(adjusted_timestamp, timezone.utc)
+        # update cache
+        cached_offset_between_ntp_server_and_local_system = (
+            ntp_server_time - datetime.now(timezone.utc)
+        )
+        logger.debug(
+            "Successfully retrieved time from NTP server {srv}. time:{time}, offset:{offset}".format(
+                srv=ntp_server,
+                time=ntp_server_time,
+                offset=cached_offset_between_ntp_server_and_local_system.total_seconds(),
+            )
+        )
+        return ntp_server_time
+
+    except Exception as e:
+        logger.debug(
+            "Failed to retrieve time from NTP server {srv}. error {err}".format(
+                srv=ntp_server, err=str(e)
+            )
+        )
+
+    return datetime.now(timezone.utc)
+
+
+def reset_cached_offset():
+    global cached_offset_between_ntp_server_and_local_system
+    cached_offset_between_ntp_server_and_local_system = None
+
+
+def get_cached_offset():
+    global cached_offset_between_ntp_server_and_local_system
+    return cached_offset_between_ntp_server_and_local_system

setup.py

@@ -61,5 +61,6 @@ def find_meta(meta):
         "dateparser",
         "filelock",
         "six>=1.13.0",
+        "ntplib",
     ],
 )

tests/test_ntp_now.py

@@ -0,0 +1,31 @@
+"""Make sure to run these tests on a machine with correct os time
+"""
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+from autouri.ntp_now import get_cached_offset, now_utc, reset_cached_offset
+
+
+def test_now_utc():
+    reset_cached_offset()
+
+    assert abs((now_utc() - datetime.now(timezone.utc)).total_seconds()) < 0.01
+
+
+class MockedDataTime(datetime):
+    @classmethod
+    def now(cls, timezone):
+        return datetime.now(timezone) - timedelta(0, 25)
+
+
+def test_now_utc_wrong_os_time():
+    reset_cached_offset()
+    with patch("autouri.ntp_now.datetime", MockedDataTime):
+        ntp_now_utc = now_utc()
+
+    # should be accurate even though system time is 25 second behind NTP server time
+    assert abs((ntp_now_utc - datetime.now(timezone.utc)).total_seconds()) < 0.01
+
+    # cache offset should be 25 second
+    cached_offset_in_seconds = get_cached_offset().total_seconds()
+    assert cached_offset_in_seconds > 24.99 and cached_offset_in_seconds < 25.01