Class ::nsmcp::UnifiedAccessAnalytics (public)

 ::nx::Class ::nsmcp::UnifiedAccessAnalytics

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

Testcases:

No testcase defined.

Source code:

    :property settings:required
    :variable legacyDatabase {}
    :variable sourceReady false
    :property {ingestBytes:integer 16777216}
    :property {ingestMilliseconds:integer 2000}
    :property {retentionBatchRows:integer 1000}
    :property {retentionMilliseconds:integer 1000}
    :property {retentionInterval 30}
    :method init {} {
        if {${:ingestBytes}<131072 || ${:ingestMilliseconds}<10 || ${:ingestMilliseconds}>10000} {error "invalid continuous ingestion limits"}
        if {${:retentionBatchRows}<1 || ${:retentionBatchRows}>10000 || ${:retentionMilliseconds}<10 || ${:retentionMilliseconds}>10000 || ${:retentionInterval}<1} {error "invalid retention cleanup limits"}
        set :path [file normalize ${:path}]
        dict set :settings sourcePath ${:path}
        set :mutex [ns_mutex create "nsmcp:unified-access:[ns_info server]"]
        set :retentionSeconds [dict get ${:settings} retention]
        set :legacyDatabase ${:database}
        set :database [::nsmcp::UnifiedAccessDatabase create [self]::database -settings ${:settings} -maxBytes [dict get ${:settings} maxBytes]]
        try {:initializeSource} on error {message options} {
            set code [dict get $options -errorcode]
            if {[lrange $code 0 1] ne {NSMCP RECORD_STORE}
                || [lindex $code 2] ni {disk-limit disk-full database-locked database-busy pool-busy pool-timeout}} {return -options $options $message}
            accessUnifiedFailure ${:settings} $options startup
        }
    }
    :method initializeSource {} {
        set init [accessRecordStoreOpen ${:settings}]
        if {![$init available]} {$init close; return -code error -errorcode {NSMCP RECORD_STORE pool-busy} "access store initialization pool unavailable"}
        set db [$init session]
        try {
            set source [$db scalar {SELECT coalesce((SELECT value FROM nsmcp_records.state WHERE key='sourcePath'),'')}]
            if {$source ne {} && $source ne ${:path}} {error "access store belongs to another configured source; use a separate record directory"}
            set path ${:path}
            $db execute {INSERT OR IGNORE INTO nsmcp_records.state(key,value) VALUES('sourcePath',$path)}
        } finally {$init close}
        set :sourceReady true
    }
    :public method summary {options} {
        return [accessRecordRetry ${:settings} [list next [list $options]]]
    }
    :public method correlateWarnings {samples redactor} {
        return [accessRecordRetry ${:settings} [list next [list $samples $redactor]]]
    }
    :method peerSchema {db} {} ;# The compatibility view is already current.
    :method state {} {
        set report [${:db} overview]
        set initial [dict merge [:initial] [:recentState]]
        return [dict merge $initial $report]
    }
    :method summaryLocked {options} {
        set result [next [list $options]]
        # The original raw-log retention history is never certified by a DB
        # snapshot. Report ingestion completion separately from that assurance.
        set coverage [dict get $result coverage]
        dict set coverage windowComplete false
        set reasons [dict get $coverage incompleteReasons]; lappend reasons source-retention-unverified
        dict set coverage incompleteReasons [lsort -unique $reasons]
        dict set coverage continuousStore true
        set overview [${:db} overview]
        set capacity [dict get $overview storage]
        foreach {field key} {storageMaxBytes maxBytes storageMainMaxBytes mainMaxBytes storageMainAllocatedBytes mainAllocatedBytes storageDiskBytes diskBytes storageLimitExceeded limitExceeded storageMinimumBudgetBytes minimumBudgetBytes} {
            dict set coverage $field [dict get $capacity $key]
        }
        set failed [nsv_exists nsmcp:access-store:[ns_info server] lastFailure]
        dict set coverage writerFailure $failed
        set reasons [dict get $coverage incompleteReasons]
        if {$failed} {lappend reasons writer-paused}
        if {[dict get $capacity limitExceeded]} {lappend reasons storage-limit-exceeded}
        dict set coverage incompleteReasons [lsort -unique $reasons]
        dict set coverage writerFailureClass [expr {$failed ? [lindex [dict get [nsv_get nsmcp:access-store:[ns_info server] lastFailure] code] 2] : {}}]
        dict set coverage ingestionLagSeconds [dict get $overview ingestionLagSeconds]
        dict set result coverage $coverage
        return $result
    }
    :public method ingest {{from {}} {allowHistory true}} {
        # Compatibility entry point for explicit administrative refreshes only.
        return [:refresh]
    }
    :public method refresh {} {
        set gate nsmcp:unified-writer:[ns_info server]
        if {[nsv_incr $gate busy]!=1} {nsv_incr $gate busy -1; return {busy true}}
        set run {}; set outcome error; set result {}
        try {
            set run [accessIngestionBegin]
            set result [accessRecordRetry ${:settings} [list :refreshLocked $run] {} [list ::nsmcp::accessIngestionProgress $run retry]]
            set outcome [expr {[dict exists $result busy] ? "pool-busy" : "ok"}]
            if {![dict exists $result busy]} {nsv_unset -nocomplain nsmcp:access-store:[ns_info server] lastFailure}
            return $result
        } on error {message options} {
            set result [accessUnifiedFailure ${:settings} $options]
            return $result
        } finally {
            try {accessIngestionFinish $run $outcome $result} finally {nsv_incr $gate busy -1}
        }
    }
    :public method ingestionStatus {} {return [accessIngestionStatus]}
    :public method retentionStatus {} {return [accessRetentionStatus]}
    :public method retentionTick {} {
        # A single threaded one-shot chain avoids phase-locking a periodic
        # maintenance timer to the ingestion timer. Busy work retries later.
        set delay ${:retentionInterval}
        try {
            set result [:retain]
            if {[dict exists $result busy]} {set delay 1}
            return $result
        } finally {catch {ns_schedule_proc -once -thread $delay [self] retentionTick}}
    }
    :public method retain {} {
        # Share the writer gate: maintenance never overlaps ingestion and never
        # waits for a running importer. Readers retain their WAL snapshots.
        set gate nsmcp:unified-writer:[ns_info server]
        if {[nsv_incr $gate busy]!=1} {nsv_incr $gate busy -1; return {busy true}}
        set key nsmcp:access-retention:[ns_info server]
        set token [nsv_incr $key sequence]
        set began [clock clicks -milliseconds]; set store {}; set outcome error
        set counts {requests 0 diagnostics 0 minutes 0 peers 0}; set batches 0; set code {}
        set last [expr {[nsv_exists $key state] ? [dict get [nsv_get $key state] lastRun] : {}}]
        nsv_set $key state [dict create token $token running true startedAt [clock seconds] startedTicks $began phaseTicks $began phase open-store deleted $counts lastRun $last]
        try {
            set store [accessRecordStoreOpen ${:settings} writer [list ::nsmcp::accessRetentionPhase $token]]
            if {![$store available]} {set outcome pool-busy; return {busy true}}
            while {$batches==0 || [clock clicks -milliseconds]-$began<${:retentionMilliseconds}} {
                accessRetentionPhase $token prune
                set removed [$store prune ${:retentionBatchRows}]
                incr batches; set total 0
                dict for {table n} $removed {dict incr counts $table $n; incr total $n}
                set state [nsv_get $key state]; dict set state deleted $counts; nsv_set $key state $state
                if {$total==0} break
            }
            set outcome ok
            return [dict create deleted $counts batches $batches]
        } on error {message options} {
            set code {NSMCP RETENTION operation-failed}
            set candidate [dict get $options -errorcode]
            if {[lrange $candidate 0 1] eq {NSMCP RECORD_STORE} && [regexp {^[a-z][a-z0-9-]{0,63}$} [lindex $candidate 2]]} {set code [lrange $candidate 0 2]}
            ns_log warning "nsmcp: retention cleanup paused ($code); ingestion checkpoints preserved"
            return [dict create failed true code $code]
        } finally {
            try {if {$store ne {}} {accessRetentionPhase $token close-store; $store close}} finally {
                set state [nsv_get $key state]
                set last [dict create startedAt [dict get $state startedAt] finishedAt [clock seconds] durationMs [expr {max(0,[clock clicks -milliseconds]-$began)}] outcome $outcome deleted $counts batches $batches]
                if {$code ne {}} {dict set last code $code}
                dict set state running false; dict set state lastRun $last; nsv_set $key state $state
                nsv_incr $gate busy -1
            }
        }
    }
    :method retireLegacy {} {
        if {${:legacyDatabase} eq {} || ![nsf::object::exists ${:legacyDatabase}]} {return}
        set db [${:legacyDatabase} connect]
        try {
            if {[$db scalar {SELECT count(*) FROM sqlite_master WHERE type='table' AND name='nsmcp_access_requests'}]} {
                set cutoff [expr {[clock seconds]-min(86400,[dict get ${:settings} peerRetention])}]
                $db execute {DELETE FROM nsmcp_access_requests WHERE rowid IN(SELECT rowid FROM nsmcp_access_requests WHERE start<$cutoff LIMIT 10000)}
            }
        } finally {$db close}
    }
    :method refreshLocked {{run {}}} {
        accessIngestionProgress $run planning initialize-source
        if {!${:sourceReady}} {:initializeSource}
        accessIngestionProgress $run planning retire-legacy
        :retireLegacy
        set start [clock milliseconds]; set now [clock seconds]
        accessIngestionProgress $run planning open-store
        set store [accessRecordStoreOpen ${:settings} writer [list ::nsmcp::accessIngestionProgress $run planning]]
        if {![$store available]} {$store close; return {busy true}}
        try {
            accessIngestionProgress $run planning discover-live
            set control [$store stateGet continuous {}]
            catch {accessIngestionCheckpoint $control}
            set oldLive [expr {[dict exists $control live] ? [dict get $control live] : {}}]
            set floor [expr {max([$store stateGet retentionFloor 0],wide(floor(($now-[dict get ${:settings} retention])/60.0))*60)}]
            set options [accessFailureOptions [dict create from [expr {max($floor,$now-300)}] to [expr {$now+1}] timezone UTC pathLiteral {} includeRotations false selection ordered mode summary maxGroups 20 ingestOnly true]]
            set files [dict get [accessScanDiscover [list ${:path}] false 1] files]
            set live {}; set earliest [expr {[dict exists $control recentFrom] ? [dict get $control recentFrom] : $now-300}]
            foreach file $files {
                accessIngestionProgress $run planning prepare-source
                set fid [$store prepare $file]
                if {[dict exists $oldLive $fid]} {
                    dict set file offset [expr {min([dict get $oldLive $fid],[dict get $file size])}]
                } else {
                    accessIngestionProgress $run planning seek-recent
                    set planning {files {} index 0 examinedBytes 0 stats {}}
                    accessOrderedPlan file planning $options [accessScanDefaultLimits] {}
                    if {[dict get $file status] eq "outside-window"} {dict set file offset 0}
                }
                dict set file status running; dict set file archiveFid $fid
                dict set file archiveLo [dict get $file offset]
                dict set file scanEnd [dict get $file size]
                dict set file scanStart [dict get $file offset]
                lappend live $file
            }
        } finally {$store close}
        set limits [dict merge [accessScanDefaultLimits] [accessAggregateSettings] [dict create slice ${:ingestBytes} milliseconds ${:ingestMilliseconds}]]
        set state [dict create files $live omittedFiles 0 unavailableFiles 0 ignoredCandidates 0 index 0 stats [accessScanEmpty] examinedBytes 0 parsedRecordBytes 0 reusedBytes 0 nativeSearch false recordStore ${:settings} limits $limits failedTrafficSelection [accessFailureSelection $options]]
        dict set options scanState $state
        set step [accessRecordScan [list ${:path}] [dict get ${:settings} directory] $options ::nsmcp::unifiedRedactor $limits {} ${:settings} [list ::nsmcp::accessIngestionProgress $run live]]
        if {[llength $step]!=4} {error "continuous ingestion could not retain its checkpoint"}
        set state [lindex $step 3]; set live {}; set sourceBytes 0; set cursor 0; set caught true; set pending 0
        foreach file [dict get $state files] {
            if {[dict exists $file archivePending]} {incr pending}
            dict set live [dict get $file archiveFid] [dict get $file offset]
            incr sourceBytes [dict get $file size]; incr cursor [dict get $file offset]
            if {[dict get $file offset]<[dict get $file size] && ![dict exists $file archivePending]} {set caught false}
        }
        dict set control live $live; dict set control observedSize $sourceBytes; dict set control cursor $cursor
        dict set control recentFrom $earliest; dict set control updatedAt $now; dict set control caughtUp $caught; dict set control pending $pending
        set backfill [expr {[dict exists $control backfill] ? [dict get $control backfill] : {}}]
        set finished [expr {$backfill ne {} && [dict exists $backfill ingestComplete]}]
        set due [expr {!$finished || ![dict exists $control backfillAt] || $now-[dict get $control backfillAt]>=60}]
        if {$finished && $due} {set backfill {}}
        if {$caught && $due && [clock milliseconds]-$start<${:ingestMilliseconds}} {
            set historical [accessFailureOptions [dict create from $floor to [expr {$now+1}] timezone UTC pathLiteral {} includeRotations true selection ordered mode summary maxGroups 20 ingestOnly true]]
            if {$backfill ne {} && ![dict exists $backfill ingestComplete]} {
                # Freeze the previous backfill's window until that pass completes.
                set historical [dict get $control backfillOptions]
                if {[catch {accessUnifiedRestore $backfill} restored]} {set backfill {}} else {dict set historical scanState $restored}
            }
            set remaining [expr {max(10,${:ingestMilliseconds}-([clock milliseconds]-$start))}]
            set histLimits [dict replace $limits milliseconds $remaining]
            if {[dict exists $historical scanState]} {dict set historical scanState limits milliseconds $remaining}
            if {[catch {accessRecordScan [list ${:path}] [dict get ${:settings} directory] $historical ::nsmcp::unifiedRedactor $histLimits {} ${:settings} [list ::nsmcp::accessIngestionProgress $run history]} step]} {
                set backfill {}; dict set control lastBackfillError [clock seconds]
            } elseif {[llength $step]==4} {
                set backfill [lindex $step 3]
                dict unset historical scanState
                dict set control backfillOptions $historical
                dict set control backfill [accessUnifiedPack $backfill]
                dict set control backfillAt $now
            }
        }
        set remaining 0; set unsupported 0
        if {$backfill eq {}} {set remaining 1} else {
            foreach file [dict get $backfill files] {
                if {[dict get $file compressed]} {incr unsupported}
                if {[dict get $file status] ni {complete outside-window unsupported-compressed}} {incr remaining [expr {max(0,[dict get $file size]-[dict get $file offset])}]}
            }
        }
        dict set control historyUnreadBytes $remaining; dict set control unsupportedCompressed $unsupported
        accessIngestionProgress $run finalizing open-store
        set store [accessRecordStoreOpen ${:settings} writer [list ::nsmcp::accessIngestionProgress $run finalizing]]
        if {![$store available]} {$store close; return {busy true}}
        try {
            accessIngestionProgress $run finalizing save-control
            $store statePut continuous $control
            catch {accessIngestionCheckpoint $control}
            accessIngestionProgress $run finalizing rollup-backfill
            if {[clock milliseconds]-$start<${:ingestMilliseconds}} {$store rollupBackfill 1}
        } finally {$store close}
        return [dict create caughtUp $caught historyUnreadBytes $remaining parsedBytes [dict get $state parsedRecordBytes] wallMs [expr {[clock milliseconds]-$start}]]
    }

XQL Not present:

Generic, PostgreSQL, Oracle