From 38891213ce4178aee1b603490eb4cd194527bfea Mon Sep 17 00:00:00 2001 From: Minseok Kim Date: Tue, 11 Aug 2026 10:07:14 +0900 Subject: [PATCH 1/2] Read a metastore page without losing the status code `Get` decodes the body and, when that fails, replaces the real status with -1. An unauthorized response has no body at all, so "this server does not serve the endpoint" and "you may not read it here" arrive identically. A caller that has to tell those apart - the next commit does - cannot. `GetStream` keeps the status and decodes off the connection rather than reading the whole body into memory first, which is what a paged dump of a large table wants anyway. `DumpMetastore(page, size)` reads `/graph/v2/metastore/global`, ordered by `id` rather than `k`: `id` is the primary key, so the server's OFFSET pagination walks an index it already has, and sorting by `k` would order by a base64 string for no benefit. `MetastoreDumpRow` declares only the id, the key and the server's decode of the row. The encoded value, the property map, the timestamp and the direction are left out so the decoder walks past them instead of allocating a string per row for data that would be discarded. Co-Authored-By: Claude Opus 5 (1M context) --- cli/internal/client/actionbase_client.go | 20 +++++++++++ cli/internal/client/http_client.go | 46 ++++++++++++++++++++++++ cli/internal/client/model/metastore.go | 30 ++++++++++++++++ 3 files changed, 96 insertions(+) diff --git a/cli/internal/client/actionbase_client.go b/cli/internal/client/actionbase_client.go index 02399c6b9..76c96a2ae 100644 --- a/cli/internal/client/actionbase_client.go +++ b/cli/internal/client/actionbase_client.go @@ -2,6 +2,7 @@ package client import ( "net/url" + "strconv" "github.com/kakao/actionbase/internal/client/model" ) @@ -121,3 +122,22 @@ func (a *ActionbaseClient) PostRaw(path string, body any) (int, string) { func (a *ActionbaseClient) GetHost() string { return a.client.baseUrl } + +// MetastoreDumpPath is served by 0.4.x and was removed on later servers, so a caller has to be +// ready for a 404 rather than assume it is there. +const MetastoreDumpPath = "/graph/v2/metastore/global" + +// DumpMetastore reads one page of the metastore, ordered by primary key. +// +// Ordered by id rather than k: it is the primary key, so the server's OFFSET pagination walks an +// index it already has. Sorting by k would order by a base64 string for no benefit here. +func (a *ActionbaseClient) DumpMetastore(page, size int) *Response[model.MetastoreDumpPage] { + u := &url.URL{Path: MetastoreDumpPath} + q := u.Query() + q.Set("page", strconv.Itoa(page)) + q.Set("size", strconv.Itoa(size)) + q.Set("sort", "id,asc") + u.RawQuery = q.Encode() + + return GetStream[model.MetastoreDumpPage](a.client, u.String()) +} diff --git a/cli/internal/client/http_client.go b/cli/internal/client/http_client.go index 494de21be..0ac1da4b6 100644 --- a/cli/internal/client/http_client.go +++ b/cli/internal/client/http_client.go @@ -69,6 +69,52 @@ func Get[T any](c *HTTPClient, uri string) *Response[T] { return call[T](c, request, nil) } +// GetStream is Get with two differences that a paged, version-dependent endpoint needs. +// +// It keeps the real status code when the body does not decode. Get replaces it with -1, and an +// unauthorized response carries no body at all, so a caller cannot otherwise tell "this server does +// not serve that path" from "this token may not". And it decodes straight off the connection rather +// than reading the whole body first, so a page is not held twice. +func GetStream[T any](c *HTTPClient, uri string) *Response[T] { + url := fmt.Sprintf("%s%s", c.baseUrl, uri) + request, err := http.NewRequest("GET", url, nil) + if err != nil { + var empty T + return NewResponse[T](-1, &empty, fmt.Errorf("failed to create request: %w", err)) + } + + request.Header.Set("Content-Type", "application/json") + if c.authKey != nil { + request.Header.Set("Authorization", *c.authKey) + } + + if c.context.IsDebugEnabled { + slog.Debug(fmt.Sprintf("→ %s %s", request.Method, request.URL.RequestURI())) + } + + response, err := c.client.Do(request) + if err != nil { + var empty T + return NewResponse[T](-1, &empty, fmt.Errorf("failed to execute request: %w", err)) + } + defer func(body io.ReadCloser) { + _, _ = io.Copy(io.Discard, body) + _ = body.Close() + }(response.Body) + + if c.context.IsDebugEnabled { + slog.Debug(fmt.Sprintf("← %s", response.Status)) + } + + var responseBody T + if err := json.NewDecoder(response.Body).Decode(&responseBody); err != nil { + var empty T + return NewResponse[T](response.StatusCode, &empty, fmt.Errorf("failed to read response Body: %w", err)) + } + + return NewResponse(response.StatusCode, &responseBody, nil) +} + func Post[T any, R any](c *HTTPClient, uri string, requestBody T) *Response[R] { url := fmt.Sprintf("%s%s", c.baseUrl, uri) requestBodyJson, err := json.Marshal(requestBody) diff --git a/cli/internal/client/model/metastore.go b/cli/internal/client/model/metastore.go index 99b2e6afb..69f55a05e 100644 --- a/cli/internal/client/model/metastore.go +++ b/cli/internal/client/model/metastore.go @@ -101,6 +101,36 @@ type DdlPage[T any] struct { Content []T `json:"content"` } +// MetastoreDumpPage is one page of the metastore dump, which the server returns as a Spring Page. +type MetastoreDumpPage struct { + Content []MetastoreDumpRow `json:"content"` + TotalElements int64 `json:"totalElements"` + NumberOfElements int `json:"numberOfElements"` + Last bool `json:"last"` +} + +// MetastoreDumpRow deliberately omits the fields the census does not read. +// +// The server sends the whole encoded value in "v", plus a decoded property map, a timestamp and a +// direction. None of them are declared here, so the decoder walks past them instead of allocating a +// string per row for data that would be thrown away - the whole point is to count a table that is +// too large to hold. +type MetastoreDumpRow struct { + ID int64 `json:"id"` + K string `json:"k"` + Decoded *MetastoreDecodedEdge `json:"decoded"` +} + +// MetastoreDecodedEdge is the server's decode of one row. Src and Tgt are typed as any because the +// server declares them as Object; metadata schemas make them strings, and anything else is still +// printable. +type MetastoreDecodedEdge struct { + Active bool `json:"active"` + Src any `json:"src"` + Tgt any `json:"tgt"` + LabelID int64 `json:"labelId"` +} + type StorageCreateRequest struct { Desc string `json:"desc"` Type string `json:"type"` From 3df02f6c8ac5c1d2b6ae98a58773de6de9734203 Mon Sep 17 00:00:00 2001 From: Minseok Kim Date: Tue, 11 Aug 2026 10:07:14 +0900 Subject: [PATCH 2/2] Report what is left in the JDBC metastore before a scan truncates A metadata `DELETE` only tombstones. The row is written back with `Active.INACTIVE` and stays, and `JdbcHashLabel.scanStorage` applies the SQL `LIMIT` before anything filters on active, so tombstones spend the budget a live scan needs. Past that limit the page is truncated and nothing says so: `AbstractLabel.scan` computes `hasNext` and `DdlService.getAll` drops it building a `DdlPage`. There was no way to ask a server how close it is. `jdbc-metastore remaining` walks `/graph/v2/metastore/global`, the only endpoint showing the rows a metadata scan actually reads - tombstones included, since it does not filter on active - and counts every scan window. A window is `(src, labelId)`, which is what `getAll` builds its scan key from and what the limit applies to; both come off the server's own decode of each row, so nothing here re-implements the codec. `--capacity` is required, not defaulted. `DEFAULT_METADATA_LIMIT` is 1000, but a deployment overrides it with `graph.metadata-fetch-limit` and the server masks the value in actuator. Guessing the OSS default reported windows OVER that had half their room left, with a non-zero exit, which is the worst possible answer to hand a monitor. It is `capacity` rather than `limit` because unlike `--max-rows` and `--max-pages` beside it, it is not a bound this command imposes but the size of the thing being measured. Every window prints, grouped by namespace. Sources are `namespace:service`, and a metastore table can be shared - several deployments pointing at one table, each seeing the others' rows - so without grouping the report leads with somebody else's window. And no threshold decides which windows are worth showing: filtering to the fullest of each kind hid a window holding half the capacity behind one holding slightly more, which was the one still holding live rows. Bounded, because the table being counted is too large to hold and the server pays per request - the controller zips every page with `getTotalCount`, so each request runs a full `COUNT(*)` on top of its own `LIMIT/OFFSET` query. Rows fold into fixed-size counters per page and are dropped. Pages are 500 by default, since server work scales with request count. Five stop conditions: `last`, an empty page, `--max-rows`, `--max-pages` - which covers what the row cap cannot, a server answering with short pages that never say they are last - and reading as many rows as the table claims to hold. Whichever cap fires is named and exits non-zero, so an incomplete census never reads as a complete one. Ctrl-C stops at the next page and reports what it counted; nothing here writes, so there is nothing to unwind. Not using the server's own `prefix=` count: `buildQuery` turns it into `k LIKE '%'` with no escaping, and the keys are base64url, whose alphabet includes `_` - a single-character LIKE wildcard. Roughly a third of prefixes carry one, and each makes the count match rows from other windows. The endpoint was removed after 0.4.x, so the command probes first and says so rather than failing obscurely. Co-Authored-By: Claude Opus 5 (1M context) --- cli/internal/command/command_type.go | 5 + .../command/jdbc_metastore_remaining.go | 442 ++++++++++++++++++ .../command/jdbc_metastore_remaining_test.go | 435 +++++++++++++++++ cli/internal/runner/actionbase_runner.go | 1 + 4 files changed, 883 insertions(+) create mode 100644 cli/internal/command/jdbc_metastore_remaining.go create mode 100644 cli/internal/command/jdbc_metastore_remaining_test.go diff --git a/cli/internal/command/command_type.go b/cli/internal/command/command_type.go index 9f005ef6d..67bd42b1b 100644 --- a/cli/internal/command/command_type.go +++ b/cli/internal/command/command_type.go @@ -85,6 +85,11 @@ var ( name: "migrate", command: "migrate ]|apply [-i ]>", } + + TypeJdbcMetastore = Type{ + name: "jdbc-metastore", + command: "jdbc-metastore --capacity [--page-size ] [--max-pages ] [--max-rows ]", + } ) func (t Type) GetName() string { diff --git a/cli/internal/command/jdbc_metastore_remaining.go b/cli/internal/command/jdbc_metastore_remaining.go new file mode 100644 index 000000000..98bcb8c8a --- /dev/null +++ b/cli/internal/command/jdbc_metastore_remaining.go @@ -0,0 +1,442 @@ +package command + +import ( + "fmt" + "net/http" + "os" + "os/signal" + "sort" + "strconv" + "strings" + "syscall" + + "github.com/kakao/actionbase/internal/client" + clientModel "github.com/kakao/actionbase/internal/client/model" + "github.com/kakao/actionbase/internal/command/model" + "github.com/kakao/actionbase/internal/util" +) + +const ( + // The OSS default for the server's metadataFetchLimit. A metadata scan asks the metastore for that + // many rows and filters inactive ones out afterwards, so tombstones spend the budget a live scan + // needs. Past it, a page is truncated and nothing says so. + // + // Named here only to say so in the error when --limit is missing. It is deliberately not a default + // for --limit: a deployment overrides it with graph.metadata-fetch-limit and the server does not + // expose the value - actuator masks it - so guessing produces confident, wrong verdicts against + // exactly the servers worth asking about. + ossMetadataFetchLimit = 1000 + + // Large on purpose, and the knob that decides what the server pays. Every page request runs a + // full COUNT(*) on top of its own LIMIT/OFFSET query, and the OFFSET cost grows with depth, so a + // table is far cheaper in few large pages than in many small ones. The census retains nothing per + // row, so the page only has to decode comfortably. + defaultRemainingPageSize = 500 + maxRemainingPageSize = 1000 + + // Two caps, because they bound different things. Rows cover a table larger than expected; pages + // cover what rows cannot - a server answering with short pages that never say they are last would + // advance the row count a row at a time, with a COUNT(*) behind every request. + defaultRemainingMaxRows = 500000 + defaultRemainingMaxPages = 2000 + + // The census holds one counter per scan window. More distinct windows than this is not a table + // this command can summarise, so it stops growing rather than grow without bound. + maxRemainingWindows = 10000 +) + +// metadataLabels says which metadata a scan of each label reads, in the order the report prints +// them. Declaration order is the report order, so there is one list rather than a map and a parallel +// slice that could disagree. +// +// A label's id is ValueUtils.stringHash(fullQualifiedName) - xxhash32 with seed 0 over a fixed name +// - so these are constants rather than something to recompute here. Printed from the engine, and +// they hold on 0.4.x too, which shares the names, the seed and the derivation. +var metadataLabels = []struct { + id int64 + name string +}{ + {1773822659, "service"}, + {-1834823921, "storage"}, + {300431312, "label"}, + {-641671585, "alias"}, + {-1531313765, "info"}, + {1397921084, "online_meta"}, + {-1296613462, "nil"}, +} + +// window is one metadata scan prefix: everything a single getAll(name) reads. getAll builds its scan +// key from encodeHashEdgeKeyPrefix(src, labelId) and the limit applies per key, so (src, labelId) is +// exactly the unit the limit bounds. Both come off the server's own decode of each row. +type window struct { + kind string + namespace string + src string + total int + tombstoned int +} + +func (w window) live() int { return w.total - w.tombstoned } + +type windowKey struct { + src string + labelID int64 +} + +// census is fixed-size state. Rows are folded in and dropped; none are retained. +type census struct { + windows map[windowKey]*window + rows int + undecodable int + badKey string + dropped int +} + +func newCensus() *census { + return &census{windows: make(map[windowKey]*window)} +} + +func (c *census) add(rows []clientModel.MetastoreDumpRow) { + for i := range rows { + c.rows++ + decoded := rows[i].Decoded + if decoded == nil { + // The server could not read it. It still occupies a scan window, but there is no way to + // say which one. + c.undecodable++ + if c.badKey == "" { + c.badKey = rows[i].K + } + continue + } + + key := windowKey{src: fmt.Sprintf("%v", decoded.Src), labelID: decoded.LabelID} + w := c.windows[key] + if w == nil { + if len(c.windows) >= maxRemainingWindows { + c.dropped++ + continue + } + w = &window{kind: kindOf(decoded.LabelID), namespace: namespaceOf(key.src), src: key.src} + c.windows[key] = w + } + w.total++ + if !decoded.Active { + w.tombstoned++ + } + } +} + +// namespaceOf splits the owner off the front of a scan source. Sources are `namespace:service`, and +// the namespace is the tenant that owns the metadata - which matters because a metastore table can +// be shared: several deployments point at one table and each sees the others' rows. Reporting +// without it hands an operator another tenant's window as if it were theirs. +// +// A source with no separator is its own namespace rather than an error. The report is a diagnostic; +// an unfamiliar source shape should still be counted and shown. +func namespaceOf(src string) string { + namespace, _, _ := strings.Cut(src, ":") + return namespace +} + +// namespaceTotal is what a namespace's heading says, which has to be known before its first window +// is printed - and is also what orders the namespaces. +type namespaceTotal struct{ rows, windows int } + +// ordered is the report, in print order: namespaces biggest first, known kinds in their declared +// order and the rest after by name, fullest window first within a kind. +// +// One sorted slice rather than a tree of groups. Every window prints, so the report is a single walk +// down this list that starts a heading whenever the namespace changes. +func (c *census) ordered() ([]*window, map[string]namespaceTotal) { + totals := make(map[string]namespaceTotal) + out := make([]*window, 0, len(c.windows)) + for _, w := range c.windows { + t := totals[w.namespace] + t.rows += w.total + t.windows++ + totals[w.namespace] = t + out = append(out, w) + } + + sort.Slice(out, func(i, j int) bool { + a, b := out[i], out[j] + switch { + case a.namespace != b.namespace && totals[a.namespace].rows != totals[b.namespace].rows: + return totals[a.namespace].rows > totals[b.namespace].rows + case a.namespace != b.namespace: + return a.namespace < b.namespace + case kindRank(a.kind) != kindRank(b.kind): + return kindRank(a.kind) < kindRank(b.kind) + case a.kind != b.kind: + return a.kind < b.kind + case a.total != b.total: + return a.total > b.total + default: + return a.src < b.src + } + }) + return out, totals +} + +// kindRank orders the known kinds ahead of anything unrecognised, which sorts after them by name. +func kindRank(kind string) int { + for i, known := range metadataLabels { + if known.name == kind { + return i + } + } + return len(metadataLabels) +} + +func kindOf(labelID int64) string { + for _, known := range metadataLabels { + if known.id == labelID { + return known.name + } + } + return fmt.Sprintf("labelId=%d", labelID) +} + +type JdbcMetastoreRemaining struct { + client *client.ActionbaseClient +} + +func NewJdbcMetastoreRemaining(c *client.ActionbaseClient) *JdbcMetastoreRemaining { + return &JdbcMetastoreRemaining{client: c} +} + +type remainingOptions struct { + capacity int + pageSize int + maxRows int + maxPages int +} + +func (m *JdbcMetastoreRemaining) Execute(args []string) *model.Response { + if len(args) < 1 || args[0] != "remaining" { + return model.Fail(fmt.Sprintf("Usage: %s", m.GetType().GetCommand())) + } + + opts, failure := parseRemainingOptions(args[1:]) + if failure != nil { + return failure + } + + // One cheap page first, both to learn the table size and to find out whether this server serves + // the endpoint at all. It was removed after 0.4.x, so a 404 is an answer, not a failure. + probe := m.client.DumpMetastore(0, 1) + switch { + case probe.StatusCode == http.StatusNotFound: + return model.Fail(fmt.Sprintf( + "No %s on this server (0.4.x only), so this cannot be measured here.", + client.MetastoreDumpPath)) + case probe.StatusCode == http.StatusUnauthorized: + return model.Fail("Unauthorized. Set --authKey or ACT_API_KEY.") + case probe.IsError(): + return model.Fail(fmt.Sprintf("Failed to read the JDBC metastore: %s", describeError(probe))) + } + + totalRows := probe.Body.TotalElements + // One line, because an operator reading it already knows what a metastore is. The page count is + // here because each page costs the server a COUNT(*). + util.Println(fmt.Sprintf("%d rows, capacity %d/window, pages of %d (max %d). Ctrl-C reports partial.", + totalRows, opts.capacity, opts.pageSize, opts.maxPages)) + + stop, stopped := interruptible() + defer stopped() + + c, stoppedBy, err := walk(m.fetch, opts, totalRows, stop) + if err != nil { + return model.Fail(fmt.Sprintf("Failed to read the JDBC metastore: %s", err)) + } + + return report(c, opts, totalRows, stoppedBy) +} + +func (m *JdbcMetastoreRemaining) fetch(page, size int) (*clientModel.MetastoreDumpPage, error) { + resp := m.client.DumpMetastore(page, size) + if resp.IsError() { + return nil, fmt.Errorf("page %d: %s", page, describeError(resp)) + } + return resp.Body, nil +} + +func parseRemainingOptions(args []string) (remainingOptions, *model.Response) { + parser := util.ParseArgs(args) + opts := remainingOptions{ + pageSize: defaultRemainingPageSize, + maxRows: defaultRemainingMaxRows, + maxPages: defaultRemainingMaxPages, + } + + // Required rather than defaulted. Every verdict here is a comparison against this number, and the + // server will not say what its own is, so a default would decide the answer while looking like a + // detail. + if _, ok := parser.Get("capacity"); !ok { + return opts, model.Fail(fmt.Sprintf( + "--capacity is required: the server's graph.metadata-fetch-limit, which it does not expose"+ + " (actuator masks it). OSS default %d.", ossMetadataFetchLimit)) + } + + for _, flag := range []struct { + name string + into *int + max int + }{ + {"capacity", &opts.capacity, 0}, + {"page-size", &opts.pageSize, maxRemainingPageSize}, + {"max-rows", &opts.maxRows, 0}, + {"max-pages", &opts.maxPages, 0}, + } { + v, ok := parser.Get(flag.name) + if !ok { + continue + } + parsed, err := strconv.Atoi(v) + if err != nil || parsed < 1 || (flag.max > 0 && parsed > flag.max) { + bound := "a positive integer" + if flag.max > 0 { + bound = fmt.Sprintf("between 1 and %d", flag.max) + } + return opts, model.Fail(fmt.Sprintf("--%s must be %s", flag.name, bound)) + } + *flag.into = parsed + } + + return opts, nil +} + +// interruptible reports whether Ctrl-C has been pressed, and stops listening when done. The walk +// checks it once per page and returns on the first true, so there is nothing to remember - it stops +// and reports what it has, which beats losing the census to a killed process. Nothing here writes, +// so there is nothing to unwind. +func interruptible() (aborted func() bool, stop func()) { + signals := make(chan os.Signal, 1) + signal.Notify(signals, os.Interrupt, syscall.SIGTERM) + return func() bool { + select { + case <-signals: + return true + default: + return false + } + }, func() { + signal.Stop(signals) + } +} + +type fetchPage func(page, size int) (*clientModel.MetastoreDumpPage, error) + +// walk pages to the end of the table, folding each page into the census and dropping it. +// +// Five ways to stop, and it needs all five. A last page or an empty one are the ordinary two. The +// row and page caps are the two ceilings, and the page cap is the one that matters against a server +// handing back short pages that never say they are last. Reading as many rows as the table claims to +// hold ends it without trusting the server's own `last`. Whichever cap fired is named, so an +// incomplete census never reads as a complete one. +func walk(fetch fetchPage, opts remainingOptions, totalRows int64, aborted func() bool) (*census, string, error) { + c := newCensus() + for page := 0; ; page++ { + switch { + case aborted != nil && aborted(): + return c, "Ctrl-C", nil + case page >= opts.maxPages: + return c, fmt.Sprintf("--max-pages %d", opts.maxPages), nil + case c.rows >= opts.maxRows: + return c, fmt.Sprintf("--max-rows %d", opts.maxRows), nil + } + + body, err := fetch(page, opts.pageSize) + if err != nil { + return c, "", err + } + c.add(body.Content) + + if body.Last || len(body.Content) == 0 || (totalRows > 0 && int64(c.rows) >= totalRows) { + return c, "", nil + } + } +} + +func report(c *census, opts remainingOptions, totalRows int64, stoppedBy string) *model.Response { + capacity := opts.capacity + windows, totals := c.ordered() + over, tight := 0, 0 + + // Every window, always. Printing only the fullest of each kind plus whatever crossed a threshold + // meant the report decided what mattered, and it decided wrong: a window holding half the + // capacity sat behind one holding a little more and never appeared. The list is short - a table + // here has tens of windows, not thousands - and the cap on the census bounds it. + namespace, kind := "", "" + for _, w := range windows { + if w.total > capacity { + over++ + } else if w.total*10 >= capacity*9 { + tight++ // within a tenth of capacity + } + + if w.namespace != namespace { + namespace, kind = w.namespace, "" + t := totals[namespace] + util.Println(fmt.Sprintf("%-10s %5d rows %s", namespace, t.rows, plural(t.windows, "window"))) + } + label := "" + if w.kind != kind { + kind, label = w.kind, w.kind + } + util.Println(fmt.Sprintf(" %-12s %-11s %-22s %5d/%-5d live %-6d tomb %d", + label, room(w, capacity), w.src, w.total, capacity, w.live(), w.tombstoned)) + } + + if c.undecodable > 0 { + // Flush left rather than indented with a namespace's windows: an undecodable row has no + // namespace to belong to, so showing it inside one would read as a claim about that namespace. + util.Println(fmt.Sprintf("%d undecodable (window unknown, e.g. k=%q)", c.undecodable, c.badKey)) + } + if c.dropped > 0 { + util.Println(fmt.Sprintf("%d rows uncounted, over %d windows", c.dropped, maxRemainingWindows)) + } + + summary := fmt.Sprintf("%d/%d rows, %s, %s.", c.rows, totalRows, + plural(len(c.windows), "window"), plural(len(totals), "namespace")) + switch { + case stoppedBy != "": + return model.Fail(fmt.Sprintf("%s Stopped at %s - incomplete, %d over is a lower bound.", + summary, stoppedBy, over)) + case over > 0: + return model.Fail(fmt.Sprintf("%s %d over, %d near. An over-capacity window is already"+ + " truncated and does not say so.", summary, over, tight)) + default: + return model.SuccessWithResult(fmt.Sprintf("%s 0 over, %d near.", summary, tight)) + } +} + +func plural(n int, noun string) string { + if n == 1 { + return fmt.Sprintf("1 %s", noun) + } + return fmt.Sprintf("%d %ss", n, noun) +} + +func room(w *window, capacity int) string { + if w.total > capacity { + return fmt.Sprintf("OVER by %d", w.total-capacity) + } + return fmt.Sprintf("%d left", capacity-w.total) +} + +func describeError[T any](r *client.Response[T]) string { + if r.Error != nil { + return r.Error.Error() + } + return fmt.Sprintf("HTTP %d", r.StatusCode) +} + +func (m *JdbcMetastoreRemaining) GetDescription() string { + return "Report how much room each metadata scan window has left in the JDBC metastore" +} + +func (m *JdbcMetastoreRemaining) GetType() Type { + return TypeJdbcMetastore +} diff --git a/cli/internal/command/jdbc_metastore_remaining_test.go b/cli/internal/command/jdbc_metastore_remaining_test.go new file mode 100644 index 000000000..1fc436b6a --- /dev/null +++ b/cli/internal/command/jdbc_metastore_remaining_test.go @@ -0,0 +1,435 @@ +package command + +import ( + "errors" + "strconv" + "strings" + "testing" + + clientModel "github.com/kakao/actionbase/internal/client/model" +) + +const ( + serviceLabel = 1773822659 + storageLabel = -1834823921 + tableLabel = 300431312 + aliasLabel = -641671585 +) + +func row(src, tgt string, labelID int64, active bool) clientModel.MetastoreDumpRow { + return clientModel.MetastoreDumpRow{ + K: src + ":" + tgt, + Decoded: &clientModel.MetastoreDecodedEdge{ + Active: active, Src: src, Tgt: tgt, LabelID: labelID, + }, + } +} + +// page builds n rows in one window. Only the walk's bounds care about the contents. +func page(n int, total int64, last bool) *clientModel.MetastoreDumpPage { + rows := make([]clientModel.MetastoreDumpRow, 0, n) + for i := 0; i < n; i++ { + rows = append(rows, row("prod:wish", "t", tableLabel, true)) + } + return &clientModel.MetastoreDumpPage{ + Content: rows, TotalElements: total, NumberOfElements: n, Last: last, + } +} + +func onlyWindow(t *testing.T, c *census) *window { + t.Helper() + if len(c.windows) != 1 { + t.Fatalf("expected 1 window, got %d", len(c.windows)) + } + for _, w := range c.windows { + return w + } + return nil +} + +func TestCensusCountsTombstonesTowardTheWindow(t *testing.T) { + c := newCensus() + c.add([]clientModel.MetastoreDumpRow{ + row("prod:wish", "likes", tableLabel, true), + row("prod:wish", "gone", tableLabel, false), + row("prod:wish", "also_gone", tableLabel, false), + }) + + w := onlyWindow(t, c) + if w.total != 3 || w.tombstoned != 2 || w.live() != 1 { + t.Errorf("total/tombstoned/live = %d/%d/%d, want 3/2/1", w.total, w.tombstoned, w.live()) + } +} + +func TestCensusSeparatesWindowsBySourceAndLabel(t *testing.T) { + c := newCensus() + c.add([]clientModel.MetastoreDumpRow{ + row("prod:wish", "likes", tableLabel, true), + row("prod:talk", "likes", tableLabel, true), // same label, other service + row("prod:wish", "likes", aliasLabel, true), // same service, other label + }) + + if len(c.windows) != 3 { + t.Fatalf("expected 3 windows, got %d", len(c.windows)) + } +} + +func TestCensusFoldsAcrossPagesWithoutRetainingRows(t *testing.T) { + c := newCensus() + for i := 0; i < 5; i++ { + rows := []clientModel.MetastoreDumpRow{ + row("prod:wish", "a", tableLabel, true), + row("prod:wish", "b", tableLabel, false), + } + c.add(rows) + // Overwriting the slice the caller handed in must not disturb the counters. + for j := range rows { + rows[j] = clientModel.MetastoreDumpRow{} + } + } + + w := onlyWindow(t, c) + if c.rows != 10 || w.total != 10 || w.tombstoned != 5 { + t.Errorf("rows/total/tombstoned = %d/%d/%d, want 10/10/5", c.rows, w.total, w.tombstoned) + } +} + +func TestCensusCountsUndecodableRowsSeparately(t *testing.T) { + c := newCensus() + c.add([]clientModel.MetastoreDumpRow{ + row("prod:wish", "likes", tableLabel, true), + {K: "not-an-encoded-key", Decoded: nil}, + }) + + if c.undecodable != 1 || c.badKey != "not-an-encoded-key" { + t.Errorf("undecodable/badKey = %d/%q, want 1/not-an-encoded-key", c.undecodable, c.badKey) + } + if len(c.windows) != 1 { + t.Errorf("windows = %d, want 1: an undecodable row joins none", len(c.windows)) + } + if c.rows != 2 { + t.Errorf("rows = %d, want 2: it was still read", c.rows) + } +} + +func TestCensusStopsGrowingAtTheWindowCap(t *testing.T) { + c := newCensus() + rows := make([]clientModel.MetastoreDumpRow, 0, maxRemainingWindows+10) + for i := 0; i < maxRemainingWindows+10; i++ { + rows = append(rows, row("prod:svc", "t", int64(i), true)) + } + c.add(rows) + + if len(c.windows) != maxRemainingWindows { + t.Errorf("windows = %d, want the cap %d", len(c.windows), maxRemainingWindows) + } + if c.dropped != 10 { + t.Errorf("dropped = %d, want 10", c.dropped) + } + if c.rows != maxRemainingWindows+10 { + t.Errorf("rows = %d, want every row read", c.rows) + } +} + +// printOrder is the report as a list of "kind src", which is the whole of what ordered() decides. +func printOrder(c *census) []string { + var out []string + windows, _ := c.ordered() + for _, w := range windows { + out = append(out, w.kind+" "+w.src) + } + return out +} + +func TestCensusOrdersKnownKindsFirstThenFullestWindow(t *testing.T) { + c := newCensus() + c.add([]clientModel.MetastoreDumpRow{ + row("prod:wish", "a", aliasLabel, true), + row("prod:small", "b", tableLabel, true), + row("prod:big", "c", tableLabel, true), + row("prod:big", "d", tableLabel, false), + row("prod:origin", "e", storageLabel, true), + row("prod:origin", "f", serviceLabel, true), + row("prod:wish", "g", 424242, true), + }) + + want := []string{ + "service prod:origin", + "storage prod:origin", + "label prod:big", // two rows, so it leads the one-row window + "label prod:small", + "alias prod:wish", + "labelId=424242 prod:wish", // unknown kinds sort after the known ones + } + if got := printOrder(c); strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("order =\n %s\nwant\n %s", strings.Join(got, "\n "), strings.Join(want, "\n ")) + } +} + +func TestKindOfNamesTheKnownMetadataLabels(t *testing.T) { + for labelID, want := range map[int64]string{ + serviceLabel: "service", + storageLabel: "storage", + tableLabel: "label", + aliasLabel: "alias", + 12345: "labelId=12345", + } { + if got := kindOf(labelID); got != want { + t.Errorf("kindOf(%d) = %q, want %q", labelID, got, want) + } + } +} + +func TestCensusKeepsANonStringSourcePrintable(t *testing.T) { + c := newCensus() + c.add([]clientModel.MetastoreDumpRow{{ + K: "k", + Decoded: &clientModel.MetastoreDecodedEdge{Active: true, Src: float64(42), Tgt: float64(7), LabelID: 3}, + }}) + + if w := onlyWindow(t, c); w.src != "42" { + t.Errorf("src = %q, want %q", w.src, "42") + } +} + +func TestWalkStopsWhenTheServerSaysThePageIsLast(t *testing.T) { + requests := 0 + fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) { + requests++ + return page(2, 4, p == 1), nil + } + + c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 1000, maxPages: 100}, 4, nil) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if stoppedBy != "" { + t.Errorf("stoppedBy = %q, want empty: a complete walk hit no cap", stoppedBy) + } + if requests != 2 || c.rows != 4 { + t.Errorf("requests/rows = %d/%d, want 2/4", requests, c.rows) + } +} + +func TestWalkStopsOnAnEmptyPageEvenIfLastNeverArrives(t *testing.T) { + fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) { + if p == 0 { + return page(2, 100, false), nil + } + return page(0, 100, false), nil + } + + c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 1000, maxPages: 100}, 100, nil) + + if err != nil || stoppedBy != "" || c.rows != 2 { + t.Errorf("err/stoppedBy/rows = %v/%q/%d, want nil/empty/2", err, stoppedBy, c.rows) + } +} + +// The case the row cap alone does not cover: a server answering every request with one row and never +// admitting to being last. Without a page cap this is a request storm, and each of those requests +// costs the server a COUNT(*). +func TestWalkStopsAtThePageCapWhenPagesStayShort(t *testing.T) { + requests := 0 + fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) { + requests++ + return page(1, 1_000_000_000, false), nil + } + + c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 500, maxRows: 500000, maxPages: 20}, 1_000_000_000, nil) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if requests != 20 || c.rows != 20 { + t.Errorf("requests/rows = %d/%d, want 20/20: the page cap, not the row cap", requests, c.rows) + } + if stoppedBy != "--max-pages 20" { + t.Errorf("stoppedBy = %q, want the page cap named", stoppedBy) + } +} + +func TestWalkStopsAtTheRowCapWhenPagesAreFull(t *testing.T) { + fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) { + return page(size, 1_000_000_000, false), nil + } + + c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 100, maxRows: 250, maxPages: 1000}, 1_000_000_000, nil) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if c.rows != 300 { + t.Errorf("rows = %d, want 300: the cap is checked between pages", c.rows) + } + if stoppedBy != "--max-rows 250" { + t.Errorf("stoppedBy = %q, want the row cap named", stoppedBy) + } +} + +// The table's own row count bounds the walk, so a `last` that never comes cannot extend it past the +// end of the table. +func TestWalkStopsOnceItHasReadTheWholeTable(t *testing.T) { + requests := 0 + fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) { + requests++ + return page(size, 10, false), nil + } + + c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 5, maxRows: 100000, maxPages: 1000}, 10, nil) + + if err != nil || stoppedBy != "" { + t.Errorf("err/stoppedBy = %v/%q, want nil/empty", err, stoppedBy) + } + if requests != 2 || c.rows != 10 { + t.Errorf("requests/rows = %d/%d, want 2/10", requests, c.rows) + } +} + +func TestWalkStopsOnInterruptAndKeepsWhatItRead(t *testing.T) { + requests := 0 + fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) { + requests++ + return page(2, 1_000_000, false), nil + } + + c, stoppedBy, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 500000, maxPages: 1000}, 1_000_000, func() bool { return requests >= 3 }) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if stoppedBy != "Ctrl-C" { + t.Errorf("stoppedBy = %q, want Ctrl-C", stoppedBy) + } + if c.rows != 6 { + t.Errorf("rows = %d, want the 6 rows read before the interrupt", c.rows) + } +} + +func TestWalkReturnsWhatItReadWhenAPageFails(t *testing.T) { + fetch := func(p, size int) (*clientModel.MetastoreDumpPage, error) { + if p == 0 { + return page(2, 100, false), nil + } + return nil, errors.New("page 1: HTTP 500") + } + + c, _, err := walk(fetch, remainingOptions{pageSize: 2, maxRows: 1000, maxPages: 100}, 100, nil) + + if err == nil { + t.Fatal("expected the failure to surface") + } + if c.rows != 2 { + t.Errorf("rows = %d, want the 2 rows already read", c.rows) + } +} + +func TestParseRemainingOptionsRequiresTheCapacity(t *testing.T) { + _, failure := parseRemainingOptions(nil) + if failure == nil { + t.Fatal("a census with no --capacity should have been rejected") + } + if !strings.Contains(*failure.ErrorMessage, "graph.metadata-fetch-limit") { + t.Errorf("error = %q, want it to name the server property", *failure.ErrorMessage) + } +} + +func TestParseRemainingOptionsDefaultsAndRejects(t *testing.T) { + opts, failure := parseRemainingOptions([]string{"--capacity", "2000"}) + if failure != nil { + t.Fatalf("defaults should parse: %v", *failure.ErrorMessage) + } + if opts.capacity != 2000 { + t.Errorf("opts = %+v, want capacity 2000", opts) + } + if opts.pageSize != defaultRemainingPageSize || opts.maxPages != defaultRemainingMaxPages { + t.Errorf("defaults = %+v, want page size %d and max pages %d", + opts, defaultRemainingPageSize, defaultRemainingMaxPages) + } + + for _, args := range [][]string{ + {"--capacity", "2000", "--page-size", "0"}, + {"--capacity", "2000", "--page-size", strconv.Itoa(maxRemainingPageSize + 1)}, + {"--capacity", "2000", "--max-pages", "0"}, + {"--capacity", "2000", "--max-rows", "-1"}, + {"--capacity", "abc"}, + } { + if _, failure := parseRemainingOptions(args); failure == nil { + t.Errorf("%v should have been rejected", args) + } + } +} + +func TestNamespaceOfSplitsTheOwnerOffTheSource(t *testing.T) { + for src, want := range map[string]string{ + "stg:origin": "stg", + "crm:aircrm": "crm", + "a:b:c": "a", + "no-separator": "no-separator", + ":leading-colon": "", + } { + if got := namespaceOf(src); got != want { + t.Errorf("namespaceOf(%q) = %q, want %q", src, got, want) + } + } +} + +// A shared metastore table holds every namespace pointed at it, so the census has to keep them +// apart. This is the shape of `ab_default`, where several deployments read one table. +func sharedTable() *census { + c := newCensus() + rows := []clientModel.MetastoreDumpRow{} + for i := 0; i < 30; i++ { + rows = append(rows, row("stg:origin", "s"+strconv.Itoa(i), storageLabel, false)) + } + for i := 0; i < 8; i++ { + rows = append(rows, row("crm:aircrm", "t"+strconv.Itoa(i), tableLabel, true)) + } + for i := 0; i < 2; i++ { + rows = append(rows, row("bmt:origin", "s"+strconv.Itoa(i), storageLabel, true)) + } + c.add(rows) + return c +} + +func TestCensusOrdersBiggestNamespaceFirst(t *testing.T) { + c := sharedTable() + + windows, totals := c.ordered() + if len(totals) != 3 || totals["stg"].rows != 30 || totals["stg"].windows != 1 { + t.Fatalf("totals = %+v, want 3 namespaces and stg holding 30 rows in 1 window", totals) + } + + var got []string + for _, w := range windows { + got = append(got, w.namespace) + } + if want := "stg,crm,bmt"; strings.Join(got, ",") != want { + t.Errorf("namespaces = %v, want %s", got, want) + } +} + +// Every window prints, not just the fullest of each kind. The report used to decide what mattered +// and hid a window holding half the capacity behind one holding a little more. +func TestReportPrintsEveryWindow(t *testing.T) { + c := sharedTable() + + got := report(c, remainingOptions{capacity: 2000}, 45, "") + if !got.IsSuccess { + t.Fatalf("nothing is over 2000: %v", *got.ErrorMessage) + } + if !strings.Contains(*got.Result, "3 windows, 3 namespaces") { + t.Errorf("summary = %q, want every window and namespace counted", *got.Result) + } +} + +func TestReportFailsOnAnOverCapacityWindow(t *testing.T) { + got := report(sharedTable(), remainingOptions{capacity: 10}, 45, "") + if got.IsSuccess { + t.Fatal("a 30-row window against a capacity of 10 should fail") + } + if !strings.Contains(*got.ErrorMessage, "over-capacity") { + t.Errorf("error = %q, want it to name the truncation", *got.ErrorMessage) + } +} diff --git a/cli/internal/runner/actionbase_runner.go b/cli/internal/runner/actionbase_runner.go index 9b6dffcdf..15f109943 100644 --- a/cli/internal/runner/actionbase_runner.go +++ b/cli/internal/runner/actionbase_runner.go @@ -74,6 +74,7 @@ func NewActionbaseCommandLineRunner(version, host string, authKey *string, curre runner.RegisterCommand(command.TypeGuide.GetName(), command.NewGuide(runner, actionbaseClient)) runner.RegisterCommand(command.TypeDoctor.GetName(), command.NewDoctor(actionbaseClient)) runner.RegisterCommand(command.TypeMigrate.GetName(), command.NewMigrate(actionbaseClient)) + runner.RegisterCommand(command.TypeJdbcMetastore.GetName(), command.NewJdbcMetastoreRemaining(actionbaseClient)) return runner }