Better logging

Author: iVishalr

.gitignore

@@ -2,6 +2,8 @@
 *logs.txt
 compile-test/
 output/
+configs/
+logs/
 # answer/A1T1
 # answer/A1T2
 *logs.txt

config/evaluator.json

@@ -1,5 +1,12 @@
 {
+	"backend": {
+		"name": "backend",
+		"log_dir": "/home/vishalr/Desktop/backend-2022/logs/",
+		"config_dir": "/home/vishalr/Desktop/backend-2022/configs/"
+	},
+
 	"executor": {
+		"name": "Assignment2",
 		"num_backends": 2,
 		"num_workers": 1,
 		"fetch_port": 9000,
@@ -9,7 +16,8 @@
 		"threshold": 8,
 		"global_queue_thread": true,
 		"global_prefetch_thread": true,
-		"timeout": 30
+		"timeout": 30,
+		"log_dir": "/home/vishalr/Desktop/backend-2022/logs/"
 	},
 
 	"docker": {
@@ -23,6 +31,8 @@
 		"shared_output_dir": "/home/vishalramesh01/backend-2022/output/",
 		"docker_output_dir": "/output",
 		"docker_memswapiness": 0,
-		"spawn_wait": 40
+		"spawn_wait": 40,
+		"blacklist_threshold": 5,
+		"blacklist_duration": 120
 	}
 }

doc/architecture.png doc/images/architecture.png

@@ -17,10 +17,6 @@
 <!-- Put site-specific property overrides in this file. -->
 
 <configuration>
-    <property>
-        <name>mapreduce.framework.name</name>
-        <value>yarn</value>
-    </property>
     <property>
         <name>mapreduce.framework.name</name>
         <value>yarn</value>

doc/website-backend.png doc/images/website-backend.png

@@ -122,8 +122,15 @@ def create_hdfs_directory(dirname: str) -> int:
     logger.mark(f"Created Directory - hdfs:{dirname}")
     return res
 
-def delete_hdfs_directories(dirname: str) -> int:
-    process = subprocess.Popen([f"{HDFS} dfs -rm -r {dirname}"], shell=True, text=True)
+def delete_hdfs_directories(dirname: List[str]) -> int:
+    if type(dirname) != list:
+        dirname = [dirname]
+    command = []
+    for dirs in dirname:
+        cmd = f"{HDFS} dfs -rm -r {dirs}"
+        command.append(cmd)
+    command = "&&".join(command)
+    process = subprocess.Popen([command], shell=True, text=True)
     res = process.wait()
     logger.mark(f"Deleted Directory - hdfs:{dirname}")
     return res
