Repository: jcrist/skein (python)

Issue:
`'skein.crt' file already exists, use `force` to overwrite` error 
I have been running multiple Dask-Yarn cluster applications on EMR. I am now getting this error when trying to start a new cluster application:

```
2019-02-21 23:09:05,489 [ERROR] __main__.forecast_: 'skein.crt' file already exists, use `force` to overwrite
Traceback (most recent call last):
 File "/usr/local/azkaban_projects/ovforecast/projects/de_ghd_ovf/forecast/ovf.py", line 302, in forecast_
   testset, custom_group_file, use_yarn=use_yarn)
 File "/usr/local/lib/python2.7/site-packages/ovforecast/forecast.py", line 87, in forecast
   yarn_app_name=app_name)
 File "/usr/local/lib/python2.7/site-packages/functools32/functools32.py", line 400, in wrapper
   result = user_function(*args, **kwds)
 File "/usr/local/lib/python2.7/site-packages/ovforecast/yarn.py", line 274, in get_dask_cluster
   cluster = YarnCluster.from_specification(spec)
 File "/usr/local/lib/python2.7/site-packages/dask_yarn/core.py", line 325, in from_specification
   self._start_cluster(spec, skein_client)
 File "/usr/local/lib/python2.7/site-packages/dask_yarn/core.py", line 335, in _start_cluster
   skein_client = _get_skein_client(skein_client)
 File "/usr/local/lib/python2.7/site-packages/dask_yarn/core.py", line 44, in _get_skein_client
   return skein.Client(security=security)
 File "/usr/local/lib/python2.7/site-packages/skein/core.py", line 326, in __init__
   security = Security.from_default()
 File "/usr/local/lib/python2.7/site-packages/skein/model.py", line 342, in from_default
   return cls.new_credentials().to_directory(directory=properties.config_dir)
 File "/usr/local/lib/python2.7/site-packages/skein/model.py", line 392, in to_directory
   raise context.FileExistsError(msg)
FileExistsError: 'skein.crt' file already exists, use `force` to overwrite
```

This error seems to have started when I moved to explicitly closing my Client and shuting down the cluster before my python application closes. My theory is that once one application closes and another is trying to start that the `skein.crt` file is being locked or not probably removed during shutdown?

My application is defined as:

```
worker = skein.Service(
instances=WORKER_INSTANCES,
max_restarts=-1,
resources=skein.Resources(
    memory=WORKER_MEMORY,
    vcores=WORKER_VCORES
),
files={
    './cachedvolume': skein.File(
        source='/mnt/tmp/cached_volume.sqlite3',
        type='file'
    )
},
env={'THEANO_FLAGS': 'base_compiledir=/tmp/.theano/',
     'WORKER_VOL_LOCATION': './cachedvolume',
     'PRESTO_URL': 'presto://{HOSTNAME}:8889'.format(
         HOSTNAME=os.environ.get('HOSTNAME')),
     'EC2_TEMPDIR': '/mnt/tmp',},
script='dask-yarn services worker',
depends=['dask.scheduler']
)

scheduler = skein.Service(
resources=skein.Resources(
    memory=SCHEDULER_MEMORY,
    vcores=SCHEDULER_VCORES
),
script='dask-yarn services scheduler'
)

spec = skein.ApplicationSpec(
name=yarn_app_name,
queue='default',
services={
    'dask.worker': worker,
    'dask.scheduler': scheduler
}
)
```

I am running Python 2.7 and
```
skein==0.5.1
dask==1.0.0
dask-yarn==0.5.0
```

Pull request description:
Make generating new credentials threadsafe
Previously if multiple processes ran `Security.from_default()` for the first time concurrently, there'd be a race condition generating the new credentials. Now we use a file lock to synchronize between processes.

Fixes #144.

Patch:
diff --git a/skein/model.py b/skein/model.py
index 32cd9d6..f8941fb 100644
--- a/skein/model.py
+++ b/skein/model.py
@@ -10,8 +10,9 @@ import yaml
 from . import proto as _proto
 from .compatibility import urlparse, string, integer, math_ceil, makedirs
 from .objects import Enum, ProtobufMessage, Specification, required
