Class ::nsmcp::AccessRecordPartitions (public)

 ::nx::Trait ::nsmcp::AccessRecordPartitions

Defined in /usr/local/ns/tcl/nsmcp/log-traffic-record-01-partitions.tcl

Testcases:

No testcase defined.

Source code:

    :method partitionOpen {} {
        set stored [:stateGet partitionInterval 0]
        if {$stored!=${:partitionInterval} && ($stored>0 || ${:partitionInterval}>0)} {
            if {$stored>0} {
                error "record-store-partition-reset-required"
            }
            if {${:mode} ne "writer"} {
                error "record-store-partition-reset-required"
            }
            ${:db} transaction {
                # Check under the writer lock; never convert populated caches.
                if {[${:db} scalar {SELECT EXISTS(SELECT 1 FROM main.rows LIMIT 1)
                    OR EXISTS(SELECT 1 FROM main.files LIMIT 1)}]} {
                    error "record-store-partition-reset-required"
                }
                ${:db} execute {
                    ALTER TABLE main.rows RENAME TO row_template;
                    CREATE TABLE main.partitions(
                        start INTEGER PRIMARY KEY,stop INTEGER NOT NULL,name TEXT UNIQUE NOT NULL,
                        n INTEGER NOT NULL DEFAULT 0,first REAL,last REAL)
                }
                ${:db} execute {UPDATE main.meta SET version=4}
                :statePut partitionInterval ${:partitionInterval}
                :statePut rollupsReady 1
                set bootstrap [:dict getdef ${:settings} bootstrapSeconds 172800]
                :statePut bootstrapFrom [expr {[clock seconds]-min($bootstrap,[dict get ${:settings} retention])}]
                :partitionView
            }
        }
        if {${:partitionInterval}>0} {
            set sql [${:db} scalar {SELECT sql FROM main.sqlite_schema WHERE name='row_template' AND type='table'}]
            if {$sql eq {} || [string first ( $sql]<0} {
                error "record-store-partition-reset-required"
            }
            set :partitionDDL [string range $sql [string first ( $sql] end]
        }
    }

    :method partitionView {} {
        set sources [list {SELECT * FROM row_template WHERE 0}]
        set parts [${:db} execute {SELECT name FROM main.partitions ORDER BY start}]
        if {[llength $parts]>256} {
            error "record-store-partition-limit"
        }
        foreach part $parts {
            set name [dict get $part name]
            if {![regexp {^p_[0-9]+$} $name]} {
                error "record-store-invalid-partition"
            }
            lappend sources "SELECT * FROM $name"
        }
        ${:db} execute {DROP VIEW IF EXISTS main.rows}
        ${:db} execute "CREATE VIEW main.rows AS [join $sources { UNION ALL }]"
    }

    :method partitionEnsure {start} {
        set stop [expr {$start+${:partitionInterval}}]
        set name p_$start
        if {![${:db} scalar {SELECT EXISTS(SELECT 1 FROM main.partitions WHERE start=:start)}]} {
            if {[${:db} scalar {SELECT count(*) FROM main.partitions}]>=256} {
                error "record-store-partition-limit"
            }
            ${:db} execute "CREATE TABLE main.$name ${:partitionDDL}"
            ${:db} execute "
                CREATE INDEX main.${name}_time_status ON ${name}(t,status,fid,off,kind);
                CREATE INDEX main.${name}_correlation ON ${name}(identifier,t)
            "
            ${:db} execute {INSERT INTO main.partitions(start,stop,name) VALUES(:start,:stop,:name)}
            :partitionView
        }
        return main.$name
    }

    :method partitionIngest {fid checkpoint tuples batched} {
        set groups {}
        set firsts {}
        set lasts {}
        set floor [:retentionFloor]
        if {![:dict getdef ${:settings} historicalBackfill false]} {
            set floor [expr {max($floor,[:stateGet bootstrapFrom $floor])}]
        }
        set skipped 0
        foreach tuple $tuples {
            if {$batched} {
                set time [lindex $tuple 4]
                if {[lindex $tuple 3]!=0} {
                    error "partitioned ingestion requires valid requests"
                }
            } else {
                if {![dict exists ${:tupleTimes} $tuple]} {
                    error "partitioned SQL ingestion requires tuple routing metadata"
                }
                set time [dict get ${:tupleTimes} $tuple]
            }
            if {![string is double -strict $time] || $time<0 || $time>[clock seconds]+3600} {
                error "record-store-invalid-partition-time"
            }
            if {$time<$floor} {
                incr skipped
                continue
            }
            set bucket [expr {wide(floor($time/${:partitionInterval}))*${:partitionInterval}}]
            dict lappend groups $bucket $tuple
            if {![dict exists $firsts $bucket]} {
                dict set firsts $bucket $time
                dict set lasts $bucket $time
            } else {
                dict set firsts $bucket [expr {min([dict get $firsts $bucket],$time)}]
                dict set lasts $bucket [expr {max([dict get $lasts $bucket],$time)}]
            }
        }
        dict set checkpoint skipped [expr {[:dict getdef $checkpoint skipped 0]+$skipped}]
        if {$groups eq {}} {
            dict set groups empty {}
        }
        set committed {records 0 requestRecords 0 malformedRecords 0 oversizedRecords 0}
        set remaining [dict size $groups]
        try {
            :accessIngestionObserve begin-transaction
            ${:db} transaction {
                dict for {bucket rows} $groups {
                    set target main.row_template
                    if {$bucket ne "empty"} {
                        set target [:partitionEnsure $bucket]
                    }
                    set :recordTable $target
                    set portion $checkpoint
                    incr remaining -1
                    if {$remaining>0} {
                        dict set portion malformed 0
                        dict set portion oversized 0
                        dict set portion skipped 0
                    }
                    set added [:ingestCanonical $fid $portion $rows $batched true]
                    foreach key {records requestRecords malformedRecords oversizedRecords} {
                        dict incr committed $key [dict get $added $key]
                    }
                    if {$bucket ne "empty"} {
                        set n [dict get $added requestRecords]
                        set first [dict get $firsts $bucket]
                        set last [dict get $lasts $bucket]
                        ${:db} execute {
                            UPDATE main.partitions
                               SET first=CASE WHEN n=0 THEN :first ELSE min(first,:first) END,
                                   last=CASE WHEN n=0 THEN :last ELSE max(last,:last) END,n=n+:n
                             WHERE start=:bucket
                        }
                    }
                }
                :sqliteBeforeCommit
            }
        } finally {
            set :recordTable main.rows
            set :tupleTimes {}
        }
        dict set committed insertionBackend [expr {$batched ? "batch-api" : "sql-values"}]
        return $committed
    }

    :method partitionPrune {batchRows} {
        if {![string is integer -strict $batchRows] || $batchRows<1 || $batchRows>10000} {
            error "retention batch rows must be 1..10000"
        }
        if {[${:db} inTransaction]} {
            error "retention requires an idle database session"
        }
        set cutoff [:retentionFloor]
        set peerCutoff [expr {[clock seconds]-[dict get ${:settings} peerRetention]}]
        set now [clock seconds]
        set counts {requests 0 diagnostics 0 minutes 0 peers 0}
        :accessIngestionObserve prepare-retention
        ${:db} execute {CREATE TEMP TABLE nsmcp_retention_candidates(id INTEGER PRIMARY KEY,n INTEGER)}
        try {
            :accessIngestionObserve begin-transaction
            ${:db} transaction {
                :statePut retentionFloor $cutoff
                :accessIngestionObserve partition-selection
                set parts [${:db} execute {SELECT * FROM main.partitions WHERE stop<=:cutoff ORDER BY start LIMIT 1}]
                if {$parts ne {}} {
                    set part [lindex $parts 0]
                    set name [dict get $part name]
                    if {![regexp {^p_[0-9]+$} $name]} {
                        error "record-store-invalid-partition"
                    }
                    :accessIngestionObserve partition-drop
                    ${:db} execute {DROP VIEW main.rows}
                    ${:db} execute "DROP TABLE main.$name"
                    ${:db} execute {DELETE FROM main.partitions WHERE name=:name}
                    :partitionView
                    dict set counts requests [dict get $part n]
                    set first [dict get $part first]
                    set last [dict get $part last]
                    set n [dict get $part n]
                    ${:db} execute {
                        INSERT INTO main.evictions(dropped_at,first_time,last_time,rows,reason)
                        VALUES(:now,:first,:last,:n,'partition');
                        DELETE FROM main.evictions WHERE id NOT IN(
                            SELECT id FROM main.evictions ORDER BY id DESC LIMIT 64)
                    }
                }
                :accessIngestionObserve diagnostic-deletion
                # Loss metadata is small and independent of request partitions.
                ${:db} execute {
                    INSERT INTO temp.nsmcp_retention_candidates
                    SELECT rowid,malformed+oversized+skipped FROM main.scan_counts
                     WHERE completed<:cutoff ORDER BY completed LIMIT :batchRows
                }
                dict set counts diagnostics [${:db} scalar {SELECT coalesce(sum(n),0) FROM temp.nsmcp_retention_candidates}]
                ${:db} execute {DELETE FROM main.scan_counts WHERE rowid IN(SELECT id FROM temp.nsmcp_retention_candidates)}
                :accessIngestionObserve minute-deletion
                ${:db} execute {DELETE FROM main.minutes WHERE rowid IN(
                    SELECT rowid FROM main.minutes WHERE slot<:cutoff ORDER BY slot LIMIT :batchRows)}
                dict set counts minutes [${:db} changes]
                :accessIngestionObserve peer-deletion
                ${:db} execute {DELETE FROM main.peer_addresses WHERE rowid IN(
                    SELECT rowid FROM main.peer_addresses WHERE seen<:peerCutoff ORDER BY seen LIMIT :batchRows)}
                dict set counts peers [${:db} changes]
                :statePut droppedThrough [expr {$cutoff-1}]
                :statePut lastEvictionAt $now
                :accessIngestionObserve commit
            }
            :accessIngestionObserve committed
        } finally {
            ${:db} execute {DROP TABLE temp.nsmcp_retention_candidates}
        }
        return $counts
    }

    :method partitionStatus {} {
        if {${:partitionInterval}==0} {
            return {partitionInterval 0 bootstrapFrom 0}
        }
        set count [${:db} scalar {SELECT count(*) FROM main.partitions}]
        set bootstrap [:stateGet bootstrapFrom 0]
        set control [:stateGet continuous {}]
        return [dict create partitionInterval ${:partitionInterval} partitionCount $count  bootstrapFrom $bootstrap availableFrom [:dict getdef $control availableFrom $bootstrap]  historicalBackfillEnabled [:dict getdef ${:settings} historicalBackfill false]]
    }

XQL Not present:

Generic, PostgreSQL, Oracle