@@ -181,14 +188,14 @@ def run_hadoop_job(team_id, assignment_id, submission_id, timeout, mapper: str,
     """
 
     path = os.path.join(SUBMISSIONS, team_id)
-    if not os.path.exists(path):
-        os.mkdir(path)
+    # if not os.path.exists(path):
+    #     os.mkdir(path)
 
     logger.mark(f"Created Directory - {path}")
 
     task_path = os.path.join(path, submission_id)
     if not os.path.exists(task_path):
-        os.mkdir(task_path)
+        os.makedirs(task_path)
 
     logger.mark(f"Created Directory - {task_path}")
 
@@ -473,10 +480,17 @@ def cleanup(team_id, assign_id, task) -> int:
 
     task_path = TASK_OUTPUT_PATH[assign_id]
 
-    _ = delete_hdfs_directories(f"/{team_id}/{assign_id}/{task}/{task_path}")
-    _ = delete_hdfs_directories(f"/{team_id}/{assign_id}/{task}")
-    _ = delete_hdfs_directories(f"/{team_id}/{assign_id}")
-    _ = delete_hdfs_directories(f"/{team_id}")
+    # _ = delete_hdfs_directories(f"/{team_id}/{assign_id}/{task}/{task_path}")
+    # _ = delete_hdfs_directories(f"/{team_id}/{assign_id}/{task}")
+    # _ = delete_hdfs_directories(f"/{team_id}/{assign_id}")
+    # _ = delete_hdfs_directories(f"/{team_id}")
+
+    _ = delete_hdfs_directories([
+        f"/{team_id}/{assign_id}/{task}/{task_path}", 
+        f"/{team_id}/{assign_id}/{task}",
+        f"/{team_id}/{assign_id}",
+        f"/{team_id}"
+    ])
 
     return 0
 

docker/hadoop_config/mapred-site.xml

@@ -1,6 +1,10 @@
+import multiprocessing
 from redis import Redis, ConnectionPool
 from common.redisqueue import RedisQueue
 
 pool = ConnectionPool(host='localhost', port=6379, db=0)
 broker = Redis(connection_pool=pool)
-queue = RedisQueue(broker=broker, queue_name="sanity-queue")
+queue = RedisQueue(broker=broker, queue_name="sanity-queue")
+
+manager = multiprocessing.Manager()
+executor_table = manager.dict()

docker/hadoop_server-A2.py

@@ -12,7 +12,7 @@
 from common.utils import Tee, get_datetime
 
 from dotenv import load_dotenv
-from flask_backend import queue
+from flask_backend import queue, executor_table
 from flask_cors import cross_origin
 from flask import Flask, request, jsonify
 from signal import signal, SIGPIPE, SIG_DFL
@@ -26,12 +26,30 @@ def createApp():
     os.environ['TZ'] = 'Asia/Kolkata'
     time.tzset()
 
+    config_path = os.path.join(os.getcwd(),"config", "evaluator.json")
+
+    configs = None
+    with open(config_path, "r") as f:
+        configs = json.loads(f.read())
+
     app = Flask(__name__)  
     db = DataBase()
 
-    f = open(f'./sanity_checker_logs.txt', 'a+')
+    configs = configs["backend"]
+    backend_name = configs["name"]
+    backend_logdir = configs["log_dir"]
+    backend_config_dir = configs["config_dir"]
+
+    log_path = os.path.join(backend_logdir, backend_name)
+    if not os.path.exists(log_path):
+        os.makedirs(log_path)
+
+    f = open(os.path.join(log_path, f'sanity_checker_logs.txt'), 'a+')
     backup = sys.stdout
     sys.stdout = Tee(sys.stdout, f)
+
+    def is_registered(executor_ip) -> bool:
+        return executor_ip in executor_table
 
     def delete_files(path):
         for file in os.listdir(path):
@@ -240,13 +258,17 @@ def get_jobs():
             client_addr = request.environ['REMOTE_ADDR']
         else:
             client_addr = request.environ['HTTP_X_FORWARDED_FOR'] # if behind a proxy
+
+        if not is_registered(str(client_addr)):
+            res = {"status": 401, "msg": f"This executor has not been registered with backend server. Please register before making request"}
+            return jsonify(res)
 
         prefetch_factor = int(request.args.get("prefetch_factor"))
 
         if prefetch_factor is None: prefetch_factor = 1
 
         if len(queue) == 0:
-            res = {"msg": "Submission Queue is currently empty.", "len": len(queue), "num_submissions": 0}
+            res = {"msg": "Submission Queue is currently empty.", "len": len(queue), "num_submissions": 0, "status": 200}
             return jsonify(res)
 
         data = []
@@ -269,11 +291,10 @@ def get_jobs():
 
         r = requests.post(request_url, data=data)
 
-        res = {"msg": f"Dequeued {length} submissions from queue.", "num_submissions": length, "len": len(queue)}
+        res = {"msg": f"Dequeued {length} submissions from queue.", "num_submissions": length, "len": len(queue), "status": 200}
         # res = {"msg": "dequeued from submission queue", "len": len(queue), "server_response": res}
         return jsonify(res)
 
-
     @app.route("/queue-length", methods=["GET"])
     def queue_length():
         msg = {"length": len(queue)}
@@ -285,6 +306,57 @@ def empty_queue():
         res = {"msg": "Queue Emptied"}
         return jsonify(res)
 
+    @app.route("/register_executor", methods=["POST"])
+    def register_executor():
+        executor_addr = None
+        if request.environ.get('HTTP_X_FORWARDED_FOR') is None:
+            executor_addr = request.environ['REMOTE_ADDR']
+        else:
+            executor_addr = request.environ['HTTP_X_FORWARDED_FOR'] # if behind a proxy
+
+        executor_addr = str(executor_addr)
+
+        data = json.loads(request.data)
+
+        executor_metadata = {
+            'executor_name': data["executor_name"],
+            'executor_uuid': data["executor_uuid"],
+            'executor_log_dir': data["executor_log_dir"],
+            'num_backends': data["num_backends"],
+            'num_workers': data["num_workers"],
+            'num_threads': data["num_threads"],
+            'num_prefetch_threads': data["num_prefetch_threads"],
+            'prefetch_factor': data["prefetch_factor"],
+            'threshold': data["threshold"],
+            'timeout': data["timeout"],
+            'ipaddr': executor_addr,
+        }
+        executor_table[executor_addr] = executor_metadata
+
+        if not os.path.exists(backend_config_dir):
+            os.makedirs(backend_config_dir)
+
+        with open(os.path.join(backend_config_dir, f'{data["executor_name"]}-{data["executor_uuid"]}.json'), 'w', encoding='utf-8') as _f:
+            json.dump(executor_metadata, _f, ensure_ascii=False, indent=4)
+        
+        msg = {"status": 200, "message": "Registered!"}
+        return jsonify(msg)
+
+    @app.route("/executors", methods=["GET"])
+    def get_executors():
+        files = list(os.listdir(backend_config_dir))
+        executors = {}
+        for filename in files:
+            if ".json" not in filename:
+                continue
+            path = os.path.join(backend_config_dir, filename)
+            e = open(path,"r")
+            executor = json.loads(e.read())
+            e.close()
+            executors[executor['ipaddr']] = executor
+        
+        return jsonify(executors)
+
     return app
 
 if __name__ == "__main__":