Class ::nsmcp::AccessAnalytics (public)
::nx::Class ::nsmcp::AccessAnalytics![[i]](/resources/acs-subsite/ZoomIn16.gif)
Defined in /usr/local/ns/tcl/nsmcp/access-analytics.tcl
- Testcases:
- No testcase defined.
Source code: :property database:required :property {maxRows:integer 2000000} :property {evictionLimit:integer 64} :method init {} { if {${:maxRows}<1 || ${:evictionLimit}<1 || ${:evictionLimit}>100} {error "positive maxRows and evictionLimit between 1 and 100 required"} if {![nsf::object::exists ${:database}] || ![${:database} info has type ::nsmcp::SQLiteDatabase]} {error "AccessAnalytics requires a SQLiteDatabase object"} next set db [${:database} connect] try { # Migrate the first SQLite draft without retaining duplicate tables. set old [$db scalar {SELECT count(*) FROM sqlite_master WHERE type='table' AND name='metadata'}] set current [$db scalar {SELECT count(*) FROM sqlite_master WHERE type='table' AND name='nsmcp_access_metadata'}] if {$old && !$current && [$db scalar {SELECT count(*) FROM sqlite_master WHERE type='table' AND name IN ('metadata','requests','evictions','diagnostics')}] == 4 && [$db scalar {SELECT value FROM metadata WHERE key='schemaVersion'}] eq "1"} { set source [$db scalar {SELECT value FROM metadata WHERE key='source'}] if {$source ne [list [ns_info server] ${:path}]} {error "draft SQLite cache belongs to another server or log source"} $db transaction { $db execute { ALTER TABLE metadata RENAME TO nsmcp_access_metadata; ALTER TABLE requests RENAME TO nsmcp_access_requests; ALTER TABLE evictions RENAME TO nsmcp_access_evictions; DROP INDEX IF EXISTS requests_time; DROP INDEX IF EXISTS requests_path_time; DROP TABLE IF EXISTS diagnostics; } } } $db execute { CREATE TABLE IF NOT EXISTS nsmcp_access_metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS nsmcp_access_requests ( generation INTEGER NOT NULL, offset INTEGER NOT NULL, start REAL NOT NULL, completed TEXT NOT NULL, request_id TEXT NOT NULL, path TEXT NOT NULL, pool TEXT NOT NULL, method TEXT NOT NULL, status INTEGER NOT NULL, bytes INTEGER NOT NULL, fallback INTEGER NOT NULL, excluded INTEGER NOT NULL, accept REAL, queue REAL, filter REAL, run REAL, PRIMARY KEY(generation,offset)); CREATE INDEX IF NOT EXISTS nsmcp_access_requests_time ON nsmcp_access_requests(start); CREATE INDEX IF NOT EXISTS nsmcp_access_requests_path_time ON nsmcp_access_requests(path,start); CREATE TABLE IF NOT EXISTS nsmcp_access_evictions (id INTEGER PRIMARY KEY, dropped_at INTEGER NOT NULL, first_time REAL NOT NULL, last_time REAL NOT NULL, rows INTEGER NOT NULL, reason TEXT NOT NULL); } set version [$db scalar {SELECT value FROM nsmcp_access_metadata WHERE key='schemaVersion'}] if {$version ne "" && $version ni {1 2 3 4 5}} {error "unsupported SQLite cache schema"} :peerSchema $db set identity [list [ns_info server] ${:path}] set stored [$db scalar {SELECT value FROM nsmcp_access_metadata WHERE key='source'}] if {$stored ne "" && $stored ne $identity} {error "SQLite cache belongs to another server or log source"} $db execute {INSERT OR IGNORE INTO nsmcp_access_metadata VALUES ('source',$identity)} } finally {$db close} } :method peerSchema {db} { # Additive migration also works after definition reload on a live object. set found 0 set authFound 0 set agentFound 0 $db execute {SELECT name FROM pragma_table_info('nsmcp_access_requests')} row { if {$row(name) eq "peer"} {set found 1} if {$row(name) eq "authenticationState"} {set authFound 1} if {$row(name) eq "userAgent"} {set agentFound 1} } if {!$found} { $db execute {ALTER TABLE nsmcp_access_requests ADD COLUMN peer TEXT NOT NULL DEFAULT ''} } if {!$authFound} { $db execute {ALTER TABLE nsmcp_access_requests ADD COLUMN authenticationState TEXT NOT NULL DEFAULT 'unknown'} } if {!$agentFound} { $db execute {ALTER TABLE nsmcp_access_requests ADD COLUMN userAgent TEXT NOT NULL DEFAULT ''} } $db execute { CREATE INDEX IF NOT EXISTS nsmcp_access_requests_peer_time ON nsmcp_access_requests(peer,start); INSERT OR REPLACE INTO nsmcp_access_metadata VALUES('schemaVersion','5'); } } :method state {} { set value [${:db} scalar {SELECT value FROM nsmcp_access_metadata WHERE key='checkpoint'}] if {$value ne "" && [dict exists $value anchorHex]} { dict set value anchor [binary decode hex [dict get $value anchorHex]] dict unset value anchorHex } if {$value eq ""} {return [dict merge [:recentState] [:initial]]} if {[dict get $value version]!=1} {error "unsupported SQLite checkpoint version"} return [dict merge [dict create latestCompletionObserved {} lastMalformedAt {}] [:recentState] $value] } :method ingestLocked {from allowHistory} { set :db [${:database} connect] set :batchRows 0 set :loggedRejects 0 try { ${:db} execute {BEGIN IMMEDIATE} :peerSchema ${:db} set :rowCount [${:db} scalar {SELECT value FROM nsmcp_access_metadata WHERE key='rowCount'}] if {${:rowCount} eq ""} {set :rowCount [${:db} scalar {SELECT count(*) FROM nsmcp_access_requests}]} :maintain set result [next] if {${:loggedRejects}>5} {ns_log warning "nsmcp: suppressed [expr {${:loggedRejects}-5}] additional malformed access-record samples from ${:path}"} ${:db} execute {COMMIT} return $result } on error {message options} { catch {${:db} execute {ROLLBACK}} return -options $options $message } finally {${:db} close; unset -nocomplain :db :batchRows :rowCount :loggedRejects} } :method rejected {line offset state} { incr :loggedRejects if {${:loggedRejects}>5} {return} if {![nsf::object::exists ::nsmcp::accessParseRedactor]} {::nsmcp::LogRedactor create ::nsmcp::accessParseRedactor} set sample [string range [encoding convertfrom utf-8 $line] 0 1023] set sample [::nsmcp::accessParseRedactor message $sample] set sample [string map [list "\n" {\n} "\r" {\r} "\x00" {\0}] $sample] set reason [${:parser} failureReason [encoding convertfrom utf-8 $line]] ns_log warning "nsmcp: malformed access record source=${:path} generation=[dict get $state generation] offset=$offset reason=$reason sample=[list $sample]" } :method accumulate {entry offset changedVar stateVar cutoff} { upvar 1 $stateVar state set start [dict get $entry startTime] if {$start < $cutoff*60 || $start > [clock seconds]+60} {return} set db ${:db}; set generation [dict get $state generation] set path [dict get $entry path]; set excluded [expr {[string length $path]>512}] if {$excluded} {set path {}} set peer [dict get $entry peerAddress] if {[string length $peer]>128 || ![ns_ip valid $peer]} {set peer {}} set authenticationState [dict get $entry authenticationState] set userAgent [expr {[dict exists $entry userAgent] ? [string range [dict get $entry userAgent] 0 1023] : {}}] set completed [dict get $entry timestamp] set request_id [string range [dict get $entry requestIdentifier] 0 511] set pool [string range [dict get $entry pool] 0 127] set method [string range [dict get $entry method] 0 127] set status [dict get $entry status]; set bytes [dict get $entry responseBytes] set fallback [expr {[dict get $entry timingSource] eq "completion"}] unset -nocomplain accept queue filter run if {!$fallback} { foreach {column field} {accept acceptTime queue queueTime filter filterTime run runTime} {set $column [dict get $entry $field]} } # Proactive eviction reuses pages before reaching SQLite's hard limit. incr :batchRows set batchLimit [expr {min(1000,max(1,int([${:database} cget -maxBytes]*0.045/2048)))}] if {${:batchRows}%$batchLimit==0} {:maintain} $db execute {INSERT OR IGNORE INTO nsmcp_access_requests (generation,offset,start,completed,request_id,path,pool,method,status,bytes,fallback,excluded,accept,queue,filter,run,peer,authenticationState,userAgent) VALUES($generation,$offset,$start,$completed,$request_id,$path,$pool,$method,$status,$bytes,$fallback,$excluded,$accept,$queue,$filter,$run,$peer,$authenticationState,$userAgent)} incr :rowCount [$db changes] } :method drop {reason cutoff {limit 0}} { set db ${:db}; set now [clock seconds] if {$reason eq "age"} { set where {start < $cutoff} } else { set where {rowid IN (SELECT rowid FROM nsmcp_access_requests ORDER BY start LIMIT $limit)} } set count 0 $db execute "SELECT count(*) AS n,coalesce(min(start),0) AS first,coalesce(max(start),0) AS last FROM nsmcp_access_requests WHERE $where" row { set count $row(n); set first $row(first); set last $row(last) } if {$count==0} {return} $db execute "DELETE FROM nsmcp_access_requests WHERE $where" incr :rowCount -$count $db execute {INSERT INTO nsmcp_access_evictions(dropped_at,first_time,last_time,rows,reason) VALUES($now,$first,$last,$count,$reason)} set old [$db scalar {SELECT value FROM nsmcp_access_metadata WHERE key='droppedThrough'}] set through [expr {$old eq "" ? $last : max(double($old),$last)}] $db execute {INSERT OR REPLACE INTO nsmcp_access_metadata VALUES('droppedThrough',$through); INSERT OR REPLACE INTO nsmcp_access_metadata VALUES('lastEvictionAt',$now)} set limit ${:evictionLimit} $db execute {DELETE FROM nsmcp_access_evictions WHERE id NOT IN (SELECT id FROM nsmcp_access_evictions ORDER BY id DESC LIMIT $limit)} } :method maintain {} { set db ${:db} set cutoff [:retentionCutoff] :drop age $cutoff set limit ${:maxRows} set count ${:rowCount} if {$count>$limit} {:drop rows 0 [expr {$count-$limit}]} set pageSize [$db scalar {PRAGMA page_size}] set cap [expr {int([${:database} cget -maxBytes]*0.45)}] # Leave room for the next batch, indexes and eviction history. while {([$db scalar {PRAGMA page_count}]-[$db scalar {PRAGMA freelist_count}])*$pageSize > $cap*0.8} { if {${:rowCount}==0} {error "SQLite cache exceeds budget and no analytics rows remain to evict"} :drop size 0 5000 } } :method publish {state changed cutoff} { :maintain dict set state anchorHex [binary encode hex [dict get $state anchor]] dict unset state anchor set count ${:rowCount} ${:db} execute {INSERT OR REPLACE INTO nsmcp_access_metadata VALUES('checkpoint',$state); INSERT OR REPLACE INTO nsmcp_access_metadata VALUES('rowCount',$count)} } :public method validateWindow {options} { set extra {} foreach key {path status pool peer authenticationState windowSeconds includePeerAddresses} { if {![dict exists $options $key]} continue set value [dict get $options $key] if {$key eq "windowSeconds"} { if {![string is entier -strict $value] || $value<0 || $value>3600} {error "windowSeconds must be between 0 and 3600"} } elseif {$key eq "includePeerAddresses"} { if {![string is boolean -strict $value]} {error "includePeerAddresses must be boolean"} } elseif {$key eq "peer"} { if {![ns_ip valid $value]} {error "peer must be a valid IP address"} } elseif {$key eq "authenticationState"} { if {$value ni {anonymous authenticated unknown}} {error "Invalid authenticationState"} } elseif {$key eq "status"} { if {![string is entier -strict $value] || $value<100 || $value>599} {error "status must be between 100 and 599"} } elseif {$value eq "" || [string length $value]>512 || [string first \x00 $value]>=0} {error "invalid exact path or pool filter"} dict set extra $key $value; dict unset options $key } return [dict merge [next [list $options]] $extra] } :public method correlateWarnings {samples redactor} { # Correlate bounded warning examples with retained cache rows only. # The existing start-time index bounds lookups; no raw log is scanned # and no ingestion/backfill is triggered. Addresses are never selected. # # @param samples Up to three dictionaries with requestIdentifier and timestamp; optional userAgent and userAgentMayBeTruncated from system-log diagnostics take precedence over cached User-Agent. # @param redactor Redactor applied to returned text fields. # @return Rooted JSON triples with sampled evidence and explicit missing/ambiguous counts. # if {[llength $samples] > 3} {error "Too many warning correlation samples"} set fields {}; set matched 0; set missing 0; set ambiguous 0; set i 0 set db [${:database} connect] try { foreach sample $samples { set request_id [dict get $sample requestIdentifier] set at [dict get $sample timestamp] if {![::nsmcp::logRecordParser validConnectionIdentifier $request_id] || ![string match "-conn:[ns_info server]:*" $request_id] || ![string is double -strict $at]} {incr missing; continue} set lower [expr {$at-60}]; set upper [expr {$at+1}] set rows [$db execute {SELECT request_id,start,path,pool,method,status,authenticationState,userAgent FROM nsmcp_access_requests WHERE start >= $lower AND start <= $upper AND request_id=$request_id LIMIT 2}] if {[llength $rows] == 0} {incr missing; continue} if {[llength $rows] > 1} {incr ambiguous; continue} set row [lindex $rows 0]; set item {} set uaSource access-cache set authSource access-cache if {[dict exists $sample authenticationState] && [dict get $sample authenticationState] in {anonymous authenticated}} { dict set row authenticationState [dict get $sample authenticationState] set authSource system-log } set uaTruncated [expr {[string length [dict get $row userAgent]] >= 1024}] if {[dict exists $sample userAgent]} { dict set row userAgent [dict get $sample userAgent] set uaSource system-log set uaTruncated [dict get $sample userAgentMayBeTruncated] } foreach key {path pool method authenticationState userAgent} { lappend item $key string [$redactor message [dict get $row $key]] } lappend item requestIdentifier string $request_id start number [dict get $row start] status number [dict get $row status] warningTimestamp number $at authenticationSource string $authSource userAgentAvailable boolean [expr {[dict get $row userAgent] ne ""}] userAgentSource string $uaSource userAgentMayBeTruncated boolean $uaTruncated lappend fields $i object $item; incr i; incr matched } } finally {$db close} return [list {} object [list available boolean true scope string sampled-warning-requests requested number [llength $samples] matched number $matched missing number $missing ambiguous number $ambiguous records array $fields limitations string {Bounded examples, not aggregate bot attribution. Exact connection identifier and start time within 60 seconds before to 1 second after the warning are required. Cache lag, eviction, long requests, legacy rows without User-Agent, and absent identifiers can prevent matches. No raw access-log scan or cache refresh was performed. Multiple candidates are not attributed. Pool and User-Agent alone do not prove bot or human identity. No peer addresses are returned.}]] } :public method summary {options} { # Use the scanner coverage and a single snapshot for all aggregates. # Hold a single SQLite snapshot for coverage and all aggregate queries. ns_mutex lock ${:mutex} try { set :db [${:database} connect] set :batchRows 0 ${:db} execute {BEGIN} set options [:validateWindow $options] :peerSchema ${:db} set plain $options foreach key {path status pool peer authenticationState windowSeconds includePeerAddresses} {dict unset plain $key} # Coverage and counts share the same read transaction. set result [:summaryLocked $plain] set filters {}; foreach key {path status pool peer authenticationState} {if {[dict exists $options $key]} {dict set filters $key [dict get $options $key]}} dict set result filters $filters set db ${:db}; set from [dict get $result from]; set to [dict get $result to] set where {start >= $from AND start < $to} foreach key {path status pool peer authenticationState} { if {[dict exists $options $key]} {set $key [dict get $options $key]; append where " AND $key = \$$key"} } $db execute "SELECT count(*) AS requestCount,coalesce(sum(bytes),0) AS responseBytes, coalesce(sum(fallback),0) AS fallbackCount,coalesce(sum(excluded),0) AS excludedPaths, count(run) AS timingCount,coalesce(sum(accept),0) AS acceptTotal, coalesce(sum(queue),0) AS queueTotal,coalesce(sum(filter),0) AS filterTotal, coalesce(sum(run),0) AS runTotal,coalesce(max(run),0) AS runMax FROM nsmcp_access_requests WHERE $where" row { foreach key $row(*) {dict set result $key $row($key)} } foreach {dimension column} {pools pool methods method statuses status paths path} { set counts {} # Bounded output, but count all matching rows in totals above. $db execute "SELECT $column AS label,count(*) AS n FROM nsmcp_access_requests WHERE $where GROUP BY $column ORDER BY n DESC LIMIT 1000" row { if {$dimension eq "paths" && $row(label) eq ""} continue dict set counts $row(label) $row(n) } dict set result $dimension $counts } set tracked 0 dict for {label count} [dict get $result paths] {incr tracked $count} dict set result excludedPaths [expr {[dict get $result requestCount]-$tracked}] set statusTimings {} $db execute "SELECT status,count(*) AS n,count(run) AS timed,coalesce(sum(filter),0) AS filters, coalesce(sum(run),0) AS runs,coalesce(max(run),0) AS maximum FROM nsmcp_access_requests WHERE $where GROUP BY status" row { lappend statusTimings [dict create status $row(status) requestCount $row(n) timingCount $row(timed) filterTotal $row(filters) runTotal $row(runs) runMax $row(maximum)] } dict set result statusTimings $statusTimings set authentication {} $db execute "SELECT authenticationState,count(*) AS n,sum(status=503) AS rejected, sum(status BETWEEN 200 AND 299) AS successful, count(CASE WHEN status BETWEEN 200 AND 299 THEN run END) AS timed, coalesce(sum(CASE WHEN status BETWEEN 200 AND 299 THEN run END),0) AS runs FROM nsmcp_access_requests WHERE $where GROUP BY authenticationState" row { dict set authentication $row(authenticationState) [dict create requestCount $row(n) status503Count $row(rejected) successfulCount $row(successful) successfulTimingCount $row(timed) successfulRunTotal $row(runs)] } dict set result authenticationStates $authentication set unknownPeers [$db scalar "SELECT count(*) FROM nsmcp_access_requests WHERE $where AND peer = ''"] dict set result coverage unknownPeerCount $unknownPeers dict set result coverage peerCountsComplete [expr {$unknownPeers==0 && [dict get $result coverage windowComplete]}] if {[dict exists $options windowSeconds] && [dict get $options windowSeconds] > 0} { # RANGE counts all equal timestamps together and uses an inclusive # [end-windowSeconds,end] range clipped to the selected period. # SQLite performs the rolling calculation rather than Tcl fetching rows. set seconds [dict get $options windowSeconds] set burst [dict create windowSeconds $seconds requestCount 0 from {} to {}] $db execute "SELECT peer,start,n FROM ( SELECT peer,start,count(*) OVER ( PARTITION BY peer ORDER BY start RANGE BETWEEN \$seconds PRECEDING AND CURRENT ROW) AS n FROM nsmcp_access_requests WHERE $where AND peer != '' ) ORDER BY n DESC,start,peer LIMIT 1" row { dict set burst requestCount $row(n) dict set burst from [expr {max(double($from),$row(start)-$seconds)}] dict set burst to $row(start) if {[dict exists $options includePeerAddresses] && [dict get $options includePeerAddresses]} { dict set burst peerAddress $row(peer) } } dict set result peerBurst $burst } set size [dict get $result bucketSeconds]; set counts [dict get $result buckets] $db execute "SELECT cast(start/\$size AS INTEGER)*\$size AS slot,count(*) AS n FROM nsmcp_access_requests WHERE $where GROUP BY slot" row { dict set counts $row(slot) $row(n) } dict set result buckets $counts dict set result pathCountsComplete [expr {[dict get $result excludedPaths]==0}] set retained [$db scalar {SELECT coalesce(min(start),'') FROM nsmcp_access_requests}] set through [$db scalar {SELECT value FROM nsmcp_access_metadata WHERE key='droppedThrough'}] set evictionAt [$db scalar {SELECT value FROM nsmcp_access_metadata WHERE key='lastEvictionAt'}] dict set result coverage source ${:path} dict set result coverage retainedRequestFrom $retained dict set result coverage droppedThrough $through dict set result coverage lastEvictionAt $evictionAt set evicted [expr {$through ne "" && $from <= $through}] dict set result coverage evictionOverlap $evicted if {$evicted} {dict set result coverage windowComplete false; set reasons [dict get $result coverage incompleteReasons]; lappend reasons eviction; dict set result coverage incompleteReasons $reasons} dict set result coverage peerCountsComplete [expr {$unknownPeers==0 && [dict get $result coverage windowComplete]}] dict set result coverage observedPeerCountsComplete [expr {$unknownPeers==0}] set peerReasons {} if {$unknownPeers} {lappend peerReasons missing-peer-fields-in-cached-records} if {![dict get $result coverage windowComplete]} { lappend peerReasons window-coverage-incomplete foreach reason [dict get $result coverage incompleteReasons] {lappend peerReasons "window:$reason"} } dict set result coverage peerCountsIncompleteReasons $peerReasons dict set result coverage databaseBytes [file size [${:database} path]] dict set result coverage storageBudgetBytes [${:database} cget -maxBytes] set events {} $db execute {SELECT * FROM nsmcp_access_evictions ORDER BY id DESC} row { lappend events [dict create droppedAt $row(dropped_at) from $row(first_time) to $row(last_time) rowCount $row(rows) reason $row(reason)] } dict set result evictions $events ${:db} execute {COMMIT} return $result } finally {if {[info exists :db]} {${:db} close; unset :db}; unset -nocomplain :batchRows; ns_mutex unlock ${:mutex}} }XQL Not present: Generic, PostgreSQL, Oracle
