diff --git a/.dockerignore b/.dockerignore deleted file mode 100644 index f933e00..0000000 --- a/.dockerignore +++ /dev/null @@ -1,18 +0,0 @@ -.git/ -.gitignore - -.vscode/ -.DS_Store - -*.md - -tests/ -__pycache__/ -.pytest_cache - -docker-compose.yml -Dockerfile -.dockerignore - -.env -.env.example diff --git a/.env.example b/.env.example index 4f5d18d..d53c1a8 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,51 @@ +## docker compose + +POSTGRES_USER=postgres +POSTGRES_PASSWORD=postgres +POSTGRES_DB=remote_code +POSTGRES_PORT=5432 + +REDIS_PORT=6379 + +RABBITMQ_USER=guest +RABBITMQ_PASSWORD=guest +RABBITMQ_PORT=5672 + +HTTP_PORT=8000 +METRICS_PORT=9100 # server / producer +PROCESSOR_METRICS_PORT=9101 # consumer + +## runtime - not injected into containers, set via compose + HTTP_ADDR=:8000 -TASK_PROCESSING_TIME=2s SHUTDOWN_TIMEOUT=10s + +METRICS_ADDR=:9100 +METRICS_SHUTDOWN_TIMEOUT=5s + +DATABASE_URL=postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@localhost:${POSTGRES_PORT}/${POSTGRES_DB}?sslmode=disable +MIGRATIONS_DIR=migrations + +REDIS_ADDR=localhost:${REDIS_PORT} +REDIS_PASSWORD= +REDIS_DB=0 +SESSION_TTL=168h + +RABBITMQ_URL=amqp://${RABBITMQ_USER}:${RABBITMQ_PASSWORD}@localhost:${RABBITMQ_PORT}/ +TASK_QUEUE_NAME=tasks +TASK_QUEUE_PREFETCH=1 +TASK_QUEUE_RECONNECT_DELAY=1s + +PHILHARMONIC_URL=http://localhost:5555 +# empty means manager runs without auth +PHILHARMONIC_TOKEN= +PHILHARMONIC_IMAGE=sandbox:latest +PHILHARMONIC_TASK_TIMEOUT=30s +PHILHARMONIC_POLL_INTERVAL=1s +# unset means PHILHARMONIC_TASK_TIMEOUT + 15s +PHILHARMONIC_POLL_TIMEOUT= +PHILHARMONIC_CPU=0.5 +# bytes +PHILHARMONIC_MEMORY=268435456 +# CI publishes the sandbox to GHCR on every push to main +PHILHARMONIC_IMAGE=ghcr.io/belyaevedu/backend-final/sandbox:latest diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..f7e6e1a --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,127 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + lint: + name: vet & lint + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + + - name: vet + run: go vet ./... + + - name: golangci-lint + uses: golangci/golangci-lint-action@v8 + with: + version: v2.11.3 + + publish-sandbox: + name: publish sandbox image to ghcr + # push to main only basically + if: github.event_name == 'push' + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + steps: + - uses: actions/checkout@v4 + + - uses: docker/setup-buildx-action@v3 + + - id: repo + name: lowercase repo name by ghcr requirement + run: echo "name=${GITHUB_REPOSITORY,,}" >> "$GITHUB_OUTPUT" + + - name: login to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: build and push + uses: docker/build-push-action@v6 + with: + context: . + file: sandbox/Dockerfile + push: true + tags: | + ghcr.io/${{ steps.repo.outputs.name }}/sandbox:latest + ghcr.io/${{ steps.repo.outputs.name }}/sandbox:${{ github.sha }} + cache-from: type=gha + cache-to: type=gha,mode=max + + e2e: + name: e2e (compose + philharmonic + pytest) + runs-on: ubuntu-latest + needs: lint + env: + DATABASE_URL: postgres://postgres:postgres@localhost:5432/remote_code?sslmode=disable + REDIS_ADDR: localhost:6379 + RABBITMQ_URL: amqp://guest:guest@localhost:5672/ + PHILHARMONIC_URL: http://localhost:5555 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + + - name: infra + run: docker compose up -d --wait postgres redis rabbitmq + + - name: build sandbox image + run: docker build -t sandbox:latest -f sandbox/Dockerfile . + + - name: start philharmonic manager + worker on the runner + run: | + git clone --depth 1 --branch v0.1.1 https://github.com/belyaevedu/philharmonic /tmp/philharmonic + (cd /tmp/philharmonic && CGO_ENABLED=0 go build -o /tmp/phrm .) + + /tmp/phrm manager --host 127.0.0.1 --port 5555 --workers 127.0.0.1:5556 > /tmp/phrm-manager.log 2>&1 & + /tmp/phrm worker --host 127.0.0.1 --port 5556 --name dev-worker > /tmp/phrm-worker.log 2>&1 & + + # wait until the manager reads the worker's stats + for i in $(seq 1 30); do + curl -sf localhost:5555/nodes | grep -q '"Cores":[1-9]' && break + sleep 1 + done + curl -sf localhost:5555/nodes | grep -q '"Cores":[1-9]' + + - name: build & run server and processor + run: | + go build -o /tmp/server ./cmd/server + go build -o /tmp/processor ./cmd/processor + /tmp/server > /tmp/server.log 2>&1 & + METRICS_ADDR=:9101 /tmp/processor > /tmp/processor.log 2>&1 & + # any HTTP answer (even 401/404) means the server is serving. + # it only serves after the migrations are done + curl -s -o /dev/null --retry 30 --retry-delay 1 --retry-all-errors \ + localhost:8000/status/00000000-0000-0000-0000-000000000000 + + - name: pytest + run: | + python -m pip install pytest requests + python -m pytest tests/hw2.py tests/hw3.py -v + + - name: debug logs + if: failure() + run: | + docker compose ps + docker compose logs --tail 30 postgres redis rabbitmq || true + for f in /tmp/server.log /tmp/processor.log /tmp/phrm-manager.log /tmp/phrm-worker.log; do + if [ -f "$f" ]; then echo "== $f =="; tail -n 50 "$f"; fi + done + + - name: teardown + if: always() + run: docker compose down -v diff --git a/Dockerfile b/Dockerfile index 76d3f24..4ac879f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,24 +1,28 @@ -FROM golang:1.26-alpine AS builder +FROM golang:1.26-alpine@sha256:ce864e7223ac17b1775e6fd0b4c0db580c2eb50e7953a427916379e4b92a1628 AS builder WORKDIR /src -RUN apk add --no-cache git ca-certificates +RUN apk add --no-cache git~=2.54.0-r0 ca-certificates~=20260611-r0 COPY go.mod go.sum ./ RUN go mod download -COPY . . +COPY cmd ./cmd +COPY internal ./internal +COPY migrations ./migrations RUN CGO_ENABLED=0 GOOS=linux \ go build -trimpath -ldflags="-s -w" -o /out/server ./cmd/server -FROM gcr.io/distroless/static-debian12:nonroot AS runtime +FROM gcr.io/distroless/static-debian12:nonroot@sha256:afa5c872c891853ca7fcf1f12c3edb23f7eeef36189728842dd51042ff57f7ab AS runtime -WORKDIR / +WORKDIR /app COPY --from=builder /out/server /app/server +COPY --from=builder /src/migrations /app/migrations EXPOSE 8000 +EXPOSE 9100 USER 65532:65532 diff --git a/Dockerfile.processor b/Dockerfile.processor new file mode 100644 index 0000000..c9bf7d4 --- /dev/null +++ b/Dockerfile.processor @@ -0,0 +1,24 @@ +FROM golang:1.26-alpine@sha256:ce864e7223ac17b1775e6fd0b4c0db580c2eb50e7953a427916379e4b92a1628 AS builder + +WORKDIR /src + +RUN apk add --no-cache git~=2.54.0-r0 ca-certificates~=20260611-r0 + +COPY go.mod go.sum ./ +RUN go mod download + +COPY cmd ./cmd +COPY internal ./internal + +RUN CGO_ENABLED=0 GOOS=linux \ + go build -trimpath -ldflags="-s -w" -o /out/processor ./cmd/processor + +FROM gcr.io/distroless/static-debian12:nonroot@sha256:afa5c872c891853ca7fcf1f12c3edb23f7eeef36189728842dd51042ff57f7ab AS runtime + +WORKDIR /app + +EXPOSE 9100 + +COPY --from=builder /out/processor /app/processor + +ENTRYPOINT ["/app/processor"] diff --git a/cmd/processor/main.go b/cmd/processor/main.go new file mode 100644 index 0000000..2d1dc2e --- /dev/null +++ b/cmd/processor/main.go @@ -0,0 +1,86 @@ +package main + +import ( + "context" + "fmt" + "log" + "os/signal" + "syscall" + "time" + + "github.com/belyaevedu/remote-code-service/internal/config" + "github.com/belyaevedu/remote-code-service/internal/controller" + "github.com/belyaevedu/remote-code-service/internal/domain" + "github.com/belyaevedu/remote-code-service/internal/repository/postgres" + "github.com/belyaevedu/remote-code-service/internal/repository/queue" + "github.com/belyaevedu/remote-code-service/internal/service" +) + +func main() { + dbCfg, err := config.LoadDBConfig() + if err != nil { + log.Fatalf("invalid db config: %v", err) + } + queueCfg, err := config.LoadQueueConfig() + if err != nil { + log.Fatalf("invalid queue config: %v", err) + } + philCfg, err := config.LoadPhilharmonicConfig() + if err != nil { + log.Fatalf("invalid philharmonic config: %v", err) + } + metricsCfg, err := config.LoadMetricsConfig() + if err != nil { + log.Fatalf("invalid metrics config: %v", err) + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + db, err := postgres.New(ctx, dbCfg) + if err != nil { + log.Fatalf("postgres: %v", err) + } + defer db.Close() + + if err := db.WaitReady(ctx); err != nil { + log.Fatalf("waiting for schema: %v", err) + } + + phrmExecutor := service.NewPhilharmonicExecutor(philCfg) + + // pulling the sandbox image on every worker before starting to process + prewarmCtx, cancel := context.WithTimeout(ctx, 2*time.Minute) + if err := phrmExecutor.PreWarm(prewarmCtx); err != nil { + log.Printf("processor: sandbox pre-warm failed: %v (continuing)", err) + } else { + log.Printf("processor: sandbox image %q pre-warmed on workers", philCfg.SandboxImage) + } + cancel() + + executor := service.NewInstrumentedExecutor(phrmExecutor) + + consumer := queue.NewConsumer(queueCfg) + + metricsServer := controller.NewMetricsServer(metricsCfg.Addr, metricsCfg.ShutdownTimeout) + go func() { + if err := metricsServer.Start(ctx); err != nil { + log.Fatalf("metrics server exited with error: %v", err) + } + }() + + handle := func(ctx context.Context, msg domain.TaskMessage) error { + result, err := executor.Execute(ctx, msg) + if err != nil { + return fmt.Errorf("execute task %s: %w", msg.TaskID, err) + } + + return db.SaveTaskResult(ctx, msg.TaskID, &domain.Result{Output: result.Output}) + } + + log.Printf("processor: consuming queue %q", queueCfg.Queue) + if err := consumer.Consume(ctx, handle); err != nil { + log.Fatalf("consumer exited with error: %v", err) + } + log.Printf("processor stopped gracefully") +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 0746ffc..e9a93df 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -9,7 +9,9 @@ import ( "github.com/belyaevedu/remote-code-service/internal/config" "github.com/belyaevedu/remote-code-service/internal/controller" "github.com/belyaevedu/remote-code-service/internal/controller/handlers" - "github.com/belyaevedu/remote-code-service/internal/repository" + "github.com/belyaevedu/remote-code-service/internal/repository/postgres" + "github.com/belyaevedu/remote-code-service/internal/repository/queue" + "github.com/belyaevedu/remote-code-service/internal/repository/redis" "github.com/belyaevedu/remote-code-service/internal/service" ) @@ -29,24 +31,72 @@ import ( // // @securitydefinitions.bearerauth BearerAuth func main() { - cfg, err := config.Load() + appCfg, err := config.LoadAppConfig() if err != nil { - log.Fatalf("invalid config: %v", err) + log.Fatalf("invalid app config: %v", err) + } + dbCfg, err := config.LoadDBConfig() + if err != nil { + log.Fatalf("invalid db config: %v", err) + } + redisCfg, err := config.LoadRedisConfig() + if err != nil { + log.Fatalf("invalid redis config: %v", err) + } + queueCfg, err := config.LoadQueueConfig() + if err != nil { + log.Fatalf("invalid queue config: %v", err) + } + metricsCfg, err := config.LoadMetricsConfig() + if err != nil { + log.Fatalf("invalid metrics config: %v", err) } - repo := repository.New() + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() - taskService := service.NewTaskService(repo, cfg.ProcessingTime) - userService := service.NewUserService(repo, repo) + if err := postgres.Migrate(ctx, dbCfg); err != nil { + log.Fatalf("migrations: %v", err) + } + + db, err := postgres.New(ctx, dbCfg) + if err != nil { + log.Fatalf("postgres: %v", err) + } + defer db.Close() + + sessions, err := redis.New(ctx, redisCfg) + if err != nil { + log.Fatalf("redis: %v", err) + } + defer func() { + if err := sessions.Close(); err != nil { + log.Printf("closing redis: %v", err) + } + }() + + publisher := queue.NewPublisher(queueCfg) + defer func() { + if err := publisher.Close(); err != nil { + log.Printf("closing queue publisher: %v", err) + } + }() + + taskService := service.NewTaskService(db, publisher) + userService := service.NewUserService(db, sessions) taskHandler := handlers.NewTaskHandlers(taskService) userHandler := handlers.NewUserHandlers(userService) router := controller.NewRouter(taskHandler, userHandler, userService) - server := controller.NewApi(cfg.HTTPAddr, router, cfg.ShutdownTimeout) + server := controller.NewApi(appCfg.HTTPAddr, router, appCfg.ShutdownTimeout) - ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) - defer stop() + metricsServer := controller.NewMetricsServer(metricsCfg.Addr, metricsCfg.ShutdownTimeout) + go func() { + if err := metricsServer.Start(ctx); err != nil { + log.Fatalf("metrics server exited with error: %v", err) + } + }() if err := server.Start(ctx); err != nil { log.Fatalf("server exited with error: %v", err) diff --git a/dbconfig.yml b/dbconfig.yml new file mode 100644 index 0000000..eddbf37 --- /dev/null +++ b/dbconfig.yml @@ -0,0 +1,13 @@ +# sql-migrate CLI config + +development: + dialect: postgres + datasource: ${DATABASE_URL} + dir: migrations + table: gorp_migrations + +production: + dialect: postgres + datasource: ${DATABASE_URL} + dir: migrations + table: gorp_migrations diff --git a/docker-compose.yml b/docker-compose.yml index ed2fc91..371f1a1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,58 @@ services: + postgres: + image: postgres:18.6-alpine + container_name: remote-code-postgres + restart: unless-stopped + + ports: + - "127.0.0.1:${POSTGRES_PORT:-5432}:5432" + + environment: + POSTGRES_USER: ${POSTGRES_USER:-postgres} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-postgres} + POSTGRES_DB: ${POSTGRES_DB:-remote_code} + + volumes: + - postgres-data:/var/lib/postgresql + + healthcheck: + test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"] + interval: 5s + timeout: 3s + retries: 10 + + redis: + image: redis:8.10.1-alpine + container_name: remote-code-redis + restart: unless-stopped + + ports: + - "127.0.0.1:${REDIS_PORT:-6379}:6379" + + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 10 + + rabbitmq: + image: rabbitmq:4.3.5-alpine + container_name: remote-code-rabbitmq + restart: unless-stopped + + ports: + - "127.0.0.1:${RABBITMQ_PORT:-5672}:5672" + + environment: + RABBITMQ_DEFAULT_USER: ${RABBITMQ_USER:-guest} + RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASSWORD:-guest} + + healthcheck: + test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"] + interval: 10s + timeout: 5s + retries: 10 + remote-code: build: context: . @@ -8,12 +62,66 @@ services: restart: unless-stopped ports: - - "127.0.0.1:8000:8000" + - "127.0.0.1:${HTTP_PORT:-8000}:8000" + - "127.0.0.1:${METRICS_PORT:-9100}:9100" environment: HTTP_ADDR: ":8000" - TASK_PROCESSING_TIME: "2s" SHUTDOWN_TIMEOUT: "10s" + METRICS_ADDR: ":9100" + DATABASE_URL: postgres://${POSTGRES_USER:-postgres}:${POSTGRES_PASSWORD:-postgres}@postgres:5432/${POSTGRES_DB:-remote_code}?sslmode=disable + REDIS_ADDR: redis:6379 + RABBITMQ_URL: amqp://${RABBITMQ_USER:-guest}:${RABBITMQ_PASSWORD:-guest}@rabbitmq:5672/ + + depends_on: + postgres: + condition: service_healthy + redis: + condition: service_healthy + rabbitmq: + condition: service_healthy + + user: "65532:65532" + + read_only: true + tmpfs: + - /tmp:rw,noexec,nosuid,size=16m + + cap_drop: + - ALL + security_opt: + - no-new-privileges:true + + mem_limit: 256m + mem_reservation: 128m + cpus: "1.0" + + processor: + build: + context: . + dockerfile: Dockerfile.processor + image: remote-code-processor:latest + container_name: remote-code-processor + restart: unless-stopped + + ports: + - "127.0.0.1:${PROCESSOR_METRICS_PORT:-9101}:9100" + + environment: + METRICS_ADDR: ":9100" + DATABASE_URL: postgres://${POSTGRES_USER:-postgres}:${POSTGRES_PASSWORD:-postgres}@postgres:5432/${POSTGRES_DB:-remote_code}?sslmode=disable + RABBITMQ_URL: amqp://${RABBITMQ_USER:-guest}:${RABBITMQ_PASSWORD:-guest}@rabbitmq:5672/ + PHILHARMONIC_URL: ${PHILHARMONIC_URL:-http://localhost:5555} + PHILHARMONIC_TOKEN: ${PHILHARMONIC_TOKEN:-} + + depends_on: + postgres: + condition: service_healthy + rabbitmq: + condition: service_healthy + # the server owns the migrations, the processor already waits for db readiness + remote-code: + condition: service_started user: "65532:65532" @@ -30,3 +138,5 @@ services: mem_reservation: 128m cpus: "1.0" +volumes: + postgres-data: diff --git a/docs/docs.go b/docs/docs.go index 5caea76..d8df8c5 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -51,6 +51,17 @@ const docTemplate = `{ }, "type": "object" }, + "handlers.taskCreateRequest": { + "properties": { + "code": { + "type": "string" + }, + "translator": { + "type": "string" + } + }, + "type": "object" + }, "handlers.taskCreateResponse": { "properties": { "task_id": { @@ -396,15 +407,26 @@ const docTemplate = `{ }, "/task": { "post": { - "description": "Creating a task owned by an authenticated user", + "description": "Creating a task owned by an authenticated user and queuing it for execution", "requestBody": { "content": { "application/json": { "schema": { - "type": "object" + "oneOf": [ + { + "type": "object" + }, + { + "$ref": "#/components/schemas/handlers.taskCreateRequest", + "summary": "request", + "description": "task submission" + } + ] } } - } + }, + "description": "task submission", + "required": true }, "responses": { "201": { diff --git a/docs/swagger.json b/docs/swagger.json index 6bf7472..1c586cf 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -44,6 +44,17 @@ }, "type": "object" }, + "handlers.taskCreateRequest": { + "properties": { + "code": { + "type": "string" + }, + "translator": { + "type": "string" + } + }, + "type": "object" + }, "handlers.taskCreateResponse": { "properties": { "task_id": { @@ -389,15 +400,26 @@ }, "/task": { "post": { - "description": "Creating a task owned by an authenticated user", + "description": "Creating a task owned by an authenticated user and queuing it for execution", "requestBody": { "content": { "application/json": { "schema": { - "type": "object" + "oneOf": [ + { + "type": "object" + }, + { + "$ref": "#/components/schemas/handlers.taskCreateRequest", + "summary": "request", + "description": "task submission" + } + ] } } - } + }, + "description": "task submission", + "required": true }, "responses": { "201": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 57280b2..f2c634a 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -27,6 +27,13 @@ components: message: type: string type: object + handlers.taskCreateRequest: + properties: + code: + type: string + translator: + type: string + type: object handlers.taskCreateResponse: properties: task_id: @@ -238,12 +245,19 @@ paths: - task /task: post: - description: Creating a task owned by an authenticated user + description: Creating a task owned by an authenticated user and queuing it for + execution requestBody: content: application/json: schema: - type: object + oneOf: + - type: object + - $ref: '#/components/schemas/handlers.taskCreateRequest' + description: task submission + summary: request + description: task submission + required: true responses: "201": content: diff --git a/go.mod b/go.mod index 79a8576..cecef2b 100644 --- a/go.mod +++ b/go.mod @@ -8,12 +8,21 @@ require ( ) require ( + github.com/go-chi/metrics v0.1.1 + github.com/jackc/pgx/v5 v5.10.0 + github.com/rabbitmq/amqp091-go v1.14.0 + github.com/redis/go-redis/v9 v9.22.0 + github.com/rubenv/sql-migrate v1.8.1 github.com/swaggo/swag/v2 v2.0.0-rc5 golang.org/x/crypto v0.55.0 ) require ( github.com/KyleBanks/depth v1.2.1 // indirect + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/go-gorp/gorp/v3 v3.1.0 // indirect github.com/go-openapi/jsonpointer v1.0.0 // indirect github.com/go-openapi/jsonreference v1.0.1 // indirect github.com/go-openapi/spec v0.22.11 // indirect @@ -24,9 +33,23 @@ require ( github.com/go-openapi/swag/stringutils v0.29.1 // indirect github.com/go-openapi/swag/typeutils v0.29.1 // indirect github.com/go-openapi/swag/yamlutils v0.29.1 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_golang v1.22.0 // indirect + github.com/prometheus/client_model v0.6.1 // indirect + github.com/prometheus/common v0.62.0 // indirect + github.com/prometheus/procfs v0.15.1 // indirect + github.com/rogpeppe/go-internal v1.13.1 // indirect github.com/sv-tools/openapi v0.4.0 // indirect + go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/mod v0.40.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect golang.org/x/tools v0.49.0 // indirect + google.golang.org/protobuf v1.36.5 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 899ebd5..d47d630 100644 --- a/go.sum +++ b/go.sum @@ -1,9 +1,22 @@ github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc= github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-chi/chi/v5 v5.3.1 h1:3j4HZLGZQ3JpMCrPJF/Jl3mYJfWLKBfNJ6quurUGCf8= github.com/go-chi/chi/v5 v5.3.1/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto= +github.com/go-chi/metrics v0.1.1 h1:CXhbnkAVVjb0k73EBRQ6Z2YdWFnbXZgNtg1Mboguibk= +github.com/go-chi/metrics v0.1.1/go.mod h1:mcGTM1pPalP7WCtb+akNYFO/lwNwBBLCuedepqjoPn4= +github.com/go-gorp/gorp/v3 v3.1.0 h1:ItKF/Vbuj31dmV4jxA1qblpSwkl9g1typ24xoe70IGs= +github.com/go-gorp/gorp/v3 v3.1.0/go.mod h1:dLEjIyyRNiXvNZ8PSmzpt1GsWAUK8kjVhEpjH8TixEw= github.com/go-openapi/jsonpointer v1.0.0 h1:kR9tHqY0CtZaOPVFm622dPVNhrvYpwr4uCxgL3h1H8s= github.com/go-openapi/jsonpointer v1.0.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y= github.com/go-openapi/jsonreference v1.0.1 h1:4zJ7AmYDKNmD3aSpfPnFNCFA5E80/xMHUNKgydaLh38= @@ -31,16 +44,71 @@ github.com/go-openapi/testify/enable/yaml/v2 v2.6.1 h1:Jm+/ze2rMtbD98yen92AhATGL github.com/go-openapi/testify/enable/yaml/v2 v2.6.1/go.mod h1:YDPnwCRDu38/oJBVMBVXOUDiJ9cIeBHWvfImHaXqnv4= github.com/go-openapi/testify/v2 v2.6.1 h1:6CNJhTjMzgaeaH8WhshcsZNPIvRemiOcFpU7seO/y7Q= github.com/go-openapi/testify/v2 v2.6.1/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw= +github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE= +github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/lib/pq v1.10.7 h1:p7ZhMD+KsSRozJr34udlUrhboJwWAgCg34+/ZZNvZZw= +github.com/lib/pq v1.10.7/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/mattn/go-sqlite3 v1.14.19 h1:fhGleo2h1p8tVChob4I9HpmVFIAkKGpiukdrgQbWfGI= +github.com/mattn/go-sqlite3 v1.14.19/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= -github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/poy/onpar v1.1.2 h1:QaNrNiZx0+Nar5dLgTVp5mXkyoVFIbepjyEoGSnhbAY= +github.com/poy/onpar v1.1.2/go.mod h1:6X8FLNoxyr9kkmnlqpK6LSoiOtrO6MICtWwEuWkLjzg= +github.com/prometheus/client_golang v1.22.0 h1:rb93p9lokFEsctTys46VnV1kLCDpVZ0a/Y92Vm0Zc6Q= +github.com/prometheus/client_golang v1.22.0/go.mod h1:R7ljNsLXhuQXYZYtw6GAE9AZg8Y7vEW5scdCXrWRXC0= +github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E= +github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY= +github.com/prometheus/common v0.62.0 h1:xasJaQlnWAeyHdUBeGjXmutelfJHWMRr+Fg4QszZ2Io= +github.com/prometheus/common v0.62.0/go.mod h1:vyBcEuLSvWos9B1+CyL7JZ2up+uFzXhkqml0W5zIY1I= +github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc= +github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= +github.com/rabbitmq/amqp091-go v1.14.0 h1:RSaT7aOKt/OrkVUyswPDW29lnRz9psuGmfZFBmLqLek= +github.com/rabbitmq/amqp091-go v1.14.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= +github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= +github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= +github.com/rubenv/sql-migrate v1.8.1 h1:EPNwCvjAowHI3TnZ+4fQu3a915OpnQoPAjTXCGOy2U0= +github.com/rubenv/sql-migrate v1.8.1/go.mod h1:BTIKBORjzyxZDS6dzoiw6eAFYJ1iNlGAtjn4LGeVjS8= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/sv-tools/openapi v0.4.0 h1:UhD9DVnGox1hfTePNclpUzUFgos57FvzT2jmcAuTOJ4= github.com/sv-tools/openapi v0.4.0/go.mod h1:kD/dG+KP0+Fom1r6nvcj/ORtLus8d8enXT6dyRZDirE= github.com/swaggo/swag/v2 v2.0.0-rc5 h1:fK7d6ET9rrEsdB8IyuwXREWMcyQN3N7gawGFbbrjgHk= github.com/swaggo/swag/v2 v2.0.0-rc5/go.mod h1:kCL8Fu4Zl8d5tB2Bgj96b8wRowwrwk175bZHXfuGVFI= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= @@ -49,9 +117,17 @@ golang.org/x/mod v0.40.0 h1:hUv+3cXcdRHz08UmSiOob7sadHig73uo5bkXxQ/tvUs= golang.org/x/mod v0.40.0/go.mod h1:0/weTWkPWGBikyTWAX3dkjVztMmBA5hM0DH6BElSupE= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI= golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM= +google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/config/app.go b/internal/config/app.go new file mode 100644 index 0000000..7cc648c --- /dev/null +++ b/internal/config/app.go @@ -0,0 +1,33 @@ +package config + +import ( + "fmt" + "time" +) + +type AppConfig struct { + HTTPAddr string + ShutdownTimeout time.Duration +} + +const ( + envVarAppAddress = "HTTP_ADDR" + envVarAppShutdownTimeout = "SHUTDOWN_TIMEOUT" +) + +const ( + defaultAppAddress = ":8000" + defaultAppShutdownTimeout = 10 * time.Second +) + +func LoadAppConfig() (AppConfig, error) { + appCfg := AppConfig{ + HTTPAddr: envString(envVarAppAddress, defaultAppAddress), + ShutdownTimeout: envDuration(envVarAppShutdownTimeout, defaultAppShutdownTimeout), + } + + if appCfg.ShutdownTimeout <= 0 { + return AppConfig{}, fmt.Errorf("%s must be positive, got %s", envVarAppShutdownTimeout, appCfg.ShutdownTimeout) + } + return appCfg, nil +} diff --git a/internal/config/config.go b/internal/config/config.go deleted file mode 100644 index d739f9c..0000000 --- a/internal/config/config.go +++ /dev/null @@ -1,54 +0,0 @@ -package config - -import ( - "fmt" - "os" - "strings" - "time" -) - -type Config struct { - HTTPAddr string - ProcessingTime time.Duration - ShutdownTimeout time.Duration -} - -const ( - envVarAddress = "HTTP_ADDR" - envVarProcessing = "TASK_PROCESSING_TIME" - envVarShutdownTimeout = "SHUTDOWN_TIMEOUT" -) - -func Load() (Config, error) { - cfg := Config{ - HTTPAddr: envString(envVarAddress, ":8000"), - ProcessingTime: envDuration(envVarProcessing, 2*time.Second), - ShutdownTimeout: envDuration(envVarShutdownTimeout, 10*time.Second), - } - - if cfg.ProcessingTime <= 0 { - return Config{}, fmt.Errorf("TASK_PROCESSING_TIME must be positive, got %s", cfg.ProcessingTime) - } - if cfg.ShutdownTimeout <= 0 { - return Config{}, fmt.Errorf("SHUTDOWN_TIMEOUT must be positive, got %s", cfg.ShutdownTimeout) - } - return cfg, nil -} - -func envString(key, def string) string { - if v, ok := os.LookupEnv(key); ok && strings.TrimSpace(v) != "" { - return v - } - return def -} - -func envDuration(key string, def time.Duration) time.Duration { - if v, ok := os.LookupEnv(key); ok && strings.TrimSpace(v) != "" { - d, err := time.ParseDuration(strings.TrimSpace(v)) - if err != nil { - return def - } - return d - } - return def -} diff --git a/internal/config/db.go b/internal/config/db.go new file mode 100644 index 0000000..0d03d82 --- /dev/null +++ b/internal/config/db.go @@ -0,0 +1,47 @@ +package config + +import ( + "fmt" + "net/url" +) + +type DBConfig struct { + URL string + MigrationsDir string +} + +const ( + envVarDBURL = "DATABASE_URL" + envVarDBMigrations = "MIGRATIONS_DIR" +) + +const ( + defaultDBURL = "postgres://postgres:postgres@localhost:5432/remote_code?sslmode=disable" + defaultDBMigrations = "migrations" +) + +func LoadDBConfig() (DBConfig, error) { + dbCfg := DBConfig{ + URL: envString(envVarDBURL, defaultDBURL), + MigrationsDir: envString(envVarDBMigrations, defaultDBMigrations), + } + + if err := validateDBConfig(dbCfg); err != nil { + return DBConfig{}, err + } + return dbCfg, nil +} + +func validateDBConfig(dbCfg DBConfig) error { + if dbCfg.URL == "" { + return fmt.Errorf("%s must not be empty", envVarDBURL) + } + u, err := url.Parse(dbCfg.URL) + if err != nil || (u.Scheme != "postgres" && u.Scheme != "postgresql") || u.Host == "" { + return fmt.Errorf("%s must be a valid postgres:// URL, got %q", envVarDBURL, dbCfg.URL) + } + if dbCfg.MigrationsDir == "" { + return fmt.Errorf("%s must not be empty", envVarDBMigrations) + } + return nil +} diff --git a/internal/config/helpers.go b/internal/config/helpers.go new file mode 100644 index 0000000..688c3d8 --- /dev/null +++ b/internal/config/helpers.go @@ -0,0 +1,53 @@ +package config + +import ( + "os" + "strconv" + "strings" + "time" +) + +func envString(key, def string) string { + if v, ok := os.LookupEnv(key); ok && strings.TrimSpace(v) != "" { + return v + } + return def +} + +func envDuration(key string, def time.Duration) time.Duration { + if v, ok := os.LookupEnv(key); ok && strings.TrimSpace(v) != "" { + d, err := time.ParseDuration(strings.TrimSpace(v)) + if err != nil { + return def + } + return d + } + return def +} + +func envFloat(key string, def float64) float64 { + if v, ok := os.LookupEnv(key); ok { + if f, err := strconv.ParseFloat(strings.TrimSpace(v), 64); err == nil { + return f + } + } + return def +} + +func envInt(key string, def int) int { + if v, ok := os.LookupEnv(key); ok { + if i, err := strconv.Atoi(strings.TrimSpace(v)); err == nil { + return i + } + } + return def +} + +func envInt64(key string, def int64) int64 { + if v, ok := os.LookupEnv(key); ok { + if i, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64); err == nil { + return i + } + } + return def +} diff --git a/internal/config/metrics.go b/internal/config/metrics.go new file mode 100644 index 0000000..75d309a --- /dev/null +++ b/internal/config/metrics.go @@ -0,0 +1,33 @@ +package config + +import ( + "fmt" + "time" +) + +type MetricsConfig struct { + Addr string + ShutdownTimeout time.Duration +} + +const ( + envVarMetricsAddress = "METRICS_ADDR" + envVarMetricsShutdownTimeout = "METRICS_SHUTDOWN_TIMEOUT" +) + +const ( + defaultMetricsAddress = ":9100" + defaultMetricsShutdownTimeout = 5 * time.Second +) + +func LoadMetricsConfig() (MetricsConfig, error) { + metricsCfg := MetricsConfig{ + Addr: envString(envVarMetricsAddress, defaultMetricsAddress), + ShutdownTimeout: envDuration(envVarMetricsShutdownTimeout, defaultMetricsShutdownTimeout), + } + + if metricsCfg.ShutdownTimeout <= 0 { + return MetricsConfig{}, fmt.Errorf("%s must be positive, got %s", envVarMetricsShutdownTimeout, metricsCfg.ShutdownTimeout) + } + return metricsCfg, nil +} diff --git a/internal/config/philharmonic.go b/internal/config/philharmonic.go new file mode 100644 index 0000000..dab8579 --- /dev/null +++ b/internal/config/philharmonic.go @@ -0,0 +1,95 @@ +package config + +import ( + "fmt" + "net/http" + "net/url" + "strings" + "time" +) + +const ( + envVarPhilURL = "PHILHARMONIC_URL" + envVarPhilToken = "PHILHARMONIC_TOKEN" + envVarPhilImage = "PHILHARMONIC_IMAGE" + envVarPhilTaskTimeout = "PHILHARMONIC_TASK_TIMEOUT" + envVarPhilPollInterval = "PHILHARMONIC_POLL_INTERVAL" + envVarPhilPollTimeout = "PHILHARMONIC_POLL_TIMEOUT" + envVarPhilCpu = "PHILHARMONIC_CPU" + envVarPhilMemory = "PHILHARMONIC_MEMORY" +) + +const ( + defaultPhilURL = "http://localhost:5555" + defaultPhilImage = "sandbox:latest" + defaultPhilTaskTimeout = 30 * time.Second + defaultPhilPollInterval = time.Second + defaultPhilCpu = 0.5 + defaultPhilMemory = 256 << 20 // bytes +) + +type PhilharmonicConfig struct { + BaseURL string + Token string // empty means the manager runs without auth + SandboxImage string + TaskTimeout time.Duration + PollInterval time.Duration + PollTimeout time.Duration + + // per-task resource limits + Cpu float64 + Memory int64 // bytes + + // HTTPClient overrides the transport, mainly for tests; nil = default + HTTPClient *http.Client +} + +// called by Load +func LoadPhilharmonicConfig() (PhilharmonicConfig, error) { + phrmCfg := PhilharmonicConfig{ + BaseURL: envString(envVarPhilURL, defaultPhilURL), + Token: envString(envVarPhilToken, ""), + SandboxImage: envString(envVarPhilImage, defaultPhilImage), + TaskTimeout: envDuration(envVarPhilTaskTimeout, defaultPhilTaskTimeout), + PollInterval: envDuration(envVarPhilPollInterval, defaultPhilPollInterval), + PollTimeout: envDuration(envVarPhilPollTimeout, 0), // derived below + Cpu: envFloat(envVarPhilCpu, defaultPhilCpu), + Memory: envInt64(envVarPhilMemory, defaultPhilMemory), + } + + if phrmCfg.PollTimeout <= 0 { + phrmCfg.PollTimeout = phrmCfg.TaskTimeout + 15*time.Second + } + + if err := validatePhilharmonic(phrmCfg); err != nil { + return PhilharmonicConfig{}, err + } + return phrmCfg, nil +} + +func validatePhilharmonic(opts PhilharmonicConfig) error { + u, err := url.Parse(opts.BaseURL) + if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") { + return fmt.Errorf("%s must be a valid http(s) URL, got %q", envVarPhilURL, opts.BaseURL) + } + if strings.TrimSpace(opts.SandboxImage) == "" { + return fmt.Errorf("%s must not be empty", envVarPhilImage) + } + if opts.TaskTimeout <= 0 { + return fmt.Errorf("%s must be positive, got %s", envVarPhilTaskTimeout, opts.TaskTimeout) + } + if opts.PollInterval <= 0 { + return fmt.Errorf("%s must be positive, got %s", envVarPhilPollInterval, opts.PollInterval) + } + if opts.PollTimeout < opts.TaskTimeout { + return fmt.Errorf("%s (%s) must exceed %s (%s): the worker enforces the kill switch asynchronously", + envVarPhilPollTimeout, opts.PollTimeout, envVarPhilTaskTimeout, opts.TaskTimeout) + } + if opts.Cpu <= 0 { + return fmt.Errorf("%s must be positive, got %v", envVarPhilCpu, opts.Cpu) + } + if opts.Memory <= 0 { + return fmt.Errorf("%s must be positive (bytes), got %d", envVarPhilMemory, opts.Memory) + } + return nil +} diff --git a/internal/config/queue.go b/internal/config/queue.go new file mode 100644 index 0000000..ca10f96 --- /dev/null +++ b/internal/config/queue.go @@ -0,0 +1,59 @@ +package config + +import ( + "fmt" + "net/url" + "time" +) + +type QueueConfig struct { + URL string + Queue string + Prefetch int + ReconnectDelay time.Duration +} + +const ( + envVarQueueURL = "RABBITMQ_URL" + envVarQueueName = "TASK_QUEUE_NAME" + envVarQueuePrefetch = "TASK_QUEUE_PREFETCH" + envVarQueueReconnect = "TASK_QUEUE_RECONNECT_DELAY" +) + +const ( + defaultQueueURL = "amqp://guest:guest@localhost:5672/" + defaultQueueName = "tasks" + defaultQueuePrefetch = 1 + defaultQueueReconnect = time.Second +) + +func LoadQueueConfig() (QueueConfig, error) { + queueCfg := QueueConfig{ + URL: envString(envVarQueueURL, defaultQueueURL), + Queue: envString(envVarQueueName, defaultQueueName), + Prefetch: envInt(envVarQueuePrefetch, defaultQueuePrefetch), + ReconnectDelay: envDuration(envVarQueueReconnect, defaultQueueReconnect), + } + + if err := validateQueueConfig(queueCfg); err != nil { + return QueueConfig{}, err + } + return queueCfg, nil +} + +func validateQueueConfig(queueCfg QueueConfig) error { + u, err := url.Parse(queueCfg.URL) + if err != nil || u.Host == "" || (u.Scheme != "amqp" && u.Scheme != "amqps") { + return fmt.Errorf("%s must be a valid amqp(s) URL, got %q", envVarQueueURL, queueCfg.URL) + } + if queueCfg.Queue == "" { + return fmt.Errorf("%s must not be empty", envVarQueueName) + } + if queueCfg.Prefetch <= 0 { + return fmt.Errorf("%s must be positive, got %d", envVarQueuePrefetch, queueCfg.Prefetch) + } + if queueCfg.ReconnectDelay <= 0 { + return fmt.Errorf("%s must be positive, got %s", envVarQueueReconnect, queueCfg.ReconnectDelay) + } + return nil +} diff --git a/internal/config/redis.go b/internal/config/redis.go new file mode 100644 index 0000000..e0e0d40 --- /dev/null +++ b/internal/config/redis.go @@ -0,0 +1,53 @@ +package config + +import ( + "fmt" + "time" +) + +type RedisConfig struct { + Addr string + Password string // empty means no auth + DB int // logical database index + SessionTTL time.Duration +} + +const ( + envVarRedisAddr = "REDIS_ADDR" + envVarRedisPassword = "REDIS_PASSWORD" + envVarRedisDB = "REDIS_DB" + envVarRedisTTL = "SESSION_TTL" +) + +const ( + defaultRedisAddr = "localhost:6379" + defaultRedisDB = 0 + defaultRedisTTL = 7 * 24 * time.Hour +) + +func LoadRedisConfig() (RedisConfig, error) { + redisCfg := RedisConfig{ + Addr: envString(envVarRedisAddr, defaultRedisAddr), + Password: envString(envVarRedisPassword, ""), + DB: envInt(envVarRedisDB, defaultRedisDB), + SessionTTL: envDuration(envVarRedisTTL, defaultRedisTTL), + } + + if err := validateRedisConfig(redisCfg); err != nil { + return RedisConfig{}, err + } + return redisCfg, nil +} + +func validateRedisConfig(redisCfg RedisConfig) error { + if redisCfg.Addr == "" { + return fmt.Errorf("%s must not be empty", envVarRedisAddr) + } + if redisCfg.DB < 0 { + return fmt.Errorf("%s must not be negative, got %d", envVarRedisDB, redisCfg.DB) + } + if redisCfg.SessionTTL <= 0 { + return fmt.Errorf("%s must be positive, got %s", envVarRedisTTL, redisCfg.SessionTTL) + } + return nil +} diff --git a/internal/controller/handlers/middleware.go b/internal/controller/handlers/middleware.go index 6688786..4b374ad 100644 --- a/internal/controller/handlers/middleware.go +++ b/internal/controller/handlers/middleware.go @@ -36,7 +36,7 @@ func AuthMiddleware(auth port.AuthService) func(http.Handler) http.Handler { return } - userID, err := auth.Authenticate(token) + userID, err := auth.Authenticate(r.Context(), token) if err != nil { unauthorizedResponseHelper(w, "invalid or expired session") return diff --git a/internal/controller/handlers/task.go b/internal/controller/handlers/task.go index fe35762..1aa1812 100644 --- a/internal/controller/handlers/task.go +++ b/internal/controller/handlers/task.go @@ -1,6 +1,7 @@ package handlers import ( + "encoding/json" "errors" "net/http" @@ -23,6 +24,11 @@ func NewTaskHandlers(taskSvc port.TaskService) *TaskHandlers { } // POST /task +type taskCreateRequest struct { + Translator string `json:"translator"` + Code string `json:"code"` +} + type taskCreateResponse struct { TaskID string `json:"task_id"` } @@ -38,10 +44,11 @@ type taskResultResponse struct { } // @Summary Create a task -// @Description Creating a task owned by an authenticated user +// @Description Creating a task owned by an authenticated user and queuing it for execution // @Tags task // @Accept json // @Produce json +// @Param request body taskCreateRequest true "task submission" // @Success 201 {object} taskCreateResponse // @Failure 400 {object} ErrorResponse // @Failure 401 {object} ErrorResponse @@ -55,8 +62,23 @@ func (h *TaskHandlers) Create(w http.ResponseWriter, r *http.Request) { return } - id, err := h.taskSvc.Submit(userID) + var req taskCreateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + WriteJSON(w, http.StatusBadRequest, ErrorResponse{Error: "invalid request body"}) + return + } + + submission := domain.Submission{ + Translator: req.Translator, + Code: req.Code, + } + + id, err := h.taskSvc.Submit(r.Context(), userID, submission) if err != nil { + if errors.Is(err, domain.ErrUnsupportedTranslator) || errors.Is(err, domain.ErrInvalidSubmission) { + WriteJSON(w, http.StatusBadRequest, ErrorResponse{Error: err.Error()}) + return + } WriteJSON(w, http.StatusInternalServerError, ErrorResponse{Error: err.Error()}) return } @@ -84,7 +106,7 @@ func (h *TaskHandlers) Status(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "task_id") - status, err := h.taskSvc.Status(userID, id) + status, err := h.taskSvc.Status(r.Context(), userID, id) if err != nil { writeTaskError(w, err) return @@ -113,7 +135,7 @@ func (h *TaskHandlers) Result(w http.ResponseWriter, r *http.Request) { id := chi.URLParam(r, "task_id") - result, err := h.taskSvc.Result(userID, id) + result, err := h.taskSvc.Result(r.Context(), userID, id) if err != nil { writeTaskError(w, err) return diff --git a/internal/controller/handlers/user.go b/internal/controller/handlers/user.go index 6e07d7c..ef2c407 100644 --- a/internal/controller/handlers/user.go +++ b/internal/controller/handlers/user.go @@ -52,7 +52,7 @@ func (h *UserHandlers) Register(w http.ResponseWriter, r *http.Request) { return } - if err := h.userSvc.Register(req.Username, req.Password); err != nil { + if err := h.userSvc.Register(r.Context(), req.Username, req.Password); err != nil { if errors.Is(err, domain.ErrInvalidCredentials) { WriteJSON(w, http.StatusBadRequest, ErrorResponse{Error: err.Error()}) return @@ -87,7 +87,7 @@ func (h *UserHandlers) Login(w http.ResponseWriter, r *http.Request) { return } - token, err := h.userSvc.Login(req.Username, req.Password) + token, err := h.userSvc.Login(r.Context(), req.Username, req.Password) if err != nil { if errors.Is(err, domain.ErrInvalidCredentials) { WriteJSON(w, http.StatusUnauthorized, ErrorResponse{Error: err.Error()}) diff --git a/internal/controller/metrics.go b/internal/controller/metrics.go new file mode 100644 index 0000000..240bcc3 --- /dev/null +++ b/internal/controller/metrics.go @@ -0,0 +1,62 @@ +package controller + +import ( + "context" + "errors" + "log" + "net/http" + "time" + + chimetrics "github.com/go-chi/metrics" +) + +type MetricsServer struct { + httpServer *http.Server + shutdown time.Duration +} + +func NewMetricsServer(address string, shutdownTimeout time.Duration) *MetricsServer { + if shutdownTimeout <= 0 { + shutdownTimeout = 5 * time.Second + } + + mux := http.NewServeMux() + mux.Handle("GET /metrics", chimetrics.Handler()) + + return &MetricsServer{ + httpServer: &http.Server{ + Addr: address, + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + }, + shutdown: shutdownTimeout, + } +} + +func (s *MetricsServer) Start(ctx context.Context) error { + errCh := make(chan error, 1) + go func() { + if err := s.httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + + log.Printf("metrics server listening on %s", s.httpServer.Addr) + + select { + case err := <-errCh: + return err + case <-ctx.Done(): + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), s.shutdown) + defer cancel() + + if err := s.httpServer.Shutdown(shutdownCtx); err != nil { + log.Printf("metrics server graceful shutdown failed: %v\n", err) + return err + } + return nil +} diff --git a/internal/controller/router.go b/internal/controller/router.go index ab1a8f4..29e44e5 100644 --- a/internal/controller/router.go +++ b/internal/controller/router.go @@ -2,6 +2,7 @@ package controller import ( "github.com/go-chi/chi/v5" + chimetrics "github.com/go-chi/metrics" "github.com/belyaevedu/remote-code-service/internal/controller/handlers" "github.com/belyaevedu/remote-code-service/internal/port" @@ -10,6 +11,8 @@ import ( func NewRouter(t *handlers.TaskHandlers, u *handlers.UserHandlers, auth port.AuthService) *chi.Mux { r := chi.NewRouter() + r.Use(chimetrics.Collector(chimetrics.CollectorOpts{})) + r.Post("/register", u.Register) r.Post("/login", u.Login) diff --git a/internal/domain/task.go b/internal/domain/task.go index 826d8f3..76ad047 100644 --- a/internal/domain/task.go +++ b/internal/domain/task.go @@ -5,8 +5,11 @@ import ( ) var ( - ErrTaskNotFound = errors.New("task not found") - ErrAccessDenied = errors.New("access denied") + ErrTaskNotFound = errors.New("task not found") + ErrAccessDenied = errors.New("access denied") + ErrUnsupportedTranslator = errors.New("unsupported translator") + ErrExecutionTimeout = errors.New("execution timed out") + ErrInvalidSubmission = errors.New("invalid submission") ) type TaskStatus string @@ -21,8 +24,27 @@ type Result struct { } type Task struct { - ID string - UserID string - Status TaskStatus - Result *Result + ID string + UserID string + Status TaskStatus + Translator string + Result *Result +} + +// pre-creation api payload +type Submission struct { + Translator string + Code string +} + +type TaskMessage struct { + TaskID string `json:"task_id"` + Translator string `json:"translator"` + Code string `json:"code"` +} + +type ExecutionResult struct { + Output string + ExitCode int // -1 when unavailable + Failed bool } diff --git a/internal/port/port.go b/internal/port/port.go index d21ce0b..afca8b2 100644 --- a/internal/port/port.go +++ b/internal/port/port.go @@ -1,39 +1,57 @@ package port import ( + "context" + "github.com/belyaevedu/remote-code-service/internal/domain" ) type TaskRepository interface { - Save(task *domain.Task) error - Get(id string) (*domain.Task, error) - UpdateStatus(id string, status domain.TaskStatus) error - SaveResult(id string, result *domain.Result) error + SaveTask(ctx context.Context, task *domain.Task) error + GetTask(ctx context.Context, id string) (*domain.Task, error) + UpdateTaskStatus(ctx context.Context, id string, status domain.TaskStatus) error + SaveTaskResult(ctx context.Context, id string, result *domain.Result) error } type TaskService interface { - Submit(userID string) (string, error) - Status(userID, id string) (domain.TaskStatus, error) - Result(userID, id string) (*domain.Result, error) + Submit(ctx context.Context, userID string, sub domain.Submission) (string, error) + Status(ctx context.Context, userID, id string) (domain.TaskStatus, error) + Result(ctx context.Context, userID, id string) (*domain.Result, error) } type UserRepository interface { - SaveUser(user *domain.User) error - GetUserByID(id string) (*domain.User, error) - GetUserByLogin(login string) (*domain.User, error) + SaveUser(ctx context.Context, user *domain.User) error + GetUserByID(ctx context.Context, id string) (*domain.User, error) + GetUserByLogin(ctx context.Context, login string) (*domain.User, error) } type UserService interface { - Register(login, password string) error - Login(login, password string) (string, error) + Register(ctx context.Context, login, password string) error + Login(ctx context.Context, login, password string) (string, error) } type SessionRepository interface { - CreateSession(session *domain.Session) error - GetSession(sessionID string) (*domain.Session, error) - DeleteSession(sessionID string) error + CreateSession(ctx context.Context, session *domain.Session) error + GetSession(ctx context.Context, sessionID string) (*domain.Session, error) + DeleteSession(ctx context.Context, sessionID string) error } type AuthService interface { - Authenticate(token string) (string, error) + Authenticate(ctx context.Context, token string) (string, error) +} + +type CodeExecutor interface { + Execute(ctx context.Context, msg domain.TaskMessage) (domain.ExecutionResult, error) +} + +type TaskPublisher interface { + Publish(ctx context.Context, msg domain.TaskMessage) error +} + +// processes a single task message consumed from the queue +// returning an error requeues the message once. a message that already failed once is dropped +type TaskHandler func(ctx context.Context, msg domain.TaskMessage) error + +type TaskConsumer interface { + Consume(ctx context.Context, handler TaskHandler) error } diff --git a/internal/repository/postgres/migrate.go b/internal/repository/postgres/migrate.go new file mode 100644 index 0000000..ebc82b9 --- /dev/null +++ b/internal/repository/postgres/migrate.go @@ -0,0 +1,37 @@ +package postgres + +import ( + "context" + "database/sql" + "fmt" + "log" + + _ "github.com/jackc/pgx/v5/stdlib" + migrate "github.com/rubenv/sql-migrate" + + "github.com/belyaevedu/remote-code-service/internal/config" +) + +func Migrate(ctx context.Context, cfg config.DBConfig) error { + db, err := sql.Open("pgx", cfg.URL) + if err != nil { + return fmt.Errorf("migrate open: %w", err) + } + defer func() { + if err := db.Close(); err != nil { + log.Printf("error raised closing migration db: %v\n", err) + } + }() + + source := &migrate.FileMigrationSource{Dir: cfg.MigrationsDir} + + applied, err := migrate.ExecContext(ctx, db, "postgres", source, migrate.Up) + if err != nil { + return fmt.Errorf("migrate up: %w", err) + } + + if applied > 0 { + log.Printf("applied %d migration(s) from %s", applied, cfg.MigrationsDir) + } + return nil +} diff --git a/internal/repository/postgres/repository.go b/internal/repository/postgres/repository.go new file mode 100644 index 0000000..087ca38 --- /dev/null +++ b/internal/repository/postgres/repository.go @@ -0,0 +1,89 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/belyaevedu/remote-code-service/internal/config" + "github.com/belyaevedu/remote-code-service/internal/domain" + "github.com/belyaevedu/remote-code-service/internal/port" +) + +const ( + pgUniqueViolation = "23505" // login already taken +) + +// stores users and tasks +type Repository struct { + pool *pgxpool.Pool +} + +var ( + _ port.TaskRepository = (*Repository)(nil) + _ port.UserRepository = (*Repository)(nil) +) + +func New(ctx context.Context, cfg config.DBConfig) (*Repository, error) { + pool, err := pgxpool.New(ctx, cfg.URL) + if err != nil { + return nil, fmt.Errorf("postgres pool: %w", err) + } + + pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + if err := pool.Ping(pingCtx); err != nil { + pool.Close() + return nil, fmt.Errorf("postgres ping: %w", err) + } + + return &Repository{pool: pool}, nil +} + +func (r *Repository) Close() { + r.pool.Close() +} + +func mapPgError(err error) error { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == pgUniqueViolation { + return domain.ErrUserAlreadyExists + } + return err +} + +const ( + schemaWaitTimeout = time.Minute + schemaWaitInterval = 500 * time.Millisecond +) + +// blocks until the schema applied by the server's migrations is visible. +// an error is returned once schemaWaitTimeout lapses +func (r *Repository) WaitReady(ctx context.Context) error { + deadline := time.Now().Add(schemaWaitTimeout) + + for { + var tasksPresent bool + err := r.pool.QueryRow(ctx, + `SELECT to_regclass('public.tasks') IS NOT NULL`, + ).Scan(&tasksPresent) + + if err == nil && tasksPresent { + return nil + } + + if time.Now().After(deadline) { + return fmt.Errorf("schema not ready after %s: has the server migrated?", schemaWaitTimeout) + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(schemaWaitInterval): + } + } +} diff --git a/internal/repository/postgres/task.go b/internal/repository/postgres/task.go new file mode 100644 index 0000000..63b1af8 --- /dev/null +++ b/internal/repository/postgres/task.go @@ -0,0 +1,91 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + + "github.com/jackc/pgx/v5" + + "github.com/belyaevedu/remote-code-service/internal/domain" +) + +func (r *Repository) SaveTask(ctx context.Context, task *domain.Task) error { + _, err := r.pool.Exec(ctx, + `INSERT INTO tasks (id, user_id, status, translator) VALUES ($1, $2, $3, $4)`, + task.ID, task.UserID, task.Status, task.Translator, + ) + return err +} + +func (r *Repository) GetTask(ctx context.Context, id string) (*domain.Task, error) { + var ( + userID string + status domain.TaskStatus + translator string + result []byte + ) + + err := r.pool.QueryRow(ctx, + `SELECT user_id, status, translator, result FROM tasks WHERE id = $1`, id, + ).Scan(&userID, &status, &translator, &result) + if errors.Is(err, pgx.ErrNoRows) { + return nil, domain.ErrTaskNotFound + } + if err != nil { + return nil, err + } + + task := &domain.Task{ + ID: id, + UserID: userID, + Status: status, + Translator: translator, + } + if result != nil { + var res domain.Result + if err := json.Unmarshal(result, &res); err != nil { + return nil, err + } + task.Result = &res + } + return task, nil +} + +func (r *Repository) UpdateTaskStatus(ctx context.Context, id string, status domain.TaskStatus) error { + tag, err := r.pool.Exec(ctx, + `UPDATE tasks SET status = $1 WHERE id = $2`, status, id, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return domain.ErrTaskNotFound + } + return nil +} + +func (r *Repository) SaveTaskResult(ctx context.Context, id string, result *domain.Result) error { + if result == nil { + result = &domain.Result{} + } + + data, err := json.Marshal(result) + if err != nil { + return err + } + + tag, err := r.pool.Exec(ctx, + `UPDATE tasks + SET status = 'ready', result = $1, finished_at = now() + WHERE id = $2`, + data, id, + ) + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return domain.ErrTaskNotFound + } + return nil +} diff --git a/internal/repository/postgres/user.go b/internal/repository/postgres/user.go new file mode 100644 index 0000000..871940f --- /dev/null +++ b/internal/repository/postgres/user.go @@ -0,0 +1,49 @@ +package postgres + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" + + "github.com/belyaevedu/remote-code-service/internal/domain" +) + +func (r *Repository) SaveUser(ctx context.Context, user *domain.User) error { + // using the UNIQUE constraint on login + _, err := r.pool.Exec(ctx, + `INSERT INTO users (id, login, password_hash) VALUES ($1, $2, $3)`, + user.ID, user.Login, user.Password, + ) + return mapPgError(err) +} + +func (r *Repository) GetUserByID(ctx context.Context, id string) (*domain.User, error) { + return r.getUser(ctx, "SELECT id, login, password_hash FROM users WHERE id = $1", id) +} + +func (r *Repository) GetUserByLogin(ctx context.Context, login string) (*domain.User, error) { + return r.getUser(ctx, "SELECT id, login, password_hash FROM users WHERE login = $1", login) +} + +func (r *Repository) getUser(ctx context.Context, query, arg string) (*domain.User, error) { + var ( + id string + found string + password string + ) + + err := r.pool.QueryRow(ctx, query, arg).Scan(&id, &found, &password) + if errors.Is(err, pgx.ErrNoRows) { + return nil, domain.ErrUserNotFound + } + if err != nil { + return nil, err + } + + return &domain.User{ + ID: id, + Login: found, + Password: password, + }, nil +} diff --git a/internal/repository/queue/consumer.go b/internal/repository/queue/consumer.go new file mode 100644 index 0000000..6540191 --- /dev/null +++ b/internal/repository/queue/consumer.go @@ -0,0 +1,125 @@ +package queue + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "time" + + "github.com/rabbitmq/amqp091-go" + + "github.com/belyaevedu/remote-code-service/internal/config" + "github.com/belyaevedu/remote-code-service/internal/domain" + "github.com/belyaevedu/remote-code-service/internal/port" +) + +type Consumer struct { + cfg config.QueueConfig +} + +var _ port.TaskConsumer = (*Consumer)(nil) + +func NewConsumer(cfg config.QueueConfig) *Consumer { + if cfg.Prefetch <= 0 { + cfg.Prefetch = 1 + } + if cfg.ReconnectDelay <= 0 { + cfg.ReconnectDelay = time.Second + } + return &Consumer{cfg: cfg} +} + +// draining the task queue until the context stops +func (c *Consumer) Consume(ctx context.Context, handler port.TaskHandler) error { + for { + if ctx.Err() != nil { + return nil + } + + if err := c.consumeOnce(ctx, handler); err != nil && ctx.Err() == nil { + log.Printf("queue consumer: %v, reconnecting in %s", err, c.cfg.ReconnectDelay) + } + + if !waitReconnect(ctx, c.cfg.ReconnectDelay) { + return nil + } + } +} + +// runs a single broker session +func (c *Consumer) consumeOnce(ctx context.Context, handler port.TaskHandler) error { + conn, ch, err := connect(c.cfg) + if err != nil { + return err + } + defer func() { + closeCleanup(ch, "channel") + closeCleanup(conn, "connection") + }() + + // bounding in-flight messages + if err := ch.Qos(c.cfg.Prefetch, 0, false); err != nil { + return fmt.Errorf("queue qos: %w", err) + } + + deliveries, err := ch.ConsumeWithContext(ctx, c.cfg.Queue, "", false, false, false, false, nil) + if err != nil { + return fmt.Errorf("queue consume: %w", err) + } + + for { + select { + case <-ctx.Done(): + return nil + case d, ok := <-deliveries: + if !ok { + // the broker closed the channel + return errors.New("queue deliveries closed: connection lost") + } + c.handle(ctx, handler, d) + } + } +} + +// success acknowledges, a handler failure requeues the message once and a message that already failed once is dropped. +// malformed messages are just dropped +func (c *Consumer) handle(ctx context.Context, handler port.TaskHandler, d amqp091.Delivery) { + var msg domain.TaskMessage + if err := json.Unmarshal(d.Body, &msg); err != nil { + log.Printf("queue consumer: dropping malformed task message: %v", err) + c.nack(d, false) + return + } + if msg.TaskID == "" { + log.Printf("queue consumer: dropping task message without an id") + c.nack(d, false) + return + } + + if err := handler(ctx, msg); err != nil { + if d.Redelivered { + log.Printf("queue consumer: dropping task %s after a repeated failure: %v", msg.TaskID, err) + c.nack(d, false) + return + } + log.Printf("queue consumer: requeueing task %s after handler failure: %v", msg.TaskID, err) + c.nack(d, true) + return + } + + c.ack(d) +} + +func (c *Consumer) ack(d amqp091.Delivery) { + if err := d.Ack(false); err != nil { + log.Printf("queue consumer: acknowledging task: %v", err) + } +} + +func (c *Consumer) nack(d amqp091.Delivery, requeue bool) { + if err := d.Nack(false, requeue); err != nil { + log.Printf("queue consumer: nacking task (requeue=%t): %v", requeue, err) + } +} diff --git a/internal/repository/queue/publisher.go b/internal/repository/queue/publisher.go new file mode 100644 index 0000000..672b46c --- /dev/null +++ b/internal/repository/queue/publisher.go @@ -0,0 +1,122 @@ +package queue + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "sync" + "time" + + "github.com/rabbitmq/amqp091-go" + + "github.com/belyaevedu/remote-code-service/internal/config" + "github.com/belyaevedu/remote-code-service/internal/domain" + "github.com/belyaevedu/remote-code-service/internal/port" +) + +type Publisher struct { + cfg config.QueueConfig + + conn *amqp091.Connection + ch *amqp091.Channel + mu sync.Mutex +} + +var _ port.TaskPublisher = (*Publisher)(nil) + +func NewPublisher(cfg config.QueueConfig) *Publisher { + if cfg.Prefetch <= 0 { + cfg.Prefetch = 1 + } + if cfg.ReconnectDelay <= 0 { + cfg.ReconnectDelay = time.Second + } + return &Publisher{cfg: cfg} +} + +func (p *Publisher) Publish(ctx context.Context, msg domain.TaskMessage) error { + pub, err := newTaskPublishing(msg) + if err != nil { + return err + } + + p.mu.Lock() + defer p.mu.Unlock() + + if err := p.ensureReady(); err != nil { + return err + } + + // "" being the default exchange + if err := p.ch.PublishWithContext(ctx, "", p.cfg.Queue, false, false, pub); err != nil { + // the cached connection and/or channel went stale + // drop, redial and retry the publish exactly once + if err := p.reset(); err != nil { + log.Printf("queue publisher: discarding stale session: %v", err) + } + + if err := p.ensureReady(); err != nil { + return err + } + + if err := p.ch.PublishWithContext(ctx, "", p.cfg.Queue, false, false, pub); err != nil { + if err := p.reset(); err != nil { + log.Printf("queue publisher: discarding failed session: %v", err) + } + return fmt.Errorf("queue publish: %w", err) + } + } + return nil +} + +func newTaskPublishing(msg domain.TaskMessage) (amqp091.Publishing, error) { + body, err := json.Marshal(msg) + if err != nil { + return amqp091.Publishing{}, fmt.Errorf("queue encode: %w", err) + } + + return amqp091.Publishing{ + ContentType: "application/json", + Body: body, + DeliveryMode: amqp091.Persistent, + CorrelationId: msg.TaskID, + Timestamp: time.Now(), + Type: "task.submitted", + }, nil +} + +func (p *Publisher) Close() error { + p.mu.Lock() + defer p.mu.Unlock() + return p.reset() +} + +func (p *Publisher) ensureReady() error { + if p.ch != nil { + return nil + } + conn, ch, err := connect(p.cfg) + if err != nil { + return err + } + p.conn, p.ch = conn, ch + return nil +} + +func (p *Publisher) reset() error { + var errs []error + if p.ch != nil { + if err := p.ch.Close(); err != nil { + errs = append(errs, fmt.Errorf("closing channel: %w", err)) + } + } + if p.conn != nil { + if err := p.conn.Close(); err != nil { + errs = append(errs, fmt.Errorf("closing connection: %w", err)) + } + } + p.ch, p.conn = nil, nil + return errors.Join(errs...) +} diff --git a/internal/repository/queue/queue.go b/internal/repository/queue/queue.go new file mode 100644 index 0000000..bfe623d --- /dev/null +++ b/internal/repository/queue/queue.go @@ -0,0 +1,58 @@ +package queue + +import ( + "context" + "fmt" + "io" + "log" + "time" + + "github.com/rabbitmq/amqp091-go" + + "github.com/belyaevedu/remote-code-service/internal/config" +) + +func closeCleanup(c io.Closer, what string) { + if err := c.Close(); err != nil { + log.Printf("queue: closing %s during cleanup: %v", what, err) + } +} + +func connect(cfg config.QueueConfig) (*amqp091.Connection, *amqp091.Channel, error) { + conn, err := amqp091.Dial(cfg.URL) + if err != nil { + return nil, nil, fmt.Errorf("queue dial: %w", err) + } + + ch, err := conn.Channel() + if err != nil { + closeCleanup(conn, "connection") + return nil, nil, fmt.Errorf("queue channel: %w", err) + } + + // durable, autoDelete, exclusive, noWait, args + if _, err := ch.QueueDeclare(cfg.Queue, true, false, false, false, nil); err != nil { + closeCleanup(ch, "channel") + closeCleanup(conn, "connection") + return nil, nil, fmt.Errorf("queue declare: %w", err) + } + + return conn, ch, nil +} + +// pauses for d, reporting false as soon as ctx is done +func waitReconnect(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return ctx.Err() == nil + } + + t := time.NewTimer(d) + defer t.Stop() + + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} diff --git a/internal/repository/redis/repository.go b/internal/repository/redis/repository.go new file mode 100644 index 0000000..950a4fc --- /dev/null +++ b/internal/repository/redis/repository.go @@ -0,0 +1,83 @@ +package redis + +import ( + "context" + "errors" + "fmt" + "log" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/belyaevedu/remote-code-service/internal/config" + "github.com/belyaevedu/remote-code-service/internal/domain" + "github.com/belyaevedu/remote-code-service/internal/port" +) + +const sessionKeyPrefix = "session:" + +type Repository struct { + client *redis.Client + ttl time.Duration +} + +var _ port.SessionRepository = (*Repository)(nil) + +func New(ctx context.Context, cfg config.RedisConfig) (*Repository, error) { + client := redis.NewClient(&redis.Options{ + Addr: cfg.Addr, + Password: cfg.Password, + DB: cfg.DB, + }) + + pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + if err := client.Ping(pingCtx).Err(); err != nil { + if err := client.Close(); err != nil { + log.Printf("error raised closing redis client: %v\n", err) + } + return nil, fmt.Errorf("redis ping: %w", err) + } + + return &Repository{client: client, ttl: cfg.SessionTTL}, nil +} + +func (r *Repository) Close() error { + return r.client.Close() +} + +func (r *Repository) CreateSession(ctx context.Context, session *domain.Session) error { + return r.client.Set(ctx, + sessionKeyPrefix+session.SessionID, session.UserID, r.ttl, + ).Err() +} + +func (r *Repository) GetSession(ctx context.Context, sessionID string) (*domain.Session, error) { + userID, err := r.client.Get(ctx, + sessionKeyPrefix+sessionID, + ).Result() + if errors.Is(err, redis.Nil) { + return nil, domain.ErrSessionNotFound + } + if err != nil { + return nil, err + } + + return &domain.Session{ + UserID: userID, + SessionID: sessionID, + }, nil +} + +func (r *Repository) DeleteSession(ctx context.Context, sessionID string) error { + removed, err := r.client.Del(ctx, + sessionKeyPrefix+sessionID, + ).Result() + if err != nil { + return err + } + if removed == 0 { + return domain.ErrSessionNotFound + } + return nil +} diff --git a/internal/repository/repository.go b/internal/repository/repository.go deleted file mode 100644 index 2e4f2b1..0000000 --- a/internal/repository/repository.go +++ /dev/null @@ -1,31 +0,0 @@ -package repository - -import ( - "sync" - - "github.com/belyaevedu/remote-code-service/internal/domain" - "github.com/belyaevedu/remote-code-service/internal/port" -) - -type Repository struct { - mu sync.RWMutex - - tasks map[string]*domain.Task - users map[string]*domain.User // key - login - sessions map[string]*domain.Session // key - session id -} - -// compile-time asserts -var ( - _ port.TaskRepository = (*Repository)(nil) - _ port.UserRepository = (*Repository)(nil) - _ port.SessionRepository = (*Repository)(nil) -) - -func New() *Repository { - return &Repository{ - tasks: make(map[string]*domain.Task), - users: make(map[string]*domain.User), - sessions: make(map[string]*domain.Session), - } -} diff --git a/internal/repository/session.go b/internal/repository/session.go deleted file mode 100644 index c895c3f..0000000 --- a/internal/repository/session.go +++ /dev/null @@ -1,41 +0,0 @@ -package repository - -import ( - "github.com/belyaevedu/remote-code-service/internal/domain" -) - -func (r *Repository) CreateSession(session *domain.Session) error { - r.mu.Lock() - defer r.mu.Unlock() - - clone := *session - r.sessions[clone.SessionID] = &clone - - return nil -} - -func (r *Repository) GetSession(sessionID string) (*domain.Session, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - s, ok := r.sessions[sessionID] - if !ok { - return nil, domain.ErrSessionNotFound - } - - clone := *s - return &clone, nil -} - -func (r *Repository) DeleteSession(sessionID string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.sessions[sessionID]; !ok { - return domain.ErrSessionNotFound - } - - delete(r.sessions, sessionID) - - return nil -} diff --git a/internal/repository/task.go b/internal/repository/task.go deleted file mode 100644 index bed75a4..0000000 --- a/internal/repository/task.go +++ /dev/null @@ -1,57 +0,0 @@ -package repository - -import ( - "github.com/belyaevedu/remote-code-service/internal/domain" -) - -func (r *Repository) Save(task *domain.Task) error { - r.mu.Lock() - defer r.mu.Unlock() - - r.tasks[task.ID] = task - - return nil -} - -func (r *Repository) Get(id string) (*domain.Task, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - t, ok := r.tasks[id] - if !ok { - return nil, domain.ErrTaskNotFound - } - - // returning a clone so the caller sees an unchanging snapshot - clone := *t - return &clone, nil -} - -func (r *Repository) UpdateStatus(id string, status domain.TaskStatus) error { - r.mu.Lock() - defer r.mu.Unlock() - - t, ok := r.tasks[id] - if !ok { - return domain.ErrTaskNotFound - } - - t.Status = status - - return nil -} - -func (r *Repository) SaveResult(id string, result *domain.Result) error { - r.mu.Lock() - defer r.mu.Unlock() - - t, ok := r.tasks[id] - if !ok { - return domain.ErrTaskNotFound - } - - t.Status = domain.StatusReady - t.Result = result - - return nil -} diff --git a/internal/repository/user.go b/internal/repository/user.go deleted file mode 100644 index 66cae75..0000000 --- a/internal/repository/user.go +++ /dev/null @@ -1,46 +0,0 @@ -package repository - -import ( - "github.com/belyaevedu/remote-code-service/internal/domain" -) - -func (r *Repository) SaveUser(user *domain.User) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.users[user.Login]; ok { - return domain.ErrUserAlreadyExists - } - - clone := *user - r.users[clone.Login] = &clone - - return nil -} - -func (r *Repository) GetUserByLogin(login string) (*domain.User, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - u, ok := r.users[login] - if !ok { - return nil, domain.ErrUserNotFound - } - - clone := *u - return &clone, nil -} - -func (r *Repository) GetUserByID(id string) (*domain.User, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - for _, u := range r.users { - if u.ID == id { - clone := *u - return &clone, nil - } - } - - return nil, domain.ErrUserNotFound -} diff --git a/internal/service/metrics.go b/internal/service/metrics.go new file mode 100644 index 0000000..2048702 --- /dev/null +++ b/internal/service/metrics.go @@ -0,0 +1,81 @@ +package service + +import ( + "context" + "errors" + "time" + + chimetrics "github.com/go-chi/metrics" + + "github.com/belyaevedu/remote-code-service/internal/domain" + "github.com/belyaevedu/remote-code-service/internal/port" +) + +const ( + taskStatusSuccess = "success" // exit code 0 + taskStatusFailure = "failure" // ran but the user code failed + taskStatusTimeout = "timeout" // killed by the execution timout + taskStatusError = "error" // infra failure +) + +type taskLabels struct { + Translator string `label:"translator"` + Status string `label:"status"` +} + +type taskInFlightLabels struct { + Translator string `label:"translator"` +} + +var ( + tasksInFlight = chimetrics.GaugeWith[taskInFlightLabels]( + "tasks_in_flight", + "Number of code tasks currently executing in the sandbox.", + ) + tasksProcessedTotal = chimetrics.CounterWith[taskLabels]( + "tasks_processed_total", + "Total number of code execution tasks processed by the worker.", + ) + taskExecutionDuration = chimetrics.HistogramWith[taskLabels]( + "task_execution_duration_seconds", + "Time a code task spent executing in the sandbox.", + []float64{.05, .1, .25, .5, 1, 2.5, 5, 10, 15, 30, 60}, + ) +) + +type InstrumentedExecutor struct { + inner port.CodeExecutor +} + +var _ port.CodeExecutor = (*InstrumentedExecutor)(nil) + +func NewInstrumentedExecutor(inner port.CodeExecutor) *InstrumentedExecutor { + return &InstrumentedExecutor{inner: inner} +} + +func (e *InstrumentedExecutor) Execute(ctx context.Context, msg domain.TaskMessage) (domain.ExecutionResult, error) { + inflight := taskInFlightLabels{Translator: msg.Translator} + tasksInFlight.Inc(inflight) + defer tasksInFlight.Dec(inflight) + + start := time.Now() + result, err := e.inner.Execute(ctx, msg) + duration := time.Since(start) + + status := taskStatusSuccess + switch { + case err != nil: + status = taskStatusError + if errors.Is(err, domain.ErrExecutionTimeout) { + status = taskStatusTimeout + } + case result.Failed: + status = taskStatusFailure + } + + labels := taskLabels{Translator: msg.Translator, Status: status} + tasksProcessedTotal.Inc(labels) + taskExecutionDuration.Observe(duration.Seconds(), labels) + + return result, err +} diff --git a/internal/service/philharmonic.go b/internal/service/philharmonic.go new file mode 100644 index 0000000..3f0958d --- /dev/null +++ b/internal/service/philharmonic.go @@ -0,0 +1,441 @@ +package service + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/belyaevedu/remote-code-service/internal/config" + "github.com/belyaevedu/remote-code-service/internal/domain" + "github.com/belyaevedu/remote-code-service/internal/port" +) + +// mirroring philharmonic's task.State +const ( + phrmStatePending = 0 + phrmStateScheduled = 1 + phrmStateRunning = 2 + phrmStateCompleted = 3 + phrmStateFailed = 4 +) + +var supportedTranslators = map[string]struct{}{ + "python3": {}, + "gcc": {}, + "clang": {}, +} + +const ( + phrmDefaultPollInterval = time.Second + phrmDefaultHTTPTimeout = 10 * time.Second + phrmStopTimeout = 5 * time.Second +) + +type PhilharmonicExecutor struct { + cfg config.PhilharmonicConfig + client *http.Client +} + +var _ port.CodeExecutor = (*PhilharmonicExecutor)(nil) + +func NewPhilharmonicExecutor(cfg config.PhilharmonicConfig) *PhilharmonicExecutor { + if cfg.PollInterval <= 0 { + cfg.PollInterval = phrmDefaultPollInterval + } + if cfg.PollTimeout <= 0 { + cfg.PollTimeout = cfg.TaskTimeout + 15*time.Second + } + if cfg.HTTPClient == nil { + cfg.HTTPClient = &http.Client{Timeout: phrmDefaultHTTPTimeout} + } + cfg.BaseURL = strings.TrimSuffix(cfg.BaseURL, "/") + + return &PhilharmonicExecutor{cfg: cfg, client: cfg.HTTPClient} +} + +func (e *PhilharmonicExecutor) Execute(ctx context.Context, msg domain.TaskMessage) (domain.ExecutionResult, error) { + if _, ok := supportedTranslators[msg.Translator]; !ok { + return domain.ExecutionResult{}, fmt.Errorf("%w: %q", domain.ErrUnsupportedTranslator, msg.Translator) + } + if msg.TaskID == "" { + return domain.ExecutionResult{}, fmt.Errorf("task message without an id") + } + + // unique per orchestrator and traceable back to our task id + name := "run-" + msg.TaskID + + if err := e.submit(ctx, name, msg); err != nil { + return domain.ExecutionResult{}, fmt.Errorf("philharmonic submit: %w", err) + } + + entry, err := e.awaitTerminal(ctx, name) + if err != nil { + e.stopAndLog(name) + return domain.ExecutionResult{}, err + } + + output, exitCode, err := e.logs(ctx, name) + if err != nil { + e.stopAndLog(name) + return domain.ExecutionResult{}, fmt.Errorf("philharmonic logs: %w", err) + } + + // a 2nd stop OR a stop on a task in a terminal state removes the record from the manager + e.stopAndLog(name) + + result := domain.ExecutionResult{ + Output: output, + ExitCode: exitCode, + Failed: entry.State == phrmStateFailed, + } + if result.Failed && strings.TrimSpace(result.Output) == "" { + if entry.FailureReason != "" { + result.Output = fmt.Sprintf("execution failed: %s", entry.FailureReason) + } else { + result.Output = fmt.Sprintf("execution failed (exit code %d)", exitCode) + } + } + return result, nil +} + +// mirrors the philharmonic task.Task JSON field names +type submitTask struct { + Name string + Image string + Env []string + RestartPolicy string + Timeout int64 // seconds + Cpu float64 + Memory int64 // bytes + Security *submitSecurity +} + +// mirrors the philharmonic task.Security +type submitSecurity struct { + User string + CapDrop []string + Tmpfs []string + ReadOnlyRootfs bool + NoNewPrivileges bool + PidsLimit int64 + Ulimits []submitUlimit +} + +type submitUlimit struct { + Name string `json:"Name"` + Soft int64 `json:"Soft"` + Hard int64 `json:"Hard"` +} + +func defaultSandboxSecurity() *submitSecurity { + return &submitSecurity{ + User: "65532:65532", + CapDrop: []string{"ALL"}, + ReadOnlyRootfs: true, + Tmpfs: []string{"/tmp:rw,nosuid,nodev,size=64m,mode=1777"}, + NoNewPrivileges: true, + PidsLimit: 128, + Ulimits: []submitUlimit{ + // fd exhaustion guard + {Name: "nofile", Soft: 256, Hard: 256}, + // no core dumps + {Name: "core", Soft: 0, Hard: 0}, + }, + } +} + +type submitEvent struct { + Task submitTask `json:"Task"` +} + +func (e *PhilharmonicExecutor) submit(ctx context.Context, name string, msg domain.TaskMessage) error { + body, err := json.Marshal(submitEvent{ + Task: submitTask{ + Name: name, + Image: e.cfg.SandboxImage, + Env: []string{ + "TRANSLATOR=" + msg.Translator, + "USER_CODE_B64=" + base64.StdEncoding.EncodeToString([]byte(msg.Code)), + }, + RestartPolicy: "no", + + // ceiling-division to whole seconds + Timeout: int64((e.cfg.TaskTimeout + time.Second - 1) / time.Second), + + Cpu: e.cfg.Cpu, + Memory: e.cfg.Memory, + + Security: defaultSandboxSecurity(), + }, + }) + if err != nil { + return err + } + + resp, err := e.sendRequest(ctx, http.MethodPost, "/tasks", bytes.NewReader(body)) + if err != nil { + return err + } + return drainAndCheck(resp, http.StatusCreated) +} + +// fields of a philharmonic TaskView the client cares about +type taskListEntry struct { + Name string `json:"Name"` + State int `json:"State"` + FailureReason string `json:"FailureReason"` +} + +// polls GET /tasks until the named task reaches a terminal state (Completed/Failed) or the poll budget runs out +func (e *PhilharmonicExecutor) awaitTerminal(ctx context.Context, name string) (taskListEntry, error) { + deadline := time.Now().Add(e.cfg.PollTimeout) + + for { + resp, err := e.sendRequest(ctx, http.MethodGet, "/tasks", nil) + if err != nil { + return taskListEntry{}, fmt.Errorf("philharmonic poll: %w", err) + } + + entries, err := decodeTaskList(resp) + if err != nil { + return taskListEntry{}, fmt.Errorf("philharmonic poll: %w", err) + } + + var match *taskListEntry + found := 0 + for i := range entries { + if entries[i].Name == name { + found++ + match = &entries[i] + } + } + + switch { + case found > 1: + return taskListEntry{}, fmt.Errorf("philharmonic poll: %d tasks named %q exist", found, name) + case found == 1 && (match.State == phrmStateCompleted || match.State == phrmStateFailed): + return *match, nil + } + + if err := ctx.Err(); err != nil { + return taskListEntry{}, fmt.Errorf("philharmonic poll: %w", err) + } + + wait := time.Until(deadline) + if wait <= 0 { + return taskListEntry{}, fmt.Errorf("%w: task %q did not reach a terminal state within %s", + domain.ErrExecutionTimeout, name, e.cfg.PollTimeout) + } + if wait > e.cfg.PollInterval { + wait = e.cfg.PollInterval + } + + select { + case <-ctx.Done(): + return taskListEntry{}, fmt.Errorf("philharmonic poll: %w", ctx.Err()) + case <-time.After(wait): + } + } +} + +// fetches stdout+stderr logs from the manager +func (e *PhilharmonicExecutor) logs(ctx context.Context, name string) (string, int, error) { + resp, err := e.sendRequest(ctx, http.MethodGet, "/tasks/logs/"+url.PathEscape(name), nil) + if err != nil { + return "", 0, err + } + defer func() { + if err := resp.Body.Close(); err != nil { + log.Printf("error raised closing resp body: %v\n", err) + } + }() + + if resp.StatusCode != http.StatusOK { + return "", 0, unexpectedStatus(resp) + } + + // the orchestrator bounds the captured log size itself + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", 0, err + } + + exitCode := -1 + if v := resp.Header.Get("X-Exit-Code"); v != "" { + code, err := strconv.Atoi(v) + if err != nil { + log.Printf("philharmonic: malformed X-Exit-Code header %q for task %s: %v", v, name, err) + } else { + exitCode = code + } + } + return string(body), exitCode, nil +} + +func (e *PhilharmonicExecutor) stopAndLog(name string) { + if err := e.stop(name); err != nil { + log.Printf("philharmonic: cleanup of task %q failed: %v", name, err) + } +} + +// for terminal state tasks this removes the record from the manager store, +// for live ones it stops the container +func (e *PhilharmonicExecutor) stop(name string) error { + // fresh context: the caller's may already be expired or canceled + ctx, cancel := context.WithTimeout(context.Background(), phrmStopTimeout) + defer cancel() + + resp, err := e.sendRequest(ctx, http.MethodDelete, "/tasks/"+url.PathEscape(name), nil) + if err != nil { + return fmt.Errorf("stop request: %w", err) + } + + defer func() { + if err := resp.Body.Close(); err != nil { + log.Printf("error raised closing resp body: %v\n", err) + } + }() + + body, err := readSnippet(resp) + if err != nil { + return fmt.Errorf("reading stop response: %w", err) + } + + switch resp.StatusCode { + case http.StatusNoContent, http.StatusNotFound: + return nil + default: + return statusError(resp.StatusCode, body) + } +} + +func (e *PhilharmonicExecutor) sendRequest(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) { + httpReq, err := http.NewRequestWithContext(ctx, method, e.cfg.BaseURL+path, body) + if err != nil { + return nil, err + } + if body != nil { + httpReq.Header.Set("Content-Type", "application/json") + } + if e.cfg.Token != "" { + httpReq.Header.Set("Authorization", "Bearer "+e.cfg.Token) + } + return e.client.Do(httpReq) +} + +func decodeTaskList(resp *http.Response) ([]taskListEntry, error) { + defer func() { + if err := resp.Body.Close(); err != nil { + log.Printf("error raised closing resp body: %v\n", err) + } + }() + + if resp.StatusCode != http.StatusOK { + return nil, unexpectedStatus(resp) + } + + var entries []taskListEntry + if err := json.NewDecoder(resp.Body).Decode(&entries); err != nil { + return nil, fmt.Errorf("decoding task list: %w", err) + } + return entries, nil +} + +func readSnippet(resp *http.Response) ([]byte, error) { + return io.ReadAll(io.LimitReader(resp.Body, 4<<10)) +} + +func statusError(status int, body []byte) error { + return fmt.Errorf("unexpected status %d: %s", status, strings.TrimSpace(string(body))) +} + +// draining responses to keep the connection reusable +func drainAndCheck(resp *http.Response, want int) error { + defer func() { + if err := resp.Body.Close(); err != nil { + log.Printf("error raised closing resp body: %v\n", err) + } + }() + + body, err := readSnippet(resp) + if err != nil { + return fmt.Errorf("reading response body: %w", err) + } + if resp.StatusCode != want { + return statusError(resp.StatusCode, body) + } + return nil +} + +func unexpectedStatus(resp *http.Response) error { + body, err := readSnippet(resp) + if err != nil { + return fmt.Errorf("unexpected status %d (body unreadable: %w)", resp.StatusCode, err) + } + return statusError(resp.StatusCode, body) +} + +// wire shapes of philharmonic manager's POST /images endpoint +type pullImagesRequest struct { + Image string `json:"image"` +} + +type pullImagesReport struct { + Image string `json:"image"` + Results []pullImageResult `json:"results"` +} + +type pullImageResult struct { + Worker string `json:"worker"` + OK bool `json:"ok"` + Pulled bool `json:"pulled"` + Error string `json:"error"` +} + +// asks manager to pull the sandbox image on all workers +func (e *PhilharmonicExecutor) PreWarm(ctx context.Context) error { + body, err := json.Marshal(pullImagesRequest{Image: e.cfg.SandboxImage}) + if err != nil { + return err + } + + resp, err := e.sendRequest(ctx, http.MethodPost, "/images", bytes.NewReader(body)) + if err != nil { + return err + } + defer func() { + if err := resp.Body.Close(); err != nil { + log.Printf("error raised closing resp body: %v\n", err) + } + }() + + // the manager always answers 200 and reports the pull per worker + if resp.StatusCode != http.StatusOK { + snippet, err := readSnippet(resp) + if err != nil { + return fmt.Errorf("reading pre-warm response: %w", err) + } + return statusError(resp.StatusCode, snippet) + } + + var report pullImagesReport + if err := json.NewDecoder(resp.Body).Decode(&report); err != nil { + return fmt.Errorf("decoding pull report: %w", err) + } + + for _, res := range report.Results { + if !res.OK { + return fmt.Errorf("worker %s: %s", res.Worker, res.Error) + } + } + return nil +} diff --git a/internal/service/task.go b/internal/service/task.go index e0fe583..b9f02a1 100644 --- a/internal/service/task.go +++ b/internal/service/task.go @@ -1,8 +1,8 @@ package service import ( - "log" - "time" + "context" + "fmt" "github.com/belyaevedu/remote-code-service/internal/domain" "github.com/belyaevedu/remote-code-service/internal/port" @@ -10,61 +10,67 @@ import ( "github.com/google/uuid" ) -const ( - outputMessage = "puk" -) - type TaskService struct { - repo port.TaskRepository - processingTime time.Duration + repo port.TaskRepository + publisher port.TaskPublisher } -// compile-time assert that task's TaskService struct -// implements port's TaskService interface var _ port.TaskService = (*TaskService)(nil) -func NewTaskService(repo port.TaskRepository, processingTime time.Duration) *TaskService { - if processingTime <= 0 { - processingTime = 2 * time.Second - } +func NewTaskService(repo port.TaskRepository, publisher port.TaskPublisher) *TaskService { return &TaskService{ - repo: repo, - processingTime: processingTime, + repo: repo, + publisher: publisher, } } -func (s *TaskService) Submit(userID string) (string, error) { +func (s *TaskService) Submit(ctx context.Context, userID string, sub domain.Submission) (string, error) { if userID == "" { return "", domain.ErrAccessDenied } + if _, ok := supportedTranslators[sub.Translator]; !ok { + return "", fmt.Errorf("%w: %q", domain.ErrUnsupportedTranslator, sub.Translator) + } + if sub.Code == "" { + return "", fmt.Errorf("%w: empty code", domain.ErrInvalidSubmission) + } id := uuid.NewString() task := &domain.Task{ - ID: id, - UserID: userID, - Status: domain.StatusInProgress, + ID: id, + UserID: userID, + Status: domain.StatusInProgress, + Translator: sub.Translator, } - if err := s.repo.Save(task); err != nil { + if err := s.repo.SaveTask(ctx, task); err != nil { return "", err } - go s.process(id) + msg := domain.TaskMessage{ + TaskID: id, + Translator: sub.Translator, + Code: sub.Code, + } + if err := s.publisher.Publish(ctx, msg); err != nil { + // the row stays behind as in_progress evidence of the failure + return "", fmt.Errorf("queue publish: %w", err) + } return id, nil } -func (s *TaskService) Status(userID, id string) (domain.TaskStatus, error) { - t, err := s.getOwnedTask(userID, id) +func (s *TaskService) Status(ctx context.Context, userID, id string) (domain.TaskStatus, error) { + t, err := s.getOwnedTask(ctx, userID, id) if err != nil { return "", err } return t.Status, nil } -func (s *TaskService) Result(userID, id string) (*domain.Result, error) { - t, err := s.getOwnedTask(userID, id) +func (s *TaskService) Result(ctx context.Context, userID, id string) (*domain.Result, error) { + t, err := s.getOwnedTask(ctx, userID, id) if err != nil { return nil, err } @@ -76,12 +82,12 @@ func (s *TaskService) Result(userID, id string) (*domain.Result, error) { // fetches the task by id and verifies it belongs to a set user // task owned by someone else results in ErrAccessDenied -func (s *TaskService) getOwnedTask(userID, id string) (*domain.Task, error) { +func (s *TaskService) getOwnedTask(ctx context.Context, userID, id string) (*domain.Task, error) { if userID == "" { return nil, domain.ErrAccessDenied } - t, err := s.repo.Get(id) + t, err := s.repo.GetTask(ctx, id) if err != nil { return nil, err } @@ -92,17 +98,3 @@ func (s *TaskService) getOwnedTask(userID, id string) (*domain.Task, error) { return t, nil } - -func (s *TaskService) process(id string) { - time.Sleep(s.processingTime) - - result := &domain.Result{ - Output: outputMessage, - } - - if err := s.repo.SaveResult(id, result); err != nil { - log.Printf("Failed to save task result %s: %v\n", id, err) - return - } - log.Printf("Task finished: %s\n", id) -} diff --git a/internal/service/user.go b/internal/service/user.go index f5ed0ef..8f29da9 100644 --- a/internal/service/user.go +++ b/internal/service/user.go @@ -1,6 +1,7 @@ package service import ( + "context" "crypto/rand" "encoding/hex" "errors" @@ -28,12 +29,12 @@ func NewUserService(users port.UserRepository, sessions port.SessionRepository) return &UserService{users: users, sessions: sessions} } -func (s *UserService) Register(login, password string) error { +func (s *UserService) Register(ctx context.Context, login, password string) error { if login == "" || password == "" { return domain.ErrInvalidCredentials } - if existing, err := s.users.GetUserByLogin(login); err == nil { + if existing, err := s.users.GetUserByLogin(ctx, login); err == nil { // tests basically involve a double register // with the same login:pass, so this logic is here to pass the tests if comparePassword(existing.Password, password) != nil { @@ -55,12 +56,12 @@ func (s *UserService) Register(login, password string) error { Password: hashed, } - if err := s.users.SaveUser(user); err != nil { + if err := s.users.SaveUser(ctx, user); err != nil { if !errors.Is(err, domain.ErrUserAlreadyExists) { return err } // possible concurrent registration race - stored, gErr := s.users.GetUserByLogin(login) + stored, gErr := s.users.GetUserByLogin(ctx, login) if gErr != nil { return gErr } @@ -73,8 +74,8 @@ func (s *UserService) Register(login, password string) error { return nil } -func (s *UserService) Login(login, password string) (string, error) { - user, err := s.users.GetUserByLogin(login) +func (s *UserService) Login(ctx context.Context, login, password string) (string, error) { + user, err := s.users.GetUserByLogin(ctx, login) if err != nil { if errors.Is(err, domain.ErrUserNotFound) { return "", domain.ErrInvalidCredentials @@ -96,15 +97,15 @@ func (s *UserService) Login(login, password string) (string, error) { SessionID: token, } - if err := s.sessions.CreateSession(session); err != nil { + if err := s.sessions.CreateSession(ctx, session); err != nil { return "", err } return token, nil } -func (s *UserService) Authenticate(token string) (string, error) { - session, err := s.sessions.GetSession(token) +func (s *UserService) Authenticate(ctx context.Context, token string) (string, error) { + session, err := s.sessions.GetSession(ctx, token) if err != nil { return "", domain.ErrUnauthorized } diff --git a/migrations/001_init.sql b/migrations/001_init.sql new file mode 100644 index 0000000..e9e39e4 --- /dev/null +++ b/migrations/001_init.sql @@ -0,0 +1,23 @@ +-- +migrate Up +CREATE TYPE task_status AS ENUM ('in_progress', 'ready'); + +CREATE TABLE users ( + id uuid PRIMARY KEY, + login text NOT NULL UNIQUE, + password_hash text NOT NULL +); + +CREATE TABLE tasks ( + id uuid PRIMARY KEY, + user_id uuid NOT NULL REFERENCES users (id), + status task_status NOT NULL DEFAULT 'in_progress', + translator text NOT NULL, + result jsonb, + created_at timestamptz NOT NULL DEFAULT now(), + finished_at timestamptz +); + +-- +migrate Down +DROP TABLE tasks; +DROP TABLE users; +DROP TYPE task_status; diff --git a/sandbox/Dockerfile b/sandbox/Dockerfile new file mode 100644 index 0000000..b7be367 --- /dev/null +++ b/sandbox/Dockerfile @@ -0,0 +1,9 @@ +FROM alpine:3.24.1@sha256:28bd5fe8b56d1bd048e5babf5b10710ebe0bae67db86916198a6eec434943f8b + +RUN apk add --no-cache python3~=3.14.7-r1 gcc~=15.2.0-r5 clang22~=22.1.3-r2 musl-dev~=1.2.6-r2 libstdc++~=15.2.0-r5 + +COPY --chmod=0755 sandbox/run.sh /run.sh + +USER 65532:65532 + +ENTRYPOINT ["/bin/sh", "/run.sh"] diff --git a/sandbox/run.sh b/sandbox/run.sh new file mode 100644 index 0000000..1724a03 --- /dev/null +++ b/sandbox/run.sh @@ -0,0 +1,29 @@ +#!/bin/sh +# The image's entrypoint is fixed to run.sh. +# This wrapper takes in the TRANSLATOR env var set, decodes the user code from USER_CODE_B64 + +set -u + +if [ -z "${USER_CODE_B64:-}" ]; then + echo "USER_CODE_B64 is not set" >&2 + exit 2 +fi + +case "${TRANSLATOR:-}" in +python3) + printf '%s' "$USER_CODE_B64" | base64 -d > /tmp/main.py + exec python3 /tmp/main.py + ;; +gcc) + printf '%s' "$USER_CODE_B64" | base64 -d > /tmp/main.c + gcc /tmp/main.c -o /tmp/prog && exec /tmp/prog + ;; +clang) + printf '%s' "$USER_CODE_B64" | base64 -d > /tmp/main.cpp + clang++ /tmp/main.cpp -o /tmp/prog && exec /tmp/prog + ;; +*) + echo "unsupported translator: ${TRANSLATOR}" >&2 + exit 2 + ;; +esac diff --git a/tests/hw2.py b/tests/hw2.py index 1a32daf..8e6fe85 100644 --- a/tests/hw2.py +++ b/tests/hw2.py @@ -54,7 +54,8 @@ def test_create_task(auth_token): task_url = f"{BASE_URL}/task" headers = {'Authorization': f'Bearer {auth_token}'} - response = requests.post(task_url, headers=headers) + payload = {"translator": "python3", "code": "print('hw2 task')"} + response = requests.post(task_url, headers=headers, json=payload) assert response.status_code == 201 data = response.json() diff --git a/tests/hw3.py b/tests/hw3.py index 719648b..424aad6 100644 --- a/tests/hw3.py +++ b/tests/hw3.py @@ -54,11 +54,9 @@ def test_create_task(auth_token): task_url = f"{BASE_URL}/task" headers = {'Authorization': f'Bearer {auth_token}'} - payload = dict() - # payload = get_code_processor_payload() + payload = get_code_processor_payload() # payload = get_image_processor_payload() - if len(payload) == 0: raise NotImplemented("Choose one of the variants for payload!")