476 lines
28 KiB
Python
476 lines
28 KiB
Python
"""Internal TEST executor. Receives one bounded task action from Jenkins.
|
|
|
|
Never edits business SQL. Docker configuration is captured before the scoped
|
|
fault and restored to the same image, mounts, ports and network identity.
|
|
Only the original DEAD event can be retried through its protected HTTP API.
|
|
"""
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import http.client
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
|
|
class DockerConnection(http.client.HTTPConnection):
|
|
def connect(self):
|
|
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.sock.settimeout(self.timeout)
|
|
self.sock.connect('/var/run/docker.sock')
|
|
|
|
|
|
def docker(method, path, body=None):
|
|
conn = DockerConnection('localhost', timeout=20)
|
|
data = None if body is None else json.dumps(body).encode()
|
|
conn.request(method, path, data, {'Content-Type': 'application/json'})
|
|
response = conn.getresponse()
|
|
raw = response.read()
|
|
conn.close()
|
|
if response.status >= 300:
|
|
raise RuntimeError('DOCKER_' + str(response.status))
|
|
return json.loads(raw) if raw else None
|
|
|
|
|
|
def inspect(name):
|
|
return docker('GET', '/containers/' + urllib.parse.quote(name, safe='') + '/json')
|
|
|
|
|
|
def env_of(container):
|
|
return dict(item.split('=', 1) for item in container['Config']['Env'])
|
|
|
|
|
|
def require_test(container, name):
|
|
if container['Name'] != '/' + name or container['Config']['Labels'].get('com.docker.compose.project') != 'firstleg-test':
|
|
raise RuntimeError('NOT_TEST_CONTAINER')
|
|
|
|
|
|
def recreation(container, environment, full=False):
|
|
body = copy.deepcopy(container['Config'])
|
|
body['Env'] = [key + '=' + value for key, value in environment.items()]
|
|
body['HostConfig'] = copy.deepcopy(container['HostConfig'])
|
|
if body['HostConfig'].get('AutoRemove') or any(mount['Type'] not in ('bind', 'volume') for mount in container['Mounts']):
|
|
raise RuntimeError('UNSUPPORTED_CONTAINER_STORAGE')
|
|
# Preserve named volumes explicitly, never create a replacement anonymous volume.
|
|
if any(mount['Type'] == 'volume' and not mount.get('Name') for mount in container['Mounts']):
|
|
raise RuntimeError('UNNAMED_VOLUME')
|
|
if body['HostConfig'].get('VolumesFrom'):
|
|
raise RuntimeError('UNSUPPORTED_VOLUME_REFERENCE')
|
|
endpoints = {}
|
|
for name, network in container['NetworkSettings']['Networks'].items():
|
|
aliases = [alias for alias in network.get('Aliases') or [] if alias != container['Id'][:12]]
|
|
endpoints[name] = {'Aliases': aliases, 'IPAMConfig': {'IPv4Address': network['IPAddress']}}
|
|
if full:
|
|
ipam = copy.deepcopy(network.get('IPAMConfig') or {})
|
|
ipam['IPv4Address'] = network['IPAddress']
|
|
if network.get('GlobalIPv6Address'):
|
|
ipam['IPv6Address'] = network['GlobalIPv6Address']
|
|
endpoints[name]['IPAMConfig'] = ipam
|
|
for key in ('Links', 'DriverOpts', 'GwPriority', 'MacAddress'):
|
|
if network.get(key) is not None:
|
|
endpoints[name][key] = copy.deepcopy(network[key])
|
|
body['NetworkingConfig'] = {'EndpointsConfig': endpoints}
|
|
return body
|
|
|
|
|
|
def replace(container, body):
|
|
name = container['Name'].lstrip('/')
|
|
current = inspect(name)
|
|
if current['Id'] != container['Id']:
|
|
raise RuntimeError('CONCURRENT_CONTAINER_CHANGE')
|
|
docker('POST', '/containers/' + current['Id'] + '/stop?t=10')
|
|
docker('DELETE', '/containers/' + current['Id'] + '?v=false')
|
|
created = docker('POST', '/containers/create?name=' + name, body)
|
|
docker('POST', '/containers/' + created['Id'] + '/start')
|
|
return created['Id']
|
|
|
|
|
|
def private_write(path, value):
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, 'w') as file:
|
|
json.dump(value, file)
|
|
|
|
|
|
def scope_directory(scope, direction='LOADING'):
|
|
if direction not in ('LOADING', 'UNLOAD', 'DELIVERY'):
|
|
raise RuntimeError('INVALID_DIRECTION')
|
|
marker = scope.get('marker', '')
|
|
if not re.fullmatch(r'[A-Za-z0-9_-]{1,96}', marker):
|
|
raise RuntimeError('INVALID_SCOPE')
|
|
keys = ('customerId', 'deliveryId', 'sourceInboundId', 'sourceOrderId', 'rootOrderId') if direction == 'DELIVERY' else ('customerId', 'containerBookingId') if direction == 'UNLOAD' else ('customerId', 'loadingPlanId', 'containerBookingId')
|
|
if direction != 'LOADING' and set(scope) != {'marker', *keys}:
|
|
raise RuntimeError('INVALID_SCOPE')
|
|
for key in keys:
|
|
if not re.fullmatch(r'[1-9][0-9]{0,19}', scope.get(key, '')):
|
|
raise RuntimeError('INVALID_SCOPE')
|
|
identity = direction + ':' + marker if direction != 'LOADING' else marker
|
|
return Path(tempfile.gettempdir()) / ('chongchong-event-' + hashlib.sha256(identity.encode()).hexdigest()[:24])
|
|
|
|
|
|
def unload(state):
|
|
return state.get('direction', 'LOADING') == 'UNLOAD'
|
|
|
|
|
|
def delivery(state):
|
|
return state.get('direction') == 'DELIVERY'
|
|
|
|
|
|
def reverse_delivery(state):
|
|
return state.get('direction') in ('UNLOAD', 'DELIVERY')
|
|
|
|
|
|
def producer_name(state):
|
|
return 'firstleg-test-lastleg' if reverse_delivery(state) else 'firstleg-test-business'
|
|
|
|
|
|
def target_setting(state):
|
|
return ('FIRSTLEG_SERVICE_URL', 'http://firstleg:9202') if reverse_delivery(state) else ('LASTLEG_SERVICE_URL', 'http://lastleg-service:9303')
|
|
|
|
|
|
def fingerprint(container):
|
|
body = recreation(container, env_of(container), full=True)
|
|
# Docker may add the legacy Config.MacAddress after explicit per-network MAC
|
|
# restoration. It is redundant only when it equals an actual preserved MAC;
|
|
# every network MAC remains in the fingerprint and unrelated values still fail.
|
|
network_macs = {endpoint.get('MacAddress') for endpoint in body['NetworkingConfig']['EndpointsConfig'].values()}
|
|
if body.get('MacAddress') and body['MacAddress'] in network_macs:
|
|
body.pop('MacAddress')
|
|
body['Env'].sort()
|
|
for endpoint in body['NetworkingConfig']['EndpointsConfig'].values():
|
|
endpoint['Aliases'].sort()
|
|
body['ResolvedMounts'] = sorted(container['Mounts'], key=lambda mount: mount['Destination'])
|
|
body['NetworkIds'] = {name: network.get('NetworkID') for name, network in container['NetworkSettings']['Networks'].items()}
|
|
return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
|
|
|
|
|
def control(state, mode=None):
|
|
url = 'http://127.0.0.1:' + str(state['controlPort']) + ('/mode' if mode else '/status')
|
|
request = urllib.request.Request(url, None if mode is None else json.dumps({'mode': mode}).encode(),
|
|
{'X-Event-Fault-Control-Token': state['controlToken'], 'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def read_sql(container, sql):
|
|
# SQL is generated here from fixed templates and decimal/UUID/hash identities.
|
|
if re.search(r'\b(INSERT|UPDATE|DELETE|ALTER|DROP|TRUNCATE|CALL|INTO|OUTFILE)\b', sql, re.I):
|
|
raise RuntimeError('NON_READ_ONLY_SQL')
|
|
e = env_of(container)
|
|
prefix = {'/firstleg-test-business': 'FIRSTLEG', '/firstleg-test-lastleg': 'LASTLEG', '/firstleg-test-order': 'ORDER'}[container['Name']]
|
|
url = urllib.parse.urlsplit(e[prefix + '_DB_URL'].removeprefix('jdbc:'))
|
|
host = '127.0.0.1' if url.hostname == 'host.docker.internal' else url.hostname
|
|
transaction = 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;\n' + sql + '\nCOMMIT;'
|
|
result = subprocess.run(['mysql', '--host=' + host, '--port=' + str(url.port or 3306),
|
|
'--user=' + e[prefix + '_DB_USERNAME'], '--batch', '--raw', '--skip-column-names', url.path.lstrip('/')],
|
|
input=transaction, text=True, capture_output=True, timeout=20,
|
|
env=dict(os.environ, MYSQL_PWD=e[prefix + '_DB_PASSWORD']))
|
|
if result.returncode:
|
|
code = re.search(r'ERROR (\d+)', result.stderr)
|
|
raise RuntimeError('MYSQL_' + (code.group(1) if code else 'UNKNOWN'))
|
|
return [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
|
|
|
|
|
|
def producer(scope, direction='LOADING'):
|
|
if direction == 'DELIVERY':
|
|
rows = read_sql(inspect('firstleg-test-lastleg'),
|
|
"SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payloadHash')),'eventType',event_type,'customerId',customer_id,'aggregateId',aggregate_id,'aggregateType',aggregate_type,'eventKey',event_key,'envelope',JSON_EXTRACT(payload_json,'$'),'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE customer_id='" + scope['customerId'] + "' AND event_type='OverseasDeliveryCompleted' AND JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payload.deliveryOrderId'))='" + scope['deliveryId'] + "';")
|
|
elif direction == 'UNLOAD':
|
|
rows = read_sql(inspect('firstleg-test-lastleg'),
|
|
"SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payloadHash')),'eventType',event_type,'customerId',customer_id,'aggregateId',aggregate_id,'aggregateType',aggregate_type,'eventKey',event_key,'envelope',JSON_EXTRACT(payload_json,'$'),'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE customer_id='0' AND aggregate_id='"
|
|
+ scope['containerBookingId'] + "' AND event_type='OverseasContainerUnloaded' AND event_key='OverseasContainerUnloaded:" + scope['containerBookingId'] + "';")
|
|
else:
|
|
rows = read_sql(inspect('firstleg-test-business'),
|
|
"SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',payload_hash,'eventType',event_type,'publishStatus',publish_status,'failedAttempts',retry_count) FROM fl_inbound_outbox WHERE tenant_id=" + scope['customerId']
|
|
+ " AND aggregate_id=" + scope['loadingPlanId'] + " AND event_type='FirstLegLoadingPlanCompleted';")
|
|
if len(rows) > 1:
|
|
raise RuntimeError('AMBIGUOUS_EVENT')
|
|
return rows[0] if rows else None
|
|
|
|
|
|
def container_facts(scope):
|
|
rows = read_sql(inspect('firstleg-test-business'),
|
|
"SELECT JSON_OBJECT('id',CAST(id AS CHAR),'containerNo',container_no,'version',CAST(version AS CHAR),'unloadingCompletedAt',CAST(unloading_completed_at AS CHAR),'unloadingPhotoUploadedAt',CAST(unloading_photo_uploaded_at AS CHAR),'estimatedEmptyReportedAt',CAST(estimated_empty_reported_at AS CHAR),'actualEmptyReportedAt',CAST(actual_empty_reported_at AS CHAR),'emptyReportSourceVersion',CAST(empty_report_source_version AS CHAR),'updatedAt',CAST(updated_at AS CHAR),'updatedBy',CAST(updated_by AS CHAR),'updatedByName',updated_by_name) FROM fl_container_booking WHERE id=" + scope['containerBookingId'] + ";")
|
|
if len(rows) != 1:
|
|
raise RuntimeError('CONTAINER_SCOPE_MISMATCH')
|
|
return rows[0]
|
|
|
|
|
|
def business_ids(scope):
|
|
rows = read_sql(inspect('firstleg-test-lastleg'), "SELECT JSON_OBJECT('id',CAST(id AS CHAR)) FROM ll_order WHERE customer_id='"
|
|
+ scope['customerId'] + "' AND source_container_booking_id='" + scope['containerBookingId'] + "' ORDER BY id;")
|
|
return [row['id'] for row in rows]
|
|
|
|
|
|
def business_snapshot(scope, state, row):
|
|
if delivery(state):
|
|
return {'business': delivery_facts(scope, row['eventId'])}
|
|
return {'container': container_facts(scope)} if unload(state) else {'businessOrderIds': business_ids(scope)}
|
|
|
|
|
|
def delivery_facts(scope, event_id):
|
|
if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id):
|
|
raise RuntimeError('INVALID_EVENT_IDENTITY')
|
|
customer, source, root = scope['customerId'], scope['sourceOrderId'], scope['rootOrderId']
|
|
firstleg = inspect('firstleg-test-business')
|
|
order = inspect('firstleg-test-order')
|
|
require_test(firstleg, 'firstleg-test-business')
|
|
require_test(order, 'firstleg-test-order')
|
|
# Each service is read in its own consistent read-only transaction; no cross-schema join.
|
|
inbound_sql = "tenant_id=" + customer + " AND deleted=0 AND order_id IN (" + source + "," + root + ")"
|
|
first_rows = read_sql(firstleg,
|
|
"SELECT JSON_OBJECT('kind','inbound','id',CAST(id AS CHAR),'customerId',CAST(tenant_id AS CHAR),'orderId',CAST(order_id AS CHAR),'orderNo',order_no,'inboundCode',inbound_code,'processStatus',process_status,'orderStatusSnapshot',order_status_snapshot,'version',version,'updatedAt',CAST(updated_at AS CHAR)) FROM fl_inbound_order WHERE " + inbound_sql + " ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','inboundHistory','id',CAST(id AS CHAR),'inboundId',CAST(inbound_id AS CHAR),'action',action,'fromStatus',from_status,'toStatus',to_status,'occurredAt',CAST(occurred_at AS CHAR)) FROM fl_inbound_status_history WHERE tenant_id=" + customer + " AND inbound_id=" + scope['sourceInboundId'] + " ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','inbox','eventId',event_id,'eventType',event_type,'aggregateId',aggregate_id,'aggregateVersion',aggregate_version,'payloadHash',payload_hash,'payload',JSON_EXTRACT(payload_json,'$'),'status',status,'receivedAt',CAST(received_at AS CHAR),'processedAt',CAST(processed_at AS CHAR)) FROM fl_container_event_inbox WHERE event_id='" + event_id + "';")
|
|
tree = "tenant_id=" + customer + " AND deleted=0 AND (id=" + root + " OR root_order_id=" + root + ")"
|
|
ids = "SELECT id FROM ord_order WHERE " + tree
|
|
order_rows = read_sql(order,
|
|
"SELECT JSON_OBJECT('kind','order','id',CAST(id AS CHAR),'customerId',CAST(tenant_id AS CHAR),'rootOrderId',CAST(root_order_id AS CHAR),'orderRole',order_role,'orderNo',order_no,'inboundCode',inbound_code,'status',status,'version',version,'completedAt',CAST(completed_at AS CHAR),'updatedAt',CAST(updated_at AS CHAR)) FROM ord_order WHERE " + tree + " ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','orderHistory','id',CAST(id AS CHAR),'orderId',CAST(order_id AS CHAR),'action',action,'fromStatus',from_status,'toStatus',to_status,'occurredAt',CAST(occurred_at AS CHAR)) FROM ord_order_status_history WHERE tenant_id=" + customer + " AND order_id IN (" + ids + ") ORDER BY id;\n"
|
|
+ "SELECT JSON_OBJECT('kind','completionEvent','id',CAST(id AS CHAR),'orderId',CAST(aggregate_id AS CHAR),'eventType',event_type,'status',publish_status,'payload',JSON_EXTRACT(payload_json,'$')) FROM ord_outbox_event WHERE tenant_id=" + customer + " AND aggregate_id IN (" + ids + ") AND event_type='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' ORDER BY id;")
|
|
return {kind: [row for row in first_rows + order_rows if row['kind'] == kind] for kind in
|
|
('inbound', 'inboundHistory', 'inbox', 'order', 'orderHistory', 'completionEvent')}
|
|
|
|
|
|
def prepare(scope, directory, config):
|
|
if delivery(config) and producer(scope, 'DELIVERY') is not None:
|
|
raise RuntimeError('DELIVERY_EVENT_ALREADY_EXISTS')
|
|
directory.mkdir(mode=0o700)
|
|
original = inspect(producer_name(config))
|
|
require_test(original, producer_name(config))
|
|
consumer_name = 'firstleg-test-business' if reverse_delivery(config) else 'firstleg-test-lastleg'
|
|
consumer = inspect(consumer_name)
|
|
require_test(consumer, consumer_name)
|
|
old_env = env_of(original)
|
|
setting, direct = target_setting(config)
|
|
if old_env.get(setting) != direct:
|
|
raise RuntimeError('PRODUCER_NOT_DIRECT')
|
|
original_body = recreation(original, old_env, full=reverse_delivery(config))
|
|
token = secrets.token_urlsafe(32)
|
|
# Reserve task-specific ports and let the existing proxy select exactly this plan.
|
|
sockets = [socket.socket() for _ in range(2)]
|
|
for sock in sockets:
|
|
sock.bind(('127.0.0.1', 0))
|
|
ports = [sock.getsockname()[1] for sock in sockets]
|
|
for sock in sockets:
|
|
sock.close()
|
|
proxy = directory / 'proxy.py'
|
|
proxy.write_bytes(base64.b64decode(config['proxySource']))
|
|
ip = consumer['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress']
|
|
log = open(directory / 'proxy-output.log', 'wb')
|
|
upstream = ':9202/internal/firstleg/v1/container-events' if reverse_delivery(config) else ':9303/internal/lastleg/events'
|
|
selector = [] if delivery(config) else ['--event-type', 'OverseasContainerUnloaded', '--container-booking-id', scope['containerBookingId']] if unload(config) else ['--event-type', 'FirstLegLoadingPlanCompleted', '--loading-plan-id', scope['loadingPlanId']]
|
|
if delivery(config):
|
|
selector = ['--event-type', 'OverseasDeliveryCompleted', '--delivery-id', scope['deliveryId']]
|
|
process = subprocess.Popen(['python3', str(proxy), '--upstream', 'http://' + ip + upstream, *selector,
|
|
'--bind', '0.0.0.0', '--port', str(ports[0]), '--control-port', str(ports[1]),
|
|
'--log', str(directory / 'proxy-evidence.jsonl')],
|
|
env=dict(os.environ, EVENT_FAULT_CONTROL_TOKEN=token), stdout=log, stderr=log, start_new_session=True)
|
|
state = {'scope': scope, 'originalBody': original_body, 'originalId': original['Id'], 'image': original['Config']['Image'],
|
|
'direction': config.get('direction', 'LOADING'), 'originalImageId': original['Image'], 'originalFingerprint': fingerprint(original),
|
|
'proxyPid': process.pid, 'proxyPort': ports[0], 'controlPort': ports[1], 'controlToken': token,
|
|
'operationsRequestId': str(uuid.uuid4()), 'prepared': False, 'restored': False}
|
|
private_write(directory / 'state.json', state)
|
|
return {'created': True, 'originalImageId': state['originalImageId'], 'originalFingerprint': state['originalFingerprint']}
|
|
|
|
|
|
def activate(scope, directory, state):
|
|
snapshot = control(state, 'FAULT')
|
|
if snapshot.get('eventId'):
|
|
raise RuntimeError('EVENT_ALREADY_PINNED')
|
|
original = inspect(producer_name(state))
|
|
require_test(original, producer_name(state))
|
|
setting, direct = target_setting(state)
|
|
if original['Id'] != state['originalId'] or original['Config']['Image'] != state['image'] or env_of(original).get(setting) != direct:
|
|
raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
|
|
if reverse_delivery(state) and (original['Image'] != state['originalImageId'] or fingerprint(original) != state['originalFingerprint']):
|
|
raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
|
|
body = copy.deepcopy(state['originalBody'])
|
|
e = env_of(original)
|
|
e[setting] = 'http://host.docker.internal:' + str(state['proxyPort'])
|
|
e.update(EVENT_MAX_ATTEMPTS='12', EVENT_BASE_DELAY_SECONDS='5', EVENT_MAX_DELAY_SECONDS='5', EVENT_JITTER='0')
|
|
if len(e.get('EVENT_OPERATIONS_TOKEN', '')) < 32:
|
|
e['EVENT_OPERATIONS_TOKEN'] = secrets.token_urlsafe(32)
|
|
body['Env'] = [key + '=' + value for key, value in e.items()]
|
|
state['prepared'] = True
|
|
private_write(directory / 'state.json', state)
|
|
replace(original, body)
|
|
return {'activated': True}
|
|
|
|
|
|
def recover(scope, directory, state):
|
|
row = producer(scope, state.get('direction', 'LOADING'))
|
|
if not row or row['publishStatus'] != 'DEAD' or row['failedAttempts'] != 12:
|
|
raise RuntimeError('ORIGINAL_EVENT_NOT_DEAD')
|
|
pinned = control(state)
|
|
if pinned['eventId'] != row['eventId'] or pinned['payloadHash'] != row['payloadHash']:
|
|
raise RuntimeError('PINNED_EVENT_MISMATCH')
|
|
state['beforeRetry'] = {**row, **business_snapshot(scope, state, row)}
|
|
private_write(directory / 'state.json', state)
|
|
control(state, 'PASS')
|
|
container = inspect(producer_name(state))
|
|
e = env_of(container)
|
|
ip = container['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress']
|
|
body = {'requestId': state['operationsRequestId'], 'operator': 'admin-full-test',
|
|
'reason': scope['marker'] + ': scoped ACK-loss recovery', 'expectedPayloadHash': row['payloadHash']}
|
|
endpoint = ':9303/internal/lastleg/event-operations/' if reverse_delivery(state) else ':9202/internal/firstleg/event-operations/'
|
|
url = 'http://' + ip + endpoint + row['entryId'] + '/retry'
|
|
request = urllib.request.Request(url, json.dumps(body).encode(),
|
|
{'X-Event-Operations-Token': e['EVENT_OPERATIONS_TOKEN'], 'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
result = json.load(response)
|
|
if result.get('code') != '200':
|
|
raise RuntimeError('OPERATIONS_RETRY_REJECTED')
|
|
return {'retrySubmitted': True}
|
|
|
|
|
|
def restore(directory, state):
|
|
if state['restored']:
|
|
return {'restored': True}
|
|
try:
|
|
control(state, 'PASS')
|
|
except Exception:
|
|
if not reverse_delivery(state):
|
|
raise
|
|
# A dead proxy must not prevent restoring the producer's direct target.
|
|
if state['prepared']:
|
|
try:
|
|
current = inspect(producer_name(state))
|
|
except RuntimeError as error:
|
|
if str(error) != 'DOCKER_404':
|
|
raise
|
|
created = docker('POST', '/containers/create?name=' + producer_name(state), state['originalBody'])
|
|
docker('POST', '/containers/' + created['Id'] + '/start')
|
|
state['prepared'] = False
|
|
private_write(directory / 'state.json', state)
|
|
return {'restoreStarted': True}
|
|
if current['Id'] == state['originalId'] or (reverse_delivery(state) and current['Image'] == state['originalImageId'] and fingerprint(current) == state['originalFingerprint']):
|
|
if reverse_delivery(state) and not current['State'].get('Running'):
|
|
docker('POST', '/containers/' + current['Id'] + '/start')
|
|
state['prepared'] = False
|
|
private_write(directory / 'state.json', state)
|
|
return {'restoreStarted': True}
|
|
if current['Config']['Image'] != state['image'] or env_of(current).get(target_setting(state)[0]) != 'http://host.docker.internal:' + str(state['proxyPort']):
|
|
raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
|
|
replace(current, state['originalBody'])
|
|
state['prepared'] = False
|
|
private_write(directory / 'state.json', state)
|
|
return {'restoreStarted': True}
|
|
|
|
|
|
def proof(scope, state):
|
|
row = producer(scope, state.get('direction', 'LOADING'))
|
|
if not row or row['publishStatus'] != 'PUBLISHED' or not state.get('beforeRetry'):
|
|
raise RuntimeError('RECOVERY_NOT_PUBLISHED')
|
|
event_id, entry_id = row['eventId'], row['entryId']
|
|
if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id) or not re.fullmatch(r'[1-9][0-9]{0,19}', entry_id):
|
|
raise RuntimeError('INVALID_EVENT_IDENTITY')
|
|
event_type = 'OverseasDeliveryCompleted' if delivery(state) else 'OverseasContainerUnloaded' if unload(state) else 'FirstLegLoadingPlanCompleted'
|
|
attempts = read_sql(inspect(producer_name(state)),
|
|
"SELECT JSON_OBJECT('attemptId',attempt_id,'serviceName',service_name,'entryId',entry_id,'eventId',event_id,'targetName',target_name,'outcome',outcome) FROM event_delivery_attempt WHERE entry_id='"
|
|
+ entry_id + "' AND event_type='" + event_type + "' ORDER BY started_at,attempt_id;")
|
|
audit = read_sql(inspect(producer_name(state)),
|
|
"SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE request_id='"
|
|
+ state['operationsRequestId'] + "';" if not reverse_delivery(state) else
|
|
"SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE entry_id='" + entry_id + "' AND event_type='" + event_type + "';")
|
|
if reverse_delivery(state):
|
|
consumers = read_sql(inspect('firstleg-test-business'),
|
|
"SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status,'eventType',event_type,'aggregateId',aggregate_id,'payload',JSON_EXTRACT(payload_json,'$')) FROM fl_container_event_inbox WHERE event_id='" + event_id + "';")
|
|
else:
|
|
consumers = read_sql(inspect('firstleg-test-lastleg'),
|
|
"SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status) FROM ll_loading_event_inbox WHERE event_id='"
|
|
+ event_id + "' AND customer_id='" + scope['customerId'] + "';")
|
|
consumer = {**(consumers[0] if len(consumers) == 1 else {}), 'count': len(consumers)}
|
|
result = {**scope, 'readOnly': True, 'producer': row, 'consumer': consumer, 'attempts': attempts,
|
|
'proxyRecords': control(state)['records'], 'beforeRetry': state['beforeRetry'],
|
|
'afterRetry': {**row, **business_snapshot(scope, state, row)},
|
|
'retryAudit': audit[0] if len(audit) == 1 else None, 'retryAuditCount': len(audit), 'retryRequestId': state['operationsRequestId'], 'restored': False}
|
|
private_write(scope_directory(scope, state.get('direction', 'LOADING')) / 'proof.json', result)
|
|
return result
|
|
|
|
|
|
def finish(directory, state):
|
|
if state['restored']:
|
|
return state.get('restoration', {'restored': True})
|
|
current = inspect(producer_name(state))
|
|
require_test(current, producer_name(state))
|
|
if current['State'].get('Health', {}).get('Status') != 'healthy' or current['Config']['Image'] != state['image']:
|
|
raise RuntimeError('RESTORED_SERVICE_NOT_HEALTHY')
|
|
if sorted(current['Config']['Env']) != sorted(state['originalBody']['Env']):
|
|
raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH')
|
|
if reverse_delivery(state) and (current['Image'] != state['originalImageId'] or fingerprint(current) != state['originalFingerprint']):
|
|
raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH')
|
|
cmdline = Path('/proc/' + str(state['proxyPid']) + '/cmdline')
|
|
if cmdline.exists():
|
|
if str(directory / 'proxy.py').encode() not in cmdline.read_bytes().split(b'\x00'):
|
|
raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH')
|
|
os.kill(state['proxyPid'], signal.SIGTERM)
|
|
if reverse_delivery(state):
|
|
for _ in range(50):
|
|
if not cmdline.exists() or not cmdline.read_bytes():
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
raise RuntimeError('PROXY_NOT_STOPPED')
|
|
result = {'restored': True}
|
|
if reverse_delivery(state):
|
|
result.update(healthy=True, proxyStopped=True, producerName=producer_name(state),
|
|
originalImageId=state['originalImageId'], currentImageId=current['Image'],
|
|
originalFingerprint=state['originalFingerprint'], currentFingerprint=fingerprint(current))
|
|
state['restoration'] = result
|
|
state['restored'] = True
|
|
# Proof is already captured. Remove credentials from the retained task metadata.
|
|
state.pop('originalBody', None)
|
|
state.pop('controlToken', None)
|
|
private_write(directory / 'state.json', state)
|
|
return result
|
|
|
|
|
|
def main(config):
|
|
scope, action = config['scope'], config['action']
|
|
direction = config.get('direction', 'LOADING')
|
|
directory = scope_directory(scope, direction)
|
|
if action == 'prepare':
|
|
return prepare(scope, directory, config)
|
|
state = json.loads((directory / 'state.json').read_text())
|
|
if state['scope'] != scope or state.get('direction', 'LOADING') != direction:
|
|
raise RuntimeError('TASK_SCOPE_MISMATCH')
|
|
if action == 'activate':
|
|
return activate(scope, directory, state)
|
|
if action == 'health':
|
|
current = inspect(producer_name(state))
|
|
return {'healthy': current['State'].get('Health', {}).get('Status') == 'healthy'}
|
|
if action == 'proxy-ready':
|
|
return {'ready': control(state)['mode'] == 'PASS'}
|
|
if action == 'status':
|
|
return {'producer': producer(scope, direction), 'proxy': control(state)}
|
|
if action == 'recover':
|
|
return recover(scope, directory, state)
|
|
if action == 'proof':
|
|
return proof(scope, state)
|
|
if action == 'restore':
|
|
return restore(directory, state)
|
|
if action == 'finish':
|
|
return finish(directory, state)
|
|
raise RuntimeError('INVALID_ACTION')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
value = main(json.loads(base64.b64decode(sys.argv[1])))
|
|
print('TEST_EVENT_JSON=' + json.dumps({'ok': True, 'data': value}))
|
|
except Exception as error:
|
|
message = str(error)
|
|
safe = message if re.fullmatch(r'[A-Z_0-9]{1,80}', message) else 'TEST_EVENT_ACTION_FAILED'
|
|
print('TEST_EVENT_JSON=' + json.dumps({'ok': False, 'error': safe}))
|