diff --git a/.github/workflows/flowview-query-go.yml b/.github/workflows/flowview-query-go.yml new file mode 100644 index 0000000..1b23fc5 --- /dev/null +++ b/.github/workflows/flowview-query-go.yml @@ -0,0 +1,67 @@ +# +-------------------------------------------------------------------------+ +# | Copyright (C) 2004-2026 The Cacti Group | +# | | +# | This program is free software; you can redistribute it and/or | +# | modify it under the terms of the GNU General Public License | +# | as published by the Free Software Foundation; either version 2 | +# | of the License, or (at your option) any later version. | +# +-------------------------------------------------------------------------+ + +name: FlowView Query Service (Go) + +on: + push: + branches: + - main + - develop + paths: + - 'service/flowview-query/**' + - '.github/workflows/flowview-query-go.yml' + pull_request: + branches: + - main + - develop + paths: + - 'service/flowview-query/**' + - '.github/workflows/flowview-query-go.yml' + +permissions: + contents: read + +jobs: + go-test: + runs-on: ubuntu-latest + + name: Build, vet and test the Go query service + + defaults: + run: + working-directory: service/flowview-query + + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Set up Go + uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 + with: + go-version-file: service/flowview-query/go.mod + cache-dependency-path: service/flowview-query/go.sum + + - name: Verify formatting + run: | + unformatted="$(gofmt -l .)" + if [ -n "$unformatted" ]; then + echo "The following files are not gofmt-clean:" + echo "$unformatted" + exit 1 + fi + + - name: Vet + run: go vet ./... + + - name: Build + run: go build ./... + + - name: Test + run: go test -race -cover ./... diff --git a/CHANGELOG.md b/CHANGELOG.md index 499ae45..80b51d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ --- develop --- +* feature: Add an optional FlowView Query Service (a long-running Go service under service/flowview-query) as a selectable Parallel Query Backend under FlowView Settings (default Legacy PHP Runner); when enabled, the parallel query parent runner (flowview_runner.php) delegates the whole query to the service over localhost instead of forking a PHP worker per shard, running the same cache-aware map-reduce (sharing parallel_database_query_shard_cache) and writing results back, with automatic fallback to the legacy runner when the service is unreachable + * bug: Load the pre-1.3 report compatibility shims (reports_run/reports_queue/reports_log_and_notify) only when the running Cacti is older than 1.3, decided with cacti_version_compare(CACTI_VERSION, '1.3', ...) rather than a function_exists() probe; because the plugin's functions.php is included during plugin init before core loads lib/reports.php, the function_exists() check ran while the now-native helpers were still absent, declared the shims, and core then fatally redeclared reports_log_and_notify() on 1.3.0 develop builds (falls back to function_exists() only when the Cacti version helpers are unavailable, e.g. the unit tests) * security: Move the plugin's remaining inline event handlers to CSP-safe bindings so the pages no longer trip Cacti's Content-Security-Policy script-src-attr directive: confirmation Cancel buttons use the `cactiReturnTo` class, the database/filter selects are bound via change() in their ready blocks, and the main filter's sort/report selects and time-shift icons are bound in flowview_display_filter's ready block * dev: Replace the translation-template CI check that regenerated locales/po/cacti.pot and compared it with a diff-based gate (tests/bin/check-i18n-pot.php) requiring cacti.pot to be updated only when a pull request adds, removes, or modifies an i18n function call diff --git a/flowview_runner.php b/flowview_runner.php index 0b18237..0867050 100644 --- a/flowview_runner.php +++ b/flowview_runner.php @@ -127,7 +127,26 @@ $current_time = time(); - $stats = parallel_database_parent_runner($query_id); + /* The Query Service backend runs the whole map-reduce in-process (shards as + * goroutines, shared shard cache) and writes results back, so we do not fork + * per-shard PHP workers. Fall back to the legacy parent runner when the + * service is disabled or unreachable. */ + $backend = read_config_option('flowview_parallel_backend'); + + if ($backend == 'service' && flowview_query_service_run(array($query_id))) { + $stats = array('threads' => 0, 'shards' => 0, 'cached' => 0, 'total_size' => 0, 'cached_size' => 0); + } else { + if ($backend == 'service') { + cacti_log('WARNING: FlowView Query Service unavailable, falling back to the Legacy PHP Runner', false, 'FLOWVIEW'); + + /* a partial service run may have staged map rows and left shards + * mid-flight; reset to a clean pending state so the legacy runner + * relaunches every shard and does not hang or duplicate data */ + flowview_parallel_reset_query($query_id); + } + + $stats = parallel_database_parent_runner($query_id); + } unregister_process('flowview', "db_query_{$query_id}", 0); diff --git a/locales/po/cacti.pot b/locales/po/cacti.pot index 243bd81..8345ac0 100644 --- a/locales/po/cacti.pot +++ b/locales/po/cacti.pot @@ -8,7 +8,7 @@ msgid "" msgstr "" "Project-Id-Version: Cacti \n" "Report-Msgid-Bugs-To: developers@cacti.net\n" -"POT-Creation-Date: 2026-09-30 10:11-0400\n" +"POT-Creation-Date: 2026-10-10 20:49-0400\n" "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n" "Last-Translator: Cacti Developers >\n" "Language-Team: Cacti Developers \n" @@ -2675,6 +2675,46 @@ msgstr "" msgid "Parallel Queries" msgstr "" +#: setup.php +msgid "Parallel Query Backend" +msgstr "" + +#: setup.php +msgid "Choose how parallel FlowView queries are executed. The Legacy PHP Runner forks the flowview_runner.php process per query and shard. The FlowView Query Service dispatches the same map-reduce work to a long-running Go service over localhost, avoiding per-query process startup and database polling latency. Requires the flowview-query service (see plugins/flowview/service/flowview-query) to be installed and running." +msgstr "" + +#: setup.php +msgid "Legacy PHP Runner" +msgstr "" + +#: setup.php +msgid "FlowView Query Service" +msgstr "" + +#: setup.php +msgid "Query Service URL" +msgstr "" + +#: setup.php +msgid "Base URL of the FlowView Query Service, for example http://127.0.0.1:8699. Leave blank to auto-discover the ephemeral port from the service's port file." +msgstr "" + +#: setup.php +msgid "Query Service Port File" +msgstr "" + +#: setup.php +msgid "Path to the port file the FlowView Query Service writes its bound address to. Used to discover the service endpoint when the Query Service URL above is blank." +msgstr "" + +#: setup.php +msgid "Query Service Token" +msgstr "" + +#: setup.php +msgid "Optional shared secret. When set, Cacti sends it to the FlowView Query Service as an \"Authorization: Bearer\" header and the service rejects any state-changing request that does not present the same token. Set the matching value in the service's config file (auth_token) or the FLOWVIEW_QUERY_TOKEN environment variable. Recommended whenever the service is reachable by anything other than trusted local processes." +msgstr "" + #: setup.php msgid "Max Concurrent Threads" msgstr "" diff --git a/service/flowview-query/.gitignore b/service/flowview-query/.gitignore new file mode 100644 index 0000000..cd046ef --- /dev/null +++ b/service/flowview-query/.gitignore @@ -0,0 +1,4 @@ +/flowview-query +*.exe +*.test +*.out diff --git a/service/flowview-query/README.md b/service/flowview-query/README.md new file mode 100644 index 0000000..f56ad76 --- /dev/null +++ b/service/flowview-query/README.md @@ -0,0 +1,304 @@ +# FlowView Parallel Query Service + +A small Go web service that runs FlowView's parallel (map-reduce) netflow +queries in-process, replacing the PHP fork/exec + database-polling path that is +the main source of query latency today. + +It listens on **localhost on an available port**, accepts a list of +already-planned query ids (the rows the PHP side inserts), fans the map phase +out across a bounded worker pool, reuses the existing shard cache so a cold +partition is read **once**, runs the reduce, and writes the same result set the +PHP `results` column holds today. A background scheduler goroutine performs the +cache/partition maintenance that PHP otherwise does on the request's critical +path. + +Selecting it is a one-click setting (**FlowView → Parallel Queries → Parallel +Query Backend**); the default remains the Legacy PHP Runner. + +--- + +## 1. Analysis of today's PHP parallel query path + +The current parallel engine is a map-reduce over the netflow partition tables, +coordinated entirely through database tables: + +| Table | Role | +| ----- | ---- | +| `parallel_database_query` | one row per query (signature, map/reduce SQL, status, `results`) | +| `parallel_database_query_shard` | one row per partition ("shard") to scan | +| `parallel_database_query_shard_cache` | cached map output per `(md5sum, partition)` for full-scan shards | + +Flow of a request: + +1. **Plan** — `parallel_database_query_request()` identifies the partition + tables in the time range, computes two md5s (the whole query, and the map + query *without* its time range → the "tables" signature used for caching), + and inserts one `parallel_database_query` row plus one + `parallel_database_query_shard` row per partition. A partition fully covered + by the time range is flagged `full_scan` (cacheable). +2. **Run** — `parallel_database_query_run()`: + - `exec_background()` launches **`flowview_runner.php` as a new PHP CLI + process** per request; + - then **polls** `parallel_database_query` in a `while (usleep(5000))` loop + until `status = complete`. +3. **Parent runner** — `parallel_database_parent_runner()` launches up to + `flowview_parallel_threads` **child PHP processes** (one per shard, again via + `exec_background`), polls the shard rows until all are `finished`, then runs + the **reduce** query over the intermediary map table and writes `results`. +4. **Child runner** — `parallel_database_child_runner()` per shard: if + `full_scan` and a cache row exists → reuse it; else run the map query against + the partition (optionally over a MaxScale read/write-split port) and cache + the result; insert the shard's rows into the intermediary table; mark + `finished`. + +**The caching insight** (which this service preserves): once a given map +signature has been run against a partition, its result is cached and +**that partition is never read again** for that signature — only the (small) +reduce runs. + +**Where the latency comes from** + +- **PHP process spawn** — one `flowview_runner.php` per request, then one more + child PHP process per shard. Interpreter startup + `cli_check.php` bootstrap + dominates small/cached queries. +- **Database polling** — the parent and the caller both busy-wait on table + status with `usleep`, adding fixed latency even when the work is already done. +- **Process bookkeeping** — `register_process_start`/`unregister_process`, + `processes` table rows, signal handling, per-shard `UPDATE`s. + +For a cached query the *useful* work is a single reduce, but the fixed cost of +spawning processes and polling tables can be seconds. + +## 2. What this service changes + +- **No process spawning.** The map phase is a **goroutine worker pool** capped + at `flowview_parallel_threads` (one shard per partition). No PHP CLI startup, + no `processes` table churn. +- **No polling.** The caller makes one blocking HTTP call and gets the result + when it is ready (bounded by `flowview_parallel_runlimit`). +- **Same cache, same semantics.** It reads/writes the existing + `parallel_database_query_shard_cache` table with the **same key** + `(md5sum, map_table, map_partition)`, so it interoperates with the PHP path + and a cold partition is still read only once. +- **Same reduce, same output.** The reduce runs in SQL over a temporary table + built from the shard output (mirroring the PHP intermediary map table), so the + returned rows match the PHP `results` exactly, in column order. +- **Thread count & MaxScale honoured.** `flowview_parallel_threads` bounds the + pool; when `flowview_use_maxscale` is on, map (read) queries are routed through + the MaxScale read/write-split port so shards round-robin across backends. +- **Maintenance off the critical path.** A scheduler goroutine expires queries + past their `time_to_live`, ages out cached partitions, and drops cache rows for + partitions that have been rolled off — the work PHP does inline today. + +### Map-reduce reduction (unchanged contract) + +``` +request ──► plan shards (one per partition) + │ + ├─ shard full_scan & cached? ── yes ─► reuse cached map output + │ no ─► run map query, cache it + │ (N shards run concurrently, capped at `threads`) + ▼ + stage all shard rows in a TEMPORARY table + ▼ + run the reduce query ──► results (same shape as PHP `results`) +``` + +## 3. Compilation & install + +Requires the Go toolchain (1.21+). On the Cacti host: + +```sh +# 1. Compile +cd plugins/flowview/service/flowview-query +go build -o flowview-query . + +# 2. Verify it runs with no database (in-memory map-reduce + HTTP bind + health) +./flowview-query -selftest + +# 3. Install the binary, config and service unit +sudo install -m 0755 flowview-query /usr/local/bin/flowview-query +sudo install -d -o root -g flowview -m 0750 /etc/flowview +sudo install -m 0640 config.sample.json /etc/flowview/flowview-query.json +sudoedit /etc/flowview/flowview-query.json # fill in DB host/user/password (see §4) +sudo install -m 0644 flowview-query.service /etc/systemd/system/ +``` + +Cross-compiling from a build host (no Go needed on the Cacti server): + +```sh +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o flowview-query . +# copy the static binary to the Cacti host and continue from step 3 above +``` + +Then create the service user and start it (see §5). + +## 4. Database credentials + +The service needs two logical databases (which may be the same server): + +- **Cacti control DB** — reads the `settings` table for the live tuning values + (`flowview_parallel_threads`, `flowview_parallel_runlimit`, + `flowview_parallel_time_to_live`, `flowview_use_maxscale`). +- **FlowView data DB** — the flow partition tables and the `parallel_*` tables. + When FlowView shares the Cacti database (`"use_cacti_db": true`, the default), + this is the same connection. + +Create a dedicated, least-privilege user. The service only needs: read the flow +partitions, read/write the three `parallel_*` tables, create/drop its own +`TEMPORARY`/intermediary tables, and read `settings` + `information_schema`. + +```sql +-- Dedicated service account (adjust host/password/schemas to your deployment) +CREATE USER 'flowview_svc'@'127.0.0.1' IDENTIFIED BY 'STRONG_PASSWORD'; + +-- Live tuning values from Cacti +GRANT SELECT ON cacti.settings TO 'flowview_svc'@'127.0.0.1'; + +-- Flow data + the parallel query tables (read flows, read/write parallel_*, +-- create/drop the intermediary + temporary reduce tables). +GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, CREATE TEMPORARY TABLES + ON flowview.* TO 'flowview_svc'@'127.0.0.1'; + +FLUSH PRIVILEGES; +``` + +If FlowView shares the Cacti schema, grant the data privileges on `cacti.*` +instead of `flowview.*` and set `"use_cacti_db": true`. + +If you enable MaxScale, point the service at the **read/write-split listener** +(e.g. port 3307/3308 from `maxscale/maxscale.cnf`) and ensure the service user +exists on all backend servers. + +Store the credentials in the config file and keep it `root:flowview 0640`: + +```sh +sudo install -d -o root -g flowview -m 0750 /etc/flowview +sudo install -m 0640 config.sample.json /etc/flowview/flowview-query.json +sudoedit /etc/flowview/flowview-query.json # fill in host/user/password/database +``` + +Credentials may also be supplied/overridden by environment variables +(useful with systemd `EnvironmentFile=` or a secrets manager): +`CACTI_DB_HOST/PORT/USER/PASS/NAME` and `FLOWVIEW_DB_HOST/PORT/USER/PASS/NAME`. + +## 5. Run + +```sh +sudo useradd --system --no-create-home --shell /usr/sbin/nologin flowview +sudo install -m 0644 flowview-query.service /etc/systemd/system/ +sudo systemctl daemon-reload +sudo systemctl enable --now flowview-query +journalctl -u flowview-query -f +``` + +On start the service logs the bound address and writes it to `port_file` +(default `/var/run/flowview/flowview-query.port`) so the PHP side can discover +the ephemeral port. + +## 6. PHP integration + +Integration is **selectable from the FlowView settings page** — no code change +is required to switch: + +> **Console → Configuration → Settings → FlowView → Parallel Queries → +> Parallel Query Backend** +> +> - **Legacy PHP Runner** (default) — forks `flowview_runner.php` per query/shard. +> - **FlowView Query Service** — dispatches to this service over localhost. + +When the service backend is selected, the parent runner (`flowview_runner.php`) +delegates the whole query to the service via `POST /run` **instead of forking a +PHP worker per shard**. `parallel_database_query_run()` is unchanged: it still +launches `flowview_runner.php`, which now returns as soon as the service has +read the query's shard rows (`parallel_database_query` + +`parallel_database_query_shard`), run the same cache-aware, thread-capped +map-reduce, and written `results` + `status = 'complete'` back. If the service +cannot be reached, the parent runner resets the query to a clean pending state +(`flowview_parallel_reset_query()` — truncates any partially-staged map rows, +returns all shards to `pending`, zeroes the finished/cached counters) and then +falls back to the legacy in-PHP `parallel_database_parent_runner()`, so a +half-finished service attempt never hangs or duplicates data. + +Endpoint discovery: set **Query Service URL** explicitly (e.g. +`http://127.0.0.1:8699`), or leave it blank and set **Query Service Port File** +to the path the service writes its bound address to (default +`/var/run/flowview/flowview-query.port`). + +Request body (`POST /run`): + +```json +{ "query_ids": [ 1337, 1338 ] } +``` + +Response: + +```json +{ "completed": [1337, 1338], "stats": { "1337": { "threads": 4, "shards": 4, "cached": 3 } } } +``` + +Because the service shares the `parallel_database_query_shard_cache` table with +the legacy path, a partition cached by one backend is reused by the other. + +## 7. Endpoints + +| Method & path | Auth | Purpose | +| ------------- | ---- | ------- | +| `POST /run` | token | Run the given `{"query_ids":[...]}` to completion; returns per-query stats. | +| `GET /health` | none | Liveness + DB ping + current thread setting. | +| `POST /maintenance` | token | Run one scheduler pass now (expiry/cleanup). | +| `POST /cache/invalidate?md5sum=` | token | Drop cached shards for a deleted query/filter signature. | + +All state-changing endpoints require `POST`; a `GET` to them returns +`405 Method Not Allowed`. When an auth token is configured (see §9) they also +require a matching `Authorization: Bearer ` header and otherwise return +`401 Unauthorized`. `/health` is always open (and never touches state) so it can +be used as a liveness probe. + +## 8. Scheduler + +A goroutine runs every `scheduler_interval` (and once at startup): + +- expires `parallel_database_query` rows past `time_to_live` and drops their + intermediary map tables; +- ages out `parallel_database_query_shard_cache` rows older than + `flowview_parallel_time_to_live`; +- drops cache rows whose partition table no longer exists (rolled-off data). + +Deleting a saved query or filter should call `POST /cache/invalidate` with that +signature so its cache does not linger until TTL. + +## 9. Security & network exposure + +The service is unauthenticated by default and is intended to listen on +**loopback only**: + +- `listen_host` **must** resolve to a loopback address (`127.0.0.1`, `::1`, + `localhost`). The service refuses to start on a routable address so it cannot + be accidentally exposed without authentication. +- For any deployment where the loopback assumption does not hold (e.g. a shared + host, or when you front the service with a reverse proxy on another + interface), set a **shared secret**: `auth_token` in the config file (or the + `FLOWVIEW_QUERY_TOKEN` environment variable) **and** the matching **FlowView → + Parallel Queries → Query Service Token** setting. State-changing requests then + require `Authorization: Bearer `; Cacti sends it automatically. +- The **database/MaxScale connections may target separate hosts.** The systemd + unit therefore does **not** apply `IPAddressAllow`/`IPAddressDeny` egress + filters (those would block reaching a remote DB). Only the HTTP *listener* is + loopback-bound. If your database is local and you want to additionally sandbox + egress, you can add `IPAddressAllow=localhost` / `IPAddressDeny=any` to the + unit yourself. + +## Files + +| File | Purpose | +| ---- | ------- | +| `main.go` | flags, config/DB wiring, listener, `-selftest` | +| `types.go` | request/response + ordered-row JSON | +| `config.go` | config load, DB pools, live Cacti settings | +| `query.go` | map-reduce engine, worker pool, shard cache, reduce | +| `scheduler.go` | background cache/partition maintenance | +| `server.go` | HTTP handlers | +| `demo.go` | in-memory self-test | +| `config.sample.json` | sample configuration | +| `flowview-query.service` | systemd unit | diff --git a/service/flowview-query/config.go b/service/flowview-query/config.go new file mode 100644 index 0000000..a4a0a60 --- /dev/null +++ b/service/flowview-query/config.go @@ -0,0 +1,263 @@ +package main + +import ( + "database/sql" + "encoding/json" + "fmt" + "net" + "os" + "strconv" + "sync" + "time" + + "github.com/go-sql-driver/mysql" +) + +// DBConfig holds the credentials for one database endpoint. +type DBConfig struct { + Host string `json:"host"` + Port int `json:"port"` + User string `json:"user"` + Password string `json:"password"` + Database string `json:"database"` + TLS string `json:"tls"` // "", "true", "skip-verify", or a registered name +} + +// Config is the service configuration, loaded from a JSON file and overridable +// by environment variables. It mirrors Cacti's config.php / flowview +// config.php.dist split: a Cacti control database (settings, processes) and a +// FlowView data database (flow partitions + the parallel_* tables), which may +// be the same server or separate. +type Config struct { + // Listen address. Host is forced to loopback; Port 0 selects a free port. + ListenHost string `json:"listen_host"` + ListenPort int `json:"listen_port"` + + // PortFile, when set, receives the chosen "host:port" so the PHP side can + // discover the endpoint (the service binds an ephemeral port by default). + PortFile string `json:"port_file"` + + // Cacti is the Cacti control DB (settings, processes tables). + Cacti DBConfig `json:"cacti"` + + // FlowView is the flow data DB. When UseCactiDB is true it is ignored and + // the Cacti connection is reused (matching $flowview_use_cacti_db). + UseCactiDB bool `json:"use_cacti_db"` + FlowView DBConfig `json:"flowview"` + + // MaxScale is the optional read/write-split endpoint used to round-robin + // the map (read) queries across MariaDB backends. When Enabled is false + // the FlowView connection is used for reads too. + MaxScale struct { + Enabled bool `json:"enabled"` + Host string `json:"host"` + Port int `json:"port"` + } `json:"maxscale"` + + // SchedulerInterval controls how often the maintenance goroutine runs. + SchedulerInterval Duration `json:"scheduler_interval"` + + // SettingsRefresh controls how often live Cacti settings are re-read. + SettingsRefresh Duration `json:"settings_refresh"` + + // AuthToken, when set, is required on the state-changing endpoints via an + // "Authorization: Bearer " header. The PHP client sends the same + // token (flowview_query_service_token). Loopback alone is not treated as an + // authorization boundary. + AuthToken string `json:"auth_token"` +} + +// Duration is a JSON-friendly time.Duration ("30s", "5m"). +type Duration time.Duration + +func (d *Duration) UnmarshalJSON(b []byte) error { + var s string + if err := json.Unmarshal(b, &s); err != nil { + return err + } + v, err := time.ParseDuration(s) + if err != nil { + return err + } + *d = Duration(v) + return nil +} + +func (d Duration) D() time.Duration { return time.Duration(d) } + +// isLoopbackHost reports whether host resolves only to loopback, so the +// unauthenticated-by-default service cannot be bound to a routable interface. +func isLoopbackHost(host string) bool { + switch host { + case "", "localhost": + return true + } + if ip := net.ParseIP(host); ip != nil { + return ip.IsLoopback() + } + return false +} + +// LoadConfig reads the JSON config file and applies environment overrides. +func LoadConfig(path string) (*Config, error) { + c := &Config{ + ListenHost: "127.0.0.1", + SchedulerInterval: Duration(60 * time.Second), + SettingsRefresh: Duration(60 * time.Second), + } + + if path != "" { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read config: %w", err) + } + if err := json.Unmarshal(raw, c); err != nil { + return nil, fmt.Errorf("parse config: %w", err) + } + } + + applyEnvDB("CACTI", &c.Cacti) + applyEnvDB("FLOWVIEW", &c.FlowView) + + if v := os.Getenv("FLOWVIEW_QUERY_TOKEN"); v != "" { + c.AuthToken = v + } + + if c.ListenHost == "" { + c.ListenHost = "127.0.0.1" + } + if c.SchedulerInterval.D() == 0 { + c.SchedulerInterval = Duration(60 * time.Second) + } + if c.SettingsRefresh.D() == 0 { + c.SettingsRefresh = Duration(60 * time.Second) + } + + return c, nil +} + +func applyEnvDB(prefix string, db *DBConfig) { + if v := os.Getenv(prefix + "_DB_HOST"); v != "" { + db.Host = v + } + if v := os.Getenv(prefix + "_DB_PORT"); v != "" { + if p, err := strconv.Atoi(v); err == nil { + db.Port = p + } + } + if v := os.Getenv(prefix + "_DB_USER"); v != "" { + db.User = v + } + if v := os.Getenv(prefix + "_DB_PASS"); v != "" { + db.Password = v + } + if v := os.Getenv(prefix + "_DB_NAME"); v != "" { + db.Database = v + } +} + +func (db DBConfig) dsn() string { + port := db.Port + if port == 0 { + port = 3306 + } + cfg := mysql.NewConfig() + cfg.User = db.User + cfg.Passwd = db.Password + cfg.Net = "tcp" + cfg.Addr = fmt.Sprintf("%s:%d", db.Host, port) + cfg.DBName = db.Database + cfg.ParseTime = false + cfg.Params = map[string]string{"charset": "utf8mb4"} + if db.TLS != "" { + cfg.TLSConfig = db.TLS + } + return cfg.FormatDSN() +} + +// open opens and verifies a pooled connection. +func (db DBConfig) open(maxOpen int) (*sql.DB, error) { + h, err := sql.Open("mysql", db.dsn()) + if err != nil { + return nil, err + } + h.SetMaxOpenConns(maxOpen) + h.SetMaxIdleConns(maxOpen) + h.SetConnMaxLifetime(5 * time.Minute) + return h, h.Ping() +} + +// Settings holds the live Cacti settings the engine honours, read from the +// Cacti `settings` table (read_config_option equivalents). +type Settings struct { + mu sync.RWMutex + threads int + runLimit time.Duration + timeToLive time.Duration + useMaxScale bool +} + +func defaultSettings() *Settings { + return &Settings{ + threads: 4, + runLimit: 300 * time.Second, + timeToLive: 21600 * time.Second, + } +} + +func (s *Settings) Threads() int { + s.mu.RLock() + defer s.mu.RUnlock() + if s.threads < 1 { + return 1 + } + return s.threads +} + +func (s *Settings) TimeToLive() time.Duration { + s.mu.RLock() + defer s.mu.RUnlock() + return s.timeToLive +} + +func (s *Settings) RunLimit() time.Duration { + s.mu.RLock() + defer s.mu.RUnlock() + return s.runLimit +} + +func (s *Settings) UseMaxScale() bool { + s.mu.RLock() + defer s.mu.RUnlock() + return s.useMaxScale +} + +// refresh re-reads the relevant options from the Cacti settings table. +func (s *Settings) refresh(cacti *sql.DB) { + get := func(name string, def int) int { + var v sql.NullString + err := cacti.QueryRow("SELECT value FROM settings WHERE name = ?", name).Scan(&v) + if err != nil || !v.Valid || v.String == "" { + return def + } + n, err := strconv.Atoi(v.String) + if err != nil { + return def + } + return n + } + + threads := get("flowview_parallel_threads", 4) + runLimit := get("flowview_parallel_runlimit", 300) + ttl := get("flowview_parallel_time_to_live", 21600) + + var ms sql.NullString + _ = cacti.QueryRow("SELECT value FROM settings WHERE name = ?", "flowview_use_maxscale").Scan(&ms) + + s.mu.Lock() + s.threads = threads + s.runLimit = time.Duration(runLimit) * time.Second + s.timeToLive = time.Duration(ttl) * time.Second + s.useMaxScale = ms.Valid && ms.String == "on" + s.mu.Unlock() +} diff --git a/service/flowview-query/config.sample.json b/service/flowview-query/config.sample.json new file mode 100644 index 0000000..6918965 --- /dev/null +++ b/service/flowview-query/config.sample.json @@ -0,0 +1,37 @@ +{ + "listen_host": "127.0.0.1", + "listen_port": 0, + "port_file": "/var/run/flowview/flowview-query.port", + + "_comment_auth_token": "Optional shared secret. When set, state-changing endpoints (/run, /maintenance, /cache/invalidate) require 'Authorization: Bearer '. Must match the FlowView 'Query Service Token' setting. May also be supplied via the FLOWVIEW_QUERY_TOKEN environment variable.", + "auth_token": "", + + "use_cacti_db": true, + + "cacti": { + "host": "127.0.0.1", + "port": 3306, + "user": "flowview_svc", + "password": "CHANGE_ME", + "database": "cacti", + "tls": "" + }, + + "flowview": { + "host": "127.0.0.1", + "port": 3306, + "user": "flowview_svc", + "password": "CHANGE_ME", + "database": "flowview", + "tls": "" + }, + + "maxscale": { + "enabled": false, + "host": "127.0.0.1", + "port": 3307 + }, + + "scheduler_interval": "60s", + "settings_refresh": "60s" +} diff --git a/service/flowview-query/config_test.go b/service/flowview-query/config_test.go new file mode 100644 index 0000000..cc7eee2 --- /dev/null +++ b/service/flowview-query/config_test.go @@ -0,0 +1,120 @@ +package main + +import ( + "os" + "strings" + "testing" + "time" +) + +func TestDSNDisablesParseTime(t *testing.T) { + db := DBConfig{Host: "127.0.0.1", Port: 3306, User: "u", Password: "p", Database: "d"} + dsn := db.dsn() + if strings.Contains(dsn, "parseTime=true") { + t.Fatalf("parseTime must be disabled so DATETIME arrives as a SQL string: %s", dsn) + } + if !strings.Contains(dsn, "charset=utf8mb4") { + t.Fatalf("charset missing from dsn: %s", dsn) + } +} + +func TestDSNDefaultsPort(t *testing.T) { + db := DBConfig{Host: "db", User: "u"} + if !strings.Contains(db.dsn(), "db:3306") { + t.Fatalf("expected default port 3306: %s", db.dsn()) + } +} + +func TestLoadConfigDefaults(t *testing.T) { + c, err := LoadConfig("") + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if c.ListenHost != "127.0.0.1" { + t.Fatalf("default listen host should be loopback, got %q", c.ListenHost) + } + if c.SchedulerInterval.D() != 60*time.Second || c.SettingsRefresh.D() != 60*time.Second { + t.Fatalf("default intervals wrong: %v / %v", c.SchedulerInterval.D(), c.SettingsRefresh.D()) + } +} + +func TestLoadConfigTokenEnvOverride(t *testing.T) { + t.Setenv("FLOWVIEW_QUERY_TOKEN", "s3cret") + c, err := LoadConfig("") + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if c.AuthToken != "s3cret" { + t.Fatalf("env token not applied: %q", c.AuthToken) + } +} + +func TestLoadConfigFromFile(t *testing.T) { + dir := t.TempDir() + path := dir + "/cfg.json" + body := `{"listen_host":"127.0.0.1","listen_port":9123,"scheduler_interval":"30s","auth_token":"file-tok"}` + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + c, err := LoadConfig(path) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if c.ListenPort != 9123 || c.AuthToken != "file-tok" || c.SchedulerInterval.D() != 30*time.Second { + t.Fatalf("file config not applied: %#v", c) + } +} + +func TestDurationUnmarshal(t *testing.T) { + var d Duration + if err := d.UnmarshalJSON([]byte(`"2m"`)); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if d.D() != 2*time.Minute { + t.Fatalf("want 2m, got %v", d.D()) + } + if err := d.UnmarshalJSON([]byte(`"not-a-duration"`)); err == nil { + t.Fatalf("expected error for bad duration") + } +} + +func TestSettingsRefresh(t *testing.T) { + cacti, mock, cleanup := newMock(t) + defer cleanup() + + mock.ExpectQuery("SELECT value FROM settings"). + WithArgs("flowview_parallel_threads"). + WillReturnRows(mockSetting("8")) + mock.ExpectQuery("SELECT value FROM settings"). + WithArgs("flowview_parallel_runlimit"). + WillReturnRows(mockSetting("120")) + mock.ExpectQuery("SELECT value FROM settings"). + WithArgs("flowview_parallel_time_to_live"). + WillReturnRows(mockSetting("600")) + mock.ExpectQuery("SELECT value FROM settings"). + WithArgs("flowview_use_maxscale"). + WillReturnRows(mockSetting("on")) + + s := defaultSettings() + s.refresh(cacti) + + if s.Threads() != 8 { + t.Fatalf("threads = %d, want 8", s.Threads()) + } + if s.RunLimit() != 120*time.Second { + t.Fatalf("runlimit = %v, want 120s", s.RunLimit()) + } + if s.TimeToLive() != 600*time.Second { + t.Fatalf("ttl = %v, want 600s", s.TimeToLive()) + } + if !s.UseMaxScale() { + t.Fatalf("maxscale should be on") + } +} + +func TestSettingsThreadsFloor(t *testing.T) { + s := &Settings{threads: 0} + if s.Threads() != 1 { + t.Fatalf("Threads() should floor at 1, got %d", s.Threads()) + } +} diff --git a/service/flowview-query/demo.go b/service/flowview-query/demo.go new file mode 100644 index 0000000..8243103 --- /dev/null +++ b/service/flowview-query/demo.go @@ -0,0 +1,127 @@ +package main + +import ( + "fmt" + "sort" + "sync" + "sync/atomic" + "time" +) + +// demoMapReduce exercises the engine's concurrency + partition-cache design +// entirely in memory (no database) so the service can be verified end-to-end on +// a host without MariaDB. It mirrors the real flow: one shard per partition, a +// bounded worker pool, a full-scan partition cache (cold -> cached), and a +// reduce that re-aggregates the per-shard group-by output. +func demoMapReduce(threads int) { + partitions := map[string][]demoFlow{ + "plugin_flowview_arin_2026010100": {{"10.0.0.1", 1200}, {"10.0.0.2", 300}, {"10.0.0.1", 800}}, + "plugin_flowview_arin_2026010101": {{"10.0.0.2", 500}, {"10.0.0.3", 900}, {"10.0.0.1", 100}}, + "plugin_flowview_arin_2026010102": {{"10.0.0.3", 150}, {"10.0.0.1", 50}, {"10.0.0.2", 700}}, + "plugin_flowview_arin_2026010103": {{"10.0.0.1", 2000}, {"10.0.0.3", 300}}, + } + + cache := &demoCache{m: map[string][]demoAgg{}} + + run := func(label string) { + start := time.Now() + var cachedShards int64 + + names := make([]string, 0, len(partitions)) + for n := range partitions { + names = append(names, n) + } + sort.Strings(names) + + sem := make(chan struct{}, threads) + var wg sync.WaitGroup + var mu sync.Mutex + combined := map[string]int64{} + + for _, name := range names { + wg.Add(1) + sem <- struct{}{} + + go func(name string) { + defer wg.Done() + defer func() { <-sem }() + + agg, hit := cache.get(name) + if hit { + atomic.AddInt64(&cachedShards, 1) + } else { + agg = mapPartition(partitions[name]) // the map phase + cache.put(name, agg) + } + + mu.Lock() + for _, a := range agg { // partial reduce merge + combined[a.key] += a.bytes + } + mu.Unlock() + }(name) + } + wg.Wait() + + type kv struct { + key string + bytes int64 + } + final := make([]kv, 0, len(combined)) + for k, v := range combined { + final = append(final, kv{k, v}) + } + sort.Slice(final, func(i, j int) bool { return final[i].bytes > final[j].bytes }) + + fmt.Printf(" [%s] threads=%d shards=%d cached=%d elapsed=%s\n", + label, threads, len(names), cachedShards, time.Since(start).Round(time.Microsecond)) + for _, f := range final { + fmt.Printf(" %-12s %6d bytes\n", f.key, f.bytes) + } + } + + fmt.Println("Self-test map-reduce (in-memory, no database):") + run("cold run ") + run("warm run ") // partition cache hit: every shard served from cache +} + +type demoFlow struct { + src string + bytes int64 +} + +type demoAgg struct { + key string + bytes int64 +} + +func mapPartition(rows []demoFlow) []demoAgg { + time.Sleep(10 * time.Millisecond) // simulate the cost of scanning a partition + m := map[string]int64{} + for _, r := range rows { + m[r.src] += r.bytes + } + out := make([]demoAgg, 0, len(m)) + for k, v := range m { + out = append(out, demoAgg{k, v}) + } + return out +} + +type demoCache struct { + mu sync.Mutex + m map[string][]demoAgg +} + +func (c *demoCache) get(k string) ([]demoAgg, bool) { + c.mu.Lock() + defer c.mu.Unlock() + v, ok := c.m[k] + return v, ok +} + +func (c *demoCache) put(k string, v []demoAgg) { + c.mu.Lock() + defer c.mu.Unlock() + c.m[k] = v +} diff --git a/service/flowview-query/extra_test.go b/service/flowview-query/extra_test.go new file mode 100644 index 0000000..e237435 --- /dev/null +++ b/service/flowview-query/extra_test.go @@ -0,0 +1,65 @@ +package main + +import ( + "context" + "os" + "testing" + + "github.com/DATA-DOG/go-sqlmock" +) + +func TestDemoMapReduceRuns(t *testing.T) { + // Smoke-test the in-memory self-test path so a DB-less host can validate + // the binary; it must not panic. + demoMapReduce(2) +} + +func TestMaxInt(t *testing.T) { + if maxInt(3, 7) != 7 || maxInt(9, 2) != 9 || maxInt(4, 4) != 4 { + t.Fatalf("maxInt wrong") + } +} + +func TestWritePortFile(t *testing.T) { + if err := writePortFile("", "ignored"); err != nil { + t.Fatalf("empty path should be a no-op, got %v", err) + } + + path := t.TempDir() + "/port" + if err := writePortFile(path, "127.0.0.1:9999"); err != nil { + t.Fatalf("writePortFile: %v", err) + } + b, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if string(b) != "127.0.0.1:9999\n" { + t.Fatalf("port file content = %q", b) + } +} + +func TestInvalidateCacheRequiresMD5(t *testing.T) { + eng := NewEngine(nil, nil, nil, defaultSettings()) + if _, err := eng.InvalidateCache(context.Background(), ""); err == nil { + t.Fatalf("empty md5sum should error") + } +} + +func TestSchedulerRunOnce(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + s := &Scheduler{eng: eng} + + // expireQueries: no expired rows + mock.ExpectQuery("FROM parallel_database_query\\s+WHERE time_to_live"). + WillReturnRows(sqlmock.NewRows([]string{"id", "map_table"})) + // expireShardCache + mock.ExpectExec("DELETE FROM parallel_database_query_shard_cache"). + WillReturnResult(sqlmock.NewResult(0, 0)) + // dropOrphanCache: nothing orphaned + mock.ExpectQuery("information_schema.TABLES"). + WillReturnRows(sqlmock.NewRows([]string{"map_table"})) + + s.RunOnce(context.Background()) +} diff --git a/service/flowview-query/flowview-query.service b/service/flowview-query/flowview-query.service new file mode 100644 index 0000000..65686c3 --- /dev/null +++ b/service/flowview-query/flowview-query.service @@ -0,0 +1,33 @@ +[Unit] +Description=FlowView Parallel Query Service +Documentation=https://github.com/Cacti/plugin_flowview +After=network-online.target mariadb.service +Wants=network-online.target + +[Service] +Type=simple +# Dedicated unprivileged user; see README for the DB grants it needs. +User=flowview +Group=flowview +RuntimeDirectory=flowview +RuntimeDirectoryMode=0750 + +# The config file holds the database credentials; keep it root:flowview 0640. +ExecStart=/usr/local/bin/flowview-query -config /etc/flowview/flowview-query.json +Restart=on-failure +RestartSec=3 + +# Hardening: minimal privileges and filesystem exposure. +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=true +PrivateTmp=true +ReadWritePaths=/var/run/flowview +# NOTE: The HTTP listener binds loopback only (listen_host in the config), but we +# deliberately do NOT add IPAddressAllow/IPAddressDeny here: those egress filters +# would also block the service from reaching a database or MaxScale node on a +# separate host. If you want to additionally sandbox egress and your DB is local, +# you may add: IPAddressAllow=localhost / IPAddressDeny=any + +[Install] +WantedBy=multi-user.target diff --git a/service/flowview-query/go.mod b/service/flowview-query/go.mod new file mode 100644 index 0000000..5ec9b08 --- /dev/null +++ b/service/flowview-query/go.mod @@ -0,0 +1,10 @@ +module flowview-query + +go 1.21 + +require ( + github.com/DATA-DOG/go-sqlmock v1.5.2 + github.com/go-sql-driver/mysql v1.8.1 +) + +require filippo.io/edwards25519 v1.1.0 // indirect diff --git a/service/flowview-query/go.sum b/service/flowview-query/go.sum new file mode 100644 index 0000000..361e8b4 --- /dev/null +++ b/service/flowview-query/go.sum @@ -0,0 +1,7 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= +github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU= +github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y= +github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg= +github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE= diff --git a/service/flowview-query/helpers_test.go b/service/flowview-query/helpers_test.go new file mode 100644 index 0000000..3db0297 --- /dev/null +++ b/service/flowview-query/helpers_test.go @@ -0,0 +1,176 @@ +package main + +import ( + "encoding/json" + "math" + "strings" + "testing" + "time" +) + +func TestNormalizeVal(t *testing.T) { + ts := time.Date(2024, 3, 2, 13, 4, 5, 0, time.UTC) + + cases := []struct { + name string + in any + want any + }{ + {"bytes to string", []byte("hello"), "hello"}, + {"time to sql datetime", ts, "2024-03-02 13:04:05"}, + {"int passthrough", int64(7), int64(7)}, + {"nil passthrough", nil, nil}, + {"string passthrough", "x", "x"}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := normalizeVal(c.in); got != c.want { + t.Fatalf("normalizeVal(%v) = %v, want %v", c.in, got, c.want) + } + }) + } +} + +func TestNormalizeParam(t *testing.T) { + if got := normalizeParam(float64(42)); got != int64(42) { + t.Fatalf("whole float64 should become int64, got %v (%T)", got, got) + } + if got := normalizeParam(float64(1.5)); got != float64(1.5) { + t.Fatalf("fractional float64 should stay float, got %v", got) + } + if got := normalizeParam(math.Inf(1)); got != math.Inf(1) { + t.Fatalf("inf float64 should pass through, got %v", got) + } + if got := normalizeParam(json.Number("13")); got != int64(13) { + t.Fatalf("integral json.Number should become int64, got %v (%T)", got, got) + } + if got := normalizeParam(json.Number("2.5")); got != float64(2.5) { + t.Fatalf("decimal json.Number should become float64, got %v (%T)", got, got) + } + if got := normalizeParam(json.Number("not-a-number")); got != "not-a-number" { + t.Fatalf("unparsable json.Number should fall back to its string, got %v", got) + } + if got := normalizeParam("keep"); got != "keep" { + t.Fatalf("string should pass through, got %v", got) + } +} + +func TestNormalizeParams(t *testing.T) { + out := normalizeParams([]any{float64(3), "a", json.Number("4")}) + if len(out) != 3 || out[0] != int64(3) || out[1] != "a" || out[2] != int64(4) { + t.Fatalf("normalizeParams unexpected: %#v", out) + } +} + +func TestQuoteTable(t *testing.T) { + cases := map[string]string{ + "flows": "`flows`", + "cacti.flows": "`cacti`.`flows`", + "we`ird": "`we``ird`", + "db.tab`le": "`db`.`tab``le`", + } + for in, want := range cases { + if got := quoteTable(in); got != want { + t.Fatalf("quoteTable(%q) = %q, want %q", in, got, want) + } + } +} + +func TestAppendClause(t *testing.T) { + var b strings.Builder + b.WriteString("SELECT 1") + appendClause(&b, "") + appendClause(&b, "WHERE x = 1") + if got := b.String(); got != "SELECT 1 WHERE x = 1" { + t.Fatalf("appendClause produced %q", got) + } +} + +func TestDecodeParams(t *testing.T) { + if got := decodeParams(nil); got != nil { + t.Fatalf("nil input should decode to nil, got %#v", got) + } + if got := decodeParams([]byte(" ")); got != nil { + t.Fatalf("blank input should decode to nil, got %#v", got) + } + if got := decodeParams([]byte("not json")); got != nil { + t.Fatalf("invalid json should decode to nil, got %#v", got) + } + got := decodeParams([]byte(`[1,"two",3]`)) + if len(got) != 3 || got[1] != "two" { + t.Fatalf("decodeParams unexpected: %#v", got) + } +} + +func TestDecodeRows(t *testing.T) { + if rows, err := decodeRows(nil); err != nil || rows != nil { + t.Fatalf("empty blob: rows=%v err=%v", rows, err) + } + if rows, err := decodeRows([]byte(`{"not":"array"}`)); err != nil || rows != nil { + t.Fatalf("non-array blob should yield nil rows, got rows=%v err=%v", rows, err) + } + + rows, err := decodeRows([]byte(`[{"a":1,"b":"x"},{"a":2,"b":"y"}]`)) + if err != nil { + t.Fatalf("decodeRows error: %v", err) + } + if len(rows) != 2 { + t.Fatalf("want 2 rows, got %d", len(rows)) + } + // Column order is preserved from the JSON object. + if got := strings.Join(rows[0].Cols, ","); got != "a,b" { + t.Fatalf("column order not preserved: %q", got) + } + if rows[1].Vals["b"] != "y" { + t.Fatalf("value mismatch: %#v", rows[1].Vals) + } +} + +func TestOrderedRowMarshalJSON(t *testing.T) { + row := &OrderedRow{ + Cols: []string{"z", "a"}, + Vals: map[string]any{"z": 1, "a": "x"}, + } + blob, err := json.Marshal(row) + if err != nil { + t.Fatalf("marshal: %v", err) + } + // Order must follow Cols, not Go's map key sorting. + if string(blob) != `{"z":1,"a":"x"}` { + t.Fatalf("ordered marshal wrong: %s", blob) + } +} + +func TestIsLoopbackHost(t *testing.T) { + cases := map[string]bool{ + "": true, + "localhost": true, + "127.0.0.1": true, + "::1": true, + "0.0.0.0": false, + "10.0.0.5": false, + "example.com": false, + } + for host, want := range cases { + if got := isLoopbackHost(host); got != want { + t.Fatalf("isLoopbackHost(%q) = %v, want %v", host, got, want) + } + } +} + +func TestBearer(t *testing.T) { + cases := map[string]string{ + "Bearer abc": "abc", + "Bearer abc ": "abc", + "bearer abc": "", + "Token abc": "", + "": "", + } + for header, want := range cases { + r := newRequest("GET", "/", header) + if got := bearer(r); got != want { + t.Fatalf("bearer(%q) = %q, want %q", header, got, want) + } + } +} diff --git a/service/flowview-query/main.go b/service/flowview-query/main.go new file mode 100644 index 0000000..94c5048 --- /dev/null +++ b/service/flowview-query/main.go @@ -0,0 +1,206 @@ +package main + +import ( + "context" + "database/sql" + "flag" + "fmt" + "io" + "log" + "net" + "net/http" + "os" + "os/signal" + "strconv" + "syscall" + "time" + + _ "github.com/go-sql-driver/mysql" +) + +var logger = log.New(os.Stderr, "flowview-query: ", log.LstdFlags|log.Lmsgprefix) + +func logf(format string, args ...any) { logger.Printf(format, args...) } + +func main() { + var ( + configPath = flag.String("config", "", "path to the JSON config file") + selftest = flag.Bool("selftest", false, "run an in-memory self-test (no database) and exit") + addr = flag.String("addr", "", "override listen address host:port (port 0 selects a free port)") + ) + flag.Parse() + + cfg, err := LoadConfig(*configPath) + if err != nil { + logger.Fatalf("config: %v", err) + } + + if *addr != "" { + host, port, perr := net.SplitHostPort(*addr) + if perr != nil { + logger.Fatalf("addr: %v", perr) + } + cfg.ListenHost = host + if p, cerr := strconv.Atoi(port); cerr == nil { + cfg.ListenPort = p + } + } + + if *selftest { + if err := runSelftest(cfg); err != nil { + logger.Fatalf("selftest: %v", err) + } + return + } + + if err := runService(cfg); err != nil { + logger.Fatalf("%v", err) + } +} + +// runService opens the databases, wires the engine/scheduler/server, binds a +// loopback port and serves until signalled. +func runService(cfg *Config) error { + settings := defaultSettings() + + cacti, err := cfg.Cacti.open(8) + if err != nil { + return fmt.Errorf("connect cacti db: %w", err) + } + defer cacti.Close() + + settings.refresh(cacti) + + var flow *sql.DB + if cfg.UseCactiDB { + flow = cacti + } else { + flow, err = cfg.FlowView.open(16) + if err != nil { + return fmt.Errorf("connect flowview db: %w", err) + } + defer flow.Close() + } + + var maxscale *sql.DB + if cfg.MaxScale.Enabled { + ms := cfg.FlowView + if cfg.UseCactiDB { + ms = cfg.Cacti + } + if cfg.MaxScale.Host != "" { + ms.Host = cfg.MaxScale.Host + } + if cfg.MaxScale.Port != 0 { + ms.Port = cfg.MaxScale.Port + } + maxscale, err = ms.open(maxInt(settings.Threads(), 4)) + if err != nil { + return fmt.Errorf("connect maxscale: %w", err) + } + defer maxscale.Close() + logf("MaxScale pool ready at %s:%d (map reads follow the live flowview_use_maxscale setting)", ms.Host, ms.Port) + } + + eng := NewEngine(flow, maxscale, cacti, settings) + sched := &Scheduler{eng: eng, interval: cfg.SchedulerInterval.D()} + srv := &Server{eng: eng, sched: sched, authToken: cfg.AuthToken} + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + // Live-refresh Cacti settings in the background. + go func() { + t := time.NewTicker(cfg.SettingsRefresh.D()) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + settings.refresh(cacti) + } + } + }() + + // Scheduler goroutine for cache/partition maintenance. + go sched.Run(ctx) + + if !isLoopbackHost(cfg.ListenHost) { + return fmt.Errorf("listen_host %q is not a loopback address: the query service is unauthenticated by default and must bind to loopback (use a reverse proxy or set auth_token for remote access)", cfg.ListenHost) + } + + ln, err := net.Listen("tcp", net.JoinHostPort(cfg.ListenHost, strconv.Itoa(cfg.ListenPort))) + if err != nil { + return fmt.Errorf("listen: %w", err) + } + + bound := ln.Addr().String() + logf("listening on http://%s (threads=%d)", bound, settings.Threads()) + if err := writePortFile(cfg.PortFile, bound); err != nil { + logf("warning: could not write port file: %v", err) + } + + httpSrv := &http.Server{Handler: srv.routes()} + + go func() { + <-ctx.Done() + shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = httpSrv.Shutdown(shutCtx) + }() + + if err := httpSrv.Serve(ln); err != nil && err != http.ErrServerClosed { + return err + } + logf("stopped") + return nil +} + +// runSelftest demonstrates the service without a database: it exercises the +// map-reduce/worker-pool/cache mechanics in memory and verifies the HTTP +// listener binds a free loopback port and answers /health. +func runSelftest(cfg *Config) error { + demoMapReduce(4) + + eng := NewEngine(nil, nil, nil, defaultSettings()) + srv := &Server{eng: eng, sched: &Scheduler{eng: eng}} + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return err + } + defer ln.Close() + + httpSrv := &http.Server{Handler: srv.routes()} + go httpSrv.Serve(ln) + defer httpSrv.Close() + + base := "http://" + ln.Addr().String() + fmt.Printf("\nHTTP listener bound at %s\n", base) + + resp, err := http.Get(base + "/health") + if err != nil { + return err + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + fmt.Printf("GET /health -> %d %s", resp.StatusCode, string(body)) + + fmt.Println("Self-test OK") + return nil +} + +func writePortFile(path, addr string) error { + if path == "" { + return nil + } + return os.WriteFile(path, []byte(addr+"\n"), 0o640) +} + +func maxInt(a, b int) int { + if a > b { + return a + } + return b +} diff --git a/service/flowview-query/query.go b/service/flowview-query/query.go new file mode 100644 index 0000000..2886b85 --- /dev/null +++ b/service/flowview-query/query.go @@ -0,0 +1,544 @@ +package main + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "fmt" + "math" + "strings" + "sync" + "time" +) + +// Engine runs parallel map-reduce queries against the FlowView database, +// mirroring parallel_database_parent_runner / parallel_database_child_runner. +type Engine struct { + flow *sql.DB // primary flow connection (intermediary + cache writes, reduce) + maxscale *sql.DB // optional MaxScale RW-split pool for map reads (nil = disabled) + cacti *sql.DB // Cacti control DB (settings) + settings *Settings + sem chan struct{} // shared, engine-wide worker limiter (global concurrency cap) +} + +// NewEngine builds an engine with a shared worker limiter sized to the current +// flowview_parallel_threads, so total database concurrency is bounded across +// all concurrent /run requests rather than per request. +func NewEngine(flow, maxscale, cacti *sql.DB, settings *Settings) *Engine { + threads := settings.Threads() + if threads < 1 { + threads = 1 + } + return &Engine{ + flow: flow, + maxscale: maxscale, + cacti: cacti, + settings: settings, + sem: make(chan struct{}, threads), + } +} + +// readPool returns the pool for map (read) queries, honouring the live +// flowview_use_maxscale setting when a MaxScale pool is configured. +func (e *Engine) readPool() *sql.DB { + if e.maxscale != nil && e.settings.UseMaxScale() { + return e.maxscale + } + return e.flow +} + +// RunQueries executes each requested query to completion (writing results and +// marking it complete), returning per-query stats. +func (e *Engine) RunQueries(ctx context.Context, ids []int64) (*RunResponse, error) { + resp := &RunResponse{Stats: map[string]Stats{}} + + for _, id := range ids { + st, err := e.RunQuery(ctx, id) + if err != nil { + return resp, fmt.Errorf("query %d: %w", id, err) + } + resp.Completed = append(resp.Completed, id) + resp.Stats[fmt.Sprintf("%d", id)] = st + } + + return resp, nil +} + +// RunQuery runs one query_id: the map phase fans out across a bounded worker +// pool (one shard per partition, cache-aware), then the reduce phase writes +// the final result set and marks the query complete. +func (e *Engine) RunQuery(ctx context.Context, id int64) (Stats, error) { + threads := e.settings.Threads() + + q, err := e.fetchQuery(ctx, id) + if err != nil { + return Stats{}, err + } + + shards, err := e.fetchShards(ctx, id) + if err != nil { + return Stats{}, err + } + + var ( + wg sync.WaitGroup + mu sync.Mutex + cached int + firstErr error + ) + + for i := range shards { + s := shards[i] + + wg.Add(1) + + go func(s shardRow) { + defer wg.Done() + + // Shared engine limiter: caps concurrency across all /run requests. + select { + case e.sem <- struct{}{}: + defer func() { <-e.sem }() + case <-ctx.Done(): + mu.Lock() + if firstErr == nil { + firstErr = ctx.Err() + } + mu.Unlock() + return + } + + wasCached, err := e.runShard(ctx, q, s) + + mu.Lock() + defer mu.Unlock() + if err != nil { + if firstErr == nil { + firstErr = err + } + return + } + if wasCached { + cached++ + } + }(s) + } + + wg.Wait() + + if firstErr != nil { + return Stats{}, firstErr + } + + if err := e.reduce(ctx, q); err != nil { + return Stats{}, err + } + + return Stats{Threads: threads, Shards: len(shards), Cached: cached}, nil +} + +// runShard runs (or cache-serves) one partition's map query and stages the +// output in the query's intermediary map table. +func (e *Engine) runShard(ctx context.Context, q *queryRow, s shardRow) (bool, error) { + if _, err := e.flow.ExecContext(ctx, + `UPDATE parallel_database_query_shard SET status = 'running' WHERE query_id = ? AND shard_id = ?`, + q.ID, s.ShardID); err != nil { + return false, err + } + + rows, cached, err := e.mapShard(ctx, q, s) + if err != nil { + return false, fmt.Errorf("map shard %s: %w", s.MapTable, err) + } + + if len(rows) > 0 { + if err := insertRows(ctx, e.flow, q.MapTable, rows); err != nil { + return false, fmt.Errorf("stage shard %s: %w", s.MapTable, err) + } + } + + if _, err := e.flow.ExecContext(ctx, + `UPDATE parallel_database_query_shard SET status = 'finished', completed = CURRENT_TIMESTAMP WHERE query_id = ? AND shard_id = ?`, + q.ID, s.ShardID); err != nil { + return false, err + } + + if _, err := e.flow.ExecContext(ctx, + `UPDATE parallel_database_query SET finished_shards = finished_shards + 1 WHERE id = ?`, q.ID); err != nil { + return false, err + } + + return cached, nil +} + +// mapShard returns the shard's map output, serving a full-scan shard from the +// shared cache when present (and populating it on a miss). +func (e *Engine) mapShard(ctx context.Context, q *queryRow, s shardRow) ([]*OrderedRow, bool, error) { + if s.FullScan { + if rows, ok, err := e.cacheGet(ctx, q.MD5SumTables, s.MapTable, s.MapPartition); err != nil { + return nil, false, err + } else if ok { + if _, uerr := e.flow.ExecContext(ctx, + `UPDATE parallel_database_query SET cached_shards = cached_shards + 1 WHERE id = ?`, q.ID); uerr != nil { + return nil, false, uerr + } + return rows, true, nil + } + } + + rows, err := e.queryDB(ctx, e.readPool(), s.MapQuery, decodeParams(s.MapParams)) + if err != nil { + return nil, false, err + } + + if s.FullScan { + if err := e.cachePut(ctx, q.MD5SumTables, s, rows); err != nil { + logf("cache write %s/%s failed: %v", s.MapTable, s.MapPartition, err) + } + } + + return rows, false, nil +} + +// reduce runs the reduce query over the intermediary map table, stores the +// results and marks the query complete (then clears its shards). +func (e *Engine) reduce(ctx context.Context, q *queryRow) error { + var stru reduceStruct + if err := json.Unmarshal(q.ReduceQuery, &stru); err != nil { + return fmt.Errorf("decode reduce_query: %w", err) + } + + var b strings.Builder + b.WriteString(stru.SQLQuery) + b.WriteString(" FROM ") + b.WriteString(quoteTable(q.MapTable)) + appendClause(&b, stru.SQLWhere) + appendClause(&b, stru.SQLGroupBy) + appendClause(&b, stru.SQLHaving) + appendClause(&b, stru.SQLOrder) + appendClause(&b, stru.SQLLimit) + + data, err := e.queryDB(ctx, e.flow, b.String(), stru.SQLParams) + if err != nil { + return fmt.Errorf("reduce: %w", err) + } + + blob, err := json.Marshal(data) + if err != nil { + return err + } + + if _, err := e.flow.ExecContext(ctx, + `UPDATE parallel_database_query SET results = ?, status = 'complete' WHERE id = ?`, blob, q.ID); err != nil { + return err + } + + if _, err := e.flow.ExecContext(ctx, + `DELETE FROM parallel_database_query_shard WHERE query_id = ?`, q.ID); err != nil { + return err + } + + return nil +} + +func (e *Engine) fetchQuery(ctx context.Context, id int64) (*queryRow, error) { + q := &queryRow{} + err := e.flow.QueryRowContext(ctx, + `SELECT id, md5sum_tables, map_table, reduce_query FROM parallel_database_query WHERE id = ?`, id). + Scan(&q.ID, &q.MD5SumTables, &q.MapTable, &q.ReduceQuery) + if err != nil { + return nil, err + } + return q, nil +} + +func (e *Engine) fetchShards(ctx context.Context, id int64) ([]shardRow, error) { + rows, err := e.flow.QueryContext(ctx, + `SELECT shard_id, full_scan, map_table, map_partition, map_query, map_params + FROM parallel_database_query_shard WHERE query_id = ? ORDER BY shard_id`, id) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []shardRow + for rows.Next() { + var s shardRow + var full int + if err := rows.Scan(&s.ShardID, &full, &s.MapTable, &s.MapPartition, &s.MapQuery, &s.MapParams); err != nil { + return nil, err + } + s.FullScan = full != 0 + out = append(out, s) + } + return out, rows.Err() +} + +// cacheGet looks up a cached full-scan shard result. +func (e *Engine) cacheGet(ctx context.Context, md5sum, table, partition string) ([]*OrderedRow, bool, error) { + var blob []byte + err := e.flow.QueryRowContext(ctx, + `SELECT results FROM parallel_database_query_shard_cache + WHERE md5sum = ? AND map_table = ? AND map_partition = ?`, + md5sum, table, partition).Scan(&blob) + + if err == sql.ErrNoRows { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + + rows, derr := decodeRows(blob) + if derr != nil { + return nil, false, nil // corrupt cache row -> treat as miss + } + return rows, true, nil +} + +// cachePut stores a full-scan shard result keyed exactly like the PHP cache. +func (e *Engine) cachePut(ctx context.Context, md5sum string, s shardRow, rows []*OrderedRow) error { + var mn, mx sql.NullString + _ = e.flow.QueryRowContext(ctx, + fmt.Sprintf("SELECT MIN(start_time), MAX(end_time) FROM %s", quoteTable(s.MapTable))).Scan(&mn, &mx) + + minDate := mn.String + maxDate := mx.String + if minDate == "" { + minDate = "0000-00-00 00:00:00" + } + if maxDate == "" { + maxDate = minDate + } + + blob, err := json.Marshal(rows) + if err != nil { + return err + } + + _, err = e.flow.ExecContext(ctx, + `INSERT INTO parallel_database_query_shard_cache + (md5sum, map_table, map_partition, min_date, max_date, results) + VALUES (?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE results = VALUES(results), date_created = CURRENT_TIMESTAMP`, + md5sum, s.MapTable, s.MapPartition, minDate, maxDate, blob) + + return err +} + +func (e *Engine) queryDB(ctx context.Context, db *sql.DB, query string, params []any) ([]*OrderedRow, error) { + rows, err := db.QueryContext(ctx, query, normalizeParams(params)...) + if err != nil { + return nil, err + } + defer rows.Close() + return scanRows(rows) +} + +// --- helpers --- + +func appendClause(b *strings.Builder, clause string) { + if clause != "" { + b.WriteString(" ") + b.WriteString(clause) + } +} + +func scanRows(rows *sql.Rows) ([]*OrderedRow, error) { + cols, err := rows.Columns() + if err != nil { + return nil, err + } + + out := make([]*OrderedRow, 0, 64) + + for rows.Next() { + vals := make([]any, len(cols)) + ptrs := make([]any, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + if err := rows.Scan(ptrs...); err != nil { + return nil, err + } + + row := &OrderedRow{Cols: append([]string(nil), cols...), Vals: make(map[string]any, len(cols))} + for i, c := range cols { + row.Vals[c] = normalizeVal(vals[i]) + } + out = append(out, row) + } + + return out, rows.Err() +} + +func insertRows(ctx context.Context, db *sql.DB, table string, rows []*OrderedRow) error { + if len(rows) == 0 { + return nil + } + + cols := rows[0].Cols + const batch = 500 + + for i := 0; i < len(rows); i += batch { + end := i + batch + if end > len(rows) { + end = len(rows) + } + chunk := rows[i:end] + + var b strings.Builder + b.WriteString("INSERT INTO ") + b.WriteString(quoteTable(table)) + b.WriteString(" (") + for j, c := range cols { + if j > 0 { + b.WriteString(", ") + } + b.WriteString("`") + b.WriteString(c) + b.WriteString("`") + } + b.WriteString(") VALUES ") + + params := make([]any, 0, len(chunk)*len(cols)) + for r, row := range chunk { + if r > 0 { + b.WriteString(", ") + } + b.WriteString("(") + for j, c := range cols { + if j > 0 { + b.WriteString(", ") + } + b.WriteString("?") + params = append(params, normalizeParam(row.Vals[c])) + } + b.WriteString(")") + } + + if _, err := db.ExecContext(ctx, b.String(), params...); err != nil { + return err + } + } + + return nil +} + +func decodeParams(raw []byte) []any { + if len(bytes.TrimSpace(raw)) == 0 { + return nil + } + var out []any + if err := json.Unmarshal(raw, &out); err != nil { + return nil + } + return out +} + +func decodeRows(blob []byte) ([]*OrderedRow, error) { + if len(bytes.TrimSpace(blob)) == 0 { + return nil, nil + } + + dec := json.NewDecoder(bytes.NewReader(blob)) + dec.UseNumber() + + tok, err := dec.Token() + if err != nil { + return nil, err + } + if d, ok := tok.(json.Delim); !ok || d != '[' { + return nil, nil + } + + var out []*OrderedRow + for dec.More() { + row, err := decodeObject(dec) + if err != nil { + return nil, err + } + out = append(out, row) + } + return out, nil +} + +func decodeObject(dec *json.Decoder) (*OrderedRow, error) { + tok, err := dec.Token() + if err != nil { + return nil, err + } + if d, ok := tok.(json.Delim); !ok || d != '{' { + return nil, fmt.Errorf("expected object") + } + + row := &OrderedRow{Vals: map[string]any{}} + for dec.More() { + keyTok, err := dec.Token() + if err != nil { + return nil, err + } + key, _ := keyTok.(string) + + var val any + if err := dec.Decode(&val); err != nil { + return nil, err + } + row.Cols = append(row.Cols, key) + row.Vals[key] = val + } + _, _ = dec.Token() // consume '}' + + return row, nil +} + +func normalizeVal(v any) any { + switch t := v.(type) { + case []byte: + return string(t) + case time.Time: + // Match the PHP backend's SQL datetime strings (not RFC3339) so detailed + // reports and warm-cache DATETIME reinsertion keep the same format. + return t.Format("2006-01-02 15:04:05") + default: + return v + } +} + +func normalizeParam(v any) any { + switch n := v.(type) { + case float64: + if n == math.Trunc(n) && !math.IsInf(n, 0) { + return int64(n) + } + return n + case json.Number: + if i, err := n.Int64(); err == nil { + return i + } + if f, err := n.Float64(); err == nil { + return f + } + return n.String() + default: + return v + } +} + +func normalizeParams(in []any) []any { + out := make([]any, len(in)) + for i, v := range in { + out[i] = normalizeParam(v) + } + return out +} + +// quoteTable backtick-quotes a (possibly schema-qualified) table identifier. +func quoteTable(name string) string { + parts := strings.Split(name, ".") + for i, p := range parts { + parts[i] = "`" + strings.ReplaceAll(p, "`", "``") + "`" + } + return strings.Join(parts, ".") +} diff --git a/service/flowview-query/query_test.go b/service/flowview-query/query_test.go new file mode 100644 index 0000000..8360934 --- /dev/null +++ b/service/flowview-query/query_test.go @@ -0,0 +1,200 @@ +package main + +import ( + "context" + "testing" + + "github.com/DATA-DOG/go-sqlmock" +) + +func TestRunQuerySingleShardNoCache(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + + settings := defaultSettings() + settings.threads = 1 + eng := NewEngine(flow, nil, nil, settings) + + // fetchQuery + reduce := `{"sql_query":"SELECT a, b","sql_params":[]}` + mock.ExpectQuery("SELECT id, md5sum_tables"). + WithArgs(int64(5)). + WillReturnRows(sqlmock.NewRows([]string{"id", "md5sum_tables", "map_table", "reduce_query"}). + AddRow(int64(5), "sig", "map_5", []byte(reduce))) + + // fetchShards: one non-full-scan shard + mock.ExpectQuery("FROM parallel_database_query_shard WHERE query_id"). + WithArgs(int64(5)). + WillReturnRows(sqlmock.NewRows([]string{"shard_id", "full_scan", "map_table", "map_partition", "map_query", "map_params"}). + AddRow(int64(1), 0, "map_5", "part0", "SELECT x FROM part0", []byte("[]"))) + + // runShard: status=running + mock.ExpectExec("SET status = 'running'"). + WithArgs(int64(5), int64(1)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + // mapShard: the map query (non-full-scan, no cache) + mock.ExpectQuery("SELECT x FROM part0"). + WillReturnRows(sqlmock.NewRows([]string{"x"}).AddRow(int64(11))) + + // insertRows into the intermediary table + mock.ExpectExec("INSERT INTO .map_5."). + WithArgs(int64(11)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + // status=finished + mock.ExpectExec("SET status = 'finished'"). + WithArgs(int64(5), int64(1)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + // finished_shards++ + mock.ExpectExec("finished_shards = finished_shards"). + WithArgs(int64(5)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + // reduce query + mock.ExpectQuery("SELECT a, b FROM .map_5."). + WillReturnRows(sqlmock.NewRows([]string{"a", "b"}).AddRow(int64(1), "y")) + + // store results + mark complete + mock.ExpectExec("SET results = \\?, status = 'complete'"). + WillReturnResult(sqlmock.NewResult(0, 1)) + + // clear shards + mock.ExpectExec("DELETE FROM parallel_database_query_shard WHERE query_id"). + WithArgs(int64(5)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + st, err := eng.RunQuery(context.Background(), 5) + if err != nil { + t.Fatalf("RunQuery: %v", err) + } + if st.Shards != 1 || st.Cached != 0 || st.Threads != 1 { + t.Fatalf("unexpected stats: %#v", st) + } +} + +func TestRunQueriesWrapsStats(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + + settings := defaultSettings() + settings.threads = 1 + eng := NewEngine(flow, nil, nil, settings) + + reduce := `{"sql_query":"SELECT a","sql_params":[]}` + mock.ExpectQuery("SELECT id, md5sum_tables"). + WithArgs(int64(9)). + WillReturnRows(sqlmock.NewRows([]string{"id", "md5sum_tables", "map_table", "reduce_query"}). + AddRow(int64(9), "sig", "map_9", []byte(reduce))) + // No shards. + mock.ExpectQuery("FROM parallel_database_query_shard WHERE query_id"). + WithArgs(int64(9)). + WillReturnRows(sqlmock.NewRows([]string{"shard_id", "full_scan", "map_table", "map_partition", "map_query", "map_params"})) + mock.ExpectQuery("SELECT a FROM .map_9."). + WillReturnRows(sqlmock.NewRows([]string{"a"}).AddRow(int64(1))) + mock.ExpectExec("SET results = \\?, status = 'complete'"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("DELETE FROM parallel_database_query_shard WHERE query_id"). + WithArgs(int64(9)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + resp, err := eng.RunQueries(context.Background(), []int64{9}) + if err != nil { + t.Fatalf("RunQueries: %v", err) + } + if len(resp.Completed) != 1 || resp.Completed[0] != 9 { + t.Fatalf("completed mismatch: %#v", resp.Completed) + } + if _, ok := resp.Stats["9"]; !ok { + t.Fatalf("stats missing key: %#v", resp.Stats) + } +} + +func TestMapShardCacheHit(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + + q := &queryRow{ID: 3, MD5SumTables: "sig", MapTable: "map_3"} + s := shardRow{ShardID: 1, FullScan: true, MapTable: "map_3", MapPartition: "p0", MapQuery: "SELECT 1"} + + cached := `[{"x":7}]` + mock.ExpectQuery("FROM parallel_database_query_shard_cache"). + WithArgs("sig", "map_3", "p0"). + WillReturnRows(sqlmock.NewRows([]string{"results"}).AddRow([]byte(cached))) + mock.ExpectExec("cached_shards = cached_shards"). + WithArgs(int64(3)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + rows, wasCached, err := eng.mapShard(context.Background(), q, s) + if err != nil { + t.Fatalf("mapShard: %v", err) + } + if !wasCached { + t.Fatalf("expected cache hit") + } + if len(rows) != 1 || rows[0].Vals["x"] == nil { + t.Fatalf("unexpected cached rows: %#v", rows) + } +} + +func TestMapShardCacheMissPopulates(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + + q := &queryRow{ID: 3, MD5SumTables: "sig", MapTable: "map_3"} + s := shardRow{ShardID: 1, FullScan: true, MapTable: "map_3", MapPartition: "p0", MapQuery: "SELECT col FROM raw"} + + // cache miss + mock.ExpectQuery("FROM parallel_database_query_shard_cache"). + WithArgs("sig", "map_3", "p0"). + WillReturnRows(sqlmock.NewRows([]string{"results"})) + // map query + mock.ExpectQuery("SELECT col FROM raw"). + WillReturnRows(sqlmock.NewRows([]string{"col"}).AddRow(int64(42))) + // cachePut: MIN/MAX then INSERT + mock.ExpectQuery("SELECT MIN\\(start_time\\), MAX\\(end_time\\)"). + WillReturnRows(sqlmock.NewRows([]string{"mn", "mx"}).AddRow("2024-01-01 00:00:00", "2024-01-02 00:00:00")) + mock.ExpectExec("INSERT INTO parallel_database_query_shard_cache"). + WillReturnResult(sqlmock.NewResult(0, 1)) + + rows, wasCached, err := eng.mapShard(context.Background(), q, s) + if err != nil { + t.Fatalf("mapShard: %v", err) + } + if wasCached { + t.Fatalf("expected cache miss") + } + if len(rows) != 1 { + t.Fatalf("expected one mapped row, got %d", len(rows)) + } +} + +func TestReadPoolHonoursLiveSetting(t *testing.T) { + flow, _, cleanupF := newMock(t) + defer cleanupF() + ms, _, cleanupM := newMock(t) + defer cleanupM() + + settings := defaultSettings() + eng := NewEngine(flow, ms, nil, settings) + + if eng.readPool() != flow { + t.Fatalf("maxscale off -> should read from flow pool") + } + + settings.mu.Lock() + settings.useMaxScale = true + settings.mu.Unlock() + if eng.readPool() != ms { + t.Fatalf("maxscale on -> should read from maxscale pool") + } + + // No maxscale pool configured -> always flow, even if the setting is on. + eng2 := NewEngine(flow, nil, nil, settings) + if eng2.readPool() != flow { + t.Fatalf("no maxscale pool -> must fall back to flow") + } +} diff --git a/service/flowview-query/scheduler.go b/service/flowview-query/scheduler.go new file mode 100644 index 0000000..e1a68ae --- /dev/null +++ b/service/flowview-query/scheduler.go @@ -0,0 +1,179 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "time" +) + +// Scheduler performs background cache/partition maintenance that the PHP path +// does inline (and thus on the request's critical path): expiring query +// results and aged partition cache, and dropping orphaned intermediary tables. +type Scheduler struct { + eng *Engine + interval time.Duration +} + +// Run ticks until the context is cancelled, performing one maintenance pass +// per interval. +func (s *Scheduler) Run(ctx context.Context) { + // One pass at startup so a restart immediately reclaims stale state. + s.RunOnce(ctx) + + t := time.NewTicker(s.interval) + defer t.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.RunOnce(ctx) + } + } +} + +// RunOnce performs a single maintenance pass. Safe to call from the /maintenance +// endpoint as well as the ticker. +func (s *Scheduler) RunOnce(ctx context.Context) { + if err := s.expireQueries(ctx); err != nil { + logf("scheduler: expire queries: %v", err) + } + if err := s.expireShardCache(ctx); err != nil { + logf("scheduler: expire shard cache: %v", err) + } + if err := s.dropOrphanCache(ctx); err != nil { + logf("scheduler: drop orphan cache: %v", err) + } +} + +// expireQueries removes completed/stale parallel_database_query rows whose TTL +// has passed and drops their intermediary map tables. +func (s *Scheduler) expireQueries(ctx context.Context) error { + rows, err := s.eng.flow.QueryContext(ctx, + `SELECT id, map_table FROM parallel_database_query + WHERE time_to_live > 0 AND time_to_live < UNIX_TIMESTAMP()`) + if err != nil { + return err + } + + type q struct { + id int64 + table string + } + var expired []q + for rows.Next() { + var item q + var tbl sql.NullString + if err := rows.Scan(&item.id, &tbl); err != nil { + rows.Close() + return err + } + item.table = tbl.String + expired = append(expired, item) + } + rows.Close() + if err := rows.Err(); err != nil { + return err + } + + for _, item := range expired { + if item.table != "" { + if _, err := s.eng.flow.ExecContext(ctx, "DROP TABLE IF EXISTS "+quoteTable(item.table)); err != nil { + // Keep the query row so a later pass can retry the drop; deleting + // it now would orphan the intermediary table. + logf("scheduler: drop map table %s (will retry): %v", item.table, err) + continue + } + } + if _, err := s.eng.flow.ExecContext(ctx, + "DELETE FROM parallel_database_query_shard WHERE query_id = ?", item.id); err != nil { + logf("scheduler: delete shards %d: %v", item.id, err) + } + if _, err := s.eng.flow.ExecContext(ctx, + "DELETE FROM parallel_database_query WHERE id = ?", item.id); err != nil { + logf("scheduler: delete query %d: %v", item.id, err) + } + } + + if len(expired) > 0 { + logf("scheduler: expired %d queries", len(expired)) + } + return nil +} + +// expireShardCache ages out cached partition results older than the configured +// time-to-live so hot partitions stay cached while cold data is reclaimed. +func (s *Scheduler) expireShardCache(ctx context.Context) error { + ttl := int64(s.eng.settings.TimeToLive().Seconds()) + if ttl <= 0 { + return nil + } + + res, err := s.eng.flow.ExecContext(ctx, + `DELETE FROM parallel_database_query_shard_cache + WHERE date_created < (NOW() - INTERVAL ? SECOND)`, ttl) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n > 0 { + logf("scheduler: expired %d cached shards", n) + } + return nil +} + +// dropOrphanCache removes cache rows whose underlying partition table no longer +// exists (e.g. a raw partition that has been rolled off). +func (s *Scheduler) dropOrphanCache(ctx context.Context) error { + rows, err := s.eng.flow.QueryContext(ctx, + `SELECT DISTINCT c.map_table + FROM parallel_database_query_shard_cache c + LEFT JOIN information_schema.TABLES t + ON t.table_schema = DATABASE() AND t.table_name = c.map_table + WHERE t.table_name IS NULL`) + if err != nil { + return err + } + + var orphans []string + for rows.Next() { + var tbl string + if err := rows.Scan(&tbl); err != nil { + rows.Close() + return err + } + orphans = append(orphans, tbl) + } + rows.Close() + if err := rows.Err(); err != nil { + return err + } + + for _, tbl := range orphans { + if _, err := s.eng.flow.ExecContext(ctx, + "DELETE FROM parallel_database_query_shard_cache WHERE map_table = ?", tbl); err != nil { + logf("scheduler: delete orphan cache %s: %v", tbl, err) + } + } + + if len(orphans) > 0 { + logf("scheduler: dropped cache for %d missing partitions", len(orphans)) + } + return nil +} + +// InvalidateCache removes cached shard results for a query/filter signature. +// Called when a saved query or filter is deleted so its cache does not linger. +func (e *Engine) InvalidateCache(ctx context.Context, md5sum string) (int64, error) { + if md5sum == "" { + return 0, fmt.Errorf("md5sum required") + } + res, err := e.flow.ExecContext(ctx, + "DELETE FROM parallel_database_query_shard_cache WHERE md5sum = ?", md5sum) + if err != nil { + return 0, err + } + n, _ := res.RowsAffected() + return n, nil +} diff --git a/service/flowview-query/scheduler_test.go b/service/flowview-query/scheduler_test.go new file mode 100644 index 0000000..cef34bc --- /dev/null +++ b/service/flowview-query/scheduler_test.go @@ -0,0 +1,96 @@ +package main + +import ( + "context" + "errors" + "testing" + + "github.com/DATA-DOG/go-sqlmock" +) + +func TestExpireQueriesDropsTableAndRows(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + s := &Scheduler{eng: eng} + + mock.ExpectQuery("FROM parallel_database_query\\s+WHERE time_to_live"). + WillReturnRows(sqlmock.NewRows([]string{"id", "map_table"}).AddRow(int64(1), "map_1")) + mock.ExpectExec("DROP TABLE IF EXISTS .map_1."). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("DELETE FROM parallel_database_query_shard WHERE query_id"). + WithArgs(int64(1)). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("DELETE FROM parallel_database_query WHERE id"). + WithArgs(int64(1)). + WillReturnResult(sqlmock.NewResult(0, 1)) + + if err := s.expireQueries(context.Background()); err != nil { + t.Fatalf("expireQueries: %v", err) + } +} + +// When the DROP fails we must keep the query row so a later pass can retry, +// otherwise the intermediary table is orphaned. +func TestExpireQueriesKeepsRowOnDropFailure(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + s := &Scheduler{eng: eng} + + mock.ExpectQuery("FROM parallel_database_query\\s+WHERE time_to_live"). + WillReturnRows(sqlmock.NewRows([]string{"id", "map_table"}).AddRow(int64(2), "map_2")) + mock.ExpectExec("DROP TABLE IF EXISTS .map_2."). + WillReturnError(errors.New("boom")) + // No DELETE expectations: the row must be left in place. + + if err := s.expireQueries(context.Background()); err != nil { + t.Fatalf("expireQueries: %v", err) + } +} + +func TestExpireShardCache(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + s := &Scheduler{eng: eng} + + mock.ExpectExec("DELETE FROM parallel_database_query_shard_cache"). + WithArgs(int64(eng.settings.TimeToLive().Seconds())). + WillReturnResult(sqlmock.NewResult(0, 3)) + + if err := s.expireShardCache(context.Background()); err != nil { + t.Fatalf("expireShardCache: %v", err) + } +} + +func TestExpireShardCacheSkippedWhenTTLZero(t *testing.T) { + flow, _, cleanup := newMock(t) + defer cleanup() + settings := defaultSettings() + settings.timeToLive = 0 + eng := NewEngine(flow, nil, nil, settings) + s := &Scheduler{eng: eng} + + // No expectations: a zero TTL must skip the delete entirely. + if err := s.expireShardCache(context.Background()); err != nil { + t.Fatalf("expireShardCache: %v", err) + } +} + +func TestDropOrphanCache(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + s := &Scheduler{eng: eng} + + mock.ExpectQuery("information_schema.TABLES"). + WillReturnRows(sqlmock.NewRows([]string{"map_table"}).AddRow("orphan_1")) + mock.ExpectExec("DELETE FROM parallel_database_query_shard_cache WHERE map_table"). + WithArgs("orphan_1"). + WillReturnResult(sqlmock.NewResult(0, 1)) + + if err := s.dropOrphanCache(context.Background()); err != nil { + t.Fatalf("dropOrphanCache: %v", err) + } +} diff --git a/service/flowview-query/server.go b/service/flowview-query/server.go new file mode 100644 index 0000000..b9e4a99 --- /dev/null +++ b/service/flowview-query/server.go @@ -0,0 +1,120 @@ +package main + +import ( + "context" + "crypto/subtle" + "encoding/json" + "net/http" + "strings" + "time" +) + +// Server exposes the engine over localhost HTTP. +type Server struct { + eng *Engine + sched *Scheduler + authToken string +} + +func (s *Server) routes() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/health", s.handleHealth) + mux.HandleFunc("/run", s.handleRun) + mux.HandleFunc("/maintenance", s.handleMaintenance) + mux.HandleFunc("/cache/invalidate", s.handleInvalidate) + return mux +} + +// guard enforces POST and, when an auth token is configured, a matching +// "Authorization: Bearer " header. Loopback is not treated as an +// authorization boundary. +func (s *Server) guard(w http.ResponseWriter, r *http.Request) bool { + if r.Method != http.MethodPost { + http.Error(w, "POST required", http.StatusMethodNotAllowed) + return false + } + if s.authToken != "" && subtle.ConstantTimeCompare([]byte(bearer(r)), []byte(s.authToken)) != 1 { + writeJSON(w, http.StatusUnauthorized, map[string]any{"error": "unauthorized"}) + return false + } + return true +} + +func bearer(r *http.Request) string { + const p = "Bearer " + h := r.Header.Get("Authorization") + if strings.HasPrefix(h, p) { + return strings.TrimSpace(h[len(p):]) + } + return "" +} + +func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { + status := "ok" + code := http.StatusOK + + if s.eng.flow != nil { + ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + if err := s.eng.flow.PingContext(ctx); err != nil { + status = "degraded: " + err.Error() + code = http.StatusServiceUnavailable + } + } + + writeJSON(w, code, map[string]any{ + "status": status, + "threads": s.eng.settings.Threads(), + }) +} + +func (s *Server) handleRun(w http.ResponseWriter, r *http.Request) { + if !s.guard(w, r) { + return + } + + var req RunRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]any{"error": "bad request: " + err.Error()}) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), s.eng.settings.RunLimit()) + defer cancel() + + resp, err := s.eng.RunQueries(ctx, req.QueryIDs) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]any{"error": err.Error(), "completed": resp.Completed}) + return + } + + writeJSON(w, http.StatusOK, resp) +} + +func (s *Server) handleMaintenance(w http.ResponseWriter, r *http.Request) { + if !s.guard(w, r) { + return + } + s.sched.RunOnce(r.Context()) + writeJSON(w, http.StatusOK, map[string]any{"status": "ok"}) +} + +func (s *Server) handleInvalidate(w http.ResponseWriter, r *http.Request) { + if !s.guard(w, r) { + return + } + md5sum := r.URL.Query().Get("md5sum") + n, err := s.eng.InvalidateCache(r.Context(), md5sum) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]any{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, map[string]any{"invalidated": n}) +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + enc := json.NewEncoder(w) + _ = enc.Encode(v) +} diff --git a/service/flowview-query/server_test.go b/service/flowview-query/server_test.go new file mode 100644 index 0000000..445f6d7 --- /dev/null +++ b/service/flowview-query/server_test.go @@ -0,0 +1,95 @@ +package main + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/DATA-DOG/go-sqlmock" +) + +func TestHealthIsOpen(t *testing.T) { + srv := &Server{eng: NewEngine(nil, nil, nil, defaultSettings()), authToken: "tok"} + w := httptest.NewRecorder() + srv.handleHealth(w, newRequest("GET", "/health", "")) + if w.Code != http.StatusOK { + t.Fatalf("health status = %d, want 200", w.Code) + } +} + +func TestGuardRejectsNonPost(t *testing.T) { + srv := &Server{} + w := httptest.NewRecorder() + if srv.guard(w, newRequest("GET", "/run", "")) { + t.Fatalf("guard should reject GET") + } + if w.Code != http.StatusMethodNotAllowed { + t.Fatalf("status = %d, want 405", w.Code) + } +} + +func TestGuardAllowsPostWhenNoToken(t *testing.T) { + srv := &Server{} + w := httptest.NewRecorder() + if !srv.guard(w, newRequest("POST", "/run", "")) { + t.Fatalf("guard should allow POST when no token configured") + } +} + +func TestGuardRequiresToken(t *testing.T) { + srv := &Server{authToken: "sekret"} + + w := httptest.NewRecorder() + if srv.guard(w, newRequest("POST", "/run", "")) { + t.Fatalf("missing token should be rejected") + } + if w.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401", w.Code) + } + + w = httptest.NewRecorder() + if srv.guard(w, newRequest("POST", "/run", "Bearer wrong")) { + t.Fatalf("wrong token should be rejected") + } + + w = httptest.NewRecorder() + if !srv.guard(w, newRequest("POST", "/run", "Bearer sekret")) { + t.Fatalf("correct token should be accepted") + } +} + +func TestHandleRunBadJSON(t *testing.T) { + srv := &Server{} + w := httptest.NewRecorder() + r := httptest.NewRequest("POST", "/run", http.NoBody) + srv.handleRun(w, r) + if w.Code != http.StatusBadRequest { + t.Fatalf("bad body status = %d, want 400", w.Code) + } +} + +func TestHandleMaintenanceRequiresPost(t *testing.T) { + srv := &Server{} + w := httptest.NewRecorder() + srv.handleMaintenance(w, newRequest("GET", "/maintenance", "")) + if w.Code != http.StatusMethodNotAllowed { + t.Fatalf("status = %d, want 405", w.Code) + } +} + +func TestHandleInvalidate(t *testing.T) { + flow, mock, cleanup := newMock(t) + defer cleanup() + eng := NewEngine(flow, nil, nil, defaultSettings()) + srv := &Server{eng: eng, sched: &Scheduler{eng: eng}} + + mock.ExpectExec("DELETE FROM parallel_database_query_shard_cache WHERE md5sum"). + WithArgs("abc"). + WillReturnResult(sqlmock.NewResult(0, 2)) + + w := httptest.NewRecorder() + srv.handleInvalidate(w, newRequest("POST", "/cache/invalidate?md5sum=abc", "")) + if w.Code != http.StatusOK { + t.Fatalf("invalidate status = %d, want 200", w.Code) + } +} diff --git a/service/flowview-query/testutil_test.go b/service/flowview-query/testutil_test.go new file mode 100644 index 0000000..74ed433 --- /dev/null +++ b/service/flowview-query/testutil_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "database/sql" + "net/http" + "net/http/httptest" + "testing" + + "github.com/DATA-DOG/go-sqlmock" +) + +// newRequest builds an *http.Request with an optional Authorization header. +func newRequest(method, target, auth string) *http.Request { + r := httptest.NewRequest(method, target, nil) + if auth != "" { + r.Header.Set("Authorization", auth) + } + return r +} + +// newMock returns a *sql.DB backed by sqlmock with unordered expectation +// matching (the engine fans shards out across goroutines) plus a cleanup +// function that asserts every expectation was met. +func newMock(t *testing.T) (*sql.DB, sqlmock.Sqlmock, func()) { + t.Helper() + conn, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New: %v", err) + } + mock.MatchExpectationsInOrder(false) + cleanup := func() { + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unmet sqlmock expectations: %v", err) + } + conn.Close() + } + return conn, mock, cleanup +} + +// mockSetting builds a single-column "value" result row for a settings lookup. +func mockSetting(value string) *sqlmock.Rows { + return sqlmock.NewRows([]string{"value"}).AddRow(value) +} diff --git a/service/flowview-query/types.go b/service/flowview-query/types.go new file mode 100644 index 0000000..c9c10f1 --- /dev/null +++ b/service/flowview-query/types.go @@ -0,0 +1,109 @@ +// Package main implements the FlowView parallel query service. +// +// It is a drop-in replacement for the PHP fork/exec + database-polling +// parallel query path (flowview_runner.php -> +// parallel_database_parent_runner / parallel_database_child_runner). It: +// +// - listens on localhost on an available port, +// - accepts a list of already-planned query ids (the rows the PHP side +// inserts into parallel_database_query / parallel_database_query_shard), +// - fans the map phase out across a bounded worker pool (one shard per +// flow partition table, capped by flowview_parallel_threads), +// - reuses the shared parallel_database_query_shard_cache so a cold +// partition is read once and never re-read for the same query signature, +// - runs the reduce phase and writes the same "results" the PHP runner +// writes today, then marks the query complete. +// +// Because it reads and writes the same tables, the PHP side only needs to +// replace its exec_background() launch with one HTTP call; the existing +// result retrieval (and multi-request UNION reduce) is unchanged. +// +// A separate scheduler goroutine performs cache/partition maintenance +// (TTL expiry, aged-out partitions, query/filter deletion). +package main + +import "encoding/json" + +// RunRequest is the payload POSTed to /run: the ids of the +// parallel_database_query rows to execute. +type RunRequest struct { + QueryIDs []int64 `json:"query_ids"` +} + +// RunResponse is returned by /run. +type RunResponse struct { + Completed []int64 `json:"completed"` + Stats map[string]Stats `json:"stats"` +} + +// Stats mirrors the PARALLEL STATS the PHP runner logs. +type Stats struct { + Threads int `json:"threads"` + Shards int `json:"shards"` + Cached int `json:"cached"` +} + +// reduceStruct is the "outer" (reduce phase) query definition decoded from the +// parallel_database_query.reduce_query column ($stru_outer on the PHP side). +type reduceStruct struct { + SQLQuery string `json:"sql_query"` + SQLWhere string `json:"sql_where"` + SQLHaving string `json:"sql_having"` + SQLGroupBy string `json:"sql_groupby"` + SQLOrder string `json:"sql_order"` + SQLLimit string `json:"sql_limit"` + SQLParams []any `json:"sql_params"` +} + +// queryRow is one parallel_database_query row. +type queryRow struct { + ID int64 + MD5SumTables string + MapTable string + ReduceQuery []byte +} + +// shardRow is one parallel_database_query_shard row. +type shardRow struct { + ShardID int64 + FullScan bool + MapTable string + MapPartition string + MapQuery string + MapParams []byte +} + +// OrderedRow is one result row that marshals to a JSON object preserving the +// SQL column order (Go maps would otherwise sort keys, changing column order +// relative to the PHP assoc-array output). +type OrderedRow struct { + Cols []string + Vals map[string]any +} + +// MarshalJSON renders the row as an ordered JSON object. +func (r *OrderedRow) MarshalJSON() ([]byte, error) { + buf := []byte{'{'} + + for i, c := range r.Cols { + if i > 0 { + buf = append(buf, ',') + } + + key, err := json.Marshal(c) + if err != nil { + return nil, err + } + + val, err := json.Marshal(r.Vals[c]) + if err != nil { + return nil, err + } + + buf = append(buf, key...) + buf = append(buf, ':') + buf = append(buf, val...) + } + + return append(buf, '}'), nil +} diff --git a/setup.php b/setup.php index 7e53144..abaa507 100644 --- a/setup.php +++ b/setup.php @@ -642,6 +642,40 @@ function flowview_config_settings(): void { 'method' => 'spacer', 'collapsible' => 'true' ), + 'flowview_parallel_backend' => array( + 'friendly_name' => __('Parallel Query Backend', 'flowview'), + 'description' => __('Choose how parallel FlowView queries are executed. The Legacy PHP Runner forks the flowview_runner.php process per query and shard. The FlowView Query Service dispatches the same map-reduce work to a long-running Go service over localhost, avoiding per-query process startup and database polling latency. Requires the flowview-query service (see plugins/flowview/service/flowview-query) to be installed and running.', 'flowview'), + 'method' => 'drop_array', + 'array' => array( + 'legacy' => __('Legacy PHP Runner', 'flowview'), + 'service' => __('FlowView Query Service', 'flowview') + ), + 'default' => 'legacy' + ), + 'flowview_query_service_url' => array( + 'friendly_name' => __('Query Service URL', 'flowview'), + 'description' => __('Base URL of the FlowView Query Service, for example http://127.0.0.1:8699. Leave blank to auto-discover the ephemeral port from the service\'s port file.', 'flowview'), + 'method' => 'textbox', + 'default' => '', + 'max_length' => 128, + 'size' => 60 + ), + 'flowview_query_service_portfile' => array( + 'friendly_name' => __('Query Service Port File', 'flowview'), + 'description' => __('Path to the port file the FlowView Query Service writes its bound address to. Used to discover the service endpoint when the Query Service URL above is blank.', 'flowview'), + 'method' => 'textbox', + 'default' => '/var/run/flowview/flowview-query.port', + 'max_length' => 255, + 'size' => 60 + ), + 'flowview_query_service_token' => array( + 'friendly_name' => __('Query Service Token', 'flowview'), + 'description' => __('Optional shared secret. When set, Cacti sends it to the FlowView Query Service as an "Authorization: Bearer" header and the service rejects any state-changing request that does not present the same token. Set the matching value in the service\'s config file (auth_token) or the FLOWVIEW_QUERY_TOKEN environment variable. Recommended whenever the service is reachable by anything other than trusted local processes.', 'flowview'), + 'method' => 'textbox', + 'default' => '', + 'max_length' => 255, + 'size' => 60 + ), 'flowview_parallel_threads' => array( 'friendly_name' => __('Max Concurrent Threads', 'flowview'), 'description' => __('The maximum number of threads that will be dispatched to run the FlowView queries. Note that you can have at most 1 thread per database partition, and you should be careful not to overload your database server with having too many concurrent threads running.', 'flowview'), @@ -762,6 +796,132 @@ function flowview_poller_bottom(): void { exec_background($php, $config['base_path'] . '/plugins/flowview/flowview_process.php'); } +/** + * flowview_query_service_endpoint - Resolve the base URL of the FlowView Query + * Service, using the configured URL or, when blank, the port file the service + * writes its bound loopback address to. + * + * @return string The base URL without a trailing slash, or '' when unavailable. + */ +function flowview_query_service_endpoint(): string { + $url = trim((string) read_config_option('flowview_query_service_url')); + + if ($url != '') { + return rtrim($url, '/'); + } + + $portfile = trim((string) read_config_option('flowview_query_service_portfile')); + + if ($portfile == '') { + $portfile = '/var/run/flowview/flowview-query.port'; + } + + if (is_readable($portfile)) { + $addr = trim((string) file_get_contents($portfile)); + + if ($addr != '') { + return 'http://' . $addr; + } + } + + return ''; +} + +/** + * flowview_query_service_run - Dispatch the given query ids to the FlowView + * Query Service, which runs the same map-reduce and writes the results back to + * the parallel_database_query table. + * + * @param array $request_ids The parallel_database_query ids to run. + * + * @return bool True when the service accepted and completed the queries, false + * on any error so the caller can fall back to the legacy runner. + */ +function flowview_query_service_run($request_ids): bool { + $base = flowview_query_service_endpoint(); + + if ($base == '' || !function_exists('curl_init')) { + return false; + } + + $timeout = (int) read_config_option('flowview_parallel_runlimit'); + + if ($timeout <= 0) { + $timeout = 300; + } + + $payload = json_encode(array('query_ids' => array_map('intval', $request_ids))); + + $headers = array('Content-Type: application/json'); + + $token = trim((string) read_config_option('flowview_query_service_token')); + + if ($token != '') { + $headers[] = 'Authorization: Bearer ' . $token; + } + + $ch = curl_init($base . '/run'); + + curl_setopt_array($ch, array( + CURLOPT_POST => true, + CURLOPT_POSTFIELDS => $payload, + CURLOPT_HTTPHEADER => $headers, + CURLOPT_RETURNTRANSFER => true, + CURLOPT_CONNECTTIMEOUT => 2, + CURLOPT_TIMEOUT => $timeout, + )); + + $response = curl_exec($ch); + $code = (int) curl_getinfo($ch, CURLINFO_HTTP_CODE); + $err = curl_error($ch); + + curl_close($ch); + + if ($response === false || $code != 200) { + cacti_log(sprintf('WARNING: FlowView Query Service call failed (code:%d err:%s)', $code, $err), false, 'FLOWVIEW'); + + return false; + } + + return true; +} + +/** + * flowview_parallel_reset_query - Restore a parallel query to a clean, re-runnable + * state before the Legacy PHP Runner takes over from a failed Query Service + * attempt. A partial service run may have staged map rows, advanced the + * finished/cached counters and left shards in the 'running'/'finished' state; + * the legacy runner only launches 'pending' shards and loops until every shard + * is 'finished', so without this reset it could hang or return duplicated data. + * + * @param mixed $query_id The parallel_database_query id to reset. + * + * @return void + */ +function flowview_parallel_reset_query($query_id): void { + $map_table = flowview_db_fetch_cell_prepared('SELECT map_table + FROM parallel_database_query + WHERE id = ?', + [$query_id]); + + /* discard any rows the service staged into the intermediary table */ + if ($map_table != '') { + flowview_db_execute_prepared("TRUNCATE TABLE $map_table"); + } + + /* return every shard to 'pending' so the legacy runner relaunches them all */ + flowview_db_execute_prepared('UPDATE parallel_database_query_shard + SET status = ?, completed = NULL + WHERE query_id = ?', + ['pending', $query_id]); + + /* reset the progress counters and mark the query schedulable again */ + flowview_db_execute_prepared('UPDATE parallel_database_query + SET finished_shards = 0, cached_shards = 0, status = ? + WHERE id = ?', + ['scheduled', $query_id]); +} + /** * @return void */ diff --git a/tests/bin/patch-coverage.php b/tests/bin/patch-coverage.php index 87ee296..ecbcd77 100644 --- a/tests/bin/patch-coverage.php +++ b/tests/bin/patch-coverage.php @@ -177,6 +177,7 @@ function changed_lines($base_ref) { 'flowview_drop_raw_tables.php', 'flowview_filters.php', 'flowview_process.php', + 'flowview_runner.php', 'flowview_schedules.php', 'flowview_upgrade.php', 'flowview_upgrade_nat_columns.php',