diff --git a/README.md b/README.md index a45d752..8c935c4 100644 --- a/README.md +++ b/README.md @@ -24,11 +24,11 @@ not PQL) is forwarded verbatim. | Path | Behaviour | |---|---| -| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract`/`count` query is **summed** instead. | -| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. An `extract`/`count` query is **summed** instead. | -| `GET /pdb/query/v4/facts/[/]` | Same fan-out and merge as `/facts`, and an `extract`/`count` query is **summed** the same way. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added — except on the fact's own path, which is **synthesised** from the `/facts` merge (see provenance). | +| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract` query with a `function` column is **combined** instead. | +| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. An `extract` query with a `function` column is **combined** instead. | +| `GET /pdb/query/v4/facts/[/]` | Same fan-out and merge as `/facts`, and an `extract` query with a `function` column is **combined** the same way. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added — except on the fact's own path, which is **synthesised** from the `/facts` merge (see provenance). | | `GET /pdb/query/v4/fact-names` | Fan out to all and serve the **union** of the flat name arrays, deduped and re-sorted, re-paged across backends, plus the `pdbmux_source` name while injection is on. `order_by` is only valid on `name`. | -| `GET /pdb/query/v4/resources` | An `extract`/`count` query is fanned out and **summed**; any other query is an unmerged pass-through. | +| `GET /pdb/query/v4/resources` | An `extract` query with a `function` column is fanned out and **combined**; any other query is an unmerged pass-through. | | `GET /pdb/query/v4/reports` | Fan out to all and serve the **union**, deduped by report `hash`, re-ordered and re-paged across backends. | | `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. | | `GET /pdb/query/v4/event-counts` | Fan out to all and **sum** each subject's counts into one row per subject. | @@ -87,36 +87,79 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see events, which carry no id of their own, dedupe on the verbatim record (a node reporting to more than one backend stores identical records in each). - **Aggregates** — `extract`/`group_by` rows are counts, not records, so each - backend returns a partial answer that has to be **added**, not deduped. This + backend returns a partial answer that has to be **combined**, not deduped. This covers `/event-counts`, `/aggregate-event-counts`, and any `/reports`, `/nodes`, `/resources`, `/facts` or `/facts/[/]` query whose `extract` carries a `["function", ...]` column. - - The grouping key is the row's non-aggregate fields: for `/reports`, - `/nodes`, `/resources`, `/facts` and `/facts/` they come from the - query — the plain `extract` fields plus any `group_by` clause — and for the - event-count endpoints from the row itself (`subject_type`/`subject`, or - `summarize_by`), whose remaining fields are all counts. + - The grouping key is the row's full set of non-aggregate columns: for + `/reports`, `/nodes`, `/resources`, `/facts` and `/facts/` they come + from the query — the plain `extract` fields, the row-function columns and + any `group_by` clause — and for the event-count endpoints from the row + itself (`subject_type`/`subject`, or `summarize_by`), whose remaining fields + are all counts. - On `/nodes`, `/facts` and `/facts/[/]` this takes precedence over the `certname` merge: an aggregate row has no `certname`, so deduping would collapse every backend's rows into one backend's numbers. A query with no `function` column — including a plain `extract` projection — still merges by `certname`. - PuppetDB accepts `count`, `sum`, `avg`, `min`, `max`, `to_string` and - `jsonb_typeof` as `extract` functions. Only `count` and `sum` are additive, - so only those two merge correctly. `avg`, `min` and `max` are folded like any - other numeric column and their merged value is **wrong**; `to_string` and - `jsonb_typeof` are row functions whose non-numeric column is carried through - from the first backend to report the key. Query a single backend directly - for any of those five. + `jsonb_typeof` as `extract` functions, and names each response column after + the function itself. Each is combined by its own rule rather than by a + blanket sum: + + | function | merged across backends by | + | --- | --- | + | `count` | adding | + | `sum` | adding | + | `min` | the smallest value any backend reported, on text columns as well as numeric ones | + | `max` | the largest value any backend reported, likewise | + | `avg` | rewriting the upstream query into `sum` + `count` of the same column and dividing the totals, so the answer is the estate's true weighted average, not an average of averages | + | `to_string` | nothing — it is a row function, so it groups like a plain projected column | + | `jsonb_typeof` | likewise | + + - Because each column is named after its function, an `extract` that projects + the **same function twice** — any of them — names one response column twice. + openvoxdb aliases the repeat as `_2` (then `_3`, and so on), which is + neither a grouping key nor an aggregate `pdbmux` knows to fold, so the first + backend's value would freeze into the merged row. Such a query is refused + with **400** naming the clashing column, as is one whose plain `extract` + field takes the name a projected function would use. Repeating a plain field + is not a clash: the copy holds the same value as the key it duplicates. + - The `avg` rewrite is invisible to the client: the request still answers under + the `avg` key. It needs the `sum` and `count` response columns for itself, so + an `extract` that also projects a `sum` or a `count` is refused with **400** + naming the clash rather than answered with a wrong number. An `avg` over no + rows stays `null`, as upstream. An `order_by` on `avg` is applied to the + merged rows here, not upstream. + - `avg` is folded as `sum / count` in float64, while a single openvoxdb divides + in Postgres `numeric`, which is arbitrary-precision. Whole-number averages + round-trip exactly; a fractional one can differ from a single backend's + answer in the low-order digits, as can a `sum` beyond 2^53. + - An `extract` function `pdbmux` has no combiner for is refused with **400** + rather than folded on a guess. - `/resources` has no cross-backend record identity to dedupe on, so only its aggregate queries merge; everything else stays an unmerged pass-through. - - Rows sharing a key collapse into one with their numeric columns summed. A key - only one backend reported is passed through byte-for-byte. An aggregate column - that is absent or non-numeric in a row is skipped, never zeroed, so the - backends that did report a number still count. + - Rows sharing a key collapse into one with each aggregate column combined by + its own rule. A key only one backend reported is passed through + byte-for-byte. An aggregate column that is absent or `null` in a row is + skipped, never zeroed or treated as an extreme, so the backends that did + report a value still count. + - `to_string` and `jsonb_typeof` compile to scalar expressions upstream, so + they return one row per record rather than an aggregate. They form part of + the grouping key alongside the plain `extract` fields and the `group_by` + clause — including a `group_by` that names the function itself. An `extract` + of nothing but row functions has no aggregate to fold, so every backend's + rows are kept as they came — and, having one row per record rather than per + group, they keep the upstream `limit` that bounds them. + - `limit` and `offset` are **not** forwarded for an `extract` that folds: a + backend's own first N groups are not the merged result's first N, and a group + truncated away on one backend would fold to a wrong value. Every group is + fetched and the window cut after the fold, which an aggregate's row count — + one per distinct group value — keeps affordable. `include_total` still + reports the merged group count. - A `/reports` query with no `function` column is a projection of real reports, not an aggregate, and stays on the union path. - - `include_total=true` on a summed endpoint reports the **merged** row count, + - `include_total=true` on a combined endpoint reports the **merged** row count, not the sum of the backends' `X-Records`, since shared keys collapse. ### Provenance: the `pdbmux_source` fact @@ -191,7 +234,7 @@ answered `[]` from the configured names alone, with no fan-out at all, and a value naming one filters a record set fetched under a key the value is not part of. The record set is a property of the estate rather than of the filter, so every value of it — and the unfiltered path — share one entry and one fetch. -An `extract`/`count` query still takes the summing branch, and the gated query +An `extract` query with a `function` column still takes the combining branch, and the gated query shapes above still answer `[]`, as does every form while injection is off — with the name kept out of `/fact-names`, since nothing then produces it. @@ -251,9 +294,11 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so the merged set's existing order). A record missing an ordered field sorts first. - Backends are asked for the first `offset + limit` records — never an `offset` — and the requested window is then cut from the merged, re-sorted set. +- A folded `extract` aggregate is the exception: neither `limit` nor `offset` is + forwarded, since a group truncated on one backend cannot be folded correctly. - `include_total=true` on a union endpoint makes `pdbmux` sum each backend's `X-Records` header into one merged header. Deduped records are counted once per - backend, so the total is an upper bound. Summed endpoints report the merged row + backend, so the total is an upper bound. Combined endpoints report the merged row count instead. - A malformed `limit`, `offset` or `order_by` gets a `400` rather than being forwarded. @@ -364,7 +409,7 @@ not answering. `/fact-names` record sets **in memory** so a busy Puppetboard does not re-fan-out the same query every few seconds — its facts overview and fact drilldown are two of the pages that hit hardest. Everything else runs uncached — including -`extract`/`count` aggregates on those paths, and +`extract` aggregates on those paths, and the `/pdb/meta/v1/*` and `/metrics/*` endpoints, which are served live on every request. The cache is an interface, and `/reports` gets its own (S3-backed) backend later without further handler changes. diff --git a/aggregate.go b/aggregate.go index 9592421..41da8b2 100644 --- a/aggregate.go +++ b/aggregate.go @@ -2,83 +2,244 @@ package main import ( "encoding/json" + "fmt" "sort" "strconv" "strings" ) -// aggregateSpec names the columns of an extract/group_by result set: keys -// identify a row across backends, sums are the numeric aggregate columns added -// together. -type aggregateSpec struct { - keys []string - sums []string +// combineOp is how one aggregate column's per-backend values become one value. +type combineOp int + +const ( + opAdd combineOp = iota + opMin + opMax +) + +// aggColumn is one aggregate column of a result row: the JSON key openvoxdb +// names it and the operation that folds the backends' values together. +type aggColumn struct { + name string + op combineOp } -// parseAggregate reads a PuppetDB AST query and returns the aggregate shape of -// its response, or nil when the query is not an aggregate — only a top-level -// `extract` carrying at least one `["function", ...]` column produces summable -// rows. Key columns are the plain (non-function) extract fields, unioned with an -// explicit `group_by` clause when the query has one. -func parseAggregate(query string) *aggregateSpec { +// rowShape splits a result row into the columns that identify it and the +// aggregate columns that are folded. finish, when set, rewrites the merged row +// before it is encoded and forces re-encoding even for a single-backend row. +type rowShape struct { + keys []string + aggs []aggColumn + finish func(map[string]json.RawMessage) +} + +// openvoxdb's accepted extract functions are the keys of pdb-fns->pg-fns, +// src/puppetlabs/puppetdb/query_eng/engine.clj:234-241, and each response column +// is named after the function itself (compile-fnexpression, engine.clj:1488-1495). +// count and sum add across backends; min and max take the extreme, on text +// columns as well as numeric ones, which openvoxdb allows because the numeric +// guard applies only to comparison clauses (engine.clj:2483-2490). avg is not +// combinable from the shard rows at all, so it is rewritten upstream — see +// rewriteAvg. +var aggregateOps = map[string]combineOp{ + "count": opAdd, + "sum": opAdd, + "min": opMin, + "max": opMax, +} + +// to_string and jsonb_typeof compile to scalar Postgres expressions, so they +// yield one row per input row rather than an aggregate: they name an ordinary +// projected column and identify a row instead of being folded into one. +var rowFns = map[string]bool{ + "to_string": true, + "jsonb_typeof": true, +} + +const avgColumn = "avg" + +// aggregateSpec is the merge shape of an extract query's response. +type aggregateSpec struct { + keys []string + aggs []aggColumn + query string // rewritten upstream query, empty when the request's own is used + avg bool // the client asked for avg; sum and count are pdbmux's helpers +} + +// parseAggregate reads a PuppetDB AST query and returns the merge shape of its +// response, or nil when the query is not an extract carrying a `["function",...]` +// column. A non-nil error means the query is an aggregate pdbmux cannot merge +// and must be refused rather than answered with a wrong number. +// +// Key columns are the plain extract fields, the row-function columns and any +// `group_by` clause; only genuine aggregates are folded. +func parseAggregate(query string) (*aggregateSpec, error) { if strings.TrimSpace(query) == "" { - return nil + return nil, nil } var ast []json.RawMessage if json.Unmarshal([]byte(query), &ast) != nil || len(ast) < 2 { - return nil + return nil, nil } var op string if json.Unmarshal(ast[0], &op) != nil || op != "extract" { - return nil + return nil, nil } var cols []json.RawMessage if json.Unmarshal(ast[1], &cols) != nil { - return nil + return nil, nil } spec := &aggregateSpec{} + var avgArgs []json.RawMessage + sawFunction := false + projected := map[string]bool{} + fnColumns := map[string]bool{} for _, col := range cols { var name string if json.Unmarshal(col, &name) == nil { + // A field repeating another field is harmless — the copy holds the same + // value as the key it duplicates — but one taking a function's name is + // the same clash the other way round. + if fnColumns[name] { + return nil, columnClash(name) + } + projected[name] = true spec.keys = appendUnique(spec.keys, name) continue } - if fn, ok := functionName(col); ok { - spec.sums = appendUnique(spec.sums, fn) + fn, args, ok := functionColumn(col) + if !ok { + continue + } + sawFunction = true + if projected[fn] { + return nil, columnClash(fn) + } + projected[fn], fnColumns[fn] = true, true + switch { + case rowFns[fn]: + spec.keys = appendUnique(spec.keys, fn) + case fn == avgColumn: + if len(args) == 0 { + return nil, fmt.Errorf("extract function avg needs a column to average") + } + spec.avg, avgArgs = true, args + default: + combine, known := aggregateOps[fn] + if !known { + return nil, fmt.Errorf("extract function %q cannot be merged across backends", fn) + } + spec.aggs = append(spec.aggs, aggColumn{name: fn, op: combine}) } } - if len(spec.sums) == 0 { - return nil + if !sawFunction { + return nil, nil } for _, node := range ast[2:] { for _, f := range groupByFields(node) { + if hasAgg(spec.aggs, f) || (spec.avg && f == avgColumn) { + return nil, groupByClash(f) + } spec.keys = appendUnique(spec.keys, f) } } - return spec + if spec.avg { + if err := spec.rewriteAvg(ast, cols, avgArgs); err != nil { + return nil, err + } + } + return spec, nil } -// functionName returns the response column an extract function produces, which -// PuppetDB names after the function itself: ["function","count","certname"] -// yields a "count" column. -func functionName(col json.RawMessage) (string, bool) { +// columnClash refuses a projection naming one response column twice. openvoxdb +// aliases every extract column after its function, so a repeat comes back as an +// order-dependent "_2" that is neither a grouping key nor a folded +// aggregate, leaving the first backend's value frozen in the merged row. +func columnClash(name string) error { + return fmt.Errorf("extract projects the column %q more than once: openvoxdb returns the repeat as %q, which pdbmux can neither key on nor fold", name, name+"_2") +} + +// groupByClash refuses a group_by naming an aggregate column: the same key +// cannot both identify a row and be folded across backends. openvoxdb rejects +// the shape too, so refusing here is a 400 instead of a failed upstream query. +func groupByClash(name string) error { + return fmt.Errorf("group_by names the aggregate column %q: pdbmux folds that column across backends, so it cannot also be a grouping key", name) +} + +// rewriteAvg replaces the client's avg column with the sum and count of the same +// expression, so the true weighted average can be computed from the shards: +// Postgres avg(x) is sum(x)/count(x), and both of those do combine. +func (a *aggregateSpec) rewriteAvg(ast, cols []json.RawMessage, args []json.RawMessage) error { + for _, taken := range []string{"sum", "count"} { + // openvoxdb names a column after its function, so a second one of the same + // name comes back as an order-dependent "_2" rather than its own key. + if contains(a.keys, taken) || hasAgg(a.aggs, taken) { + return fmt.Errorf("avg cannot be merged alongside a %q column: pdbmux rewrites avg into an upstream sum and count, which openvoxdb would return under the same response key", taken) + } + } + sumCol, err := functionNode("sum", args) + if err != nil { + return err + } + countCol, err := functionNode("count", args) + if err != nil { + return err + } + + out := make([]json.RawMessage, 0, len(cols)+1) + for _, col := range cols { + if fn, _, ok := functionColumn(col); ok && fn == avgColumn { + out = append(out, sumCol) + continue + } + out = append(out, col) + } + out = append(out, countCol) + + rewritten := append([]json.RawMessage(nil), ast...) + encoded, err := json.Marshal(out) + if err != nil { + return err + } + rewritten[1] = encoded + query, err := json.Marshal(rewritten) + if err != nil { + return err + } + a.query = string(query) + a.aggs = append(a.aggs, aggColumn{name: "sum", op: opAdd}, aggColumn{name: "count", op: opAdd}) + return nil +} + +func functionNode(fn string, args []json.RawMessage) (json.RawMessage, error) { + parts := make([]json.RawMessage, 0, len(args)+2) + parts = append(parts, json.RawMessage(`"function"`), json.RawMessage(`"`+fn+`"`)) + parts = append(parts, args...) + return json.Marshal(parts) +} + +// functionColumn returns the response column an extract function produces, which +// openvoxdb names after the function itself — ["function","count","certname"] +// yields a "count" column — along with the function's arguments. +func functionColumn(col json.RawMessage) (string, []json.RawMessage, bool) { var parts []json.RawMessage if json.Unmarshal(col, &parts) != nil || len(parts) < 2 { - return "", false + return "", nil, false } var head, name string if json.Unmarshal(parts[0], &head) != nil || head != "function" { - return "", false + return "", nil, false } if json.Unmarshal(parts[1], &name) != nil || name == "" { - return "", false + return "", nil, false } - return name, true + return name, parts[2:], true } -// groupByFields returns the field names of a ["group_by", ...] AST node, or nil -// for any other node. +// groupByFields returns the response columns a ["group_by", ...] AST node names, +// or nil for any other node. An entry may be a plain field or a function node, +// which groups on that function's own column. func groupByFields(node json.RawMessage) []string { var parts []json.RawMessage if json.Unmarshal(node, &parts) != nil || len(parts) < 2 { @@ -93,6 +254,10 @@ func groupByFields(node json.RawMessage) []string { var name string if json.Unmarshal(p, &name) == nil { out = append(out, name) + continue + } + if fn, _, ok := functionColumn(p); ok { + out = append(out, fn) } } return out @@ -105,19 +270,46 @@ func appendUnique(s []string, v string) []string { return append(s, v) } -// columns reports which fields of a row form its grouping key and which are -// summed. A spec is fixed by the query, so the row is ignored. -func (a *aggregateSpec) columns(map[string]json.RawMessage) ([]string, []string) { - return a.keys, a.sums +func hasAgg(s []aggColumn, name string) bool { + for _, x := range s { + if x.name == name { + return true + } + } + return false } -// inferredColumns derives an event-counts row's shape from the row itself: the +// shape reports a row's merge shape. A spec is fixed by the query, so the row is +// ignored. +func (a *aggregateSpec) shape(map[string]json.RawMessage) rowShape { + sh := rowShape{keys: a.keys, aggs: a.aggs} + if a.avg { + sh.finish = finishAvg + } + return sh +} + +// finishAvg turns the summed helper columns into the avg key the client asked +// for. An empty count is openvoxdb's own answer for an average over no rows. +func finishAvg(row map[string]json.RawMessage) { + sum, sumOK := numberOf(row["sum"]) + count, countOK := numberOf(row["count"]) + delete(row, "sum") + delete(row, "count") + if !sumOK || !countOK || count == 0 { + row[avgColumn] = json.RawMessage("null") + return + } + row[avgColumn] = json.RawMessage(strconv.FormatFloat(sum/count, 'f', -1, 64)) +} + +// inferredShape derives an event-counts row's shape from the row itself: the // counts to add (successes, failures, noops, skips, total) are its numeric -// fields, plus any null one — PuppetDB nulls an aggregate column when a backend +// fields, plus any null one — openvoxdb nulls an aggregate column when a backend // matched nothing — and everything else, subject_type/subject/summarize_by, // identifies the row. Those endpoints have a fixed response shape with no // numeric key field, so nothing summable is mistaken for identity. -func inferredColumns(row map[string]json.RawMessage) ([]string, []string) { +func inferredShape(row map[string]json.RawMessage) rowShape { var keys, sums []string for name, val := range row { if isJSONNumber(val) || isJSONNull(val) { @@ -128,7 +320,11 @@ func inferredColumns(row map[string]json.RawMessage) ([]string, []string) { } sort.Strings(keys) sort.Strings(sums) - return keys, sums + aggs := make([]aggColumn, 0, len(sums)) + for _, s := range sums { + aggs = append(aggs, aggColumn{name: s, op: opAdd}) + } + return rowShape{keys: keys, aggs: aggs} } // isJSONNumber reports whether a raw JSON value is a number. @@ -144,31 +340,33 @@ func isJSONNull(raw json.RawMessage) bool { return strings.TrimSpace(string(raw)) == "null" } -// sumGroup accumulates the rows sharing one grouping key. -type sumGroup struct { - raw json.RawMessage // first contributing row, verbatim - row map[string]json.RawMessage // its decoded fields - totals map[string]float64 // running sum per aggregate column - merged bool // a second row was folded in +// mergeGroup accumulates the rows sharing one grouping key. +type mergeGroup struct { + raw json.RawMessage // first contributing row, verbatim + row map[string]json.RawMessage // its decoded fields + totals map[string]float64 // running sum per additive column + extremes map[string]json.RawMessage // running min/max per extreme column + finish func(map[string]json.RawMessage) + merged bool // a later row changed the group } -// sumRows folds each backend's aggregate rows into one row per grouping key, -// adding the numeric aggregate columns. columns decides, per row, which fields -// are the key and which are summed. +// combineRows folds each backend's aggregate rows into one row per grouping key, +// combining each aggregate column by its own operation. shape decides, per row, +// which fields are the key and how each aggregate column combines. // -// A row that is not a JSON object passes through untouched, as does the sole row -// of a key only one backend reported — those keep their upstream bytes. An -// aggregate column that is absent or non-numeric in a later row is left at the -// earlier backend's value rather than being coerced to zero. results come in -// configured backend order and the output keeps first-seen order, a tie-break -// only. -func sumRows(results []backendResult, columns func(map[string]json.RawMessage) ([]string, []string)) []json.RawMessage { +// A row that is not a JSON object passes through untouched, as does the sole +// row of a key only one backend reported and every row of a shape with no +// aggregate column to fold — those keep their upstream bytes. An aggregate +// column that is absent or null in a later row leaves the group's value alone +// rather than being coerced to zero. results come in configured backend order +// and the output keeps first-seen order, a tie-break only. +func combineRows(results []backendResult, shape func(map[string]json.RawMessage) rowShape) []json.RawMessage { type slot struct { raw json.RawMessage // passthrough row, when group is nil - group *sumGroup + group *mergeGroup } var order []slot - groups := map[string]*sumGroup{} + groups := map[string]*mergeGroup{} for _, res := range results { for _, rec := range res.records { @@ -177,32 +375,22 @@ func sumRows(results []backendResult, columns func(map[string]json.RawMessage) ( order = append(order, slot{raw: rec.Raw}) continue } - keys, sums := columns(row) - k := groupKey(row, keys) + sh := shape(row) + if len(sh.aggs) == 0 { + // Only row functions were projected, so there is nothing to fold + // and each backend's rows stand on their own. + order = append(order, slot{raw: rec.Raw}) + continue + } + k := groupKey(row, sh.keys) g, ok := groups[k] if !ok { - g = &sumGroup{raw: rec.Raw, row: row, totals: map[string]float64{}} - for _, s := range sums { - if n, ok := numberOf(row[s]); ok { - g.totals[s] = n - } - } + g = newMergeGroup(rec.Raw, row, sh) groups[k] = g order = append(order, slot{group: g}) continue } - for _, s := range sums { - n, ok := numberOf(row[s]) - if !ok { - continue - } - if _, seen := g.totals[s]; !seen { - // First numeric value for a column the earlier row lacked. - g.totals[s] = 0 - } - g.totals[s] += n - g.merged = true - } + g.fold(row, sh.aggs) } } @@ -217,10 +405,75 @@ func sumRows(results []backendResult, columns func(map[string]json.RawMessage) ( return out } +func newMergeGroup(raw json.RawMessage, row map[string]json.RawMessage, sh rowShape) *mergeGroup { + g := &mergeGroup{ + raw: raw, + row: row, + totals: map[string]float64{}, + extremes: map[string]json.RawMessage{}, + finish: sh.finish, + } + for _, c := range sh.aggs { + if c.op == opAdd { + if n, ok := numberOf(row[c.name]); ok { + g.totals[c.name] = n + } + continue + } + if v, ok := row[c.name]; ok && !isJSONNull(v) { + g.extremes[c.name] = v + } + } + return g +} + +func (g *mergeGroup) fold(row map[string]json.RawMessage, aggs []aggColumn) { + for _, c := range aggs { + if c.op == opAdd { + n, ok := numberOf(row[c.name]) + if !ok { + continue + } + if _, seen := g.totals[c.name]; !seen { + // First numeric value for a column the earlier row lacked. + g.totals[c.name] = 0 + } + g.totals[c.name] += n + g.merged = true + continue + } + v, ok := row[c.name] + if !ok || isJSONNull(v) { + continue + } + cur, seen := g.extremes[c.name] + if seen && !extremeWins(v, cur, c.op) { + continue + } + g.extremes[c.name] = v + g.merged = true + } +} + +// extremeWins reports whether candidate replaces the running min or max. Values +// are compared decoded, so min/max works on the text columns openvoxdb allows +// them on as well as on numbers. +func extremeWins(candidate, current json.RawMessage, op combineOp) bool { + var a, b any + if json.Unmarshal(candidate, &a) != nil || json.Unmarshal(current, &b) != nil { + return false + } + c := compareValues(a, b) + if op == opMin { + return c < 0 + } + return c > 0 +} + // encode renders a group back to JSON, reusing the first row's bytes when -// nothing was added to it. -func (g *sumGroup) encode() json.RawMessage { - if !g.merged { +// nothing changed it and no finish step has to rewrite it. +func (g *mergeGroup) encode() json.RawMessage { + if !g.merged && g.finish == nil { return g.raw } row := make(map[string]json.RawMessage, len(g.row)) @@ -230,6 +483,12 @@ func (g *sumGroup) encode() json.RawMessage { for col, total := range g.totals { row[col] = json.RawMessage(strconv.FormatFloat(total, 'f', -1, 64)) } + for col, v := range g.extremes { + row[col] = v + } + if g.finish != nil { + g.finish(row) + } raw, err := json.Marshal(row) if err != nil { return g.raw diff --git a/aggregate_test.go b/aggregate_test.go index f3b8979..3cbe8f4 100644 --- a/aggregate_test.go +++ b/aggregate_test.go @@ -2,8 +2,10 @@ package main import ( "encoding/json" + "net/url" "reflect" "slices" + "strings" "testing" ) @@ -29,29 +31,82 @@ func decodeRows(t *testing.T, raws []json.RawMessage) []map[string]any { return out } -func TestParseAggregate_ExtractWithGroupBy(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) - if spec == nil { - t.Fatal("expected an aggregate spec") +// mustAggregate parses a query that has to be a mergeable aggregate. +func mustAggregate(t *testing.T, q string) *aggregateSpec { + t.Helper() + spec, err := parseAggregate(q) + if err != nil { + t.Fatalf("parseAggregate(%s): %v", q, err) } + if spec == nil { + t.Fatalf("parseAggregate(%s) = nil, want an aggregate spec", q) + } + return spec +} + +func aggNames(aggs []aggColumn) []string { + out := make([]string, 0, len(aggs)) + for _, a := range aggs { + out = append(out, a.name) + } + return out +} + +func TestParseAggregate_ExtractWithGroupBy(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) if !slices.Equal(spec.keys, []string{"status"}) { t.Errorf("keys = %v, want [status]", spec.keys) } - if !slices.Equal(spec.sums, []string{"count"}) { - t.Errorf("sums = %v, want [count]", spec.sums) + if !slices.Equal(aggNames(spec.aggs), []string{"count"}) { + t.Errorf("aggs = %v, want [count]", aggNames(spec.aggs)) + } + if spec.aggs[0].op != opAdd { + t.Errorf("count op = %v, want opAdd", spec.aggs[0].op) } } func TestParseAggregate_GroupByAddsUnextractedField(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count","certname"]],["~","certname",".*"],["group_by","status"]]`) - if spec == nil { - t.Fatal("expected an aggregate spec") - } + spec := mustAggregate(t, `["extract",[["function","count","certname"]],["~","certname",".*"],["group_by","status"]]`) if !slices.Equal(spec.keys, []string{"status"}) { t.Errorf("keys = %v, want [status] from the group_by clause", spec.keys) } - if !slices.Equal(spec.sums, []string{"count"}) { - t.Errorf("sums = %v, want [count]", spec.sums) + if !slices.Equal(aggNames(spec.aggs), []string{"count"}) { + t.Errorf("aggs = %v, want [count]", aggNames(spec.aggs)) + } +} + +func TestParseAggregate_OpPerFunction(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","min","line"],["function","max","line"],["function","sum","line"],["function","count"]]]`) + want := []aggColumn{ + {name: "min", op: opMin}, + {name: "max", op: opMax}, + {name: "sum", op: opAdd}, + {name: "count", op: opAdd}, + } + if !reflect.DeepEqual(spec.aggs, want) { + t.Errorf("aggs = %v, want %v", spec.aggs, want) + } + if len(spec.keys) != 0 { + t.Errorf("keys = %v, want none", spec.keys) + } +} + +// to_string and jsonb_typeof are row functions, so they identify a row rather +// than being folded into it. +func TestParseAggregate_RowFunctionsAreKeys(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"],["function","jsonb_typeof","value"],["function","count"]]]`) + if !slices.Equal(spec.keys, []string{"to_string", "jsonb_typeof"}) { + t.Errorf("keys = %v, want [to_string jsonb_typeof]", spec.keys) + } + if !slices.Equal(aggNames(spec.aggs), []string{"count"}) { + t.Errorf("aggs = %v, want [count]", aggNames(spec.aggs)) + } +} + +func TestParseAggregate_GroupByOnAFunctionColumn(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDD"]]]`) + if !slices.Equal(spec.keys, []string{"to_string"}) { + t.Errorf("keys = %v, want [to_string] once", spec.keys) } } @@ -64,18 +119,33 @@ func TestParseAggregate_NoFunctionIsNotAggregate(t *testing.T) { `["extract"]`, `{"not":"an array"}`, } { - if spec := parseAggregate(q); spec != nil { + spec, err := parseAggregate(q) + if err != nil { + t.Errorf("parseAggregate(%q): unexpected error %v", q, err) + } + if spec != nil { t.Errorf("parseAggregate(%q) = %+v, want nil", q, spec) } } } -func TestSumRows_SharedKeysAreAdded(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) - merged := sumRows([]backendResult{ +// A function pdbmux has no combiner for is refused, never folded on a guess. +func TestParseAggregate_UnknownFunctionIsRefused(t *testing.T) { + spec, err := parseAggregate(`["extract",[["function","stddev","line"]]]`) + if err == nil { + t.Fatalf("parseAggregate = %+v, want an error", spec) + } + if !strings.Contains(err.Error(), "stddev") { + t.Errorf("error %q does not name the function", err) + } +} + +func TestCombineRows_SharedKeysAreAdded(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) + merged := combineRows([]backendResult{ {name: "a", records: rows(`{"count":3,"status":"changed"}`, `{"count":1,"status":"failed"}`)}, {name: "b", records: rows(`{"count":4,"status":"changed"}`, `{"count":2,"status":"failed"}`)}, - }, spec.columns) + }, spec.shape) want := []map[string]any{ {"count": float64(7), "status": "changed"}, @@ -86,12 +156,308 @@ func TestSumRows_SharedKeysAreAdded(t *testing.T) { } } -func TestSumRows_DisjointKeysAreKept(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) - merged := sumRows([]backendResult{ +func TestCombineRows_SumIsAdded(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","sum","line"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"sum":30}`)}, + {name: "b", records: rows(`{"sum":12}`)}, + }, spec.shape) + + want := []map[string]any{{"sum": float64(42)}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +// The bug this replaces: max folded by addition returned the sum of the +// backends' maxima, a number no backend ever held. +func TestCombineRows_MaxIsNotASum(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","max","line"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"max":20}`)}, + {name: "b", records: rows(`{"max":50}`)}, + }, spec.shape) + + got := decodeRows(t, merged) + want := []map[string]any{{"max": float64(50)}} + if !reflect.DeepEqual(got, want) { + t.Fatalf("merged = %v, want %v", got, want) + } + if got[0]["max"] == float64(70) { + t.Error("max was summed") + } +} + +func TestCombineRows_MinIsTheSmallest(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","min","line"]]]`) + for _, tc := range []struct { + name string + a, b string + want any + }{ + {"later backend wins", `{"min":20}`, `{"min":5}`, float64(5)}, + {"earlier backend wins", `{"min":5}`, `{"min":20}`, float64(5)}, + {"negative values", `{"min":-1}`, `{"min":-9}`, float64(-9)}, + } { + t.Run(tc.name, func(t *testing.T) { + merged := combineRows([]backendResult{ + {name: "a", records: rows(tc.a)}, + {name: "b", records: rows(tc.b)}, + }, spec.shape) + got := decodeRows(t, merged) + if len(got) != 1 || got[0]["min"] != tc.want { + t.Errorf("merged = %v, want min %v", got, tc.want) + } + }) + } +} + +// openvoxdb allows min/max on text columns, where the wrong answer used to be +// whichever backend answered first. +func TestCombineRows_MinMaxOnTextColumns(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","min","name"],["function","max","name"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"min":"kernel","max":"role"}`)}, + {name: "b", records: rows(`{"min":"extra_b","max":"uptime"}`)}, + }, spec.shape) + + want := []map[string]any{{"min": "extra_b", "max": "uptime"}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +// A backend that matched nothing answers null, which is not an extreme. +func TestCombineRows_MinMaxIgnoreNulls(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","max","line"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"max":null}`)}, + {name: "b", records: rows(`{"max":7}`)}, + }, spec.shape) + + want := []map[string]any{{"max": float64(7)}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +func TestCombineRows_MinMaxWithGroupBy(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","min","line"],["function","max","line"],"type"],["group_by","type"]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"min":10,"max":20,"type":"File"}`, `{"min":1,"max":2,"type":"Stage"}`)}, + {name: "b", records: rows(`{"min":30,"max":50,"type":"File"}`)}, + }, spec.shape) + + want := []map[string]any{ + {"min": float64(10), "max": float64(50), "type": "File"}, + {"min": float64(1), "max": float64(2), "type": "Stage"}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +// avg is rewritten upstream into the sum and count of the same expression. +func TestParseAggregate_AvgIsRewrittenToSumAndCount(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","avg","line"]],["=","type","File"]]`) + if !spec.avg { + t.Fatal("spec does not record the avg rewrite") + } + const want = `["extract",[["function","sum","line"],["function","count","line"]],["=","type","File"]]` + if spec.query != want { + t.Errorf("upstream query = %s, want %s", spec.query, want) + } + if !slices.Equal(aggNames(spec.aggs), []string{"sum", "count"}) { + t.Errorf("aggs = %v, want the sum and count helpers", aggNames(spec.aggs)) + } +} + +func TestParseAggregate_AvgKeepsCompanionColumnsAndGroupBy(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","avg","line"],"type"],["group_by","type"]]`) + const want = `["extract",[["function","sum","line"],"type",["function","count","line"]],["group_by","type"]]` + if spec.query != want { + t.Errorf("upstream query = %s, want %s", spec.query, want) + } + if !slices.Equal(spec.keys, []string{"type"}) { + t.Errorf("keys = %v, want [type]", spec.keys) + } +} + +func TestCombineRows_AvgIsWeightedByCount(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","avg","line"]]]`) + // Backend a averages 10 over 1 row, b averages 20 over 3: the true average + // is 70/4, not the 15 an average of averages gives. + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"sum":10,"count":1}`)}, + {name: "b", records: rows(`{"sum":60,"count":3}`)}, + }, spec.shape) + + want := []map[string]any{{"avg": float64(17.5)}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +// A single backend still gets the avg key back, not the helper columns. +func TestCombineRows_AvgFromOneBackend(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","avg","line"],"type"],["group_by","type"]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"sum":30,"count":4,"type":"File"}`)}, + {name: "b", records: nil}, + }, spec.shape) + + want := []map[string]any{{"avg": float64(7.5), "type": "File"}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +func TestCombineRows_AvgOverNoRowsIsNull(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","avg","line"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"sum":null,"count":0}`)}, + {name: "b", records: rows(`{"sum":null,"count":0}`)}, + }, spec.shape) + + want := []map[string]any{{"avg": nil}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +func TestCombineRows_AvgWithGroupBy(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","avg","line"],"type"],["group_by","type"]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"sum":10,"count":1,"type":"File"}`, `{"sum":8,"count":2,"type":"Stage"}`)}, + {name: "b", records: rows(`{"sum":60,"count":3,"type":"File"}`)}, + }, spec.shape) + + want := []map[string]any{ + {"avg": float64(17.5), "type": "File"}, + {"avg": float64(4), "type": "Stage"}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +// The helper columns take the response keys openvoxdb would give the client's +// own sum or count, so the pair cannot be served together. +func TestParseAggregate_AvgWithSumOrCountIsRefused(t *testing.T) { + for _, q := range []string{ + `["extract",[["function","avg","line"],["function","sum","line"]]]`, + `["extract",[["function","avg","line"],["function","count"]]]`, + } { + spec, err := parseAggregate(q) + if err == nil { + t.Errorf("parseAggregate(%s) = %+v, want a refusal", q, spec) + continue + } + if !strings.Contains(err.Error(), "avg") { + t.Errorf("error %q does not name avg", err) + } + } +} + +func TestParseAggregate_AvgWithoutAColumnIsRefused(t *testing.T) { + if spec, err := parseAggregate(`["extract",[["function","avg"]]]`); err == nil { + t.Errorf("parseAggregate = %+v, want a refusal", spec) + } +} + +func TestParseAggregate_TwoAvgColumnsAreRefused(t *testing.T) { + if spec, err := parseAggregate(`["extract",[["function","avg","line"],["function","avg","value"]]]`); err == nil { + t.Errorf("parseAggregate = %+v, want a refusal", spec) + } +} + +// Every extract function names its column after itself, so a second one of the +// same name clashes whatever the function is — not only for avg. +func TestParseAggregate_RepeatedFunctionNameIsRefused(t *testing.T) { + for _, fn := range []string{"count", "sum", "min", "max", "to_string", "jsonb_typeof", "avg"} { + q := `["extract",[["function","` + fn + `","line"],["function","` + fn + `","type"]],["group_by","line","type"]]` + spec, err := parseAggregate(q) + if err == nil { + t.Errorf("parseAggregate(%s) = %+v, want a refusal", q, spec) + continue + } + if !strings.Contains(err.Error(), fn) { + t.Errorf("error %q does not name the clashing column %q", err, fn) + } + } +} + +// The same clash the other way round: a plain field takes the response key a +// later function column would name. +func TestParseAggregate_FieldClashingWithAFunctionIsRefused(t *testing.T) { + const q = `["extract",["count",["function","count","certname"]],["group_by","count"]]` + spec, err := parseAggregate(q) + if err == nil { + t.Fatalf("parseAggregate(%s) = %+v, want a refusal", q, spec) + } + if !strings.Contains(err.Error(), "count") { + t.Errorf("error %q does not name the clashing column", err) + } +} + +// The ordering the sibling case does not cover: the function column comes first +// and a later plain field takes the response key it already named. +func TestParseAggregate_FunctionClashingWithALaterFieldIsRefused(t *testing.T) { + const q = `["extract",[["function","count","certname"],"count"],["=","certname","h1"]]` + spec, err := parseAggregate(q) + if err == nil { + t.Fatalf("parseAggregate(%s) = %+v, want a refusal", q, spec) + } + if !strings.Contains(err.Error(), "count") { + t.Errorf("error %q does not name the clashing column", err) + } +} + +// A group_by naming a folded column would make it a grouping key and an +// aggregate at once, so it is refused rather than sent upstream to fail. +func TestParseAggregate_GroupByOnAnAggregateColumnIsRefused(t *testing.T) { + for _, fn := range []string{"count", "sum", "min", "max", "avg"} { + q := `["extract",[["function","` + fn + `","line"]],["group_by","` + fn + `"]]` + spec, err := parseAggregate(q) + if err == nil { + t.Errorf("parseAggregate(%s) = %+v, want a refusal", q, spec) + continue + } + if !strings.Contains(err.Error(), fn) { + t.Errorf("error %q does not name the clashing column %q", err, fn) + } + } +} + +// The refusal is limited to the folded columns: grouping on a plain field or on +// a row function's own column stays legitimate. +func TestParseAggregate_GroupByOnANonAggregateColumnIsKept(t *testing.T) { + for _, q := range []string{ + `["extract",[["function","count","certname"],"status"],["group_by","status"]]`, + `["extract",[["function","to_string","producer_timestamp","FMDD"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDD"]]]`, + `["extract",[["function","avg","line"],"type"],["group_by","type"]]`, + } { + if spec, err := parseAggregate(q); err != nil || spec == nil { + t.Errorf("parseAggregate(%s) = %+v, %v, want an accepted spec", q, spec, err) + } + } +} + +// A repeated plain field projects the same value twice, so it is no clash: the +// duplicate carries nothing the grouping key has not already got. +func TestParseAggregate_RepeatedPlainFieldIsKept(t *testing.T) { + spec := mustAggregate(t, `["extract",["type","type",["function","count","certname"]],["group_by","type"]]`) + if !slices.Equal(spec.keys, []string{"type"}) { + t.Errorf("keys = %v, want [type]", spec.keys) + } +} + +func TestCombineRows_DisjointKeysAreKept(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) + merged := combineRows([]backendResult{ {name: "a", records: rows(`{"count":3,"status":"changed"}`)}, {name: "b", records: rows(`{"count":2,"status":"skipped"}`)}, - }, spec.columns) + }, spec.shape) want := []map[string]any{ {"count": float64(3), "status": "changed"}, @@ -102,25 +468,25 @@ func TestSumRows_DisjointKeysAreKept(t *testing.T) { } } -func TestSumRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) +func TestCombineRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) const raw = `{"count":3,"status":"changed","extra":{"kept":true}}` - merged := sumRows([]backendResult{ + merged := combineRows([]backendResult{ {name: "a", records: rows(raw)}, {name: "b", records: nil}, - }, spec.columns) + }, spec.shape) if len(merged) != 1 || string(merged[0]) != raw { t.Errorf("merged = %s, want the row verbatim %s", merged, raw) } } -func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) - merged := sumRows([]backendResult{ +func TestCombineRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) + merged := combineRows([]backendResult{ {name: "a", records: rows(`{"count":5,"status":"changed"}`)}, {name: "b", records: rows(`{"count":null,"status":"changed"}`)}, - }, spec.columns) + }, spec.shape) want := []map[string]any{{"count": float64(5), "status": "changed"}} if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { @@ -128,12 +494,12 @@ func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) { } } -func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) - merged := sumRows([]backendResult{ +func TestCombineRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) + merged := combineRows([]backendResult{ {name: "a", records: rows(`{"status":"changed"}`)}, {name: "b", records: rows(`{"count":6,"status":"changed"}`)}, - }, spec.columns) + }, spec.shape) want := []map[string]any{{"count": float64(6), "status": "changed"}} if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { @@ -141,29 +507,26 @@ func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) { } } -func TestSumRows_NonObjectRowsPassThrough(t *testing.T) { - spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) - merged := sumRows([]backendResult{ +func TestCombineRows_NonObjectRowsPassThrough(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) + merged := combineRows([]backendResult{ {name: "a", records: rows(`"surprise"`)}, {name: "b", records: rows(`{"count":1,"status":"changed"}`)}, - }, spec.columns) + }, spec.shape) if len(merged) != 2 || string(merged[0]) != `"surprise"` { t.Fatalf("merged = %s, want the non-object row kept as-is", merged) } } -func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) { +func TestCombineRows_NoKeyColumnsCollapseToOneRow(t *testing.T) { // ["extract",[["function","count"]],...] is a whole-estate count: one row // per backend, and the merged answer is their sum. - spec := parseAggregate(`["extract",[["function","count"]],["=","certname","h1"]]`) - if spec == nil { - t.Fatal("expected an aggregate spec") - } - merged := sumRows([]backendResult{ + spec := mustAggregate(t, `["extract",[["function","count"]],["=","certname","h1"]]`) + merged := combineRows([]backendResult{ {name: "a", records: rows(`{"count":10}`)}, {name: "b", records: rows(`{"count":32}`)}, - }, spec.columns) + }, spec.shape) want := []map[string]any{{"count": float64(42)}} if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { @@ -171,23 +534,96 @@ func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) { } } -func TestInferredColumns_SplitsCountsFromIdentity(t *testing.T) { +// to_string compiles to a scalar expression, so openvoxdb answers one row per +// input row; with nothing to fold, every backend's rows are kept rather than +// collapsing into a single empty-key bucket. +func TestCombineRows_ToStringAloneDoesNotCollapse(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"to_string":"01"}`, `{"to_string":"02"}`)}, + {name: "b", records: rows(`{"to_string":"02"}`, `{"to_string":"03"}`)}, + }, spec.shape) + + want := []map[string]any{ + {"to_string": "01"}, {"to_string": "02"}, + {"to_string": "02"}, {"to_string": "03"}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want every row kept %v", got, want) + } +} + +func TestCombineRows_ToStringGroupsItsCompanionCount(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","to_string","producer_timestamp","FMDD"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDD"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"to_string":"01","count":2}`, `{"to_string":"02","count":3}`)}, + {name: "b", records: rows(`{"to_string":"02","count":4}`)}, + }, spec.shape) + + want := []map[string]any{ + {"to_string": "01", "count": float64(2)}, + {"to_string": "02", "count": float64(7)}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +func TestCombineRows_JsonbTypeofGroups(t *testing.T) { + spec := mustAggregate(t, `["extract",[["function","jsonb_typeof","value"],["function","count"]],["group_by",["function","jsonb_typeof","value"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"jsonb_typeof":"string","count":5}`)}, + {name: "b", records: rows(`{"jsonb_typeof":"string","count":6}`, `{"jsonb_typeof":"number","count":1}`)}, + }, spec.shape) + + want := []map[string]any{ + {"jsonb_typeof": "string", "count": float64(11)}, + {"jsonb_typeof": "number", "count": float64(1)}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +// A plain projected column keeps rows apart even when the query has no group_by. +func TestCombineRows_PlainColumnsAreGroupingKeys(t *testing.T) { + spec := mustAggregate(t, `["extract",["type","title",["function","count"]]]`) + merged := combineRows([]backendResult{ + {name: "a", records: rows(`{"type":"File","title":"/tmp/a","count":1}`)}, + {name: "b", records: rows(`{"type":"File","title":"/tmp/b","count":1}`, `{"type":"File","title":"/tmp/a","count":2}`)}, + }, spec.shape) + + want := []map[string]any{ + {"type": "File", "title": "/tmp/a", "count": float64(3)}, + {"type": "File", "title": "/tmp/b", "count": float64(1)}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +func TestInferredShape_SplitsCountsFromIdentity(t *testing.T) { var row map[string]json.RawMessage if err := json.Unmarshal([]byte(`{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"skips":null}`), &row); err != nil { t.Fatal(err) } - keys, sums := inferredColumns(row) - if !slices.Equal(keys, []string{"subject", "subject_type"}) { - t.Errorf("keys = %v, want [subject subject_type]", keys) + sh := inferredShape(row) + if !slices.Equal(sh.keys, []string{"subject", "subject_type"}) { + t.Errorf("keys = %v, want [subject subject_type]", sh.keys) } // A null count is an empty aggregate, not part of the row's identity. - if !slices.Equal(sums, []string{"failures", "skips", "successes"}) { - t.Errorf("sums = %v, want [failures skips successes]", sums) + if !slices.Equal(aggNames(sh.aggs), []string{"failures", "skips", "successes"}) { + t.Errorf("aggs = %v, want [failures skips successes]", aggNames(sh.aggs)) + } + for _, a := range sh.aggs { + if a.op != opAdd { + t.Errorf("%s op = %v, want opAdd", a.name, a.op) + } } } -func TestSumRows_EventCountsPerSubject(t *testing.T) { - merged := sumRows([]backendResult{ +func TestCombineRows_EventCountsPerSubject(t *testing.T) { + merged := combineRows([]backendResult{ {name: "a", records: rows( `{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"noops":0,"skips":0}`, `{"subject_type":"certname","subject":{"title":"h2"},"failures":0,"successes":5,"noops":0,"skips":0}`, @@ -195,7 +631,7 @@ func TestSumRows_EventCountsPerSubject(t *testing.T) { {name: "b", records: rows( `{"subject_type":"certname","subject":{"title":"h1"},"failures":3,"successes":4,"noops":1,"skips":0}`, )}, - }, inferredColumns) + }, inferredShape) got := decodeRows(t, merged) want := []map[string]any{ @@ -208,3 +644,22 @@ func TestSumRows_EventCountsPerSubject(t *testing.T) { t.Errorf("merged = %v, want %v", got, want) } } + +func TestDropOrderBy(t *testing.T) { + for _, tc := range []struct { + name, in, want string + }{ + {"removes the named field", `[{"field":"avg","order":"desc"}]`, ``}, + {"keeps the others", `[{"field":"avg"},{"field":"type"}]`, `[{"field":"type"}]`}, + {"leaves an unrelated order_by alone", `[{"field":"type"}]`, `[{"field":"type"}]`}, + {"drops an unparseable order_by", `not json`, ``}, + } { + t.Run(tc.name, func(t *testing.T) { + params := url.Values{"order_by": {tc.in}} + dropOrderBy(params, "avg") + if got := params.Get("order_by"); got != tc.want { + t.Errorf("order_by = %q, want %q", got, tc.want) + } + }) + } +} diff --git a/e2e_fixture_test.go b/e2e_fixture_test.go index 0375424..20e9fb3 100644 --- a/e2e_fixture_test.go +++ b/e2e_fixture_test.go @@ -4,6 +4,7 @@ package main import ( "context" + "fmt" "time" ) @@ -48,28 +49,32 @@ type nodeFixture struct { facts map[string]any // reportEnd is the report's end_time, and so the node's report_timestamp. reportEnd string + // lines are the catalog File resources' line numbers, one resource each. They + // give /resources a numeric column whose per-backend minimum, maximum, sum and + // row count all differ, so no combiner can be mistaken for another. + lines []int } // Backend A: 2 nodes, 7 facts. Backend B: 3 nodes, 10 facts. The counts are // deliberately unequal so a summed aggregate cannot be mistaken for either // backend's own number. var fixtureA = []nodeFixture{ - {certname: nodeAlpha, reportEnd: tsAlpha, facts: map[string]any{ + {certname: nodeAlpha, reportEnd: tsAlpha, lines: []int{10, 12, 14}, facts: map[string]any{ "osfamily": "RedHat", "kernel": "Linux", "role": "web", "only_a": "yes", }}, - {certname: nodeShared, reportEnd: tsSharedOnA, facts: map[string]any{ + {certname: nodeShared, reportEnd: tsSharedOnA, lines: []int{20}, facts: map[string]any{ "osfamily": "RedHat", "kernel": "Linux", "owner": backendAName, }}, } var fixtureB = []nodeFixture{ - {certname: nodeBeta, reportEnd: tsBeta, facts: map[string]any{ + {certname: nodeBeta, reportEnd: tsBeta, lines: []int{30}, facts: map[string]any{ "osfamily": "Debian", "kernel": "Linux", "role": "db", "only_b": "yes", "extra_b": "1", }}, - {certname: nodeGamma, reportEnd: tsGamma, facts: map[string]any{ + {certname: nodeGamma, reportEnd: tsGamma, lines: []int{40}, facts: map[string]any{ "osfamily": "Debian", "kernel": "Linux", }}, - {certname: nodeShared, reportEnd: tsSharedOnB, facts: map[string]any{ + {certname: nodeShared, reportEnd: tsSharedOnB, lines: []int{50}, facts: map[string]any{ "osfamily": "Debian", "kernel": "Linux", "owner": backendBName, }}, } @@ -84,7 +89,7 @@ func loadFixtures(ctx context.Context, t fatalf, a, b *backend) { } // A deactivated node proves the merged view reflects each backend's own // filtering rather than a raw union of everything ever stored. - loadNode(ctx, t, a, nodeFixture{certname: nodeGone, reportEnd: tsGone, facts: map[string]any{"osfamily": "RedHat"}}) + loadNode(ctx, t, a, nodeFixture{certname: nodeGone, reportEnd: tsGone, lines: []int{100}, facts: map[string]any{"osfamily": "RedHat"}}) a.submit(ctx, t, cmdDeactivateNode, verDeactivateNode, nodeGone, tsDeactivation, map[string]any{ "certname": nodeGone, "producer_timestamp": tsDeactivation, @@ -105,7 +110,26 @@ func loadNode(ctx context.Context, t fatalf, b *backend, n nodeFixture) { // a catalog_environment and populate /resources, which the aggregate assertions // and Puppetboard's index both read. func catalogPayload(n nodeFixture) map[string]any { - title := "/tmp/" + n.certname + resources := []any{ + map[string]any{ + "type": "Stage", "title": "main", "aliases": []string{}, "exported": false, + "file": nil, "line": nil, "tags": []string{"stage"}, "parameters": map[string]any{}, + }, + } + edges := []any{} + for i, line := range n.lines { + title := fileTitle(n.certname, i) + resources = append(resources, map[string]any{ + "type": "File", "title": title, "aliases": []string{}, "exported": false, + "file": "/etc/puppetlabs/code/site.pp", "line": line, "tags": []string{"file"}, + "parameters": map[string]any{"ensure": "present"}, + }) + edges = append(edges, map[string]any{ + "source": map[string]any{"type": "Stage", "title": "main"}, + "target": map[string]any{"type": "File", "title": title}, + "relationship": "contains", + }) + } return map[string]any{ "certname": n.certname, "version": "1", @@ -115,27 +139,21 @@ func catalogPayload(n nodeFixture) map[string]any { "code_id": nil, "producer_timestamp": n.reportEnd, "producer": "pdbmux-e2e", - "edges": []any{ - map[string]any{ - "source": map[string]any{"type": "Stage", "title": "main"}, - "target": map[string]any{"type": "File", "title": title}, - "relationship": "contains", - }, - }, - "resources": []any{ - map[string]any{ - "type": "Stage", "title": "main", "aliases": []string{}, "exported": false, - "file": nil, "line": nil, "tags": []string{"stage"}, "parameters": map[string]any{}, - }, - map[string]any{ - "type": "File", "title": title, "aliases": []string{}, "exported": false, - "file": "/etc/puppetlabs/code/site.pp", "line": 1, "tags": []string{"file"}, - "parameters": map[string]any{"ensure": "present"}, - }, - }, + "edges": edges, + "resources": resources, } } +// The first File resource keeps the plain /tmp/ title the report's +// event names; the rest sort after every other fixture title, so one backend +// holds the estate's largest resource title and the other its smallest. +func fileTitle(certname string, i int) string { + if i == 0 { + return "/tmp/" + certname + } + return fmt.Sprintf("/tmp/zz-%s-%d", certname, i) +} + // factsPayload is the "replace facts" v5 wire format: certname, environment, // producer, producer_timestamp and the fact values. func factsPayload(n nodeFixture) map[string]any { diff --git a/e2e_query_test.go b/e2e_query_test.go index dc59eb3..f95606e 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -6,10 +6,12 @@ import ( "context" "encoding/json" "fmt" + "math" "net/http" "net/url" "reflect" "sort" + "strings" "testing" ) @@ -361,6 +363,237 @@ func TestResourcesAggregatesAreSummed(t *testing.T) { } } +// oneAggregateRow reads the single row a whole-estate aggregate returns. +func oneAggregateRow(t *testing.T, rows []map[string]any) map[string]any { + t.Helper() + if len(rows) != 1 { + t.Fatalf("want exactly one aggregate row, got %d: %v", len(rows), rows) + } + return rows[0] +} + +// aggregateNumber reads one numeric aggregate column, which openvoxdb nulls when +// the backend matched nothing. +func aggregateNumber(t *testing.T, row map[string]any, column string) float64 { + t.Helper() + n, ok := row[column].(float64) + if !ok { + t.Fatalf("aggregate row has no numeric %s: %v", column, row) + } + return n +} + +// backendAggregate runs a whole-estate aggregate against one backend directly. +func backendAggregate(ctx context.Context, t *testing.T, b *backend, path, q string) map[string]any { + t.Helper() + return oneAggregateRow(t, b.query(ctx, t, path, query(q))) +} + +// min and max are extremes, not sums: the merged answer has to be the smallest +// and largest the backends reported, both of which the fixture makes distinct. +func TestResourcesMinMaxAreExtremes(t *testing.T) { + ctx := context.Background() + const q = `["extract",[["function","min","line"],["function","max","line"]]]` + + a := backendAggregate(ctx, t, h.a, resourcesPath, q) + b := backendAggregate(ctx, t, h.b, resourcesPath, q) + minA, maxA := aggregateNumber(t, a, "min"), aggregateNumber(t, a, "max") + minB, maxB := aggregateNumber(t, b, "min"), aggregateNumber(t, b, "max") + if minA == minB || maxA == maxB { + t.Fatalf("the fixture gives both backends the same extremes (min %v/%v, max %v/%v)", minA, minB, maxA, maxB) + } + + got := oneAggregateRow(t, get(t, resourcesPath, query(q)).rows(t)) + if wantMin := math.Min(minA, minB); aggregateNumber(t, got, "min") != wantMin { + t.Errorf("/resources min = %v, want %v (%s=%v, %s=%v)", got["min"], wantMin, h.a.name, minA, h.b.name, minB) + } + wantMax := math.Max(maxA, maxB) + if gotMax := aggregateNumber(t, got, "max"); gotMax != wantMax { + t.Errorf("/resources max = %v, want %v (%s=%v, %s=%v)", gotMax, wantMax, h.a.name, maxA, h.b.name, maxB) + } else if gotMax == maxA+maxB { + t.Errorf("/resources max = %v, which is the sum of the backends' maxima", gotMax) + } +} + +// openvoxdb allows min/max on text columns, where the merged answer used to be +// whichever backend happened to answer first. The fixture puts the estate's +// smallest resource title in one backend and its largest in the other, so a +// first-wins merge gets one of the two wrong whichever way round it reads them. +func TestResourcesMinMaxOnATextColumn(t *testing.T) { + ctx := context.Background() + const q = `["extract",[["function","min","title"],["function","max","title"]],["=","type","File"]]` + + a := backendAggregate(ctx, t, h.a, resourcesPath, q) + b := backendAggregate(ctx, t, h.b, resourcesPath, q) + minA, minB := aggregateString(t, a, "min"), aggregateString(t, b, "min") + maxA, maxB := aggregateString(t, a, "max"), aggregateString(t, b, "max") + if minA == minB || maxA == maxB { + t.Fatalf("the fixture gives both backends the same extremes (min %q/%q, max %q/%q)", minA, minB, maxA, maxB) + } + + got := oneAggregateRow(t, get(t, resourcesPath, query(q)).rows(t)) + if want := min(minA, minB); got["min"] != want { + t.Errorf("/resources min(title) = %v, want %q (%s=%q, %s=%q)", got["min"], want, h.a.name, minA, h.b.name, minB) + } + if want := max(maxA, maxB); got["max"] != want { + t.Errorf("/resources max(title) = %v, want %q (%s=%q, %s=%q)", got["max"], want, h.a.name, maxA, h.b.name, maxB) + } +} + +func aggregateString(t *testing.T, row map[string]any, column string) string { + t.Helper() + s, ok := row[column].(string) + if !ok { + t.Fatalf("aggregate row has no string %s: %v", column, row) + } + return s +} + +// avg is decomposed into an upstream sum and count, so the merged answer is the +// true weighted average of the estate rather than an average of averages. +func TestResourcesAvgIsWeighted(t *testing.T) { + ctx := context.Background() + const parts = `["extract",[["function","sum","line"],["function","count","line"]]]` + + a := backendAggregate(ctx, t, h.a, resourcesPath, parts) + b := backendAggregate(ctx, t, h.b, resourcesPath, parts) + sumA, countA := aggregateNumber(t, a, "sum"), aggregateNumber(t, a, "count") + sumB, countB := aggregateNumber(t, b, "sum"), aggregateNumber(t, b, "count") + if countA == countB { + t.Fatalf("the fixture gives both backends %v rows, so a weighted average is indistinguishable from a plain one", countA) + } + want := (sumA + sumB) / (countA + countB) + + const q = `["extract",[["function","avg","line"]]]` + got := aggregateNumber(t, oneAggregateRow(t, get(t, resourcesPath, query(q)).rows(t)), "avg") + if math.Abs(got-want) > 1e-9 { + t.Fatalf("/resources avg(line) = %v, want %v ((%v+%v)/(%v+%v))", got, want, sumA, sumB, countA, countB) + } + + avgA := aggregateNumber(t, backendAggregate(ctx, t, h.a, resourcesPath, q), "avg") + avgB := aggregateNumber(t, backendAggregate(ctx, t, h.b, resourcesPath, q), "avg") + if math.Abs(got-(avgA+avgB)) < 1e-9 { + t.Errorf("/resources avg(line) = %v, which is the sum of the backends' averages", got) + } + if math.Abs(got-(avgA+avgB)/2) < 1e-9 { + t.Errorf("/resources avg(line) = %v, which is an unweighted average of averages", got) + } +} + +func TestResourcesAvgWithGroupBy(t *testing.T) { + ctx := context.Background() + const parts = `["extract",[["function","sum","line"],["function","count","line"],"type"],["group_by","type"]]` + const q = `["extract",[["function","avg","line"],"type"],["group_by","type"]]` + + sums, counts := map[string]float64{}, map[string]float64{} + for _, b := range []*backend{h.a, h.b} { + for _, row := range b.query(ctx, t, resourcesPath, query(parts)) { + typ, ok := row["type"].(string) + if !ok { + t.Fatalf("grouped row has no type: %v", row) + } + if n, ok := row["sum"].(float64); ok { + sums[typ] += n + counts[typ] += aggregateNumber(t, row, "count") + } + } + } + if len(sums) == 0 { + t.Fatal("no grouped rows carried a sum, so the weighted average is untested") + } + + for _, row := range get(t, resourcesPath, query(q)).rows(t) { + typ, ok := row["type"].(string) + if !ok { + t.Fatalf("grouped row has no type: %v", row) + } + if counts[typ] == 0 { + if row["avg"] != nil { + t.Errorf("avg(line) for %s = %v, want null over no rows", typ, row["avg"]) + } + continue + } + want := sums[typ] / counts[typ] + if got := aggregateNumber(t, row, "avg"); math.Abs(got-want) > 1e-9 { + t.Errorf("avg(line) for %s = %v, want %v", typ, got, want) + } + } +} + +// A query pdbmux cannot merge is refused outright rather than answered with a +// plausible wrong number. +func TestUnmergeableAggregateIsRefused(t *testing.T) { + resp := rawGet(t, resourcesPath, query(`["extract",[["function","avg","line"],["function","count"]]]`)) + if resp.status != http.StatusBadRequest { + t.Fatalf("status %d, want 400: %s", resp.status, resp.body) + } + if !strings.Contains(string(resp.body), "avg") { + t.Errorf("refusal %q does not name the limitation", resp.body) + } +} + +// openvoxdb names every extract column after its function and aliases a repeat +// as "_2", which pdbmux's spec never learns about: it would be neither a +// grouping key nor a folded aggregate, so the first backend's value would +// freeze into the merged row. The backend answers such a query, which is +// exactly why pdbmux has to refuse it. +func TestRepeatedFunctionColumnIsRefused(t *testing.T) { + const q = `["extract",[["function","count","certname"],["function","count","catalog_environment"]]]` + + resp := rawGet(t, nodesPath, query(q)) + if resp.status != http.StatusBadRequest { + t.Fatalf("status %d, want 400: %s", resp.status, resp.body) + } + if !strings.Contains(string(resp.body), "count") { + t.Errorf("refusal %q does not name the clashing column", resp.body) + } + + rows := h.a.query(context.Background(), t, nodesPath, query(q)) + if len(rows) != 1 { + t.Fatalf("backend %s returned %d rows for the repeated projection, want 1", h.a.name, len(rows)) + } + if _, ok := rows[0]["count_2"]; !ok { + t.Errorf("backend %s row = %v, want the aliased count_2 column the refusal exists for", h.a.name, rows[0]) + } +} + +// to_string is a scalar expression, so it yields one row per record: alone it +// must not collapse the estate into a single row, and with a companion count it +// groups. +func TestReportsToString(t *testing.T) { + ctx := context.Background() + const bare = `["extract",[["function","to_string","producer_timestamp","FMDAY"]]]` + + wantRows := len(h.a.query(ctx, t, reportsPath, query(bare))) + len(h.b.query(ctx, t, reportsPath, query(bare))) + if wantRows < 2 { + t.Fatalf("the fixture yields %d to_string rows, so a collapse would be invisible", wantRows) + } + rows := get(t, reportsPath, query(bare)).rows(t) + if len(rows) != wantRows { + t.Errorf("/reports to_string returned %d rows, want %d (every backend's rows)", len(rows), wantRows) + } + for _, row := range rows { + if _, ok := row["to_string"].(string); !ok { + t.Fatalf("to_string row has no string column: %v", row) + } + } + + const grouped = `["extract",[["function","to_string","producer_timestamp","FMDAY"],["function","count"]],["group_by",["function","to_string","producer_timestamp","FMDAY"]]]` + want := map[string]float64{} + for _, b := range []*backend{h.a, h.b} { + for _, row := range b.query(ctx, t, reportsPath, query(grouped)) { + want[row["to_string"].(string)] += aggregateNumber(t, row, "count") + } + } + got := map[string]float64{} + for _, row := range get(t, reportsPath, query(grouped)).rows(t) { + got[row["to_string"].(string)] += aggregateNumber(t, row, "count") + } + if !reflect.DeepEqual(got, want) { + t.Errorf("grouped /reports to_string counts = %v, want %v", got, want) + } +} + // A fact count row carries no certname, so the per-certname fact merge would // keep one backend's rows and drop the other's; only adding the numbers is right. func TestFactsAggregatesAreSummed(t *testing.T) { diff --git a/reports.go b/reports.go index aeb692f..e9ee47d 100644 --- a/reports.go +++ b/reports.go @@ -150,6 +150,41 @@ func valueRank(v any) int { } } +// dropOrderBy removes one field from an upstream order_by, for a column the +// rewritten query no longer projects. An unparseable or emptied order_by is +// dropped entirely; pdbmux re-sorts the merged rows on the client's own order. +func dropOrderBy(params url.Values, field string) { + raw := params.Get("order_by") + if strings.TrimSpace(raw) == "" { + return + } + var entries []map[string]any + if json.Unmarshal([]byte(raw), &entries) != nil { + params.Del("order_by") + return + } + kept := make([]map[string]any, 0, len(entries)) + for _, e := range entries { + if f, ok := e["field"].(string); ok && f == field { + continue + } + kept = append(kept, e) + } + if len(kept) == len(entries) { + return + } + if len(kept) == 0 { + params.Del("order_by") + return + } + encoded, err := json.Marshal(kept) + if err != nil { + params.Del("order_by") + return + } + params.Set("order_by", string(encoded)) +} + type paging struct { limit int // -1 when unset offset int @@ -184,10 +219,7 @@ func parsePaging(v url.Values) (paging, error) { // Backends are asked for the first offset+limit records with no offset; the offset is applied to the union instead. func (p paging) upstreamParams(in url.Values) url.Values { - out := url.Values{} - for k, vs := range in { - out[k] = append([]string(nil), vs...) - } + out := copyParams(in) out.Del("offset") if p.limit >= 0 { out.Set("limit", strconv.Itoa(p.limit+p.offset)) @@ -195,6 +227,25 @@ func (p paging) upstreamParams(in url.Values) url.Values { return out } +// unpagedParams is upstreamParams for a response whose rows are folded together: +// a backend's own first N groups are not the merged result's first N, and a +// group truncated away on one backend folds to a wrong value, so every group is +// fetched and the window is cut after the fold. +func unpagedParams(in url.Values) url.Values { + out := copyParams(in) + out.Del("offset") + out.Del("limit") + return out +} + +func copyParams(in url.Values) url.Values { + out := url.Values{} + for k, vs := range in { + out[k] = append([]string(nil), vs...) + } + return out +} + func (p paging) apply(recs []json.RawMessage) []json.RawMessage { if p.offset >= len(recs) { return []json.RawMessage{} diff --git a/routes_test.go b/routes_test.go index a63421d..d5f6300 100644 --- a/routes_test.go +++ b/routes_test.go @@ -69,6 +69,37 @@ func TestQueryRoutes_GuardedRoutesSumAggregatesUncached(t *testing.T) { } } +// The dispatch guard and the per-function combiners have to hold at once: every +// guarded route in the table folds each aggregate column by its own operation, +// so a max comes back as the larger of the backends' values rather than as the +// blanket sum a count gets. Asserting both columns of one row pins that the +// operation is chosen per column, not per request. +func TestQueryRoutes_GuardedRoutesCombinePerFunction(t *testing.T) { + const q = `["extract",[["function","count"],["function","max","report_timestamp"]],["=","environment","production"]]` + for _, rt := range queryRoutes { + if rt.unsummed != "" { + continue + } + t.Run(rt.name, func(t *testing.T) { + probe := aggregateProbes[rt.name] + a := newCountingBackend(t, map[string]string{probe: `[{"count":90,"max":90}]`}) + b := newCountingBackend(t, map[string]string{probe: `[{"count":53,"max":53}]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + rec := doGet(t, srv.Handler(), probe, q) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{143}) { + t.Errorf("count = %v, want [143]", got) + } + if got := counts(t, rec.Body.Bytes(), "max"); !slices.Equal(got, []float64{90}) { + t.Errorf("max = %v, want [90]; the column was combined by the wrong operation", got) + } + }) + } +} + // The opt-out is deliberate, so widening it has to be deliberate too. func TestQueryRoutes_UnsummedRoutesAreTheKnownOnes(t *testing.T) { want := []string{ diff --git a/server.go b/server.go index 29d0a42..2d38958 100644 --- a/server.go +++ b/server.go @@ -102,9 +102,9 @@ func (s *Server) StopProbes() { s.health.Stop() } func (s *Server) cacheFor(path string, params url.Values) (Cache, bool) { switch { case path == factsPath, path == nodesPath, path == factNamesPath, isFactsSubPath(path): - // An aggregate row is a summed count, not the merged record set the + // An aggregate row is a combined count, not the merged record set the // cache was built for, so it stays on the live path. - if parseAggregate(params.Get("query")) != nil { + if spec, err := parseAggregate(params.Get("query")); spec != nil || err != nil { return noopCache{}, false } if s.factsCache != nil { @@ -126,15 +126,15 @@ func (s *Server) Handler() http.Handler { // route is one query endpoint: how a request selects it, the path its fan-out // asks backends for, and how it answers a plain query. handleQuery diverts an -// extract/function query to serveSummed before serve runs, so aggregate rows — +// extract/function query to serveCombined before serve runs, so aggregate rows — // which carry none of the certname, hash or name the merges key on, and would // collapse into one backend's numbers — cannot reach an identity-keyed merge. // unsummed opts a route out and records why; the zero value is guarded, so a -// route added without a decision is summed rather than silently merged. +// route added without a decision is combined rather than silently merged. type route struct { name string matches func(path string) bool - // fanOut is the path backends are queried on when the guard sums; empty + // fanOut is the path backends are queried on when the guard combines; empty // means the request's own path. fanOut string serve func(*Server, http.ResponseWriter, *http.Request) @@ -207,12 +207,19 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { } rt := routeFor(r.URL.Path) if rt.unsummed == "" { - if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil { + // The one place an aggregate is read, so a query pdbmux cannot fold is + // refused here rather than by whichever handler happens to notice. + spec, err := parseAggregate(r.URL.Query().Get("query")) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if spec != nil { path := rt.fanOut if path == "" { path = r.URL.Path } - s.serveSummed(w, r, path, spec.columns) + s.serveCombined(w, r, path, spec, spec.shape) return } } @@ -449,25 +456,39 @@ func (s *Server) serveEvents(w http.ResponseWriter, r *http.Request) { s.serveUnion(w, r, eventsPath, rawKey) } +// The columns come from the row rather than a query spec, so there is no +// grouping key or rewrite for serveCombined to apply. func (s *Server) serveEventCounts(w http.ResponseWriter, r *http.Request) { - s.serveSummed(w, r, r.URL.Path, inferredColumns) + s.serveCombined(w, r, r.URL.Path, nil, inferredShape) } // Merged rows are fewer than the backends' combined records, so include_total reports the merged count rather than a sum of X-Records. -func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string, columns func(map[string]json.RawMessage) ([]string, []string)) { +func (s *Server) serveCombined(w http.ResponseWriter, r *http.Request, path string, spec *aggregateSpec, shape func(map[string]json.RawMessage) rowShape) { in := r.URL.Query() page, err := parsePaging(in) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } + upstream := page.upstreamParams(in) + if spec != nil && len(spec.aggs) > 0 { + // An aggregate returns one row per distinct group, so the whole result is + // fetched and paged locally rather than truncated per backend. + upstream = unpagedParams(in) + } + if spec != nil && spec.query != "" { + // The backends answer the rewritten query, so they no longer carry the + // column the client's order_by may name; the merged rows are sorted here. + upstream.Set("query", spec.query) + dropOrderBy(upstream, avgColumn) + } s.serveCached(w, r, path, in, func(ctx context.Context) (cachedResponse, error) { - alive, err := s.aliveResults(ctx, path, page.upstreamParams(in)) + alive, err := s.aliveResults(ctx, path, upstream) if err != nil { return cachedResponse{}, err } - merged := sumRows(alive, columns) + merged := combineRows(alive, shape) sortRecords(merged, page.order) resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1} s.countBackends(&resp, alive) diff --git a/server_test.go b/server_test.go index 8b2454f..8768c59 100644 --- a/server_test.go +++ b/server_test.go @@ -810,6 +810,114 @@ func TestHandler_ReportsAggregateSummed(t *testing.T) { } } +// The backends never see the client's avg: they are asked for the sum and count +// it decomposes into, and the weighted average is computed from those. +func TestHandler_AvgIsRewrittenUpstreamAndWeighted(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"sum":10,"count":1}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"sum":60,"count":3}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{ + "query": {`["extract",[["function","avg","line"]]]`}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + const wantUpstream = `["extract",[["function","sum","line"],["function","count","line"]]]` + for _, fb := range []*fakeBackend{a, b} { + if got := fb.gotQuery(resourcesPath); got != wantUpstream { + t.Errorf("backend query = %s, want %s", got, wantUpstream) + } + } + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + want := []map[string]any{{"avg": float64(17.5)}} + if !reflect.DeepEqual(got, want) { + t.Errorf("body = %v, want %v (70/4, not the 15 an average of averages gives)", got, want) + } +} + +// order_by names a column the rewritten query no longer projects, so it is +// dropped upstream and applied to the merged rows here instead. +func TestHandler_AvgOrderByIsDroppedUpstream(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"sum":10,"count":1,"type":"File"},{"sum":8,"count":2,"type":"Stage"}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"sum":60,"count":3,"type":"File"}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{ + "query": {`["extract",[["function","avg","line"],"type"],["group_by","type"]]`}, + "order_by": {`[{"field":"avg","order":"desc"}]`}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if params, _ := a.params(resourcesPath); params.Get("order_by") != "" { + t.Errorf("backend got order_by %q, want it dropped", params.Get("order_by")) + } + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + want := []map[string]any{ + {"avg": float64(17.5), "type": "File"}, + {"avg": float64(4), "type": "Stage"}, + } + if !reflect.DeepEqual(got, want) { + t.Errorf("body = %v, want %v sorted by avg descending", got, want) + } +} + +func TestHandler_UnmergeableAggregateIsRefused(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + for _, path := range []string{nodesPath, factsPath, reportsPath, resourcesPath, factsPath + "/uptime"} { + rec := doGetParams(t, srv.Handler(), path, url.Values{ + "query": {`["extract",[["function","avg","line"],["function","count"]]]`}, + }) + if rec.Code != http.StatusBadRequest { + t.Errorf("%s: status %d, want 400: %s", path, rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "avg") { + t.Errorf("%s: body %q does not name the limitation", path, rec.Body.String()) + } + } + if _, asked := a.params(nodesPath); asked { + t.Error("a refused query was still fanned out") + } +} + +// max used to be folded by addition, returning a number no backend held. +func TestHandler_MaxIsNotSummed(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"max":20,"min":10}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"max":50,"min":30}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{ + "query": {`["extract",[["function","max","line"],["function","min","line"]]]`}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + want := []map[string]any{{"max": float64(50), "min": float64(10)}} + if !reflect.DeepEqual(got, want) { + t.Errorf("body = %v, want %v", got, want) + } +} + func TestHandler_ReportsAggregateRecordsIsMergedRowCount(t *testing.T) { a := newFakeBackend(t, `[]`, `[]`) a.bodies[reportsPath] = `[{"count":4,"status":"changed"}]` @@ -1006,6 +1114,100 @@ func TestHandler_FactsAggregateRecordsIsMergedRowCount(t *testing.T) { } } +// aggregatePagingBackends hold group counts a per-backend limit would truncate +// to the wrong answer: backend a's own first row is Exec, so a limit pushed +// upstream drops the File rows that together make File the real top group. +func aggregatePagingBackends(t *testing.T) (*fakeBackend, *fakeBackend) { + t.Helper() + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"count":6,"type":"Exec"},{"count":5,"type":"File"}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"count":4,"type":"File"}]` + return a, b +} + +const aggregateCountByType = `["extract",[["function","count","certname"],"type"],["group_by","type"]]` + +// A group truncated away on one backend would fold to a wrong total, so the +// whole aggregate is fetched and the window cut after the fold. +func TestHandler_ResourcesAggregateLimitIsAppliedAfterTheFold(t *testing.T) { + a, b := aggregatePagingBackends(t) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{ + "query": {aggregateCountByType}, + "order_by": {`[{"field":"count","order":"desc"}]`}, + "limit": {"1"}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + want := []map[string]any{{"count": float64(9), "type": "File"}} + if !reflect.DeepEqual(got, want) { + t.Errorf("top group = %v, want %v", got, want) + } + for name, fb := range map[string]*fakeBackend{"a": a, "b": b} { + p, ok := fb.params(resourcesPath) + if !ok { + t.Fatalf("%s backend was not queried", name) + } + if p.Has("limit") || p.Has("offset") { + t.Errorf("%s backend got limit=%q offset=%q, want both applied locally", name, p.Get("limit"), p.Get("offset")) + } + } +} + +// The merged row count is the whole aggregate's, not the paged window's. +func TestHandler_ResourcesAggregateIncludeTotalWithLocalPaging(t *testing.T) { + a, b := aggregatePagingBackends(t) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{ + "query": {aggregateCountByType}, + "order_by": {`[{"field":"count","order":"desc"}]`}, + "limit": {"1"}, + "offset": {"1"}, + "include_total": {"true"}, + }) + if got := rec.Header().Get(recordsHeader); got != "2" { + t.Errorf("%s = %q, want 2 merged groups", recordsHeader, got) + } + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + want := []map[string]any{{"count": float64(6), "type": "Exec"}} + if !reflect.DeepEqual(got, want) { + t.Errorf("offset window = %v, want %v", got, want) + } +} + +// Row functions return one row per record rather than per group, so nothing is +// folded and the upstream limit that bounds them still applies. +func TestHandler_ResourcesRowFunctionKeepsUpstreamLimit(t *testing.T) { + a, b := aggregatePagingBackends(t) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), resourcesPath, url.Values{ + "query": {`["extract",[["function","to_string","line"],"type"]]`}, + "limit": {"1"}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + p, ok := a.params(resourcesPath) + if !ok { + t.Fatal("backend a was not queried") + } + if p.Get("limit") != "1" { + t.Errorf("backend got limit=%q, want it forwarded", p.Get("limit")) + } +} + func TestHandler_FactsNonAggregateStillMergedByCertname(t *testing.T) { // Regression: routing aggregates to the summing path must not divert plain // queries, including an extract projection that carries no function column.