-
Notifications
You must be signed in to change notification settings - Fork 47
/
enqueue.py
145 lines (100 loc) · 4.3 KB
/
enqueue.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
"""
This is where we should implement any and all job function for the
redis queue. The rq library requires special namespacing in order to work,
so these functions must reside in a separate file.
"""
import traceback
from redis import Redis
from rq import Queue
from anubis.env import env
from anubis.github.repos import create_assignment_github_repo
from anubis.ide.initialize import initialize_theia_session
from anubis.k8s.pipeline.create import create_submission_pipeline
from anubis.k8s.pipeline.reap import reap_pipeline_jobs
from anubis.k8s.pvc.create import create_user_pvc
from anubis.k8s.pvc.reap import reap_user_pvc
from anubis.k8s.theia.reap import reap_stale_theia_sessions
from anubis.k8s.theia.reap import reap_theia_session_by_id
from anubis.lms.assignments import make_shared_assignment
from anubis.lms.autograde import bulk_autograde
from anubis.lms.questions import assign_missing_questions
from anubis.lms.regrade import bulk_regrade_assignment
from anubis.lms.submissions import bulk_regrade_submissions, recalculate_late
from anubis.lms.courses import bulk_create_students
from anubis.utils.data import with_context
from anubis.utils.testing.seed import seed
@with_context
def _run_rpc_function(func, *args):
return func(*args)
def rpc_enqueue(func, queue=None, args=None):
"""
Enqueues a job on the redis cache
:func callable: any callable object
:args tuple: ordered arguments for function
"""
# set defaults
if queue is None:
queue = "default"
if args is None:
args = tuple()
# If we are running in mindebug, there is
# no rq cluster to send things off to.
if env.MINDEBUG:
try:
return func(*args)
except Exception as e:
print(e)
print(traceback.format_exc())
return
with Redis(host=env.CACHE_REDIS_HOST, password=env.CACHE_REDIS_PASSWORD) as conn:
q = Queue(name=queue, connection=conn)
q.enqueue(_run_rpc_function, func, *args)
conn.close()
def enqueue_autograde_pipeline(*args, queue: str = "regrade"):
"""Enqueues a test job"""
rpc_enqueue(create_submission_pipeline, queue=queue, args=args)
def enqueue_ide_initialize(*args):
"""Enqueue an ide initialization job"""
rpc_enqueue(initialize_theia_session, queue="theia", args=args)
def enqueue_ide_stop(*args):
"""Reap theia session kube resources"""
rpc_enqueue(reap_theia_session_by_id, queue="theia", args=args)
def enqueue_ide_reap_stale(*args):
"""Reap stale ide resources"""
rpc_enqueue(reap_stale_theia_sessions, queue="theia", args=args)
def enqueue_pipeline_reap_stale(*args):
"""Reap stale pipeline job resources"""
rpc_enqueue(reap_pipeline_jobs, queue="theia", args=args)
def enqueue_seed():
"""Enqueue debug seed data"""
rpc_enqueue(seed, queue="default")
def enqueue_assign_missing_questions(*args):
"""Enqueue assign missing questions"""
rpc_enqueue(assign_missing_questions, queue="default", args=args)
def enqueue_make_shared_assignment(*args):
"""Enqueue make shared assignment"""
rpc_enqueue(make_shared_assignment, queue="default", args=args)
def enqueue_create_assignment_github_repo(*args):
"""Enqueue make shared assignment"""
rpc_enqueue(create_assignment_github_repo, queue="default", args=args)
def enqueue_reap_pvc_user(*args):
"""Enqueue reap pvc for user"""
rpc_enqueue(reap_user_pvc, queue='default', args=args)
def enqueue_create_pvc_user(*args):
"""Enqueue create pvc for user"""
rpc_enqueue(create_user_pvc, queue='default', args=args)
def enqueue_bulk_autograde(*args):
"""Enqueue bulk autograde of assignment"""
rpc_enqueue(bulk_autograde, queue="regrade", args=args)
def enqueue_bulk_regrade_assignment(*args):
"""Enqueue bulk autograde of assignment"""
rpc_enqueue(bulk_regrade_assignment, queue="regrade", args=args)
def enqueue_bulk_regrade_submissions(*args):
"""Enqueue bulk autograde of assignment"""
rpc_enqueue(bulk_regrade_submissions, queue="regrade", args=args)
def enqueue_bulk_create_students(*args):
"""Enqueue create users for course"""
rpc_enqueue(bulk_create_students, queue='default', args=args)
def enqueue_recalculate_late(*args):
"""Enqueue recalculate late for assignment"""
rpc_enqueue(recalculate_late, queue='default', args=args)