#!/usr/bin/env python3
+"""File-based locking helper."""
+
import datetime
import json
import logging
cfg.add_argument(
'--lockfile_held_duration_warning_threshold_sec',
type=float,
- default=10.0,
+ default=60.0,
metavar='SECONDS',
help='If a lock is held for longer than this threshold we log a warning',
)
class LockFileException(Exception):
+ """An exception related to lock files."""
+
pass
@dataclass
class LockFileContents:
+ """The contents we'll write to each lock file."""
+
pid: int
commandline: str
expiration_timestamp: Optional[float]
expiration_timestamp: Optional[float] = None,
override_command: Optional[str] = None,
) -> None:
- self.is_locked = False
- self.lockfile = lockfile_path
- self.override_command = override_command
+ self.is_locked: bool = False
+ self.lockfile: str = lockfile_path
+ self.locktime: Optional[int] = None
+ self.override_command: Optional[str] = override_command
if do_signal_cleanup:
signal.signal(signal.SIGINT, self._signal)
signal.signal(signal.SIGTERM, self._signal)
return not os.path.exists(self.lockfile)
def try_acquire_lock_once(self) -> bool:
- logger.debug(f"Trying to acquire {self.lockfile}.")
+ logger.debug("Trying to acquire %s.", self.lockfile)
try:
# Attempt to create the lockfile. These flags cause
# os.open to raise an OSError if the file already
contents = self._get_lockfile_contents()
logger.debug(contents)
f.write(contents)
- logger.debug(f'Success; I own {self.lockfile}.')
+ logger.debug('Success; I own %s.', self.lockfile)
self.is_locked = True
return True
except OSError:
pass
- msg = f'Could not acquire {self.lockfile}.'
- logger.warning(msg)
+ logger.warning('Couldn\'t acquire %s.', self.lockfile)
return False
def acquire_with_retries(
logger.warning(msg)
raise LockFileException(msg)
- def __exit__(self, type, value, traceback):
+ def __exit__(self, _, value, traceback):
if self.locktime:
ts = datetime.datetime.now().timestamp()
duration = ts - self.locktime
line = lines[0]
line_dict = json.loads(line)
contents = LockFileContents(**line_dict)
- logger.debug(f'Blocking lock contents="{contents}"')
+ logger.debug('Blocking lock contents="%s"', contents)
# Does the PID exist still?
try: