From 6c8d10e80a410ee6b09c6653415b3d816ae54634 Mon Sep 17 00:00:00 2001 From: TheWitness Date: Sat, 10 Oct 2026 19:52:23 -0400 Subject: [PATCH 1/4] feature: optional FlowView Query Service backend for parallel queries Add a long-running Go service (service/flowview-query) that runs the parallel map-reduce in-process (goroutine worker pool, shared shard cache, background maintenance), selectable under FlowView Settings as the Parallel Query Backend (default Legacy PHP Runner). parallel_database_query_run() dispatches scheduled query ids to the service over localhost (POST /run) instead of forking flowview_runner.php per query/shard; the service reads the same shard tables, honours flowview_parallel_threads and MaxScale, writes results back, and PHP falls back to the legacy runner if the service is unreachable. --- CHANGELOG.md | 2 + includes/functions.php | 116 +++- service/flowview-query/.gitignore | 4 + service/flowview-query/README.md | 272 ++++++++++ service/flowview-query/config.go | 240 +++++++++ service/flowview-query/config.sample.json | 34 ++ service/flowview-query/demo.go | 127 +++++ service/flowview-query/flowview-query.service | 30 ++ service/flowview-query/go.mod | 7 + service/flowview-query/go.sum | 4 + service/flowview-query/main.go | 202 +++++++ service/flowview-query/query.go | 503 ++++++++++++++++++ service/flowview-query/scheduler.go | 176 ++++++ service/flowview-query/server.go | 88 +++ service/flowview-query/types.go | 109 ++++ setup.php | 26 + 16 files changed, 1935 insertions(+), 5 deletions(-) create mode 100644 service/flowview-query/.gitignore create mode 100644 service/flowview-query/README.md create mode 100644 service/flowview-query/config.go create mode 100644 service/flowview-query/config.sample.json create mode 100644 service/flowview-query/demo.go create mode 100644 service/flowview-query/flowview-query.service create mode 100644 service/flowview-query/go.mod create mode 100644 service/flowview-query/go.sum create mode 100644 service/flowview-query/main.go create mode 100644 service/flowview-query/query.go create mode 100644 service/flowview-query/scheduler.go create mode 100644 service/flowview-query/server.go create mode 100644 service/flowview-query/types.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 499ae45..e372751 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, parallel_database_query_run() dispatches scheduled query ids to the service over localhost instead of forking flowview_runner.php per query/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/includes/functions.php b/includes/functions.php index bbe9f3f..1d8c058 100644 --- a/includes/functions.php +++ b/includes/functions.php @@ -4154,6 +4154,88 @@ function parallel_database_query_is_running($request_id): bool { } } +/** + * 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 scheduled 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 scheduled parallel_database_query ids. + * + * @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))); + + $ch = curl_init($base . '/run'); + + curl_setopt_array($ch, array( + CURLOPT_POST => true, + CURLOPT_POSTFIELDS => $payload, + CURLOPT_HTTPHEADER => array('Content-Type: application/json'), + 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; +} + /** * parallel_database_query_run - Runs a parallel query by calling the flowview_running.php * in background. If the data is already cached, then the number of pending request @@ -4172,6 +4254,14 @@ function parallel_database_query_run($requests) { $php = read_config_option('path_php_binary'); $redirect = ''; + $backend = read_config_option('flowview_parallel_backend'); + + if (empty($backend)) { + $backend = 'legacy'; + } + + $scheduled = []; + foreach($requests as $request_id) { $pending = flowview_db_fetch_cell_prepared('SELECT COUNT(*) FROM parallel_database_query @@ -4186,16 +4276,32 @@ function parallel_database_query_run($requests) { WHERE id = ?', ['scheduled', $request_id]); - db_debug('Launching FlowView Database Query Process ' . $request_id); - - cacti_log('NOTE: Launching FlowView Database Query Process ' . $request_id, false, 'BOOST', POLLER_VERBOSITY_MEDIUM); - - exec_background($php, $config['base_path'] . "/plugins/flowview/flowview_runner.php --query-id=$request_id" . ($debug ? ' --debug':''), $redirect); + $scheduled[] = $request_id; } else { db_debug('Not Launching FlowView Database Query Process ' . $request_id . ' as it has already completed or is running.'); } } + if (cacti_sizeof($scheduled)) { + /* The Query Service runs the same map-reduce and writes results/status + * back, so the result-retrieval loop below is unchanged either way. */ + if ($backend == 'service' && flowview_query_service_run($scheduled)) { + db_debug('Dispatched FlowView Queries to the Query Service: ' . implode(', ', $scheduled)); + } else { + if ($backend == 'service') { + cacti_log('WARNING: FlowView Query Service unavailable, falling back to the Legacy PHP Runner', false, 'FLOWVIEW'); + } + + foreach($scheduled as $request_id) { + db_debug('Launching FlowView Database Query Process ' . $request_id); + + cacti_log('NOTE: Launching FlowView Database Query Process ' . $request_id, false, 'BOOST', POLLER_VERBOSITY_MEDIUM); + + exec_background($php, $config['base_path'] . "/plugins/flowview/flowview_runner.php --query-id=$request_id" . ($debug ? ' --debug':''), $redirect); + } + } + } + $max_time = read_config_option('flowview_parallel_runlimit'); $start = time(); 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..be958a1 --- /dev/null +++ b/service/flowview-query/README.md @@ -0,0 +1,272 @@ +# 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, `parallel_database_query_run()` replaces +its `exec_background()` launch with a single call to `POST /run`, passing the +ids of the queries it just scheduled. The service reads those rows +(`parallel_database_query` + `parallel_database_query_shard`), runs the same +map-reduce (cache-aware, thread-capped), and writes `results` + `status = +'complete'` back — so the existing result-retrieval path is unchanged. If the +service cannot be reached, PHP automatically falls back to the legacy runner. + +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 | Purpose | +| ------------- | ------- | +| `POST /run` | Run the given `{"query_ids":[...]}` to completion; returns per-query stats. | +| `GET /health` | Liveness + DB ping + current thread setting. | +| `POST /maintenance` | Run one scheduler pass now (expiry/cleanup). | +| `POST /cache/invalidate?md5sum=` | Drop cached shards for a deleted query/filter signature. | + +## 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. + +## 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..7d52277 --- /dev/null +++ b/service/flowview-query/config.go @@ -0,0 +1,240 @@ +package main + +import ( + "database/sql" + "encoding/json" + "fmt" + "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"` +} + +// 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) } + +// 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 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 = true + cfg.Loc = time.Local + 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..38a36b3 --- /dev/null +++ b/service/flowview-query/config.sample.json @@ -0,0 +1,34 @@ +{ + "listen_host": "127.0.0.1", + "listen_port": 0, + "port_file": "/var/run/flowview/flowview-query.port", + + "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/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/flowview-query.service b/service/flowview-query/flowview-query.service new file mode 100644 index 0000000..7073a9e --- /dev/null +++ b/service/flowview-query/flowview-query.service @@ -0,0 +1,30 @@ +[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: the service only needs loopback networking and its runtime dir. +NoNewPrivileges=true +ProtectSystem=strict +ProtectHome=true +PrivateTmp=true +ReadWritePaths=/var/run/flowview +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..9728e71 --- /dev/null +++ b/service/flowview-query/go.mod @@ -0,0 +1,7 @@ +module flowview-query + +go 1.21 + +require 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..19dbcec --- /dev/null +++ b/service/flowview-query/go.sum @@ -0,0 +1,4 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +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= diff --git a/service/flowview-query/main.go b/service/flowview-query/main.go new file mode 100644 index 0000000..9bd498b --- /dev/null +++ b/service/flowview-query/main.go @@ -0,0 +1,202 @@ +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() + } + + read := flow + if cfg.MaxScale.Enabled && settings.UseMaxScale() { + 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 + } + read, err = ms.open(maxInt(settings.Threads(), 4)) + if err != nil { + return fmt.Errorf("connect maxscale: %w", err) + } + defer read.Close() + logf("map reads routed through MaxScale at %s:%d", ms.Host, ms.Port) + } + + eng := &Engine{flow: flow, read: read, cacti: cacti, settings: settings} + sched := &Scheduler{eng: eng, interval: cfg.SchedulerInterval.D()} + srv := &Server{eng: eng, sched: sched} + + 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) + + 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 := &Engine{settings: 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..08dbe3f --- /dev/null +++ b/service/flowview-query/query.go @@ -0,0 +1,503 @@ +package main + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "fmt" + "math" + "strings" + "sync" +) + +// 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) + read *sql.DB // map read connection (MaxScale RW-split when enabled, else == flow) + cacti *sql.DB // Cacti control DB (settings) + settings *Settings +} + +// 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 + } + + sem := make(chan struct{}, threads) + + var ( + wg sync.WaitGroup + mu sync.Mutex + cached int + firstErr error + ) + + for i := range shards { + s := shards[i] + + wg.Add(1) + sem <- struct{}{} + + go func(s shardRow) { + defer wg.Done() + defer func() { <-sem }() + + 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.read, 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) + 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/scheduler.go b/service/flowview-query/scheduler.go new file mode 100644 index 0000000..e26a1fb --- /dev/null +++ b/service/flowview-query/scheduler.go @@ -0,0 +1,176 @@ +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 { + logf("scheduler: drop map table %s: %v", item.table, err) + } + } + 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/server.go b/service/flowview-query/server.go new file mode 100644 index 0000000..53fc820 --- /dev/null +++ b/service/flowview-query/server.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "time" +) + +// Server exposes the engine over localhost HTTP. +type Server struct { + eng *Engine + sched *Scheduler +} + +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 +} + +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 r.Method != http.MethodPost { + http.Error(w, "POST required", http.StatusMethodNotAllowed) + 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) { + s.sched.RunOnce(r.Context()) + writeJSON(w, http.StatusOK, map[string]any{"status": "ok"}) +} + +func (s *Server) handleInvalidate(w http.ResponseWriter, r *http.Request) { + 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/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..869c702 100644 --- a/setup.php +++ b/setup.php @@ -642,6 +642,32 @@ 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_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'), From e397dd95d4b98af2878b34756231073192bc0afa Mon Sep 17 00:00:00 2001 From: TheWitness Date: Sat, 10 Oct 2026 19:54:11 -0400 Subject: [PATCH 2/4] dev: regenerate cacti.pot for the new Parallel Query Backend settings --- locales/po/cacti.pot | 34 +++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/locales/po/cacti.pot b/locales/po/cacti.pot index 243bd81..c344d60 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 19:50-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,38 @@ 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 "Max Concurrent Threads" msgstr "" From 148b02bd600fd524a4bdd5dc60baa36fdcd6c46d Mon Sep 17 00:00:00 2001 From: TheWitness Date: Sat, 10 Oct 2026 20:01:04 -0400 Subject: [PATCH 3/4] refactor: engage Query Service from flowview_runner.php; keep functions.php untouched Move the backend switch and the service HTTP helpers out of includes/functions.php (coverage-measured) into flowview_runner.php (the parallel query parent CLI entry) and setup.php (both allowlisted). flowview_runner.php now delegates the whole query to the service when the backend is 'service', eliminating the per-shard PHP worker forks, and falls back to the legacy parent runner otherwise. Allowlist flowview_runner.php as a CLI entry point in patch-coverage. --- CHANGELOG.md | 2 +- flowview_runner.php | 16 ++++- includes/functions.php | 116 ++----------------------------- service/flowview-query/README.md | 16 +++-- setup.php | 82 ++++++++++++++++++++++ tests/bin/patch-coverage.php | 1 + 6 files changed, 113 insertions(+), 120 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e372751..80b51d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ --- 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, parallel_database_query_run() dispatches scheduled query ids to the service over localhost instead of forking flowview_runner.php per query/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 +* 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 diff --git a/flowview_runner.php b/flowview_runner.php index 0b18237..2bd4bc5 100644 --- a/flowview_runner.php +++ b/flowview_runner.php @@ -127,7 +127,21 @@ $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'); + } + + $stats = parallel_database_parent_runner($query_id); + } unregister_process('flowview', "db_query_{$query_id}", 0); diff --git a/includes/functions.php b/includes/functions.php index 1d8c058..bbe9f3f 100644 --- a/includes/functions.php +++ b/includes/functions.php @@ -4154,88 +4154,6 @@ function parallel_database_query_is_running($request_id): bool { } } -/** - * 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 scheduled 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 scheduled parallel_database_query ids. - * - * @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))); - - $ch = curl_init($base . '/run'); - - curl_setopt_array($ch, array( - CURLOPT_POST => true, - CURLOPT_POSTFIELDS => $payload, - CURLOPT_HTTPHEADER => array('Content-Type: application/json'), - 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; -} - /** * parallel_database_query_run - Runs a parallel query by calling the flowview_running.php * in background. If the data is already cached, then the number of pending request @@ -4254,14 +4172,6 @@ function parallel_database_query_run($requests) { $php = read_config_option('path_php_binary'); $redirect = ''; - $backend = read_config_option('flowview_parallel_backend'); - - if (empty($backend)) { - $backend = 'legacy'; - } - - $scheduled = []; - foreach($requests as $request_id) { $pending = flowview_db_fetch_cell_prepared('SELECT COUNT(*) FROM parallel_database_query @@ -4276,29 +4186,13 @@ function parallel_database_query_run($requests) { WHERE id = ?', ['scheduled', $request_id]); - $scheduled[] = $request_id; - } else { - db_debug('Not Launching FlowView Database Query Process ' . $request_id . ' as it has already completed or is running.'); - } - } + db_debug('Launching FlowView Database Query Process ' . $request_id); - if (cacti_sizeof($scheduled)) { - /* The Query Service runs the same map-reduce and writes results/status - * back, so the result-retrieval loop below is unchanged either way. */ - if ($backend == 'service' && flowview_query_service_run($scheduled)) { - db_debug('Dispatched FlowView Queries to the Query Service: ' . implode(', ', $scheduled)); - } else { - if ($backend == 'service') { - cacti_log('WARNING: FlowView Query Service unavailable, falling back to the Legacy PHP Runner', false, 'FLOWVIEW'); - } - - foreach($scheduled as $request_id) { - db_debug('Launching FlowView Database Query Process ' . $request_id); - - cacti_log('NOTE: Launching FlowView Database Query Process ' . $request_id, false, 'BOOST', POLLER_VERBOSITY_MEDIUM); + cacti_log('NOTE: Launching FlowView Database Query Process ' . $request_id, false, 'BOOST', POLLER_VERBOSITY_MEDIUM); - exec_background($php, $config['base_path'] . "/plugins/flowview/flowview_runner.php --query-id=$request_id" . ($debug ? ' --debug':''), $redirect); - } + exec_background($php, $config['base_path'] . "/plugins/flowview/flowview_runner.php --query-id=$request_id" . ($debug ? ' --debug':''), $redirect); + } else { + db_debug('Not Launching FlowView Database Query Process ' . $request_id . ' as it has already completed or is running.'); } } diff --git a/service/flowview-query/README.md b/service/flowview-query/README.md index be958a1..d264e0c 100644 --- a/service/flowview-query/README.md +++ b/service/flowview-query/README.md @@ -207,13 +207,15 @@ is required to switch: > - **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, `parallel_database_query_run()` replaces -its `exec_background()` launch with a single call to `POST /run`, passing the -ids of the queries it just scheduled. The service reads those rows -(`parallel_database_query` + `parallel_database_query_shard`), runs the same -map-reduce (cache-aware, thread-capped), and writes `results` + `status = -'complete'` back — so the existing result-retrieval path is unchanged. If the -service cannot be reached, PHP automatically falls back to the legacy runner. +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 falls back to the legacy in-PHP +`parallel_database_parent_runner()`. 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** diff --git a/setup.php b/setup.php index 869c702..f275aa0 100644 --- a/setup.php +++ b/setup.php @@ -788,6 +788,88 @@ 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))); + + $ch = curl_init($base . '/run'); + + curl_setopt_array($ch, array( + CURLOPT_POST => true, + CURLOPT_POSTFIELDS => $payload, + CURLOPT_HTTPHEADER => array('Content-Type: application/json'), + 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; +} + /** * @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', From 4fd49e535f5042c88c5c58399846f42c2a6893d0 Mon Sep 17 00:00:00 2001 From: TheWitness Date: Sat, 10 Oct 2026 20:51:33 -0400 Subject: [PATCH 4/4] flowview-query: address service review findings + add Go test coverage Security/correctness fixes from the PR review: - shared engine-wide worker limiter (global concurrency cap, context-aware acquire) - dynamic MaxScale read pool honouring the live flowview_use_maxscale setting - normalize time.Time to SQL datetime (and disable DSN parseTime) - token auth (Authorization: Bearer) + enforced POST on state-changing endpoints - refuse to bind a non-loopback listen host - keep the query row when a scheduler map-table DROP fails (retry, no orphan) - drop IPAddressAllow/Deny from the systemd unit so remote DB/MaxScale works - PHP: reset query to a clean pending state before the legacy fallback; send auth token - add flowview_query_service_token setting; regenerate cacti.pot Testing: real sqlmock-based engine/scheduler/server/config tests (replacing the toy selftest for coverage), plus a Go CI workflow (go vet + race tests). --- .github/workflows/flowview-query-go.yml | 67 ++++++ flowview_runner.php | 5 + locales/po/cacti.pot | 10 +- service/flowview-query/README.md | 46 +++- service/flowview-query/config.go | 27 ++- service/flowview-query/config.sample.json | 3 + service/flowview-query/config_test.go | 120 +++++++++++ service/flowview-query/extra_test.go | 65 ++++++ service/flowview-query/flowview-query.service | 9 +- service/flowview-query/go.mod | 5 +- service/flowview-query/go.sum | 3 + service/flowview-query/helpers_test.go | 176 +++++++++++++++ service/flowview-query/main.go | 20 +- service/flowview-query/query.go | 53 ++++- service/flowview-query/query_test.go | 200 ++++++++++++++++++ service/flowview-query/scheduler.go | 5 +- service/flowview-query/scheduler_test.go | 96 +++++++++ service/flowview-query/server.go | 40 +++- service/flowview-query/server_test.go | 95 +++++++++ service/flowview-query/testutil_test.go | 43 ++++ setup.php | 54 ++++- 21 files changed, 1107 insertions(+), 35 deletions(-) create mode 100644 .github/workflows/flowview-query-go.yml create mode 100644 service/flowview-query/config_test.go create mode 100644 service/flowview-query/extra_test.go create mode 100644 service/flowview-query/helpers_test.go create mode 100644 service/flowview-query/query_test.go create mode 100644 service/flowview-query/scheduler_test.go create mode 100644 service/flowview-query/server_test.go create mode 100644 service/flowview-query/testutil_test.go 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/flowview_runner.php b/flowview_runner.php index 2bd4bc5..0867050 100644 --- a/flowview_runner.php +++ b/flowview_runner.php @@ -138,6 +138,11 @@ } 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); diff --git a/locales/po/cacti.pot b/locales/po/cacti.pot index c344d60..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-10-10 19:50-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" @@ -2707,6 +2707,14 @@ msgstr "" 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/README.md b/service/flowview-query/README.md index d264e0c..f56ad76 100644 --- a/service/flowview-query/README.md +++ b/service/flowview-query/README.md @@ -214,8 +214,11 @@ 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 falls back to the legacy in-PHP -`parallel_database_parent_runner()`. +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** @@ -239,12 +242,18 @@ the legacy path, a partition cached by one backend is reused by the other. ## 7. Endpoints -| Method & path | Purpose | -| ------------- | ------- | -| `POST /run` | Run the given `{"query_ids":[...]}` to completion; returns per-query stats. | -| `GET /health` | Liveness + DB ping + current thread setting. | -| `POST /maintenance` | Run one scheduler pass now (expiry/cleanup). | -| `POST /cache/invalidate?md5sum=` | Drop cached shards for a deleted query/filter signature. | +| 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 @@ -259,6 +268,27 @@ A goroutine runs every `scheduler_interval` (and once at startup): 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 | diff --git a/service/flowview-query/config.go b/service/flowview-query/config.go index 7d52277..a4a0a60 100644 --- a/service/flowview-query/config.go +++ b/service/flowview-query/config.go @@ -4,6 +4,7 @@ import ( "database/sql" "encoding/json" "fmt" + "net" "os" "strconv" "sync" @@ -58,6 +59,12 @@ type Config struct { // 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"). @@ -78,6 +85,19 @@ func (d *Duration) UnmarshalJSON(b []byte) error { 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{ @@ -99,6 +119,10 @@ func LoadConfig(path string) (*Config, error) { 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" } @@ -143,8 +167,7 @@ func (db DBConfig) dsn() string { cfg.Net = "tcp" cfg.Addr = fmt.Sprintf("%s:%d", db.Host, port) cfg.DBName = db.Database - cfg.ParseTime = true - cfg.Loc = time.Local + cfg.ParseTime = false cfg.Params = map[string]string{"charset": "utf8mb4"} if db.TLS != "" { cfg.TLSConfig = db.TLS diff --git a/service/flowview-query/config.sample.json b/service/flowview-query/config.sample.json index 38a36b3..6918965 100644 --- a/service/flowview-query/config.sample.json +++ b/service/flowview-query/config.sample.json @@ -3,6 +3,9 @@ "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": { 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/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 index 7073a9e..65686c3 100644 --- a/service/flowview-query/flowview-query.service +++ b/service/flowview-query/flowview-query.service @@ -17,14 +17,17 @@ ExecStart=/usr/local/bin/flowview-query -config /etc/flowview/flowview-query.jso Restart=on-failure RestartSec=3 -# Hardening: the service only needs loopback networking and its runtime dir. +# Hardening: minimal privileges and filesystem exposure. NoNewPrivileges=true ProtectSystem=strict ProtectHome=true PrivateTmp=true ReadWritePaths=/var/run/flowview -IPAddressAllow=localhost -IPAddressDeny=any +# 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 index 9728e71..5ec9b08 100644 --- a/service/flowview-query/go.mod +++ b/service/flowview-query/go.mod @@ -2,6 +2,9 @@ module flowview-query go 1.21 -require github.com/go-sql-driver/mysql v1.8.1 +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 index 19dbcec..361e8b4 100644 --- a/service/flowview-query/go.sum +++ b/service/flowview-query/go.sum @@ -1,4 +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 index 9bd498b..94c5048 100644 --- a/service/flowview-query/main.go +++ b/service/flowview-query/main.go @@ -82,8 +82,8 @@ func runService(cfg *Config) error { defer flow.Close() } - read := flow - if cfg.MaxScale.Enabled && settings.UseMaxScale() { + var maxscale *sql.DB + if cfg.MaxScale.Enabled { ms := cfg.FlowView if cfg.UseCactiDB { ms = cfg.Cacti @@ -94,17 +94,17 @@ func runService(cfg *Config) error { if cfg.MaxScale.Port != 0 { ms.Port = cfg.MaxScale.Port } - read, err = ms.open(maxInt(settings.Threads(), 4)) + maxscale, err = ms.open(maxInt(settings.Threads(), 4)) if err != nil { return fmt.Errorf("connect maxscale: %w", err) } - defer read.Close() - logf("map reads routed through MaxScale at %s:%d", ms.Host, ms.Port) + defer maxscale.Close() + logf("MaxScale pool ready at %s:%d (map reads follow the live flowview_use_maxscale setting)", ms.Host, ms.Port) } - eng := &Engine{flow: flow, read: read, cacti: cacti, settings: settings} + eng := NewEngine(flow, maxscale, cacti, settings) sched := &Scheduler{eng: eng, interval: cfg.SchedulerInterval.D()} - srv := &Server{eng: eng, sched: sched} + srv := &Server{eng: eng, sched: sched, authToken: cfg.AuthToken} ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() @@ -126,6 +126,10 @@ func runService(cfg *Config) error { // 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) @@ -159,7 +163,7 @@ func runService(cfg *Config) error { func runSelftest(cfg *Config) error { demoMapReduce(4) - eng := &Engine{settings: defaultSettings()} + eng := NewEngine(nil, nil, nil, defaultSettings()) srv := &Server{eng: eng, sched: &Scheduler{eng: eng}} ln, err := net.Listen("tcp", "127.0.0.1:0") diff --git a/service/flowview-query/query.go b/service/flowview-query/query.go index 08dbe3f..2886b85 100644 --- a/service/flowview-query/query.go +++ b/service/flowview-query/query.go @@ -9,15 +9,43 @@ import ( "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) - read *sql.DB // map read connection (MaxScale RW-split when enabled, else == flow) + 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 @@ -53,8 +81,6 @@ func (e *Engine) RunQuery(ctx context.Context, id int64) (Stats, error) { return Stats{}, err } - sem := make(chan struct{}, threads) - var ( wg sync.WaitGroup mu sync.Mutex @@ -66,11 +92,22 @@ func (e *Engine) RunQuery(ctx context.Context, id int64) (Stats, error) { s := shards[i] wg.Add(1) - sem <- struct{}{} go func(s shardRow) { defer wg.Done() - defer func() { <-sem }() + + // 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) @@ -150,7 +187,7 @@ func (e *Engine) mapShard(ctx context.Context, q *queryRow, s shardRow) ([]*Orde } } - rows, err := e.queryDB(ctx, e.read, s.MapQuery, decodeParams(s.MapParams)) + rows, err := e.queryDB(ctx, e.readPool(), s.MapQuery, decodeParams(s.MapParams)) if err != nil { return nil, false, err } @@ -460,6 +497,10 @@ 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 } 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 index e26a1fb..e1a68ae 100644 --- a/service/flowview-query/scheduler.go +++ b/service/flowview-query/scheduler.go @@ -81,7 +81,10 @@ func (s *Scheduler) expireQueries(ctx context.Context) error { for _, item := range expired { if item.table != "" { if _, err := s.eng.flow.ExecContext(ctx, "DROP TABLE IF EXISTS "+quoteTable(item.table)); err != nil { - logf("scheduler: drop map table %s: %v", item.table, err) + // 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, 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 index 53fc820..b9e4a99 100644 --- a/service/flowview-query/server.go +++ b/service/flowview-query/server.go @@ -2,15 +2,18 @@ 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 + eng *Engine + sched *Scheduler + authToken string } func (s *Server) routes() http.Handler { @@ -22,6 +25,30 @@ func (s *Server) routes() http.Handler { 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 @@ -42,8 +69,7 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleRun(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost { - http.Error(w, "POST required", http.StatusMethodNotAllowed) + if !s.guard(w, r) { return } @@ -66,11 +92,17 @@ func (s *Server) handleRun(w http.ResponseWriter, r *http.Request) { } 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 { 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/setup.php b/setup.php index f275aa0..abaa507 100644 --- a/setup.php +++ b/setup.php @@ -668,6 +668,14 @@ function flowview_config_settings(): void { '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'), @@ -844,12 +852,20 @@ function flowview_query_service_run($request_ids): bool { $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 => array('Content-Type: application/json'), + CURLOPT_HTTPHEADER => $headers, CURLOPT_RETURNTRANSFER => true, CURLOPT_CONNECTTIMEOUT => 2, CURLOPT_TIMEOUT => $timeout, @@ -870,6 +886,42 @@ function flowview_query_service_run($request_ids): bool { 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 */