612 lines
18 KiB
Go
612 lines
18 KiB
Go
package store
|
|
|
|
import (
|
|
"database/sql"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
auditMigrationBatch = 256
|
|
auditSplitPendingMigration = "split-audit-storage-v1-pending"
|
|
auditSplitDoneMigration = "split-audit-storage-v1"
|
|
)
|
|
|
|
type portableCommandAudit struct {
|
|
id, proxyID, targetID, source, protocol, kind, decision, at string
|
|
content, result []byte
|
|
estimate int64
|
|
}
|
|
|
|
type portableEventAudit struct {
|
|
id int64
|
|
proxyID, source, category, detail, at string
|
|
}
|
|
|
|
// migratePortableAudit moves audit payloads out of the configuration database.
|
|
// Destination commits always happen before narrow source-ID deletes, making an
|
|
// interrupted migration resumable without losing a command or event.
|
|
func (s *Store) migratePortableAudit() error {
|
|
s.auditOps.Lock()
|
|
defer s.auditOps.Unlock()
|
|
|
|
done, err := s.auditMigrationExists(auditSplitDoneMigration)
|
|
if err != nil || done {
|
|
return err
|
|
}
|
|
pending, err := s.auditMigrationExists(auditSplitPendingMigration)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
commands, err := auditSourceTableExists(s.db, "commands")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
results, err := auditSourceTableExists(s.db, "command_results")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
outputs, err := auditSourceTableExists(s.db, "command_outputs")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
events, err := auditSourceTableExists(s.db, "events")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !commands && !results && !outputs && !events {
|
|
if pending {
|
|
if _, err = s.db.Exec("VACUUM"); err != nil {
|
|
return fmt.Errorf("compact configuration database after audit split: %w", err)
|
|
}
|
|
}
|
|
return s.completeAuditSplitMigration()
|
|
}
|
|
if !pending {
|
|
if _, err = s.db.Exec("INSERT INTO migrations(name) VALUES(?)", auditSplitPendingMigration); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if commands {
|
|
if err = s.splitPortableCommands(results, outputs); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if events {
|
|
if err = s.splitPortableEvents(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err = s.dropPortableAuditTables(commands, results, outputs, events); err != nil {
|
|
return err
|
|
}
|
|
// Dropping the payload tables separates the logical stores; VACUUM also
|
|
// returns their pages so linkport.sqlite reflects configuration size.
|
|
if _, err = s.db.Exec("VACUUM"); err != nil {
|
|
return fmt.Errorf("compact configuration database after audit split: %w", err)
|
|
}
|
|
return s.completeAuditSplitMigration()
|
|
}
|
|
|
|
func (s *Store) auditMigrationExists(name string) (bool, error) {
|
|
var count int
|
|
err := s.db.QueryRow("SELECT COUNT(*) FROM migrations WHERE name=?", name).Scan(&count)
|
|
return count != 0, err
|
|
}
|
|
|
|
func (s *Store) completeAuditSplitMigration() error {
|
|
pending, err := s.auditMigrationExists(auditSplitPendingMigration)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if pending {
|
|
result, err := s.db.Exec("DELETE FROM migrations WHERE name=?", auditSplitPendingMigration)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if affected != 1 {
|
|
return fmt.Errorf("complete audit split migration: removed %d pending markers", affected)
|
|
}
|
|
}
|
|
_, err = s.db.Exec("INSERT INTO migrations(name) VALUES(?) ON CONFLICT(name) DO NOTHING", auditSplitDoneMigration)
|
|
return err
|
|
}
|
|
|
|
func auditSourceTableExists(db *sql.DB, name string) (bool, error) {
|
|
var count int
|
|
err := db.QueryRow("SELECT COUNT(*) FROM sqlite_schema WHERE type='table' AND name=?", name).Scan(&count)
|
|
return count != 0, err
|
|
}
|
|
|
|
func (s *Store) splitPortableCommands(hasResults, hasOutputs bool) error {
|
|
locations, err := s.auditCommandLocations()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
batch, err := s.readPortableCommandBatch(hasResults, hasOutputs)
|
|
if err != nil || len(batch) == 0 {
|
|
return err
|
|
}
|
|
for start := 0; start < len(batch); {
|
|
first := batch[start]
|
|
key := locations[first.id]
|
|
end := start + 1
|
|
estimate := max(first.estimate, int64(1))
|
|
if key != "" {
|
|
for end < len(batch) && locations[batch[end].id] == key {
|
|
estimate += max(batch[end].estimate, int64(1))
|
|
end++
|
|
}
|
|
} else {
|
|
date, parseErr := time.Parse(time.RFC3339Nano, first.at)
|
|
if parseErr != nil {
|
|
return parseErr
|
|
}
|
|
key = date.UTC().Format("2006-01-02")
|
|
groupLimit := s.auditMaxBytes / 8
|
|
if groupLimit < 1<<20 {
|
|
groupLimit = 1 << 20
|
|
}
|
|
if groupLimit > 32<<20 {
|
|
groupLimit = 32 << 20
|
|
}
|
|
for end < len(batch) && locations[batch[end].id] == "" {
|
|
nextDate, parseErr := time.Parse(time.RFC3339Nano, batch[end].at)
|
|
if parseErr != nil || nextDate.UTC().Format("2006-01-02") != key || estimate+batch[end].estimate > groupLimit {
|
|
break
|
|
}
|
|
estimate += max(batch[end].estimate, int64(1))
|
|
end++
|
|
}
|
|
}
|
|
var destination *sql.DB
|
|
if locations[first.id] != "" {
|
|
destination, err = s.auditDB(key)
|
|
} else {
|
|
destination, err = s.auditDBForWrite(key, estimate)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
group := batch[start:end]
|
|
if err = s.copyPortableCommandGroup(destination, group, hasOutputs); err != nil {
|
|
return err
|
|
}
|
|
ids := make([]string, 0, len(group))
|
|
for _, row := range group {
|
|
locations[row.id] = key
|
|
ids = append(ids, row.id)
|
|
}
|
|
if err = s.deletePortableCommands(ids, hasResults, hasOutputs); err != nil {
|
|
return err
|
|
}
|
|
start = end
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Store) readPortableCommandBatch(hasResults, hasOutputs bool) ([]portableCommandAudit, error) {
|
|
resultColumn, resultJoin := "NULL", ""
|
|
if hasResults {
|
|
resultColumn = "r.content"
|
|
resultJoin = " LEFT JOIN command_results r ON r.command_id=c.id"
|
|
}
|
|
outputSize := "0"
|
|
if hasOutputs {
|
|
outputSize = "COALESCE((SELECT SUM(length(o.content)+o.size+128) FROM command_outputs o WHERE o.command_id=c.id),0)"
|
|
}
|
|
query := `SELECT c.id,c.proxy_id,c.target_id,c.source,c.protocol,c.kind,c.decision,c.content,c.at,` + resultColumn + `,
|
|
length(c.content)+COALESCE(length(` + resultColumn + `),0)+` + outputSize + `+1024
|
|
FROM commands c` + resultJoin + ` ORDER BY c.at,c.rowid LIMIT ?`
|
|
rows, err := s.db.Query(query, auditMigrationBatch)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
batch := make([]portableCommandAudit, 0, auditMigrationBatch)
|
|
for rows.Next() {
|
|
var row portableCommandAudit
|
|
if err = rows.Scan(&row.id, &row.proxyID, &row.targetID, &row.source, &row.protocol, &row.kind, &row.decision, &row.content, &row.at, &row.result, &row.estimate); err != nil {
|
|
return nil, err
|
|
}
|
|
batch = append(batch, row)
|
|
}
|
|
return batch, rows.Err()
|
|
}
|
|
|
|
func (s *Store) auditCommandLocations() (map[string]string, error) {
|
|
keys, err := s.auditDays()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
locations := map[string]string{}
|
|
for _, key := range keys {
|
|
db, err := s.auditDB(key)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := db.Query("SELECT id FROM commands")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for rows.Next() {
|
|
var id string
|
|
if err = rows.Scan(&id); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
if previous := locations[id]; previous != "" && previous != key {
|
|
rows.Close()
|
|
return nil, fmt.Errorf("command audit %q exists in multiple shards", id)
|
|
}
|
|
locations[id] = key
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return locations, nil
|
|
}
|
|
|
|
func (s *Store) copyPortableCommandGroup(destination *sql.DB, group []portableCommandAudit, hasOutputs bool) error {
|
|
tx, err := destination.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
for _, row := range group {
|
|
result, err := tx.Exec("INSERT OR IGNORE INTO commands(id,proxy_id,target_id,source,protocol,kind,decision,content,at) VALUES(?,?,?,?,?,?,?,?,?)", row.id, row.proxyID, row.targetID, row.source, row.protocol, row.kind, row.decision, row.content, row.at)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if affected, _ := result.RowsAffected(); affected == 0 {
|
|
var matches int
|
|
err = tx.QueryRow("SELECT COUNT(*) FROM commands WHERE id=? AND proxy_id=? AND target_id=? AND source=? AND protocol=? AND kind=? AND decision=? AND content=? AND at=?", row.id, row.proxyID, row.targetID, row.source, row.protocol, row.kind, row.decision, row.content, row.at).Scan(&matches)
|
|
if err != nil || matches != 1 {
|
|
return fmt.Errorf("conflicting command audit %q", row.id)
|
|
}
|
|
}
|
|
if row.result != nil {
|
|
result, err = tx.Exec("INSERT OR IGNORE INTO command_results(command_id,content) VALUES(?,?)", row.id, row.result)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if affected, _ := result.RowsAffected(); affected == 0 {
|
|
var matches int
|
|
err = tx.QueryRow("SELECT COUNT(*) FROM command_results WHERE command_id=? AND content=?", row.id, row.result).Scan(&matches)
|
|
if err != nil || matches != 1 {
|
|
return fmt.Errorf("conflicting command result %q", row.id)
|
|
}
|
|
}
|
|
}
|
|
if hasOutputs {
|
|
if err = s.copyPortableCommandOutputs(tx, row.id); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) copyPortableCommandOutputs(destination *sql.Tx, commandID string) error {
|
|
rows, err := s.db.Query("SELECT id,stream,content,size FROM command_outputs WHERE command_id=? ORDER BY seq", commandID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var id, stream string
|
|
var content []byte
|
|
var size int64
|
|
if err = rows.Scan(&id, &stream, &content, &size); err != nil {
|
|
return err
|
|
}
|
|
result, insertErr := destination.Exec("INSERT OR IGNORE INTO command_outputs(id,command_id,stream,content,size) VALUES(?,?,?,?,?)", id, commandID, stream, content, size)
|
|
if insertErr != nil {
|
|
return insertErr
|
|
}
|
|
if affected, _ := result.RowsAffected(); affected == 0 {
|
|
var matches int
|
|
insertErr = destination.QueryRow("SELECT COUNT(*) FROM command_outputs WHERE id=? AND command_id=? AND stream=? AND content=? AND size=?", id, commandID, stream, content, size).Scan(&matches)
|
|
if insertErr != nil || matches != 1 {
|
|
return fmt.Errorf("conflicting command output %q", id)
|
|
}
|
|
}
|
|
}
|
|
return rows.Err()
|
|
}
|
|
|
|
func (s *Store) deletePortableCommands(ids []string, hasResults, hasOutputs bool) error {
|
|
if len(ids) == 0 {
|
|
return nil
|
|
}
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(ids)), ",")
|
|
args := make([]any, len(ids))
|
|
for index, id := range ids {
|
|
args[index] = id
|
|
}
|
|
if hasOutputs {
|
|
if _, err = tx.Exec("DELETE FROM command_outputs WHERE command_id IN ("+placeholders+")", args...); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if hasResults {
|
|
if _, err = tx.Exec("DELETE FROM command_results WHERE command_id IN ("+placeholders+")", args...); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
result, err := tx.Exec("DELETE FROM commands WHERE id IN ("+placeholders+")", args...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if affected != int64(len(ids)) {
|
|
return fmt.Errorf("delete migrated commands: affected %d of %d", affected, len(ids))
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) splitPortableEvents() error {
|
|
locations, err := s.auditEventLocations()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
batch, err := s.readPortableEventBatch()
|
|
if err != nil || len(batch) == 0 {
|
|
return err
|
|
}
|
|
for start := 0; start < len(batch); {
|
|
date, err := time.Parse(time.RFC3339Nano, batch[start].at)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
day := date.UTC().Format("2006-01-02")
|
|
end := start + 1
|
|
estimate := int64(len(batch[start].detail) + 512)
|
|
for end < len(batch) {
|
|
next, parseErr := time.Parse(time.RFC3339Nano, batch[end].at)
|
|
if parseErr != nil || next.UTC().Format("2006-01-02") != day || estimate+int64(len(batch[end].detail)+512) > 8<<20 {
|
|
break
|
|
}
|
|
estimate += int64(len(batch[end].detail) + 512)
|
|
end++
|
|
}
|
|
group := batch[start:end]
|
|
pending := make([]portableEventAudit, 0, len(group))
|
|
copied := make([]int64, 0, len(group))
|
|
for _, event := range group {
|
|
present, err := s.portableEventAlreadyCopied(event, locations)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if present {
|
|
copied = append(copied, event.id)
|
|
} else {
|
|
pending = append(pending, event)
|
|
}
|
|
}
|
|
if len(pending) > 0 {
|
|
destination, err := s.auditDBForWrite(day, estimate)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = copyPortableEventGroup(destination, pending); err != nil {
|
|
return err
|
|
}
|
|
for _, event := range pending {
|
|
locations[event.id] = append(locations[event.id], day)
|
|
if alias, ok := portableEventAliasID(event.id); ok {
|
|
locations[alias] = append(locations[alias], day)
|
|
}
|
|
copied = append(copied, event.id)
|
|
}
|
|
}
|
|
if err = s.deletePortableEvents(copied); err != nil {
|
|
return err
|
|
}
|
|
start = end
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Store) readPortableEventBatch() ([]portableEventAudit, error) {
|
|
rows, err := s.db.Query("SELECT id,proxy_id,source,category,detail,at FROM events ORDER BY at,id LIMIT ?", auditMigrationBatch)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
batch := make([]portableEventAudit, 0, auditMigrationBatch)
|
|
for rows.Next() {
|
|
var event portableEventAudit
|
|
if err = rows.Scan(&event.id, &event.proxyID, &event.source, &event.category, &event.detail, &event.at); err != nil {
|
|
return nil, err
|
|
}
|
|
batch = append(batch, event)
|
|
}
|
|
return batch, rows.Err()
|
|
}
|
|
|
|
func (s *Store) auditEventLocations() (map[int64][]string, error) {
|
|
keys, err := s.auditDays()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
locations := map[int64][]string{}
|
|
for _, key := range keys {
|
|
db, err := s.auditDB(key)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := db.Query("SELECT id FROM events")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for rows.Next() {
|
|
var id int64
|
|
if err = rows.Scan(&id); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
locations[id] = append(locations[id], key)
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return locations, nil
|
|
}
|
|
|
|
func portableEventAliasID(id int64) (int64, bool) {
|
|
if id <= 0 || id >= auditShardStride/2 {
|
|
return 0, false
|
|
}
|
|
return auditShardStride - id, true
|
|
}
|
|
|
|
func portableEventCandidateIDs(id int64) ([]int64, error) {
|
|
if id <= 0 || id >= auditShardStride {
|
|
return nil, fmt.Errorf("event audit id %d exceeds shard cursor range", id)
|
|
}
|
|
ids := []int64{id}
|
|
if alias, ok := portableEventAliasID(id); ok {
|
|
ids = append(ids, alias)
|
|
}
|
|
return ids, nil
|
|
}
|
|
|
|
func (s *Store) portableEventAlreadyCopied(event portableEventAudit, locations map[int64][]string) (bool, error) {
|
|
ids, err := portableEventCandidateIDs(event.id)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
keys := map[string]bool{}
|
|
for _, id := range ids {
|
|
for _, key := range locations[id] {
|
|
keys[key] = true
|
|
}
|
|
}
|
|
for key := range keys {
|
|
db, err := s.auditDB(key)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
var matches int
|
|
err = db.QueryRow("SELECT COUNT(*) FROM events WHERE id IN (?,?) AND proxy_id=? AND source=? AND category=? AND detail=? AND at=?", ids[0], ids[len(ids)-1], event.proxyID, event.source, event.category, event.detail, event.at).Scan(&matches)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if matches == 1 {
|
|
return true, nil
|
|
}
|
|
if matches > 1 {
|
|
return false, fmt.Errorf("event audit %d exists more than once in shard %s", event.id, key)
|
|
}
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
func copyPortableEventGroup(destination *sql.DB, events []portableEventAudit) error {
|
|
tx, err := destination.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
for _, event := range events {
|
|
ids, err := portableEventCandidateIDs(event.id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
result, err := tx.Exec("INSERT OR IGNORE INTO events(id,proxy_id,source,category,detail,at) VALUES(?,?,?,?,?,?)", event.id, event.proxyID, event.source, event.category, event.detail, event.at)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if affected, _ := result.RowsAffected(); affected == 0 {
|
|
var matches int
|
|
err = tx.QueryRow("SELECT COUNT(*) FROM events WHERE id=? AND proxy_id=? AND source=? AND category=? AND detail=? AND at=?", event.id, event.proxyID, event.source, event.category, event.detail, event.at).Scan(&matches)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if matches == 0 {
|
|
if len(ids) != 2 {
|
|
return fmt.Errorf("conflicting event audit %d", event.id)
|
|
}
|
|
result, err = tx.Exec("INSERT OR IGNORE INTO events(id,proxy_id,source,category,detail,at) VALUES(?,?,?,?,?,?)", ids[1], event.proxyID, event.source, event.category, event.detail, event.at)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if affected, _ := result.RowsAffected(); affected == 0 {
|
|
err = tx.QueryRow("SELECT COUNT(*) FROM events WHERE id=? AND proxy_id=? AND source=? AND category=? AND detail=? AND at=?", ids[1], event.proxyID, event.source, event.category, event.detail, event.at).Scan(&matches)
|
|
if err != nil || matches != 1 {
|
|
return fmt.Errorf("conflicting event audit %d", event.id)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) deletePortableEvents(ids []int64) error {
|
|
if len(ids) == 0 {
|
|
return nil
|
|
}
|
|
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(ids)), ",")
|
|
args := make([]any, len(ids))
|
|
for index, id := range ids {
|
|
args[index] = id
|
|
}
|
|
result, err := s.db.Exec("DELETE FROM events WHERE id IN ("+placeholders+")", args...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
affected, err := result.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if affected != int64(len(ids)) {
|
|
return fmt.Errorf("delete migrated events: affected %d of %d", affected, len(ids))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) dropPortableAuditTables(commands, results, outputs, events bool) error {
|
|
for _, item := range []struct {
|
|
name string
|
|
exists bool
|
|
}{{"command_outputs", outputs}, {"command_results", results}, {"commands", commands}, {"events", events}} {
|
|
if !item.exists {
|
|
continue
|
|
}
|
|
var count int64
|
|
if err := s.db.QueryRow("SELECT COUNT(*) FROM " + item.name).Scan(&count); err != nil {
|
|
return err
|
|
}
|
|
if count != 0 {
|
|
return fmt.Errorf("cannot remove non-empty migrated audit table %s", item.name)
|
|
}
|
|
if _, err := s.db.Exec("DROP TABLE " + item.name); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|