Class ::nsmcp::AccessCardinalityStore (public)

 ::nx::Class ::nsmcp::AccessCardinalityStore

Defined in /usr/local/ns/tcl/nsmcp/log-traffic-aggregate.tcl

Testcases:

No testcase defined.

Source code:

    :property directory:required
    :property checkpoint:required
    :property limits:required
    :property {readResults:boolean false}
    :variable data
    :variable db {}
    :variable handle {}
    :variable attached false
    :variable buffer {}
    :variable events 0
    :variable peerPending 0
    :variable agentPending 0

    :method init {} {
        set :data ${:checkpoint}
        set fresh [expr {${:data} eq {}}]
        if {$fresh} {
            set :data [dict create backend sqlite status active reason {} path {} owner [ns_crypto::randombytes -encoding hex 16]  pool [dict get ${:limits} aggregatePool] maxBytes [dict get ${:limits} aggregateBytes]  peerOmitted 0 agentOmitted 0 peakBufferedKeys 0 labelTruncatedRecords 0 labelRedactedRecords 0 diskBytes 0]
        }
        # Old checkpoints with a spool predate explicit schema-ready tracking.
        set defaults [dict create schemaReady [expr {!$fresh && [dict get ${:data} path] ne {}}]  poolWaitCount 0 poolWaitMilliseconds 0 poolWaitLimitMilliseconds  [:dict getdef ${:limits} aggregateWaitMs 30000] failure {}]
        set :data [dict merge $defaults ${:data}]
        if {[dict get ${:data} status] in {unavailable limited} && !${:readResults}} {
            return
        }
        if {${:readResults} && ![dict get ${:data} schemaReady] && [dict get ${:data} status] eq "unavailable"} {
            return
        }
        if {${:readResults} && [dict get ${:data} failure] ne {}
            && [dict get ${:data} failure class] in {invalid-database corrupt-database spool-missing owner-mismatch wrong-driver}} {
            return
        }
        set phase acquire-pool
        try {
            set remaining [expr {[dict get ${:data} poolWaitLimitMilliseconds]-[dict get ${:data} poolWaitMilliseconds]}]
            if {$remaining<=0} {
                error "aggregation-pool-wait-limit"
            }
            set started [clock milliseconds]
            set :handle [ns_db gethandle -timeout [expr {min(100,$remaining)/1000.0}] [dict get ${:data} pool]]
            # A timed-out ns_db gethandle returns an empty list, not an error.
            # Yield BEFORE parsing or advancing any log bytes; the same slice
            # can therefore be retried without omissions or duplicate counts.
            if {${:handle} eq {}} {
                dict incr :data poolWaitCount
                dict incr :data poolWaitMilliseconds [expr {max(1,[clock milliseconds]-$started)}]
                if {[dict get ${:data} poolWaitMilliseconds]>=[dict get ${:data} poolWaitLimitMilliseconds]} {
                    error "aggregation-pool-wait-limit"
                }
                if {[dict get ${:data} status] ne "waiting"} {
                    dict set :data retryStatus [dict get ${:data} status]
                    dict set :data retryReason [dict get ${:data} reason]
                }
                dict set :data status waiting
                dict set :data reason aggregation-pool-busy
                return
            }
            if {[dict get ${:data} status] eq "waiting"} {
                dict set :data status [:dict getdef ${:data} retryStatus "active"]
                dict set :data reason [:dict getdef ${:data} retryReason {}]
            }
            set phase validate-pool
            if {[ns_db dbtype ${:handle}] ne "sqlite"} {
                error "aggregation-wrong-driver"
            }
            set :db [::nsmcp::DatabaseSession new -handle ${:handle}]
            set phase create-spool
            set create [expr {![dict get ${:data} schemaReady]}]
            if {[dict get ${:data} path] eq {}} {
                set f [file tempfile path [file join ${:directory} access-aggregate-]]
                close $f
                dict set :data path $path
                file attributes $path -permissions 0600
            } else {
                set path [dict get ${:data} path]
            }
            set phase check-spool
            if {![file isfile $path]} {
                error "aggregation-spool-missing"
            }
            set phase detach-spool
            foreach row [${:db} execute {PRAGMA database_list}] {
                if {[dict get $row name] eq "nsmcp_access_aggregate"} {
                    ${:db} execute {DETACH DATABASE nsmcp_access_aggregate}
                }
            }
            set phase attach-spool
            ${:db} execute {ATTACH DATABASE $path AS nsmcp_access_aggregate}
            set :attached true
            set phase configure-spool
            ${:db} execute {PRAGMA nsmcp_access_aggregate.journal_mode=DELETE;
                PRAGMA nsmcp_access_aggregate.synchronous=OFF;
                PRAGMA nsmcp_access_aggregate.cache_size=-1024;
                PRAGMA nsmcp_access_aggregate.mmap_size=0;
                PRAGMA nsmcp_access_aggregate.journal_size_limit=0}
            set phase disk-budget
            set pageSize [${:db} scalar {PRAGMA nsmcp_access_aggregate.page_size}]
            set pages [expr {wide([dict get ${:data} maxBytes]*0.4)/$pageSize}]
            if {[${:db} scalar {PRAGMA nsmcp_access_aggregate.page_count}]>$pages} {
                error "aggregation-disk-limit"
            }
            ${:db} execute "PRAGMA nsmcp_access_aggregate.max_page_count=$pages"
            if {$create} {
                set phase create-schema
                set owner [dict get ${:data} owner]
                ${:db} execute {CREATE TABLE nsmcp_access_aggregate.owner (id TEXT PRIMARY KEY);
                    INSERT INTO nsmcp_access_aggregate.owner VALUES($owner);
                    CREATE TABLE nsmcp_access_aggregate.counts (
                        kind INTEGER NOT NULL,k TEXT NOT NULL,label TEXT NOT NULL,
                        clipped INTEGER NOT NULL,redacted INTEGER NOT NULL,n INTEGER NOT NULL,
                        PRIMARY KEY(kind,k)) WITHOUT ROWID;
                    CREATE INDEX nsmcp_access_aggregate.ranking ON counts(kind,n DESC,k)}
                dict set :data schemaReady true
            } else {
                set phase verify-owner
                if {[${:db} scalar {SELECT id FROM nsmcp_access_aggregate.owner}] ne [dict get ${:data} owner]} {
                    error "aggregation-owner-mismatch"
                }
            }
        } on error {message options} {
            set reason sqlite-initialization-failed
            if {$message in {aggregation-spool-missing aggregation-disk-limit aggregation-pool-wait-limit aggregation-owner-mismatch aggregation-wrong-driver}} {
                set reason $message
            }
            :fail $phase $reason $message $options
            :disconnect true
        }
    }

    :method fail {phase reason message options {status unavailable}} {
        if {[dict get ${:data} failure] eq {}} {
            dict set :data failure [::nsmcp::accessAggregationDiagnostic $phase $message $options ${:handle}]
            dict set :data reason $reason
        }
        dict set :data status $status
    }

    :method disconnect {{force false}} {
        if {${:db} ne ""} {
            if {$force || (${:attached} && [catch {
                ${:db} execute {DETACH DATABASE nsmcp_access_aggregate}
            }])} {
                catch {
                    ns_db disconnect ${:handle}
                }
            }
            ${:db} close
            set :db {}
            set :handle {}
            set :attached false
        } elseif {${:handle} ne ""} {
            catch {
                ns_db releasehandle ${:handle}
            }
            set :handle {}
        }
    }

    :public method add {row redactor} {
        #
        # Accumulate peer and user-agent observations from a request.
        #
        # Apply the redactor and configured capacity limits before buffering
        # distinct-observation updates.
        #
        # @param row Parsed access-request record.
        # @param redactor Redactor object applied before returning diagnostic
        #        evidence.
        #
        # @return The observation update result.
        #

        set peer [dict get $row peerAddress]
        set knownPeer [expr {$peer ni {{} - {[REDACTED]}}}]
        set knownAgent [expr {[dict exists $row userAgent] && [dict get $row userAgent] ni {{} -}}]
        if {[dict get ${:data} status] ne "active"} {
            if {$knownPeer} {
                dict incr :data peerOmitted
            }
            if {$knownAgent} {
                dict incr :data agentOmitted
            }
            return
        }
        if {$knownPeer} {
            set key [ns_crypto::md string -digest sha256 -- $peer]
            :addKey 0 $key {} 0 0
            incr :peerPending
        }
        if {$knownAgent} {
            set agent [dict get $row userAgent]
            set key [ns_crypto::md string -digest sha256 -- $agent]
            set label [$redactor message $agent]
            set redacted [expr {$label ne $agent}]
            set bytes [encoding convertto utf-8 $label]
            set clipped [expr {[string length $bytes]>1024}]
            if {$clipped} {
                set label [encoding convertfrom utf-8 [string range $bytes 0 1023]]
                while {[string length [encoding convertto utf-8 $label]]>1024} {
                    set label [string range $label 0 end-1]
                }
                dict incr :data labelTruncatedRecords
            }
            if {$redacted} {
                dict incr :data labelRedactedRecords
            }
            :addKey 1 $key $label $clipped $redacted
            incr :agentPending
        }
        incr :events
        if {${:events}>=512} {
            :flush
        }
    }

    :method addKey {kind key label clipped redacted} {
        set id $kind:$key
        if {[dict exists ${:buffer} $id]} {
            set row [dict get ${:buffer} $id]
            dict incr row n
        } else {
            set row [dict create kind $kind k $key label $label clipped $clipped redacted $redacted n 1]
        }
        dict set :buffer $id $row
        set peak [:dict getdef ${:data} peakBufferedKeys 0]
        if {[dict size ${:buffer}]>$peak} {
            dict set :data peakBufferedKeys [dict size ${:buffer}]
        }
    }

    :method diskBytes {} {
        set bytes 0
        set path [dict get ${:data} path]
        if {$path eq {}} {
            return 0
        }
        foreach suffix {{} -journal -wal -shm} {
            if {[file exists $path$suffix]} {
                incr bytes [file size $path$suffix]
            }
        }
        return $bytes
    }

    :public method flush {} {
        #
        # Persist buffered cardinality observations.
        #
        # Write pending observations to the optional cache and retain explicit
        # capacity or storage failure state.
        #
        # @return The flush result, or an empty result when no observations are
        #         pending.
        #

        if {${:events}==0} {
            return
        }
        if {${:buffer} ne "" && ${:db} ne "" && [dict get ${:data} status] eq "active"} {
            set values {}
            dict for {id row} ${:buffer} {
                lappend values "([ns_dbquotelist [list [dict get $row kind] [dict get $row k] [dict get $row label] [dict get $row clipped] [dict get $row redacted] [dict get $row n]]])"
            }
            try {
                ns_db dml ${:handle} BEGIN
                # Values are quoted individually. Do not cache this bounded
                # batch as a distinct SQL template in DatabaseSession.
                ns_db dml ${:handle} "INSERT INTO nsmcp_access_aggregate.counts VALUES [join $values ,] ON CONFLICT(kind,k) DO UPDATE SET n=n+excluded.n"
                if {[:diskBytes]>[dict get ${:data} maxBytes]} {
                    error "aggregation-disk-limit"
                }
                ns_db dml ${:handle} COMMIT
            } on error {message options} {
                set exception [ns_db exception ${:handle}]
                # Closing the native connection rolls back atomically even
                # when SQLite FULL has already ended the transaction. Avoid a
                # speculative ROLLBACK that logs a second spurious error.
                set disk [expr {$message eq "aggregation-disk-limit" || [string match -nocase {*full*} $exception]}]
                :fail write-batch [expr {$disk ? "aggregation-disk-limit" : "sqlite-write-failed"}] $message $options [expr {$disk ? "limited" : "unavailable"}]
                :disconnect true
                dict incr :data peerOmitted ${:peerPending}
                dict incr :data agentOmitted ${:agentPending}
            }
        }
        set :buffer {}
        set :events 0
        set :peerPending 0
        set :agentPending 0
        dict set :data diskBytes [:diskBytes]
    }

    :public method snapshot {} {
        #
        # Return current cardinality accumulator state.
        #
        # Flush pending observations before reading the stored counters and
        # completeness flags.
        #
        # @return The cardinality state dictionary.
        #

        :flush
        return ${:data}
    }

    :public method results {limit expectedPeers expectedAgents} {
        #
        # Build bounded peer and agent cardinality evidence.
        #
        # Flush observations and compare observed diversity with expected counts to
        # report lower bounds and omissions.
        #
        # @param limit Maximum number of records or entries to return.
        # @param expectedPeers Expected peer observation count used to assess
        #        completeness.
        # @param expectedAgents Expected user-agent observation count used to assess
        #        completeness.
        #
        # @return The cardinality result dictionary.
        #

        :flush
        set result [dict create backend sqlite status [dict get ${:data} status] reason [dict get ${:data} reason]  maxBytes [dict get ${:data} maxBytes] diskBytes [:diskBytes] peakBufferedKeys [:dict getdef ${:data} peakBufferedKeys 0] peerOmitted [dict get ${:data} peerOmitted] agentOmitted [dict get ${:data} agentOmitted]  labelTruncatedRecords [dict get ${:data} labelTruncatedRecords] labelRedactedRecords [dict get ${:data} labelRedactedRecords]]
        foreach key {failure poolWaitCount poolWaitMilliseconds poolWaitLimitMilliseconds} {
            dict set result $key [dict get ${:data} $key]
        }
        foreach {field kind} {peers 0 agents 1} {
            set rows {}
            set totals [dict create distinct 0 records 0 top1 0 top10 0 top100 0 rankedRecords 0 clippedRecords 0 redactedRecords 0]
            if {${:db} ne "" && [catch {
                ${:db} execute -arrayName r -script {
                    dict set totals distinct $r(distinctCount)
                    dict set totals records $r(records)
                    dict set totals clippedRecords $r(clippedRecords)
                    dict set totals redactedRecords $r(redactedRecords)
                } "SELECT count(*) AS distinctCount,coalesce(sum(n),0) AS records,coalesce(sum(n*clipped),0) AS clippedRecords,coalesce(sum(n*redacted),0) AS redactedRecords FROM nsmcp_access_aggregate.counts WHERE kind=$kind"
                set i 0
                ${:db} execute -arrayName r -script {
                    incr i
                    if {$i==1} {
                        dict set totals top1 $r(n)
                    }
                    if {$i<=10} {
                        dict incr totals top10 $r(n)
                    }
                    dict incr totals top100 $r(n)
                    if {$i<=$limit} {
                        dict incr totals rankedRecords $r(n)
                        lappend rows [dict create value $r(label) count $r(n) valueTruncated $r(clipped) valueRedacted $r(redacted)]
                    }
                } "SELECT label,clipped,redacted,n FROM nsmcp_access_aggregate.counts WHERE kind=$kind ORDER BY n DESC,k LIMIT 100"
            } readError readOptions]} {
                :fail read-results sqlite-read-failed $readError $readOptions
                foreach key {status reason failure} {
                    dict set result $key [dict get ${:data} $key]
                }
                set rows {}
                set totals {distinct 0 records 0 top1 0 top10 0 top100 0 rankedRecords 0 clippedRecords 0 redactedRecords 0}
                :disconnect true
            }
            dict set result $field $rows
            dict set result ${field}Totals $totals
        }
        dict set result labelTruncatedRecords [dict get $result agentsTotals clippedRecords]
        dict set result labelRedactedRecords [dict get $result agentsTotals redactedRecords]
        if {[dict get $result peersTotals records]+[dict get $result peerOmitted]!=$expectedPeers
            || [dict get $result agentsTotals records]+[dict get $result agentOmitted]!=$expectedAgents} {
            if {[dict get $result status] ne "unavailable"} {
                dict set result reason aggregation-count-mismatch
            }
            dict set result status unavailable
            foreach {field expected omitted} [list peers $expectedPeers peerOmitted agents $expectedAgents agentOmitted] {
                dict set result $field {}
                dict set result ${field}Totals {distinct 0 records 0 top1 0 top10 0 top100 0 rankedRecords 0 clippedRecords 0 redactedRecords 0}
                dict set result $omitted $expected
            }
        }
        return $result
    }

    :public method readable {} {
        #
        # Report whether the cardinality cache owns a readable database session.
        #
        # This is a current-state check, not a new pool checkout.
        #
        # @return True when a database session is present, otherwise false.
        #

        return [expr {${:db} ne {}}]
    }

    :public method close {} {
        #
        # Release cardinality-store resources.
        #
        # Destroy the accumulator and allow its destructor to flush or close owned
        # resources.
        #
        # @return The object destruction result.
        #

        :destroy
    }

    :method destroy {} {
        :disconnect
        next
    }

XQL Not present:

Generic, PostgreSQL, Oracle