diff --git a/src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh b/src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh index b8efc18b..42692e52 100644 --- a/src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh +++ b/src/MissionParallelCatchup/parallel_catchup_helm/files/worker.sh @@ -9,6 +9,9 @@ if [ -z "$FAILED_QUEUE" ]; then echo "FAILED_QUEUE not set"; exit 1; fi if [ -z "$SUCCESS_QUEUE" ]; then echo "SUCCESS_QUEUE not set"; exit 1; fi if [ -z "$METRICS" ]; then echo "METRICS not set"; exit 1; fi if [ -z "$JOB_OWNERS" ]; then echo "JOB_OWNERS not set"; exit 1; fi +if [ -z "$WORKER_CURRENT" ]; then echo "WORKER_CURRENT not set"; exit 1; fi +if [ -z "$ATTEMPTS" ]; then echo "ATTEMPTS not set"; exit 1; fi +if [ -z "$MAX_ATTEMPTS" ]; then echo "MAX_ATTEMPTS not set"; exit 1; fi if [ -z "$RELEASE_NAME" ]; then echo "RELEASE_NAME not set"; exit 1; fi if [ -z "$POD_NAME" ]; then echo "POD_NAME not set"; exit 1; fi @@ -21,6 +24,48 @@ fi SLEEP_INTERVAL=10 LOG_DIR="/data" +# Hand back a range that a previous incarnation of this pod was holding. +# +# If our container was OOMKilled or our pod was evicted mid-catchup, the range +# we had claimed is still sitting in PROGRESS_QUEUE with nobody working on it. +# Nothing else can tell that from a range that is progressing normally, so it +# used to stay there until every other range in the mission had finished and +# the job monitor saw the whole fleet idle -- which is how a single dead worker +# added hours to a run (stellar/supercluster#409). We are the one process that +# knows for certain we are not working on it, so we are the one that returns it. +# +# The LREM guard means we only re-enqueue a range we actually removed, so this +# is a no-op if the range was already recovered some other way. Ranges that keep +# coming back here are failed rather than retried forever, otherwise a range +# that reliably OOMs would loop (stellar/supercluster#334). +# +# KEYS: progress queue, job queue, failed queue, attempts hash, worker current hash +# ARGV: job key, pod name, max attempts +# Returns: 0 if the range was no longer in progress, n > 0 if requeued as +# attempt n, -1 if attempts were exhausted and the range was failed +RECLAIM_LUA=' +redis.call("HDEL", KEYS[5], ARGV[2]) +if redis.call("LREM", KEYS[1], -1, ARGV[1]) ~= 1 then return 0 end +local n = redis.call("HINCRBY", KEYS[4], ARGV[1], 1) +if n > tonumber(ARGV[3]) then + redis.call("LPUSH", KEYS[3], ARGV[1] .. "|" .. ARGV[2]) + return -1 +end +redis.call("LPUSH", KEYS[2], ARGV[1]) +return n' + +PREV_JOB=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" HGET "$WORKER_CURRENT" "$POD_NAME") +if [ -n "$PREV_JOB" ]; then + RECLAIMED=$(redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" EVAL "$RECLAIM_LUA" 5 \ + "$PROGRESS_QUEUE" "$JOB_QUEUE" "$FAILED_QUEUE" "$ATTEMPTS" "$WORKER_CURRENT" \ + "$PREV_JOB" "$POD_NAME" "$MAX_ATTEMPTS") + case "$RECLAIMED" in + 0) echo "Previously held range $PREV_JOB was already recovered elsewhere" ;; + -1) echo "Range $PREV_JOB has now been orphaned more than $MAX_ATTEMPTS times; failing it" ;; + *) echo "Returned orphaned range $PREV_JOB to the front of $JOB_QUEUE as attempt $RECLAIMED" ;; + esac +fi + while true; do # Fetch the next job key from the Redis queue. # Our ranges are generated in the order we want to run them from left to right, so we always pull from the left @@ -29,8 +74,16 @@ LMOVE_EXIT_CODE=$? # Only process a job if the command succeeded AND we got a non-empty job key if [ $LMOVE_EXIT_CODE -eq 0 ] && [ -n "$JOB_KEY" ]; then - # Register ownership so the monitor knows which worker owns this job - redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" HSET "$JOB_OWNERS" "$JOB_KEY" "$POD_NAME" + # Register ownership so the monitor knows which worker owns this job, and + # record it against our pod name so that if we die and come back we can tell + # which range we abandoned. Both directions are needed: JOB_OWNERS answers + # "who has this range", WORKER_CURRENT answers "what was this pod holding". + redis-cli -h "$REDIS_HOST" -p "$REDIS_PORT" <