Move agent builds to a separate job

deployer/Dockerfile

@@ -0,0 +1,27 @@
+FROM --platform=linux/amd64 python:3.11
+ENV POETRY_HOME=/opt/poetry
+ENV POETRY_VENV=/opt/poetry-venv
+ENV POETRY_DATA_DIR=/opt/poetry-data
+ENV POETRY_CONFIG_DIR=/opt/poetry-config
+# Tell Poetry where to place its cache and virtual environment
+ENV POETRY_CACHE_DIR=/opt/.cache
+
+VOLUME ./.poetry-data-cache /opt/poetry-data
+VOLUME ./.poetry-cache /opt/.cache
+
+RUN python3.11 -m pip install poetry==1.7.1
+WORKDIR /workspace
+RUN poetry config virtualenvs.create false
+
+COPY ./pyproject.toml ./pyproject.toml
+COPY ./poetry.lock ./poetry.lock
+# install only deps in dependency list first and lockfile to cache them
+RUN poetry install --no-root --only main
+
+COPY . .
+# then install our own module
+RUN poetry install --only main
+
+EXPOSE 8080
+
+ENTRYPOINT [ "poetry", "run", "start" ]

deployer/database/__init__.py

@@ -0,0 +1 @@
+from .database import Database

deployer/database/database.py

@@ -0,0 +1,76 @@
+from typing import Optional
+import os
+from supabase import create_client, Client
+from models import AppConfig, Agent, User
+
+
+class Database:
+    def __init__(self):
+        supabase_url = os.getenv("SUPABASE_URL")
+        supabase_key = os.getenv("SUPABASE_KEY")
+        self.supabase = create_client(supabase_url, supabase_key)
+
+    def get_config(self, bearer_token: str) -> Optional[AppConfig]:
+        response = (
+            self.supabase.table("user")
+            .select("*")
+            .filter("secret_key", "eq", bearer_token)
+            .execute()
+        )
+        if len(response.data) > 0:
+            row = response.data[0]
+            return AppConfig(user_id=row["id"], app_id=row["app_id"])
+        return None
+
+    def get_secret_key_for_user(self, user_id: str):
+        response = (
+            self.supabase.table("user")
+            .select("secret_key")
+            .filter("id", "eq", user_id)
+            .execute()
+        )
+        if len(response.data) > 0:
+            return response.data[0]["secret_key"]
+        return None
+
+    def upsert_agent(self, agent: Agent) -> Optional[Agent]:
+        payload = agent.dict()
+        # Remove created_at field
+        payload.pop("created_at", None)
+        response = (
+            self.supabase.table("agent")
+            .upsert(
+                payload,
+            )
+            .execute()
+        )
+        if len(response.data) > 0:
+            row = response.data[0]
+            return Agent(**row)
+        return None
+
+    def get_agent(self, config: AppConfig, id: str) -> Optional[Agent]:
+        response = (
+            self.supabase.table("agent")
+            .select("*")
+            .filter("app_id", "eq", config.app_id)
+            .filter("id", "eq", id)
+            .execute()
+        )
+        if len(response.data) > 0:
+            row = response.data[0]
+            return Agent(**row)
+        return None
+
+    def get_user(self, config: AppConfig) -> Optional[Agent]:
+        response = (
+            self.supabase.table("user")
+            .select("*")
+            .filter("app_id", "eq", config.app_id)
+            .filter("id", "eq", config.user_id)
+            .execute()
+        )
+        if len(response.data) > 0:
+            row = response.data[0]
+            return User(**row)
+        return None

deployer/deployer/__init__.py

@@ -0,0 +1 @@
+from .deployer import Deployer

deployer/deployer/deployer.py

