From 1dd713514cd54310636461663473eee0398e198d Mon Sep 17 00:00:00 2001 From: Denozordec Date: Wed, 20 May 2026 00:21:16 +0700 Subject: [PATCH] =?UTF-8?q?fix(httpapi):=20phase=201=20compliance=20?= =?UTF-8?q?=E2=80=94=20ERR-01,=20SEC-04,=20CDN=20client,=20scheduler=20doc?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- cmd/evobgp-scheduler/main.go | 3 +++ deploy/compose/docker-compose.yaml | 2 ++ internal/httpapi/bootstrap.go | 7 ++++++- internal/httpapi/problem.go | 22 ++++++++++++++++++++++ internal/httpapi/routes.go | 24 ++++++++++++------------ internal/httpapi/routes_crud.go | 10 +++++----- internal/httpapi/server.go | 2 ++ internal/httpapi/server_test.go | 16 +++++++++++++++- 8 files changed, 67 insertions(+), 19 deletions(-) diff --git a/cmd/evobgp-scheduler/main.go b/cmd/evobgp-scheduler/main.go index 2c270f0..c42b2d1 100644 --- a/cmd/evobgp-scheduler/main.go +++ b/cmd/evobgp-scheduler/main.go @@ -41,6 +41,9 @@ func main() { apiBase := strings.TrimSpace(os.Getenv("EVOBGP_CONTROL_PLANE_URL")) apiTok := strings.TrimSpace(os.Getenv("EVOBGP_SCHEDULER_BEARER")) + if os.Getenv("EVOBGP_SCHEDULER_STANDALONE") == "1" && apiBase == "" { + log.Fatal("evobgp-scheduler: EVOBGP_SCHEDULER_STANDALONE=1 requires EVOBGP_CONTROL_PLANE_URL and EVOBGP_SCHEDULER_BEARER (split deploy must not use in-process jobs.Registry)") + } deps := &scheduler.Deps{ Store: st, HTTP: &http.Client{Timeout: 45 * time.Second}, diff --git a/deploy/compose/docker-compose.yaml b/deploy/compose/docker-compose.yaml index f8862da..a8698ad 100644 --- a/deploy/compose/docker-compose.yaml +++ b/deploy/compose/docker-compose.yaml @@ -126,6 +126,8 @@ services: condition: service_started environment: <<: *env-ref + # Split deploy: scheduler must enqueue jobs via API, not in-process Registry (ARCH-04). + EVOBGP_SCHEDULER_STANDALONE: "1" EVOBGP_CONTROL_PLANE_URL: http://evobgp-api:8080 EVOBGP_SCHEDULER_BEARER: dev logging: *default-logging diff --git a/internal/httpapi/bootstrap.go b/internal/httpapi/bootstrap.go index 256f621..ddf03c4 100644 --- a/internal/httpapi/bootstrap.go +++ b/internal/httpapi/bootstrap.go @@ -15,6 +15,11 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) +// NewCDNHTTPClient returns the shared HTTP client for CDN and preview fetches (PERF-02 / ERR-03). +func NewCDNHTTPClient() *http.Client { + return &http.Client{Timeout: 45 * time.Second} +} + // BootstrapWorkers opens the same store.Backend and jobs.Registry as New (without HTTP or bundle keys). // Used by standalone worker binaries (scheduler, ingest, …) that share PostgreSQL with the API. func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.Registry, *pgxpool.Pool, error) { @@ -44,7 +49,7 @@ func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.R backend = mem } - cdnHTTP := &http.Client{Timeout: 45 * time.Second} + cdnHTTP := NewCDNHTTPClient() wk := &jobs.Worker{Store: backend, HTTPClient: cdnHTTP} reg := jobs.NewRegistry(wk.Process) wk.Registry = reg diff --git a/internal/httpapi/problem.go b/internal/httpapi/problem.go index 0f917eb..50f07cc 100644 --- a/internal/httpapi/problem.go +++ b/internal/httpapi/problem.go @@ -2,9 +2,15 @@ package httpapi import ( "encoding/json" + "log" "net/http" ) +const ( + internalErrorDetail = "an internal error occurred" + badGatewayDetail = "upstream request failed" +) + // Problem is RFC 9457 application/problem+json. type Problem struct { Type string `json:"type,omitempty"` @@ -34,3 +40,19 @@ func writeJSON(w http.ResponseWriter, status int, v any) { func writeNoContent(w http.ResponseWriter) { w.WriteHeader(http.StatusNoContent) } + +// writeInternalError logs err server-side and returns a generic 500 problem (ERR-01). +func writeInternalError(w http.ResponseWriter, operation string, err error) { + if err != nil { + log.Printf("httpapi: %s: %v", operation, err) + } + writeProblem(w, http.StatusInternalServerError, "Internal Error", internalErrorDetail) +} + +// writeBadGateway logs err server-side and returns a generic 502 problem (ERR-01). +func writeBadGateway(w http.ResponseWriter, operation string, err error) { + if err != nil { + log.Printf("httpapi: %s: %v", operation, err) + } + writeProblem(w, http.StatusBadGateway, "Bad Gateway", badGatewayDetail) +} diff --git a/internal/httpapi/routes.go b/internal/httpapi/routes.go index eb79e76..06f378f 100644 --- a/internal/httpapi/routes.go +++ b/internal/httpapi/routes.go @@ -83,7 +83,7 @@ func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) { defer cancel() if s.pgPool != nil { if err := s.pgPool.Ping(ctx); err != nil { - checks["postgres"] = err.Error() + checks["postgres"] = "unavailable" writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "checks": checks}) return } @@ -317,7 +317,7 @@ func (s *Server) handleGetModule(w http.ResponseWriter, r *http.Request) { writeProblem(w, http.StatusNotFound, "Not Found", "module not found") return } - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } writeJSON(w, http.StatusOK, moduleJSON(mod)) @@ -451,7 +451,7 @@ func (s *Server) handleModuleRefresh(w http.ResponseWriter, r *http.Request) { writeProblem(w, http.StatusNotFound, "Not Found", "module not found") return } - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } idem := r.Header.Get("Idempotency-Key") @@ -463,7 +463,7 @@ func (s *Server) handleModuleRefresh(w http.ResponseWriter, r *http.Request) { mid := mod.ID j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindModuleRefresh, idemPtr, &mid, map[string]any{"module_id": moduleID}) if err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) @@ -512,7 +512,7 @@ func (s *Server) handleTenantRefresh(w http.ResponseWriter, r *http.Request) { "trigger": "api", }) if err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) @@ -916,7 +916,7 @@ func (s *Server) handleRevisionRollback(w http.ResponseWriter, r *http.Request) writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") return } - if !s.requireAtLeast(w, a, "editor") { + if !s.requireAtLeast(w, a, "operator") { return } revID := r.PathValue("revision_id") @@ -933,7 +933,7 @@ func (s *Server) handleRevisionRollback(w http.ResponseWriter, r *http.Request) "source_revision_id": revID, }) if err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) @@ -980,7 +980,7 @@ func (s *Server) handleApply(w http.ResponseWriter, r *http.Request) { "strategy": body.Strategy, }) if err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) @@ -1031,7 +1031,7 @@ func (s *Server) handleSpeakerApply(w http.ResponseWriter, r *http.Request) { "speaker_id": spkID, }) if err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) @@ -1056,7 +1056,7 @@ func (s *Server) handleBirdReload(w http.ResponseWriter, r *http.Request) { } j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindBirdReload, idemPtr, nil, nil) if err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } w.Header().Set("Location", "/v1/jobs/"+j.ID) @@ -1191,7 +1191,7 @@ func (s *Server) handleNodeBundle(w http.ResponseWriter, r *http.Request) { } tgz, err := bundle.BuildGzippedTar(rid, sid, rev.PreviewFragments, s.bundlePriv) if err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } w.Header().Set("Content-Type", "application/gzip") @@ -1241,7 +1241,7 @@ func (s *Server) handleNodeEnroll(w http.ResponseWriter, r *http.Request) { } patch := &store.SpeakerPatch{MetaJSON: &meta} if _, err := s.store.UpdateSpeaker(a.TenantID, sid, patch); err != nil { - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "internal", err) return } writeJSON(w, http.StatusOK, map[string]any{ diff --git a/internal/httpapi/routes_crud.go b/internal/httpapi/routes_crud.go index 9b992e5..2f8706a 100644 --- a/internal/httpapi/routes_crud.go +++ b/internal/httpapi/routes_crud.go @@ -178,7 +178,7 @@ func writeStoreErr(w http.ResponseWriter, err error) { writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", err.Error()) return } - writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error()) + writeInternalError(w, "store", err) } func (s *Server) handleListCDNSources(w http.ResponseWriter, r *http.Request) { @@ -251,20 +251,20 @@ func (s *Server) handlePreviewCDNSource(w http.ResponseWriter, r *http.Request) writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "invalid url") return } - resp, err := http.DefaultClient.Do(req) + resp, err := s.cdnHTTP.Do(req) if err != nil { - writeProblem(w, http.StatusBadGateway, "Bad Gateway", err.Error()) + writeBadGateway(w, "cdn preview fetch", err) return } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { _, _ = io.Copy(io.Discard, resp.Body) - writeProblem(w, http.StatusBadGateway, "Bad Gateway", fmt.Sprintf("upstream status: %s", resp.Status)) + writeBadGateway(w, "cdn preview fetch", fmt.Errorf("upstream status: %s", resp.Status)) return } raw, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) if err != nil { - writeProblem(w, http.StatusBadGateway, "Bad Gateway", err.Error()) + writeBadGateway(w, "cdn preview read body", err) return } pfxs, err := pipeline.ExtractCIDRs(string(raw), body.SourceKind, body.PrefixPath) diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 16fa5ad..306c00e 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -24,6 +24,7 @@ type Server struct { apiKeys []apiKeyRecord insecureDev bool corsOrigins []string + cdnHTTP *http.Client mux *http.ServeMux } @@ -67,6 +68,7 @@ func New(opts Options) (*Server, error) { apiKeys: parseAPIKeysSpec(opts.APIKeys), insecureDev: opts.InsecureDev && opts.SeedDemo, corsOrigins: parseCORSOrigins(opts.CORSAllowedOrigins), + cdnHTTP: NewCDNHTTPClient(), } s.mux = http.NewServeMux() s.registerRoutes() diff --git a/internal/httpapi/server_test.go b/internal/httpapi/server_test.go index 8971a7d..76cbf6c 100644 --- a/internal/httpapi/server_test.go +++ b/internal/httpapi/server_test.go @@ -28,7 +28,7 @@ func TestAPIRefreshApplyJobsBundle(t *testing.T) { } defer srv.Close() tenant, modCDN, modIP, rev, speaker := srv.Store().DemoIDs() - srv.apiKeys = parseAPIKeysSpec("nodekey|" + tenant + "|node,opkey|" + tenant + "|operator") + srv.apiKeys = parseAPIKeysSpec("nodekey|" + tenant + "|node,opkey|" + tenant + "|operator,edkey|" + tenant + "|editor") ts := httptest.NewServer(srv.Handler()) defer ts.Close() @@ -247,6 +247,20 @@ func TestAPIRefreshApplyJobsBundle(t *testing.T) { } }) + t.Run("rollback forbidden for editor", func(t *testing.T) { + req, _ := http.NewRequest(http.MethodPost, base+"/v1/revisions/"+rev+"/rollback", nil) + req.Header.Set("Authorization", "Bearer edkey") + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusForbidden { + b, _ := io.ReadAll(resp.Body) + t.Fatalf("status %d want 403: %s", resp.StatusCode, b) + } + }) + t.Run("rollback queues job", func(t *testing.T) { req, _ := http.NewRequest(http.MethodPost, base+"/v1/revisions/"+rev+"/rollback", nil) req.Header.Set("Authorization", "Bearer opkey")