tegwick
047a6a2d31
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1045b-933d-7aa1-b2db-104557487a79
59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Health-only check before a future authorized run; never reserves or executes calls."""
|
|
import argparse
|
|
import json
|
|
from urllib import error, request
|
|
from urllib.parse import urlsplit
|
|
|
|
|
|
class NoRedirect(request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
def probe(endpoint, timeout=5, opener=None):
|
|
parsed = urlsplit(endpoint)
|
|
if (parsed.scheme not in {'http', 'https'} or not parsed.hostname
|
|
or parsed.username or parsed.password or parsed.query or parsed.fragment
|
|
or parsed.path not in {'', '/'}):
|
|
raise ValueError('Use an HTTP(S) service origin without credentials, path or query')
|
|
if not 0 < timeout <= 15:
|
|
raise ValueError('Health timeout must be greater than zero and at most 15 seconds')
|
|
opener = opener or request.build_opener(NoRedirect())
|
|
result = {'endpoint': endpoint.rstrip('/'), 'ready': False,
|
|
'inference_calls': 0, 'attempts_reserved': 0,
|
|
'limit': 'Health only; profile, provider, token enforcement and authorization are unverified.'}
|
|
try:
|
|
req = request.Request(endpoint.rstrip('/') + '/health', method='GET')
|
|
with opener.open(req, timeout=timeout) as response:
|
|
body = response.read(4097)
|
|
if response.status != 200 or len(body) > 4096:
|
|
result['reason'] = 'invalid_health_response'
|
|
else:
|
|
data = json.loads(body)
|
|
result['ready'] = isinstance(data, dict) and data.get('status') == 'ok'
|
|
result['reason'] = 'health_ok' if result['ready'] else 'invalid_health_response'
|
|
except error.HTTPError as exc:
|
|
result.update(reason='health_http_error', http_status=exc.code)
|
|
except (error.URLError, TimeoutError, OSError):
|
|
result['reason'] = 'health_unreachable'
|
|
except (ValueError, UnicodeError):
|
|
result['reason'] = 'invalid_health_response'
|
|
return result
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--endpoint', default='http://127.0.0.1:18081')
|
|
parser.add_argument('--timeout', type=float, default=5)
|
|
args = parser.parse_args()
|
|
try:
|
|
result = probe(args.endpoint, args.timeout)
|
|
except ValueError as exc:
|
|
parser.error(str(exc))
|
|
print(json.dumps(result, indent=2))
|
|
return 0 if result['ready'] else 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|