diff --git a/pkg/engine/planetscale/volume.go b/pkg/engine/planetscale/volume.go index fac3f251e..c5a542701 100644 --- a/pkg/engine/planetscale/volume.go +++ b/pkg/engine/planetscale/volume.go @@ -46,7 +46,7 @@ func (e *Engine) Volume(ctx context.Context, req *engine.VolumeRequest) (*engine return &engine.VolumeResult{ Accepted: true, - PreviousVolume: 0, // Unknown — PlanetScale has no query API for current ratio + PreviousVolume: 0, // Unknown — the engine does not track the deploy request's prior ratio NewVolume: req.Volume, Message: fmt.Sprintf("Throttle ratio set to %.0f%%", ratio*100), }, nil diff --git a/pkg/localscale/handlers_actions.go b/pkg/localscale/handlers_actions.go index b3ee23fb8..7a241fb7f 100644 --- a/pkg/localscale/handlers_actions.go +++ b/pkg/localscale/handlers_actions.go @@ -226,34 +226,41 @@ func (s *Server) handleThrottleDeployRequest(w http.ResponseWriter, r *http.Requ } number := ref.number + // PlanetScale expresses the throttler ratio as a whole percentage on the + // wire; the rest of LocalScale carries it as a fraction, matching Vitess. var body struct { - ThrottleRatio float64 `json:"throttle_ratio"` + Ratio int `json:"ratio"` } if err := s.decodeJSON(r, &body); err != nil { return err } - if body.ThrottleRatio < 0 || body.ThrottleRatio > 0.95 { - return newHTTPError(http.StatusUnprocessableEntity, "throttle_ratio must be between 0.0 and 0.95, got %f", body.ThrottleRatio) + if body.Ratio < 0 || body.Ratio > maxThrottleRatioPercent { + return newHTTPError(http.StatusUnprocessableEntity, "ratio must be between 0 and %d, got %d", maxThrottleRatioPercent, body.Ratio) } + ratio := float64(body.Ratio) / 100 // Store in metadata for query purposes _, err = s.metadataDB.ExecContext(r.Context(), `UPDATE localscale_deploy_requests SET throttle_ratio = ? WHERE org = ? AND database_name = ? AND number = ?`, - body.ThrottleRatio, ref.org, ref.database, number) + ratio, ref.org, ref.database, number) if err != nil { return newHTTPError(http.StatusInternalServerError, "update throttle: %v", err) } - if err := s.applyThrottle(r.Context(), backend, number, body.ThrottleRatio); err != nil { + if err := s.applyThrottle(r.Context(), backend, number, ratio); err != nil { return newHTTPError(http.StatusInternalServerError, "apply throttle: %v", err) } s.writeJSON(w, map[string]string{"status": "ok"}) return nil } +// maxThrottleRatioPercent is the highest throttler ratio PlanetScale accepts, +// as a whole percentage. +const maxThrottleRatioPercent = 95 + // applyThrottle sets the throttle ratio for online DDL migrations across all keyspaces. // Ratio 0.0 = full speed, 0.95 = max throttle (PlanetScale caps at 0.95). // diff --git a/pkg/localscale/server.go b/pkg/localscale/server.go index b7e4884eb..242a99b24 100644 --- a/pkg/localscale/server.go +++ b/pkg/localscale/server.go @@ -1370,7 +1370,7 @@ func (s *Server) registerRoutes(mux *http.ServeMux) { mux.HandleFunc("POST /v1/organizations/{org}/databases/{db}/deploy-requests/{number}/apply-deploy", s.handleError(s.handleApplyDeployRequest)) mux.HandleFunc("POST /v1/organizations/{org}/databases/{db}/deploy-requests/{number}/revert", s.handleError(s.handleRevertDeployRequest)) mux.HandleFunc("POST /v1/organizations/{org}/databases/{db}/deploy-requests/{number}/skip-revert", s.handleError(s.handleSkipRevertDeployRequest)) - mux.HandleFunc("PUT /v1/organizations/{org}/databases/{db}/deploy-requests/{number}/throttle", s.handleError(s.handleThrottleDeployRequest)) + mux.HandleFunc("PATCH /v1/organizations/{org}/databases/{db}/deploy-requests/{number}/throttler", s.handleError(s.handleThrottleDeployRequest)) // Deploy request CRUD endpoints mux.HandleFunc("GET /v1/organizations/{org}/databases/{db}/deploy-requests/{number}", s.handleError(s.handleGetDeployRequest)) diff --git a/pkg/psclient/client.go b/pkg/psclient/client.go index 16a28e04b..32ddea06c 100644 --- a/pkg/psclient/client.go +++ b/pkg/psclient/client.go @@ -5,7 +5,11 @@ import ( "context" "encoding/json" "fmt" + "io" + "math" "net/http" + "strings" + "unicode" ps "github.com/planetscale/planetscale-go/planetscale" ) @@ -44,10 +48,11 @@ type PSClient interface { // ListDeployRequests lists all deploy requests for a database. ListDeployRequests(ctx context.Context, req *ps.ListDeployRequestsRequest) ([]*ps.DeployRequest, error) - // ThrottleDeployRequest sets the throttle ratio for a running deploy request. + // ThrottleDeployRequest sets the throttler ratio for a running deploy request. // This controls the speed of the online DDL copy phase (0.0 = full speed, - // 0.95 = max throttle). The PlanetScale API supports this endpoint but the - // Go SDK (planetscale-go) does not expose it, so we use raw HTTP via baseURL. + // 0.95 = max throttle) across every keyspace the deploy request touches. The + // PlanetScale API supports this endpoint but the Go SDK (planetscale-go) does + // not expose it, so we use raw HTTP via baseURL. // Requires NewPSClientWithBaseURL; returns an error if baseURL is not set. ThrottleDeployRequest(ctx context.Context, req *ThrottleDeployRequestRequest) error } @@ -182,31 +187,81 @@ func (w *psClientWrapper) ListDeployRequests(ctx context.Context, req *ps.ListDe return w.client.DeployRequests.List(ctx, req) } -// ThrottleDeployRequest sets the throttle ratio via a raw HTTP PUT. +// ThrottleDeployRequest sets the throttler ratio via a raw HTTP PATCH. // Not yet available in the PlanetScale SDK. +// +// The API expresses the ratio as a whole percentage (0-95) on the wire, while +// SchemaBot carries it as a fraction; the conversion happens here so the rest of +// the codebase speaks one unit. func (w *psClientWrapper) ThrottleDeployRequest(ctx context.Context, req *ThrottleDeployRequestRequest) error { if w.baseURL == "" { return fmt.Errorf("throttle not supported without base URL") } - url := fmt.Sprintf("%s/v1/organizations/%s/databases/%s/deploy-requests/%d/throttle", + url := fmt.Sprintf("%s/v1/organizations/%s/databases/%s/deploy-requests/%d/throttler", w.baseURL, req.Organization, req.Database, req.Number) - body, err := json.Marshal(map[string]float64{"throttle_ratio": req.ThrottleRatio}) + body, err := json.Marshal(map[string]int{"ratio": throttleRatioPercent(req.ThrottleRatio)}) if err != nil { - return fmt.Errorf("marshal throttle payload: %w", err) + return fmt.Errorf("marshal throttler payload: %w", err) } - httpReq, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(body)) + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPatch, url, bytes.NewReader(body)) if err != nil { - return fmt.Errorf("create throttle request: %w", err) + return fmt.Errorf("create throttler request: %w", err) } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Authorization", w.tokenName+":"+w.tokenValue) resp, err := http.DefaultClient.Do(httpReq) if err != nil { - return fmt.Errorf("throttle deploy request: %w", err) + return fmt.Errorf("throttle deploy request %d (%s/%s): %w", req.Number, req.Organization, req.Database, err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return fmt.Errorf("throttle request failed: %s", resp.Status) + // The status alone does not say why — a 404 on a moved path and a 404 on + // a deleted deploy request read identically. Carry a bounded, sanitized + // excerpt of the body so the cause survives to the logs. Callers must not + // render this into PR markdown. + return fmt.Errorf("throttle deploy request %d (%s/%s) failed: %s: %s", + req.Number, req.Organization, req.Database, resp.Status, sanitizeResponseExcerpt(resp.Body)) } return nil } + +// throttleRatioPercent converts SchemaBot's fractional throttle ratio to the +// whole percentage the PlanetScale API expects. Out-of-range values are passed +// through so the API's own validation reports them rather than being silently +// clamped into a ratio the caller did not ask for. +func throttleRatioPercent(ratio float64) int { + return int(math.Round(ratio * 100)) +} + +// maxResponseExcerptLen bounds how much of an error response body is carried in +// a wrapped error. +const maxResponseExcerptLen = 200 + +// sanitizeResponseExcerpt reads a bounded prefix of an error response body and +// makes it safe to embed in an error string: newlines collapse to spaces and +// non-printable runes are dropped, so the excerpt cannot break log or table +// formatting downstream. +func sanitizeResponseExcerpt(body io.Reader) string { + raw, err := io.ReadAll(io.LimitReader(body, maxResponseExcerptLen*4)) + if err != nil { + return "" + } + cleaned := strings.Map(func(r rune) rune { + if r == '\n' || r == '\r' || r == '\t' { + return ' ' + } + if !unicode.IsPrint(r) { + return -1 + } + return r + }, string(raw)) + cleaned = strings.TrimSpace(strings.Join(strings.Fields(cleaned), " ")) + if cleaned == "" { + return "" + } + runes := []rune(cleaned) + if len(runes) > maxResponseExcerptLen { + return string(runes[:maxResponseExcerptLen-1]) + "…" + } + return cleaned +} diff --git a/pkg/psclient/client_test.go b/pkg/psclient/client_test.go new file mode 100644 index 000000000..9428ad53f --- /dev/null +++ b/pkg/psclient/client_test.go @@ -0,0 +1,136 @@ +package psclient + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// throttleRequest captures what the client put on the wire so the test can +// assert against PlanetScale's published contract for the throttler endpoint. +type throttleRequest struct { + method string + path string + body string + auth string +} + +func newThrottleServer(t *testing.T, status int, responseBody string) (*httptest.Server, *throttleRequest) { + t.Helper() + captured := &throttleRequest{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + require.NoError(t, err) + captured.method = r.Method + captured.path = r.URL.Path + captured.body = string(body) + captured.auth = r.Header.Get("Authorization") + w.WriteHeader(status) + _, _ = w.Write([]byte(responseBody)) + })) + t.Cleanup(server.Close) + return server, captured +} + +func TestThrottleDeployRequestUsesThrottlerEndpoint(t *testing.T) { + server, captured := newThrottleServer(t, http.StatusOK, `{"keyspaces":["commerce"]}`) + + client, err := NewPSClientWithBaseURL("token-name", "token-value", server.URL) + require.NoError(t, err) + + require.NoError(t, client.ThrottleDeployRequest(t.Context(), &ThrottleDeployRequestRequest{ + Organization: "acme", + Database: "orders", + Number: 42, + ThrottleRatio: 0.85, + })) + + assert.Equal(t, http.MethodPatch, captured.method) + assert.Equal(t, "/v1/organizations/acme/databases/orders/deploy-requests/42/throttler", captured.path) + assert.JSONEq(t, `{"ratio":85}`, captured.body) + assert.Equal(t, "token-name:token-value", captured.auth) +} + +func TestThrottleDeployRequestConvertsRatioToWholePercent(t *testing.T) { + tests := []struct { + ratio float64 + want string + }{ + {0.0, `{"ratio":0}`}, + {0.05, `{"ratio":5}`}, + {0.85, `{"ratio":85}`}, + {0.95, `{"ratio":95}`}, + } + for _, tc := range tests { + server, captured := newThrottleServer(t, http.StatusOK, "{}") + client, err := NewPSClientWithBaseURL("token-name", "token-value", server.URL) + require.NoError(t, err) + + require.NoError(t, client.ThrottleDeployRequest(t.Context(), &ThrottleDeployRequestRequest{ + Organization: "acme", + Database: "orders", + Number: 7, + ThrottleRatio: tc.ratio, + })) + assert.JSONEq(t, tc.want, captured.body, "ratio %v", tc.ratio) + } +} + +// A failing throttle call must carry the API's explanation, not just the status +// line, so an operator can tell a moved endpoint from a deleted deploy request. +func TestThrottleDeployRequestErrorCarriesResponseExcerpt(t *testing.T) { + server, _ := newThrottleServer(t, http.StatusNotFound, "{\"error\":\"Not Found\",\n\"detail\":\"deploy request 42 does not exist\"}") + + client, err := NewPSClientWithBaseURL("token-name", "token-value", server.URL) + require.NoError(t, err) + + err = client.ThrottleDeployRequest(t.Context(), &ThrottleDeployRequestRequest{ + Organization: "acme", + Database: "orders", + Number: 42, + ThrottleRatio: 0.85, + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "404") + assert.Contains(t, err.Error(), "deploy request 42 does not exist") + assert.Contains(t, err.Error(), "throttle deploy request 42 (acme/orders)") + assert.NotContains(t, err.Error(), "\n", "the excerpt must not break log formatting") +} + +func TestThrottleDeployRequestRequiresBaseURL(t *testing.T) { + client := &psClientWrapper{} + err := client.ThrottleDeployRequest(t.Context(), &ThrottleDeployRequestRequest{ + Organization: "acme", + Database: "orders", + Number: 42, + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "base URL") +} + +func TestSanitizeResponseExcerpt(t *testing.T) { + tests := []struct { + name string + body string + want string + }{ + {"empty", "", ""}, + {"whitespace only", " \n\t ", ""}, + {"collapses newlines", "line one\nline two\r\nline three", "line one line two line three"}, + {"drops control runes", "before\x00\x07after", "beforeafter"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, sanitizeResponseExcerpt(strings.NewReader(tc.body))) + }) + } + + long := sanitizeResponseExcerpt(strings.NewReader(strings.Repeat("x", 5000))) + assert.Len(t, []rune(long), maxResponseExcerptLen) + assert.True(t, strings.HasSuffix(long, "…")) +}