commit 5d0d78cb521b00af2ca2d76c21cc7038cdc94c48
parent c45bd0b6e346f61b1fdb8e6fcfe030c5cf641832
Author: Asher Morgan <asher@ashermorgan.net>
Date: Thu, 20 Aug 2026 19:02:23 -0700
Teardown runners ~10m after last client connection
Diffstat:
| M | aoh/runner.py | | | 157 | ++++++++++++++++++++++++++++++++++++++++++++++++++----------------------------- |
| M | aoh/server.py | | | 32 | +++++++++++++++++++++++++++----- |
2 files changed, 126 insertions(+), 63 deletions(-)
diff --git a/aoh/runner.py b/aoh/runner.py
@@ -3,19 +3,28 @@ import os
import shutil
import tempfile
import time
+from threading import Lock
from uuid import uuid4
import ansible_runner
_CONNECTION_PLUGIN_DIR = f'{os.path.dirname(__file__)}/connection_plugins/'
+_CLIENT_TIMEOUT = 600 # 10 minutes
class AoHError(Exception):
- """Raised for AoC-specific errors."""
+ """Raised for AoH-specific errors."""
class AoHRunner:
+ """An Ansible-over-HTTP playbook runner."""
+
def __init__(self, config, playbook, host, args):
+ """Create a new AoH runner."""
+
self.id = str(uuid4())
+
+ self._lock = Lock() # Used to protect all public methods
+
self._dir = tempfile.mkdtemp(prefix='aoh-')
self._LOGS_PATH = f'{self._dir}/artifacts/{self.id}/stdout'
self._RECVBUF_PATH = f'{self._dir}/recvbuf'
@@ -23,21 +32,18 @@ class AoHRunner:
self._sendbuf = None
self._recvbuf = None
self._logs = None
+
self._thread = None
self._runner = None
+ self._last_req = time.time()
+
try:
# Create hostname file so connection plugin can verify hosts
with open(f'{self._dir}/hostname', 'w') as f:
f.write(f'{host}\n')
- # Set ansible_connection=aoh for the client host only. Note that we
- # don't use ansible-runners inventory directory because that will
- # shadow user inventory files.
- with open(f'{self._dir}/inventory.ini', 'w') as f:
- f.write(f'[aoh]\n{host} ansible_connection=aoh')
-
- self._start_runner(config, playbook, args)
+ self._start_runner(config, playbook, host, args)
# We assume that ansible-runner will eventually create its log file
while not os.path.exists(self._LOGS_PATH):
@@ -58,7 +64,15 @@ class AoHRunner:
raise
- def _start_runner(self, config, playbook, args):
+ def _start_runner(self, config, playbook, host, args):
+ """Configure and start the underlying ansible-runner."""
+
+ # Set ansible_connection=aoh for the client host only. Note that we
+ # don't use ansible-runners inventory directory because that will
+ # shadow user inventory files.
+ with open(f'{self._dir}/inventory.ini', 'w') as f:
+ f.write(f'[aoh]\n{host} ansible_connection=aoh')
+
raw_config = ansible_runner.get_ansible_config(
'dump',
config,
@@ -104,65 +118,92 @@ class AoHRunner:
def _runner_finished(self):
+ """Check whether the underlying ansible runner has finished."""
+
assert self._runner is not None
return self._runner.status not in ['starting', 'running']
+ def has_timed_out(self):
+ """Check whether the AoH runner has timed out."""
+
+ with self._lock:
+ return self._last_req < time.time() - _CLIENT_TIMEOUT
+
+
def process_client_request(self, req):
+ """Process an AoH client request."""
+
res = {}
- # We check runner status first, in case new logs come in afterwards
- res['finished'] = self._runner_finished()
- assert self._logs is not None
- res['logs'] = self._logs.read()
+ with self._lock:
+ if self.id is None:
+ # Runner has already been torn down
+ return {
+ 'finished': True,
+ 'err': 'Client timeout',
+ }
+
+ self._last_req = time.time()
+
+ # We check runner status first, in case new logs come in afterwards
+ res['finished'] = self._runner_finished()
+ assert self._logs is not None
+ res['logs'] = self._logs.read()
- if req and self._recvbuf:
- try:
- self._recvbuf.write(json.dumps(req) + '\n')
- self._recvbuf.flush()
- except BrokenPipeError:
- # Ansible should exit soon on its own
- pass
+ if req and self._recvbuf:
+ try:
+ self._recvbuf.write(json.dumps(req) + '\n')
+ self._recvbuf.flush()
+ except BrokenPipeError:
+ # Ansible should exit soon on its own
+ pass
- if self._sendbuf:
- line = self._sendbuf.readline()
- for key, val in json.loads(line or '{}').items():
- res[key] = val
+ if self._sendbuf:
+ line = self._sendbuf.readline()
+ for key, val in json.loads(line or '{}').items():
+ res[key] = val
- return res
+ return res
def teardown(self):
- self.id = None
-
- if self._logs:
- self._logs.close()
-
- if self._recvbuf:
- try:
- # Send an error in case the AoH connection plugin is currently
- # blocking on a read
- self._recvbuf.write('{"err":"Runner teardown"}\n')
- self._recvbuf.flush()
- except BrokenPipeError:
- pass
-
- try:
- self._recvbuf.close()
- except BrokenPipeError:
- pass
-
- if self._sendbuf:
- try:
- self._sendbuf.close()
- except BrokenPipeError:
- pass
-
- if self._thread:
- # The Ansible runner *should* exit if it hasn't already due to a
- # broken recvbuf pipe or the sendbuf error.
- self._thread.join(10)
- if self._thread.is_alive():
- raise AoHError('Ansible runner thread not terminated')
-
- shutil.rmtree(self._dir)
+ """Terminate the AoH runner and free its resources."""
+
+ with self._lock:
+ if self.id is None:
+ # Runner has already been torn down
+ return
+ self.id = None
+
+ if self._logs:
+ self._logs.close()
+
+ if self._recvbuf:
+ try:
+ # Send an error in case the AoH connection plugin is currently
+ # blocking on a read
+ self._recvbuf.write('{"err":"Runner teardown"}\n')
+ self._recvbuf.flush()
+ except BrokenPipeError:
+ pass
+
+ try:
+ self._recvbuf.close()
+ except BrokenPipeError:
+ pass
+
+ if self._sendbuf:
+ try:
+ self._sendbuf.close()
+ except BrokenPipeError:
+ pass
+
+ if self._thread:
+ # The Ansible runner *should* exit if it hasn't already due to a
+ # broken recvbuf pipe or the sendbuf error.
+ self._thread.join(10)
+ if self._thread.is_alive():
+ raise AoHError('Ansible runner thread not terminated')
+
+ shutil.rmtree(self._dir)
diff --git a/aoh/server.py b/aoh/server.py
@@ -2,13 +2,31 @@ import os
import yaml
from flask import Flask, abort, make_response, render_template, request, session
+from flask_apscheduler import APScheduler
from .runner import AoHRunner
+_RUNNERS = {}
+_LAST_GC = 0
+_GC_INTERVAL = 60 # 1 minute
+
app = Flask(__name__, template_folder=os.path.dirname(__file__))
app.secret_key = os.urandom(16)
-_DATA = {}
+scheduler = APScheduler()
+scheduler.init_app(app)
+scheduler.start()
+
+
+@scheduler.task('interval', seconds=_GC_INTERVAL)
+def gc():
+ """Teardown and delete timed-out runners."""
+
+ runners = list(_RUNNERS.items())
+ for id, runner in runners:
+ if runner and runner.has_timed_out():
+ runner.teardown()
+ _RUNNERS.pop(id, None)
def _get_opts(cmdline):
@@ -72,7 +90,7 @@ def job_new():
runner = AoHRunner(playbook['config'], playbook['playbook'],
request.json['host'], request.json['args'])
- _DATA[runner.id] = runner
+ _RUNNERS[runner.id] = runner
session['runner'] = runner.id
return make_response('', 201, {'Location': f'/runners/{runner.id}'})
@@ -83,10 +101,14 @@ def job_status(id):
if id != session.get('runner'):
abort(401)
- res = _DATA[id].process_client_request(request.json)
+ runner = _RUNNERS.get(id)
+ if not runner:
+ abort(400)
+
+ res = runner.process_client_request(request.json)
if res['finished']:
- _DATA[id].teardown()
- del _DATA[id]
+ runner.teardown()
+ _RUNNERS.pop(id, None)
return res
