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:

    :require trait ::nsmcp::AccessRecordSupport
    # nx::trait versions that export aliases publicly need explicit protection.
    foreach method [::nsmcp::AccessRecordSupport info methods -callprotection all] {
        ::nsf::method::property [self] $method call-private false
        ::nsf::method::property [self] $method call-protected true
    }

    :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}
    :property {retentionRetryInterval 5}

    :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 || ${:retentionRetryInterval}<1 || ${:retentionRetryInterval}>3600} {
            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 partition-reset-required}} {
                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 main.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 main.state(key,value) VALUES('sourcePath',:path)}
        } finally {
            $init close
        }
        set :sourceReady true
    }

    :public method summary {options} {
        #
        # Query unified access summaries with bounded lock retry.
        #
        # Delegate to the superclass using the same normalized options after
        # eligible transient store failures.
        #
        # @param options Tcl dictionary containing the operation options.
        #
        # @return The access summary dictionary.
        #

        return [:accessRecordRetry ${:settings} [list next [list $options]]]
    }

    :public method correlateWarnings {samples redactor} {
        #
        # Correlate warnings with unified stored requests.
        #
        # Delegate through bounded store retry while preserving the caller's sample
        # list and redactor.
        #
        # @param samples Bounded warning sample records to correlate with retained
        #        access evidence.
        # @param redactor Redactor object applied before returning diagnostic
        #        evidence.
        #
        # @return The warning correlation evidence.
        #

        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}} {
        #
        # Provide the compatibility entry point for an administrative refresh.
        #
        # Legacy range arguments remain accepted; the unified ingestor controls its
        # live and historical work through refresh.
        #
        # @param from Starting timestamp or byte position, as required by the
        #        operation.
        # @param allowHistory Whether this pass may process historical sources.
        #        Defaults to 'true'.
        #
        # @return The bounded refresh result.
        #

        # Compatibility entry point for explicit administrative refreshes only.
        return [:refresh]
    }

    :public method refresh {} {
        #
        # Run a bounded unified ingestion pass under the writer gate.
        #
        # Reject overlapping writers, publish phase and commit statistics, and
        # always finish the run token before releasing the gate.
        #
        # @return The ingestion result, or a busy dictionary when another writer
        #         owns the gate.
        #

        set gate nsmcp:unified-writer:[ns_info server]
        # Give a waiting maintenance callback one bounded opportunity to run.
        # Expiry prevents a lost callback from indefinitely stopping ingestion.
        set backlog [accessBacklogStatus]
        if {[nsv_exists $gate retentionUntil]
            && [nsv_get $gate retentionUntil]>[clock seconds]
            && (![dict get $backlog available] || [dict get $backlog caughtUp])} {
            return {busy true}
        }
        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 {} {
        #
        # Read current and last-run ingestion progress.
        #
        # Use the retained ingestion status instead of triggering a scan or
        # ingestion pass.
        #
        # @return The ingestion status dictionary.
        #

        return [accessIngestionStatus]
    }

    :public method retentionStatus {} {
        #
        # Read current retention-maintenance status.
        #
        # Return retained maintenance observations without starting another pruning
        # pass.
        #
        # @return The retention status dictionary.
        #

        return [accessRetentionStatus]
    }

    :public method retentionTick {} {
        #
        # Run and reschedule one retention-maintenance callback.
        #
        # Use a one-shot scheduling chain so maintenance retries do not phase-lock
        # with the ingestion timer.
        #
        # @return The scheduler callback result.
        #

        # A single threaded one-shot chain avoids phase-locking a periodic
        # maintenance timer to the ingestion timer. Busy work retries later.
        set delay ${:retentionInterval}
        set began [clock clicks -milliseconds]
        try {
            set result [:retain]
            if {[dict exists $result busy] || [:dict getdef $result more false]} {
                # Leave at least two run durations idle after cleanup. The cap
                # bounds adaptive backoff; the configured minimum always wins.
                set duration [expr {max([:dict getdef $result durationMs 0],
                                       [clock clicks -milliseconds]-$began)}]
                set delay [expr {max(${:retentionRetryInterval},
                                    min(${:retentionInterval},ceil(2.0*$duration/1000.0)))}]
            }
            set key nsmcp:access-retention:[ns_info server]
            if {[nsv_exists $key state]} {
                set state [nsv_get $key state]
                dict set state nextDelaySeconds $delay
                dict set state nextRunAt [expr {[clock seconds]+int(ceil($delay))}]
                nsv_set $key state $state
            }
            return $result
        } finally {
            catch {
                ns_schedule_proc -once -thread $delay [self] retentionTick
            }
        }
    }

    :public method retain {} {
        #
        # Perform bounded retention maintenance under the writer gate.
        #
        # Avoid overlapping ingestion while pruning expired store state and
        # retaining maintenance statistics.
        #
        # @return The retention result, including a busy indication when the gate is
        #         occupied.
        #

        # 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]
        set backlog [accessBacklogStatus]
        if {![dict get $backlog available] || [dict get $backlog caughtUp]} {
            nsv_set $gate retentionUntil [expr {[clock seconds]+5}]
        } else {
            nsv_unset -nocomplain $gate retentionUntil
        }
        if {[nsv_incr $gate busy]!=1} {
            nsv_incr $gate busy -1
            return {busy true}
        }
        nsv_unset -nocomplain $gate retentionUntil
        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 more false
        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 phaseMilliseconds {} 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
                }
            }
            accessRetentionPhase $token pressure-check
            set pressure [$store maintenancePressure]
            set more [dict get $pressure expired]
            set outcome ok
            return [dict create deleted $counts batches $batches more $more  durationMs [expr {max(0,[clock clicks -milliseconds]-$began)}]]
        } 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 {
                accessRetentionPhase $token finished
                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 more $more phaseMilliseconds [dict get $state phaseMilliseconds]]
                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 {[:dict getdef ${:settings} partitionInterval 0]>0} {
            return
        }
        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 partition [$store partitionInfo]
            foreach key {partitionInterval partitionCount bootstrapFrom availableFrom historicalBackfillEnabled} {
                if {[dict exists $partition $key]} {
                    dict set control $key [dict get $partition $key]
                }
            }
            set pressure [$store maintenancePressure]
            set oldLive [:dict getdef $control live {}]
            set floor [expr {max([$store stateGet retentionFloor 0],wide(floor(($now-[dict get ${:settings} retention])/60.0))*60)}]
            if {[:dict getdef ${:settings} partitionInterval 0]>0
                && ![:dict getdef ${:settings} historicalBackfill false]} {
                set floor [expr {max($floor,[$store stateGet bootstrapFrom $floor])}]
            }
            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 [:dict getdef $control recentFrom [expr {$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 [:dict getdef $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 {}
        }
        set historyAllowed [expr {![dict get $pressure expired]
                                  && ![dict get $pressure capacityPressure]}]
        dict set control historyDeferred [expr {!$historyAllowed}]
        dict set control maintenancePressure $pressure
        if {$caught && $due && $historyAllowed && [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
                if {[dict exists $backfill ingestComplete]} {
                    dict set control availableFrom [dict get $historical from]
                }
            }
        }
        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 {
            set partition [$store partitionInfo]
            foreach key {partitionInterval partitionCount bootstrapFrom historicalBackfillEnabled} {
                if {[dict exists $partition $key]} {
                    dict set control $key [dict get $partition $key]
                }
            }
            accessIngestionProgress $run finalizing save-control
            $store statePut continuous $control
            catch {
                accessIngestionCheckpoint $control
            }
            accessIngestionProgress $run finalizing rollup-backfill
            if {$historyAllowed && [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