Deactivate admin api (#133)

Stop the threads for:
Scheduled workflows
Queues
Kafka

dbos/admin_sever.py

@@ -18,6 +18,7 @@
 health_check_path = "/dbos-healthz"
 workflow_recovery_path = "/dbos-workflow-recovery"
 perf_path = "/dbos-perf"
+deactivate_path = "/deactivate"
 
 
 class AdminServer:
@@ -66,6 +67,15 @@ def do_GET(self) -> None:
             self.send_response(200)
             self._end_headers()
             self.wfile.write(json.dumps(perf_util).encode("utf-8"))
+        elif self.path == deactivate_path:
+            dbos_logger.info("Deactivating DBOS")
+            # Stop all scheduled workflows, queues, and kafka loops
+            for event in self.dbos.stop_events: 
+                event.set()
+
+            self.send_response(200)
+            self._end_headers()
+            self.wfile.write("deactivated".encode("utf-8"))
         else:
             self.send_response(404)
             self._end_headers()

tests/test_admin_server.py

@@ -48,6 +48,13 @@ def test_admin_endpoints(dbos: DBOS) -> None:
     response = requests.post("http://localhost:3001/stuff", timeout=5)
     assert response.status_code == 404
 
+    response = requests.get("http://localhost:3001/deactivate", timeout=5)
+    assert response.status_code == 200
+
+    for event in dbos.stop_events:
+        assert event.is_set(), "Event is not set!"
+
+
 
 def test_admin_recovery(dbos: DBOS) -> None:
     os.environ["DBOS__VMID"] = "testexecutor"