Rewrote pipeline to use grpc flow

2023-04-10 19:05:31 -06:00
parent cd69ff50f7
commit 719778fde3
6 changed files with 224 additions and 3 deletions
@@ -0,0 +1,79 @@
+import os
+import sys
+import grpc
+
+# HACK: need to fix how protobuf is generated to get imports working
+sys.path.append(
+    "/pipeline/.venv/lib/python3.11/site-packages/cursorius_pipeline_api_v2/"
+)
+from cursorius_pipeline_api_v2 import get_runner_pb2, get_runner_pb2_grpc
+from cursorius_pipeline_api_v2 import run_command_pb2, run_command_pb2_grpc
+from cursorius_pipeline_api_v2 import release_runner_pb2, release_runner_pb2_grpc
+
+
+def main():
+    run_id = os.environ.get("CURSORIUS_RUN_ID")
+    src_dir = os.environ.get("CURSORIUS_SRC_DIR")
+    server_url = os.environ.get("CURSORIUS_SERVER_URL")
+
+    print(f"{run_id=}")
+    print(f"{src_dir=}")
+    print(f"{server_url=}")
+
+    # create GRPc connection
+    channel = grpc.insecure_channel(server_url)
+
+    # GetRunner stub
+    get_runner_stub = get_runner_pb2_grpc.GetRunnerServiceStub(channel)
+
+    # GetRunner Request
+    request = get_runner_pb2.GetRunnerRequest()
+
+    # Call GetRunner
+    get_runner_resp = None
+    try:
+        get_runner_resp = get_runner_stub.GetRunner(request)
+    except grpc._channel._InactiveRpcError as e:
+        print(f"Error: {e.details()}")
+        sys.exit(-1)
+
+    # Print ID of allocated runner
+    print(get_runner_resp.id)
+
+    # RunCommand stub
+    run_command_stub = run_command_pb2_grpc.RunCommandServiceStub(channel)
+
+    # RunCommand request
+    request = run_command_pb2.RunCommandRequest(
+        id=get_runner_resp.id,
+        command="ls",
+        args=["-lah"],
+    )
+
+    # Call RunCommand
+    run_command_resp = None
+    try:
+        run_command_resp = run_command_stub.RunCommand(request)
+    except grpc._channel._InactiveRpcError as e:
+        print(f"Error: {e.details()}")
+        sys.exit(-1)
+
+    print(f"return code: {run_command_resp.return_code}")
+    print(f"stdout: {run_command_resp.stdout}")
+    print(f"stderr: {run_command_resp.stderr}")
+
+    # ReleaseRunner stub
+    release_runner_stub = release_runner_pb2_grpc.ReleaseRunnerServiceStub(channel)
+
+    # ReleaseRunner request
+    request = release_runner_pb2.ReleaseRunnerRequest(
+        id=get_runner_resp.id,
+    )
+
+    # Call ReleaseRunner
+    release_runner_resp = None
+    try:
+        release_runner_resp = release_runner_stub.ReleaseRunner(request)
+    except grpc._channel._InactiveRpcError as e:
+        print(f"Error: {e.details()}")
+        sys.exit(-1)