tubearchivist-server/builder/monitor.py
2022-05-20 11:41:44 +07:00

130 lines
3.7 KiB
Python

"""monitor redis for tasks to execute"""
import json
import subprocess
import os
import redis
class RedisBase:
"""connection base for redis"""
REDIS_HOST = "localhost"
REDIS_PORT = 6379
NAME_SPACE = "ta:"
TASK_KEY = NAME_SPACE + "task:buildx"
def __init__(self):
self.conn = redis.Redis(host=self.REDIS_HOST, port=self.REDIS_PORT)
class Monitor(RedisBase):
"""look for messages"""
def get_tasks(self):
"""get task list"""
response = self.conn.execute_command("JSON.GET", self.TASK_KEY)
tasks = json.loads(response.decode())
return tasks
def bootstrap(self):
"""create custom builder"""
print("validate builder")
command = ["docker", "buildx", "inspect"]
output = subprocess.run(command, check=True, capture_output=True)
inspect = output.stdout.decode()
config = {}
lines = [i for i in inspect.split("\n") if i]
for line in lines:
key, value = line.split(":", maxsplit=1)
if value:
config[key.strip()] = value.strip()
if not config["Name"].startswith("tubearchivist"):
print("create tubearchivist builder")
self._create_builder()
else:
print("tubearchivist builder already created")
@staticmethod
def _create_builder():
"""create buildx builder"""
base = ["docker", "buildx"]
subprocess.run(
base + ["create", "--name", "tubearchivist"], check=True
)
subprocess.run(base + ["use", "tubearchivist"], check=True)
subprocess.run(base + ["inspect", "--bootstrap"], check=True)
def watch(self):
"""watch for messages"""
print("waiting for tasks")
watcher = self.conn.pubsub()
watcher.subscribe(self.TASK_KEY)
for i in watcher.listen():
if i["type"] == "message":
task = i["data"].decode()
print(task)
Builder(task).run()
class Builder(RedisBase):
"""execute task"""
CLONE_BASE = "clone"
def __init__(self, task):
super().__init__()
self.task = task
self.task_detail = False
def run(self):
"""run all steps"""
self.get_task()
self.clone()
self.build()
self.remove_task()
def get_task(self):
"""get what to execute"""
print("get task from redis")
response = self.conn.execute_command("JSON.GET", self.TASK_KEY)
response_json = json.loads(response.decode())
self.task_detail = response_json["tasks"][self.task]
def clone(self):
"""clone repo to destination"""
print("clone repo")
clone = ["git", "clone", self.task_detail["clone"]]
pull = ["git", "pull", self.task_detail["clone"]]
os.chdir("clone")
try:
subprocess.run(clone, check=True)
except subprocess.CalledProcessError:
print("git pull instead")
os.chdir(self.task)
subprocess.run(pull, check=True)
os.chdir("../")
os.chdir("../")
def build(self):
"""build the container"""
build_command = ["docker", "buildx"] + self.task_detail["build"]
build_command.append(os.path.join(self.CLONE_BASE, self.task))
subprocess.run(build_command, check=True)
def remove_task(self):
"""remove task from redis queue"""
print("remove task from redis")
self.conn.json().delete(self.TASK_KEY, f".tasks.{self.task}")
if __name__ == "__main__":
handler = Monitor()
handler.bootstrap()
try:
handler.watch()
except KeyboardInterrupt:
print(" [X] cancle watch")