-from .exceptions import context, FileNotFoundError
-from .utils import implements, format_list, datetime_from_millis, runtime, xor
+from .exceptions import context, FileNotFoundError, FileExistsError
+from .utils import (implements, format_list, datetime_from_millis, runtime,
+                    xor, lock_file)
 
 __all__ = ('ApplicationSpec', 'Service', 'Resources', 'File', 'FileType',
            'FileVisibility', 'ACLs', 'Master', 'Security', 'ApplicationState',
@@ -347,9 +348,17 @@ class Security(Specification):
             return cls.from_directory(properties.config_dir)
         except FileNotFoundError:
             pass
-        context.warn("Skein global security credentials not found, writing now "
-                     "to %r." % properties.config_dir)
-        return cls.new_credentials().to_directory(directory=properties.config_dir)
+
+        new = cls.new_credentials()
+        try:
+            out = new.to_directory(properties.config_dir)
+            context.warn("Skein global security credentials not found, "
+                         "writing now to %r." % properties.config_dir)
+        except FileExistsError:
+            # Race condition between competing processes, use the credentials
+            # written by the other process.
+            out = cls.from_directory(properties.config_dir)
+        return out
 
     @classmethod
     def from_directory(cls, directory):
@@ -392,19 +401,21 @@ class Security(Specification):
         cert_bytes = self._get_bytes('cert')
         key_bytes = self._get_bytes('key')
 
-        for path, name in [(cert_path, 'skein.crt'), (key_path, 'skein.pem')]:
-            if os.path.exists(path):
-                if force:
-                    os.unlink(path)
-                else:
-                    msg = ("%r file already exists, use `%s` to overwrite" %
-                           (name, '--force' if context.is_cli else 'force'))
-                    raise context.FileExistsError(msg)
-
-        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
-        for path, data in [(cert_path, cert_bytes), (key_path, key_bytes)]:
-            with os.fdopen(os.open(path, flags, 0o600), 'wb') as fil:
-                fil.write(data)
+        lock_path = os.path.join(directory, 'skein.lock')
+        with lock_file(lock_path):
+            for path, name in [(cert_path, 'skein.crt'), (key_path, 'skein.pem')]:
+                if os.path.exists(path):
+                    if force:
+                        os.unlink(path)
+                    else:
+                        msg = ("%r file already exists, use `%s` to overwrite" %
+                               (name, '--force' if context.is_cli else 'force'))
+                        raise context.FileExistsError(msg)
+
+            flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+            for path, data in [(cert_path, cert_bytes), (key_path, key_bytes)]:
+                with os.fdopen(os.open(path, flags, 0o600), 'wb') as fil:
+                    fil.write(data)
 
         return Security(cert_file=cert_path, key_file=key_path)
 
diff --git a/skein/utils.py b/skein/utils.py
index 0d3fb6c..47f5850 100644
--- a/skein/utils.py
+++ b/skein/utils.py
@@ -1,13 +1,69 @@
 from __future__ import print_function, division, absolute_import
 
 import errno
+import fcntl
 import os
+import threading
+import weakref
 from contextlib import contextmanager
 from datetime import datetime, timedelta
 
 from .compatibility import unicode, UTC
 
 
+_paths_lock = threading.Lock()
+_paths_to_locks = weakref.WeakValueDictionary()
+
+
+def lock_file(path):
+    """File based lock on ``path``.
+
+    Creates a file based lock. When acquired, other processes or threads are
+    prevented from acquiring the same lock until it is released.
+    """
+    with _paths_lock:
+        lock = _paths_to_locks.get(path)
+        if lock is None:
+            _paths_to_locks[path] = lock = _FileLock(path)
+    return lock
+
+
+class _FileLock(object):
+    """Internal file based lock object"""
+    def __init__(self, path):
+        self._path = path
+        self._file = None
+        self._threadlock = threading.Lock()
+
+    def _acquire_file(self):
+        if self._file is None:
+            self._file = open(self._path, "wb")
+        fcntl.flock(self._file, fcntl.LOCK_EX)
+
+    def _release_file(self):
+        fcntl.flock(self._file, fcntl.LOCK_UN)
+        self._file.close()
+        self._file = None
+
+    def acquire(self):
+        self._threadlock.acquire()
+        try:
+            self._acquire_file()
+        except Exception:  # pragma: no cover
+            self._threadlock.release()
+            raise
+
+    def release(self):
+        self._release_file()
+        self._threadlock.release()
+
+    def __enter__(self):
+        self.acquire()
+
+    def __exit__(self, *args):
+        self.release()
+
+
 def pid_exists(pid):
     try:
         os.kill(pid, 0)


Test patch:
diff --git a/skein/test/test_core.py b/skein/test/test_core.py
index e2529ab..24fc77f 100644
--- a/skein/test/test_core.py
+++ b/skein/test/test_core.py
@@ -3,7 +3,10 @@ from __future__ import print_function, division, absolute_import
 import os
 import subprocess
 import time
+import warnings
 import weakref
+from contextlib import closing
+from multiprocessing.pool import Pool
 
 import pytest
 
@@ -105,6 +108,27 @@ def test_security_auto_inits(skein_config):
     assert sec == sec2
 
 
+def _pool_initializer(config_dir):
+    skein.properties._mapping['config_dir'] = config_dir
+
+
+def _from_default_count_warnings(n):
+    with warnings.catch_warnings(record=True) as rec:
+        skein.Security.from_default()
+    return len(rec)
+
+
+def test_security_auto_inits_no_race_condition(tmpdir):
+    config_dir = str(tmpdir)
+    pool = Pool(processes=4, initializer=_pool_initializer,
+                initargs=(config_dir,))
+    with closing(pool):
+        num_warnings = sum(pool.map(_from_default_count_warnings, range(8)))
+    # Only one process actually writes (and raises a warning)
+    # all others just read in credentials
+    assert num_warnings == 1
+
+
 def test_client(security, kinit, tmpdir):
     logpath = str(tmpdir.join("log.txt"))
 
@@ -638,10 +662,15 @@ def test_node_locality(client, strict):
 def test_set_application_progress(client):
     with run_application(client) as app:
         app.set_progress(0.5)
-        # Give the allocate loop time to update
-        time.sleep(2)
-        report = client.application_report(app.id)
-        assert report.progress == 0.5
+        # The report won't update until the allocator loop runs. Try a couple
+        # times.
+        for _ in range(6):
+            report = client.application_report(app.id)
+            if report.progress == 0.5:
+                break
+            time.sleep(1)
+        else:
+            assert report.progress == 0.5
 
         with pytest.raises(ValueError):
             app.set_progress(-0.5)
diff --git a/skein/test/test_utils.py b/skein/test/test_utils.py
index 56efcdc..0bcf86a 100644
--- a/skein/test/test_utils.py
+++ b/skein/test/test_utils.py
@@ -1,9 +1,10 @@
 from __future__ import absolute_import, print_function, division
 
+import multiprocessing
 from datetime import timedelta
 
 from skein.utils import (humanize_timedelta, format_table,
-                         format_comma_separated_list)
+                         format_comma_separated_list, lock_file)
 
 
 def test_humanize_timedelta():
