nsmcp::UnifiedAccessAnalytics method refreshLocked (protected)

 <instance of nsmcp::UnifiedAccessAnalytics> refreshLocked

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

Testcases:

No testcase defined.

Source code:

:retireLegacy
set start [clock milliseconds]; set now [clock seconds]
set store [accessRecordStoreOpen ${:settings}]
if {![$store available]} {$store close; return {busy true}}
try {
    $store prune
    set control [$store stateGet continuous {}]
    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 {
        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 {
            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}]
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}} 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
set store [accessRecordStoreOpen ${:settings}]
if {![$store available]} {$store close; return {busy true}}
try {
    $store statePut continuous $control
    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