nsmcp::UnifiedAccessAnalytics method refreshLocked (protected)

 <instance of nsmcp::UnifiedAccessAnalytics> refreshLocked \
    [ run ]

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

Parameters:

run (optional)

Testcases:

No testcase defined.

Source code:

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