CHG use distributed locks to distribute theia-poller job

[wabscale]

The poller job is a job thats sole purpose is to make sure that the database entries for IDEs match the resources allocated. This means looking at the kubernetes resources allocated and updating IDE database entries. Right now only one of these runs at a time. It is pretty much the only piece of Anubis that is a single point of failure. If for whatever reason the poller goes away, IDE states will not be updated. This will make launching or stopping IDEs impossible (which would be a huge problem).

We'll be able to run more than a single instance of redis poller if we can use distributed locks to ensure consistency and synchronization.

redis distributed locks: https://redis.io/docs/reference/patterns/distributed-locks/
python lib: https://github.com/brainix/pottery#redlock
poller job right now: https://github.com/AnubisLMS/Anubis/blob/a5e3f004076c108f6ec8819f9ee7c099eb639fc1/api/anubis/jobs/ide_poller.py

[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