@@ -32,3 +33,69 @@ def test_format_comma_separated_list():
     assert format_comma_separated_list([None]) == 'None'
     assert format_comma_separated_list([None, False]) == 'None or False'
     assert format_comma_separated_list([None, False, True]) == 'None, False, or True'
+
+
+def test_lock_file_single_process(tmpdir):
+    path = str(tmpdir.join('lock'))
+    acquired = False
+    with lock_file(path):
+        acquired = True
+    assert acquired
+
+
+def test_lock_file_same_object_used_for_same_path(tmpdir):
+    path = str(tmpdir.join('lock'))
+    lock1 = lock_file(path)
+    lock2 = lock_file(path)
+    assert lock1 is lock2
+
+
+class Locker(multiprocessing.Process):
+    def __init__(self, path=None):
+        super(Locker, self).__init__()
+        self.path = path
+        self.daemon = True
+        self.parent_conn, self.child_conn = multiprocessing.Pipe()
+
+    def shutdown(self):
+        if not self.parent_conn.closed:
+            self.parent_conn.send('shutdown')
+            self.parent_conn.close()
+        self.join()
+
+    def run(self):
+        lock = lock_file(self.path)
+        while True:
+            msg = self.child_conn.recv()
+            if msg == 'acquire':
+                lock.acquire()
+                self.child_conn.send('acquired')
+            elif msg == 'release':
+                lock.release()
+                self.child_conn.send('released')
+            elif msg == 'shutdown':
+                break
+        self.child_conn.close()
+
+
+def test_lock_file_multiple_processes(tmpdir):
+    path = str(tmpdir.join('lock'))
+
+    lock = lock_file(path)
+
+    locker = Locker(path=path)
+    locker.start()
+    locker.parent_conn.send('acquire')
+    assert 'acquired' == locker.parent_conn.recv()
+    locker.parent_conn.send('release')
+    assert 'released' == locker.parent_conn.recv()
+
+    lock.acquire()
+    locker.parent_conn.send('acquire')
+    assert not locker.parent_conn.poll(0.1)
+    lock.release()
+    assert 'acquired' == locker.parent_conn.recv()
+    locker.parent_conn.send('release')
+    assert 'released' == locker.parent_conn.recv()
+    locker.shutdown()
+    locker.join()