Class ::nsmcp::AccessCardinalityStore (public)
::nx::Class ::nsmcp::AccessCardinalityStore![[i]](/resources/acs-subsite/ZoomIn16.gif)
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 buffer {} :variable events 0 :variable peerPending 0 :variable agentPending 0 :method init {} { set :data ${:checkpoint} if {${:data} eq ""} { 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] } if {[dict get ${:data} status] in {unavailable limited} && !${:readResults}} {return} if {[dict get ${:data} path] eq ""} { set f [file tempfile path [file join ${:directory} access-aggregate-]] close $f; file attributes $path -permissions 0600 dict set :data path $path set create true } else {set path [dict get ${:data} path]; set create false} try { if {!$create && ![file isfile $path]} {error "aggregation-spool-missing"} set :handle [ns_db gethandle -timeout 2 [dict get ${:data} pool]] if {${:handle} eq "" || [ns_db dbtype ${:handle}] ne "sqlite"} {error "aggregation-pool-unavailable"} set :db [::nsmcp::DatabaseSession new -handle ${:handle}] # A released native handle is exclusive to this session. Detach any # orphaned private alias before using it; leave the primary DB alone. foreach row [${:db} execute {PRAGMA database_list}] { if {[dict get $row name] eq "nsmcp_access_aggregate"} {${:db} execute {DETACH DATABASE nsmcp_access_aggregate}} } ${:db} execute {ATTACH DATABASE $path AS nsmcp_access_aggregate} ${: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 pageSize [${:db} scalar {PRAGMA nsmcp_access_aggregate.page_size}] # Conservative 40% main-file ceiling reserves rollback-journal # space and overhead inside the per-job disk budget. 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} { ${:db} execute {CREATE TABLE nsmcp_access_aggregate.owner (id TEXT PRIMARY KEY)} } } on error {message options} { # Schema creation below uses an explicit owner value, never state. dict set :data status unavailable dict set :data reason [expr {$message in {aggregation-spool-missing aggregation-disk-limit} ? $message : "sqlite-initialization-failed"}] :disconnect return } if {$create} { set owner [dict get ${:data} owner] try { ${:db} execute {DELETE FROM nsmcp_access_aggregate.owner; 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)} } on error {message options} { dict set :data status unavailable; dict set :data reason sqlite-initialization-failed :disconnect } } elseif {${:db} ne ""} { if {[catch {${:db} scalar {SELECT id FROM nsmcp_access_aggregate.owner}} owner] || $owner ne [dict get ${:data} owner]} { dict set :data status unavailable; dict set :data reason aggregation-owner-mismatch; :disconnect } } } :method disconnect {{force false}} { if {${:db} ne ""} { if {$force || [catch {${:db} execute {DETACH DATABASE nsmcp_access_aggregate}}]} {catch {ns_db disconnect ${:handle}}} ${:db} close; set :db {}; set :handle {} } elseif {${:handle} ne ""} {catch {ns_db releasehandle ${:handle}}; set :handle {}} } :public method add {row redactor} { 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 [expr {[dict exists ${:data} peakBufferedKeys] ? [dict get ${: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] foreach suffix {{} -journal -wal -shm} {if {[file exists $path$suffix]} {incr bytes [file size $path$suffix]}} return $bytes } :public method flush {} { if {${:events}==0} {return} if {${:buffer} ne "" && ${:db} ne "" && [dict get ${:data} status] eq "active"} { set values {} dict for {id row} ${:buffer} { lappend values "([dict get $row kind],[${:db} quote [dict get $row k]],[${:db} quote [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. :disconnect true set disk [expr {$message eq "aggregation-disk-limit" || [string match -nocase {*full*} $exception]}] dict set :data status [expr {$disk ? "limited" : "unavailable"}] dict set :data reason [expr {$disk ? "aggregation-disk-limit" : "sqlite-write-failed"}] 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 {} {:flush; return ${:data}} :public method results {limit expectedPeers expectedAgents} { :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 [expr {[dict exists ${:data} peakBufferedKeys] ? [dict get ${:data} peakBufferedKeys] : 0}] peerOmitted [dict get ${:data} peerOmitted] agentOmitted [dict get ${:data} agentOmitted] labelTruncatedRecords [dict get ${:data} labelTruncatedRecords] labelRedactedRecords [dict get ${:data} labelRedactedRecords]] 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 "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" r { dict set totals distinct $r(distinctCount); dict set totals records $r(records) dict set totals clippedRecords $r(clippedRecords); dict set totals redactedRecords $r(redactedRecords) } set i 0 ${:db} execute "SELECT label,clipped,redacted,n FROM nsmcp_access_aggregate.counts WHERE kind=$kind ORDER BY n DESC,k LIMIT 100" r { 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)] } } }]} { dict set result status unavailable; dict set result reason sqlite-read-failed 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} dict set result $omitted $expected } } return $result } :public method close {} {:destroy} :method destroy {} {:disconnect; next}XQL Not present: Generic, PostgreSQL, Oracle