@@ -0,0 +1,96 @@
+from database import Database
+from models import AppConfig, Agent
+import os
+import zipfile
+from google.cloud.devtools import cloudbuild_v1
+from google.oauth2 import service_account
+import json
+from datetime import timedelta
+from google.cloud import run_v2
+
+
+class Deployer:
+    def __init__(self):
+        service_account_string = os.getenv("GCLOUD_SERVICE_ACCOUNT")
+        self.deployments_bucket = os.getenv("DEPLOYMENTS_BUCKET")
+        self.project_id = os.getenv("GCLOUD_PROJECT")
+        credentials = service_account.Credentials.from_service_account_info(
+            json.loads(service_account_string)
+        )
+
+        self.build_client = cloudbuild_v1.CloudBuildClient(credentials=credentials)
+        self.jobs_client = run_v2.JobsClient(credentials=credentials)
+
+    def deploy_agent(self, agent: Agent):
+        # Check if the job already exists in Cloud Run
+        try:
+            self.jobs_client.get_job(
+                name=f"projects/{self.project_id}/locations/us-central1/jobs/{Agent.get_cloud_job_id(agent)}"
+            )
+            job_exists = True
+        except Exception:
+            job_exists = False
+
+        # Define the build steps
+        build_config = self._get_build_config(agent=agent, job_exists=job_exists)
+
+        # Trigger the build
+        build = cloudbuild_v1.Build(
+            steps=build_config["steps"],
+            images=build_config["images"],
+            source=cloudbuild_v1.Source(
+                storage_source=cloudbuild_v1.StorageSource(
+                    bucket=self.deployments_bucket,
+                    object_=f"{agent.finic_id}.zip",
+                )
+            ),
+        )
+        operation = self.build_client.create_build(
+            project_id=self.project_id, build=build
+        )
+
+        # Wait for the build to complete
+        result = operation.result()
+        if result.status != cloudbuild_v1.Build.Status.SUCCESS:
+            raise Exception(f"Build failed with status: {result.status}")
+
+        print(f"Built and pushed Docker image: {agent.finic_id}")
+
+    def _get_build_config(self, agent: Agent, job_exists: bool) -> dict:
+        image_name = f"gcr.io/{self.project_id}/{agent.finic_id}:latest"
+        gcs_source = f"gs://{self.deployments_bucket}/{agent.finic_id}.zip"
+        job_command = "update" if job_exists else "create"
+        return {
+            "steps": [
+                {
+                    "name": "gcr.io/cloud-builders/gsutil",
+                    "args": ["cp", gcs_source, "/workspace/source.zip"],
+                },
+                {
+                    "name": "gcr.io/cloud-builders/gcloud",
+                    "entrypoint": "bash",
+                    "args": [
+                        "-c",
+                        "apt-get update && apt-get install -y unzip && unzip /workspace/source.zip -d /workspace/unzipped",
+                    ],
+                },
+                {
+                    "name": "gcr.io/cloud-builders/docker",
+                    "args": ["build", "-t", image_name, "/workspace/unzipped"],
+                },
+                {
+                    "name": "gcr.io/cloud-builders/docker",
+                    "args": ["push", image_name],
+                },
+                {
+                    "name": "gcr.io/google.com/cloudsdktool/cloud-sdk",
+                    "entrypoint": "bash",
+                    "args": [
+                        "-c",
+                        f"gcloud run jobs {job_command} {Agent.get_cloud_job_id(agent)} --image {image_name} --region us-central1 "
+                        f"--tasks=1 --max-retries={agent.num_retries} --task-timeout=86400s --memory=4Gi",
+                    ],
+                },
+            ],
+            "images": [image_name],
+        }

deployer/main/__init__.py

@@ -0,0 +1 @@
+from .main import main

deployer/main/main.py

@@ -0,0 +1,26 @@
+from database import Database
+import os
+from deployer.deployer import Deployer
+from models.models import AgentStatus
+from dotenv import load_dotenv
+
+load_dotenv()
+
+API_KEY = os.getenv("FINIC_API_KEY")
+AGENT_ID = os.getenv("FINIC_AGENT_ID")
+
+
+def main():
+    db = Database()
+    config = db.get_config(API_KEY)
+    agent = db.get_agent(config=config, id=AGENT_ID)
+    try:
+        deployer = Deployer()
+        deployer.deploy_agent(agent=agent)
+        agent.status = AgentStatus.deployed
+        db.upsert_agent(agent)
+        return agent
+    except Exception as e:
+        agent.status = AgentStatus.failed
+        db.upsert_agent(agent)
+        raise e

deployer/models/__init__.py

@@ -0,0 +1 @@
+from .models import AppConfig, Agent, User, AgentStatus, ExecutionStatus

deployer/models/models.py

@@ -0,0 +1,43 @@
+from enum import Enum
+import datetime
+from typing import Optional
+from pydantic import BaseModel
+
+
+class AppConfig(BaseModel):
+    user_id: str
+    app_id: str
+
+
+class User(BaseModel):
+    id: str
+    created_at: datetime.datetime
+    email: str
+    secret_key: str
+    avatar_url: str
+
+
+class AgentStatus(str, Enum):
+    deploying = "deploying"
+    deployed = "deployed"
+    failed = "deploy_failed"
+
+
+class ExecutionStatus(str, Enum):
+    running = "running"
+    successful = "successful"
+    failed = "failed"
+
+
+class Agent(BaseModel):
+    finic_id: str
+    id: str
+    app_id: str
+    description: str
+    status: AgentStatus
+    created_at: Optional[datetime.datetime] = None
+    num_retries: int = 3
+
+    @staticmethod
+    def get_cloud_job_id(agent: "Agent") -> str:
+        return f"job-{agent.finic_id}"