CHG use distributed locks to distribute theia-poller job about anubis

[wabscale]

Basically as it is done here:

Anubis/api/anubis/k8s/pipeline.py

Lines 175 to 241 in a075c56

	def reap_pipeline_jobs() -> int:
	"""
	Runs through all jobs in the namespace. If the job is finished, it will
	send a request to the kube api to delete it. Number of active jobs is
	returned.
	:return: number of active jobs
	"""
	# Get the batch v1 object so we can query for active k8s jobs
	batch_v1 = client.BatchV1Api()
	# Get all pipeline jobs in the anubis namespace
	jobs = get_active_pipeline_jobs()
	# Get the autograde pipeline timeout from config
	autograde_pipeline_timeout_minutes = get_config_int("AUTOGRADE_PIPELINE_TIMEOUT_MINUTES", default=5)
	# Iterate through all pipeline jobs
	for job in jobs:
	job: client.V1Job
	# If submission id not in labels just skip. Job ttl will delete itself.
	if 'submission-id' not in job.metadata.labels:
	logger.error(f'skipping job based off old label format: {job.metadata.name}')
	continue
	# Read submission id from labels
	submission_id = job.metadata.labels['submission-id']
	# Create a distributed lock for the submission job
	lock = Redlock(
	key=f'submission-job-{submission_id}',
	masters={redis},
	auto_release_time=3.0,
	)
	if not lock.acquire(blocking=False):
	continue
	# Log that we are inspecting the pipeline
	logger.debug(f'inspecting pipeline: {job.metadata.name}')
	# Get database record of the submission
	submission: Submission = Submission.query.filter(
	Submission.id == submission_id,
	).first()
	if submission is None:
	logger.error(f"submission from db not found {submission_id}")
	continue
	# Calculate job created time
	job_created = job.metadata.creation_timestamp.replace(tzinfo=None)
	# Delete the job if it is older than a few minutes
	if datetime.utcnow() - job_created > timedelta(minutes=autograde_pipeline_timeout_minutes):
	# Attempt to delete the k8s job
	reap_pipeline_job(job, submission)
	# If the job has finished, and was marked as successful, then
	# we can clean it up
	elif job.status.succeeded is not None and job.status.succeeded >= 1:
	# Attempt to delete the k8s job
	reap_pipeline_job(job, submission)
	lock.release()

Just need to lock the IDE resource as the submissions are locked here. We may want to formalize the way the locks are done here a bit. Maybe make a single function for creating the lock and releasing the lock. There is also a syncronize function in pottery. Maybe a decorator like this with argument based locking (like hash the values of the arguments to create the key).

>>> from pottery import synchronize
>>> @synchronize(key='synchronized-func', masters={redis}, auto_release_time=1.5, blocking=True, timeout=-1)
... def func():
...   # Only one thread can execute this function at a time.
...   return True
...
>>> func()
True
>>>

[wabscale]

53eca0b