nsmcp::AccessRecordSupport method accessRecordScan (protected)

 <instance of nsmcp::AccessRecordSupport> accessRecordScan paths \
    directory options redactor limits publish settings [ observer ]

Defined in /usr/local/ns/tcl/nsmcp/log-traffic-record-00-support.tcl

Parameters:

paths (required)

directory (required)

options (required)

redactor (required)

limits (required)

publish (required)

settings (required)

observer (optional)

Testcases:

No testcase defined.

Source code:

if {![dict exists $settings ingestion]} {
    dict set settings ingestion [ns_config ns/server/[ns_info server]/module/nsmcp accessRecordIngestion tcl]
}
if {[dict get $settings ingestion] ni {tcl c batch}} {
    error "accessRecordIngestion must be tcl or c (batch is a legacy alias for c)"
}
set options [accessFailureOptions $options]
set selection [accessFailureSelection $options]
set policy [list schema1 [::nsmcp::accessRecordParser cget -timingColumn]]
foreach key {queryParameters headerNames maskPeerAddresses maskUserIdentifiers replacement} {
    lappend policy [$redactor cget -$key]
}
if {[dict exists $settings policy] && [dict get $settings policy] ne $policy} {
    error "Record-store parsing/redaction policy changed; start a new analysis"
}
dict set settings policy $policy
set configured [:dict getdef $options scanLimits {}]
set limits [dict merge [accessScanDefaultLimits] [accessAggregateSettings] $configured $limits]
if {[dict exists $options scanState]} {
    set state [dict get $options scanState]
    if {![dict exists $state recordStore] || [dict get $state failedTrafficSelection] ne $selection} {
        error "Record-store selection changed; start a new analysis"
    }
    set settings [dict merge [dict create ingestion [dict get $settings ingestion]] [dict get $state recordStore]]
    dict set state recordStore $settings
    set limits [dict get $state limits]
} else {
    set state [accessScanDiscover $paths [dict get $options includeRotations] [dict get $limits files]]
    dict set state index 0
    dict set state stats [accessScanEmpty]
    dict set state examinedBytes 0
    dict set state parsedRecordBytes 0
    dict set state reusedBytes 0
    dict set state nativeSearch false
    dict set state recordStore $settings
    dict set state limits $limits
    dict set state failedTrafficSelection $selection
    # Compressed generations retain the established bounded scanner. No
    # partial persistent result is silently substituted for these sources.
    foreach file [dict get $state files] {
        if {[dict get $file compressed] && ![dict exists $options ingestOnly]} {
            return {}
        }
    }
    set files {}
    foreach file [dict get $state files] {
        if {[dict get $file compressed]} {
            dict set file status unsupported-compressed
        } elseif {[accessOrderedEnabled $options]} {
            accessOrderedPlan file state $options $limits $publish
        }
        dict set file archiveLo [dict get $file offset]
        lappend files $file
    }
    dict set state files $files
}
set store {}
set files [dict get $state files]
set index [dict get $state index]
try {
    set waitStart [clock milliseconds]
    :accessIngestionObserve -observer $observer open-store
    set store [:accessRecordStoreOpen $settings writer $observer]
    if {![$store available]} {
        dict incr state recordPoolWaitMilliseconds [expr {max(1,[clock milliseconds]-$waitStart)}]
        set waitMs [:dict getdef $settings waitMs 30000]
        if {[dict get $state recordPoolWaitMilliseconds]>=$waitMs} {
            error "record-store-pool-wait-limit"
        }
        accessScanProgress $publish waiting-record-store $state {}
        return [list {} object {} $state]
    }
    set batchedInsertion [$store batchInsertionAvailable]
    if {![:dict getdef $settings unified false]} {
        $store prune
    }
    set budget [expr {min([dict get $limits slice],[dict get $limits bytes]-[dict get $state examinedBytes])}]
    set started [clock milliseconds]
    set maxMs [:dict getdef $limits milliseconds 1000]
    while {$index<[llength $files] && $budget>0 && [clock milliseconds]-$started<$maxMs} {
        :accessIngestionObserve -observer $observer prepare-source
        set file [lindex $files $index]
        if {[dict get $file status] in {outside-window unsupported-compressed}} {
            incr index
            continue
        }
        accessScanUnchanged $file
        set fid [$store prepare $file]
        if {[dict exists $file archiveFid] && $fid ne [dict get $file archiveFid]} {
            error "record-store-generation-changed"
        }
        dict set file archiveFid $fid
        set offset [dict get $file offset]
        dict set file dropping [$store dropState $fid $offset]
        set end [expr {[dict exists $file scanEnd] ? [dict get $file scanEnd] : [dict get $file size]}]
        set covered [$store coveredEnd $fid $offset $end]
        if {$covered>$offset} {
            dict incr state reusedBytes [expr {$covered-$offset}]
            set offset $covered
        } else {
            set batchStarted [clock clicks -milliseconds]
            set stop [$store gapEnd $fid $offset $end]
            set begin $offset
            set tuples {}
            set pending false
            set counters {malformed 0 oversized 0 skipped 0}
            set useCScanner [expr {[dict exists $settings ingestion] && [dict get $settings ingestion] in {c batch} && ![dict get $file dropping] && [dict get $limits line]>=128 && [dict get $limits line]<=1048576}]
            if {$useCScanner} {
                lassign [$store cRead $fid $file $offset $stop $budget $limits $redactor [dict exists $options ingestOnly] $batchedInsertion] next tuples pending direct counters
                if {$direct eq "true"} {
                    return {}
                }
                # A framing budget may end before emitting a row. Retain the
                # established bounded Tcl reader for such exceptional spans.
                set useCScanner [expr {$next>$offset || $pending}]
                set offset $next
            }
            # The Tcl reader follows when C framing made no progress.
            if {!$useCScanner} {
                :accessIngestionObserve -observer $observer scan-tcl
                set f [open [dict get $file path] rb]
                seek $f $offset
                set readLimit [expr {min($stop,$offset+max([dict get $limits slice],[dict get $limits line]+1),$offset+[dict get $limits bytes]-[dict get $state examinedBytes])}]
                set reader [::nsmcp::LogSummaryReader new -channel $f -limit $readLimit -offset $offset -maxBytes [dict get $limits line]]
                set begin $offset
                set tuples {}
                set pending false
                set counters {malformed 0 oversized 0 skipped 0}
                set processed 0
                try {
                    while {$offset<$stop && $processed<512 && $offset-$begin<$budget} {
                        set line [$reader nextLine]
                        if {$line eq {}} {
                            break
                        }
                        if {[dict exists $options ingestOnly] && [dict get $file dropping]} {
                            set kind 3
                            set row {}
                            if {[dict get $line terminated]} {
                                dict set file dropping 0
                            }
                            dict incr counters skipped
                            incr processed
                            set offset [dict get $line end]
                            continue
                        }
                        if {![dict get $line terminated]} {
                            if {$readLimit<$stop} {
                                if {[dict exists $options ingestOnly] && [dict get $line truncated]} {
                                    dict set file dropping 1
                                    dict incr counters oversized
                                    incr processed
                                    set offset [dict get $line end]
                                    break
                                }
                                return {}
                            }
                            # A slice boundary can cut a line. Retry it from its
                            # original byte offset in the next slice; a true
                            # snapshot tail stays pending until a future query.
                            if {$stop==$end} {
                                set pending true
                            }
                            break
                        }
                        set next [dict get $line end]
                        set kind 0
                        set row {}
                        if {[dict get $line truncated]} {
                            set kind 2
                        } else {
                            set row [::nsmcp::accessRecordParser parse [dict get $line text]]
                            if {$row ne {} && [dict exists $row timeDependent]} {
                                if {[dict exists $options ingestOnly]} {
                                    set row {}
                                } else {
                                    return {}
                                }
                            }
                            if {$row eq {}} {
                                set kind 1
                            }
                        }
                        incr processed
                        if {$kind==0} {
                            lappend tuples [$store tuple $fid $offset $next 0 $row $redactor $file $batchedInsertion]
                        } elseif {$kind==2} {
                            dict incr counters oversized
                        } else {
                            dict incr counters malformed
                        }
                        set offset $next
                    }
                } finally {
                    $reader destroy
                    close $f
                }
            }
            accessScanUnchanged $file
            if {$offset>$begin} {
                :accessIngestionObserve -observer $observer prepare-checkpoint
                set checkpoint [dict merge [:accessRecordCheckpoint $store $fid $file $begin $offset] $counters]
                set batch [$store ingest $fid $checkpoint $tuples $batchedInsertion]
                dict set batch scannerBackend [expr {$useCScanner ? "c" : "tcl"}]
                dict set batch durationMs [expr {max(0,[clock clicks -milliseconds]-$batchStarted)}]
                dict set batch recordsPerSecond [expr {[dict get $batch durationMs]>0 ? 1000.0*[dict get $batch records]/[dict get $batch durationMs] : {}}]
                set consumed [expr {$offset-$begin}]
                dict incr state examinedBytes $consumed
                dict incr state parsedRecordBytes $consumed
                incr budget -$consumed
                dict set batch parsedBytes $consumed
                :accessIngestionObserve -observer $observer -bytes $consumed -batch $batch committed
                dict incr file scanReadBytes $consumed
            } elseif {!$pending && $stop<$end} {
                # Ensure a configured slice smaller than one complete line
                # cannot cause an endless zero-progress retry.
                error "record-store-slice-smaller-than-record"
            }
            if {$pending} {
                dict set file archivePending true
                set end $offset
            }
        }
        dict set file offset $offset
        if {$offset>=$end} {
            dict set file status complete
            incr index
        } else {
            dict set file status running
        }
        lset files [dict get $file fileIndex] $file
        dict set state files $files
        dict set state index $index
        accessScanProgress $publish ingesting-records $state $file $offset [dict get $file size]
        if {[dict get $state examinedBytes]>=[dict get $limits bytes]} {
            break
        }
    }
    if {$index<[llength $files]} {
        if {[dict get $state examinedBytes]<[dict get $limits bytes]} {
            return [list {} object {} $state]
        }
        for {set i $index} {$i<[llength $files]} {incr i} {
            set file [lindex $files $i]
            dict set file status scan-byte-limit
            lset files $i $file
        }
    }
    foreach file $files {
        if {[dict get $file status] ne "outside-window"} {
            accessScanUnchanged $file
        }
    }
    dict set state files $files
    if {[dict exists $options ingestOnly]} {
        dict set state ingestComplete true
        return [list {} object {} $state]
    }
    accessScanProgress $publish querying-records $state {}
    $store scope $files
    lassign [$store stats $options $redactor $limits] stats aggregate selected
    lassign [$store samples $selected $files $redactor] samples successes
    dict set stats samples $samples
    dict set stats successSamples $successes
    foreach file $files {
        if {[dict exists $file archivePending]} {
            dict incr stats pendingRecords
        }
    }
    dict set state stats $stats
    dict set state aggregateResult $aggregate
    set result [accessScanResult $state $options $limits $redactor]
    set fields [lindex $result 2]
    lappend fields recordStorage object [list backend string sqlite persistent boolean true parsedBytesThisJob number [dict get $state parsedRecordBytes] locatorProbeBytes number [expr {[dict get $state examinedBytes]-[dict get $state parsedRecordBytes]}]  reusedBytes number [dict get $state reusedBytes] diskBytes number [$store diskBytes] maxBytes number [dict get $settings maxBytes]  retentionSeconds number [dict get $settings retention] retentionScope string unused-source-generations temporarySortDiskIncluded boolean false reuseValidation string {File identity, original prefix and committed range boundary anchors; assumes immutable interiors of append-only log generations.}]
    lset result 2 $fields
    # Run retention only after querying this job's fixed scopes. Expiration
    # never removes current active-query rows before they are summarized.
    $store finish
    if {![:dict getdef $settings unified false]} {
        $store prune
    }
    return $result
} on error {message opts} {
    if {[lrange [dict get $opts -errorcode] 0 1] eq {NSMCP RECORD_STORE}} {
        set failure [dict create class [lindex [dict get $opts -errorcode] 2] diagnosticId [lindex [dict get $opts -errorcode] 3]]
    } elseif {$store ne {}} {
        set failure [$store diagnostic $message $opts]
    } else {
        set failure [accessAggregationDiagnostic record-store $message $opts]
    }
    if {[dict get $failure class] in {database-locked database-busy}} {
        dict incr state recordPoolWaitMilliseconds 100
        set waitMs [:dict getdef $settings waitMs 30000]
        if {[dict get $state recordPoolWaitMilliseconds]<$waitMs} {
            accessScanProgress $publish waiting-record-store $state {}
            return [list {} object {} $state]
        }
    }
    return -code error -errorcode [list NSMCP RECORD_STORE [dict get $failure class]] "Persistent access-record analysis failed ([dict get $failure class], diagnostic [dict get $failure diagnosticId]); committed ranges remain available."
} finally {
    if {$store ne {}} {
        $store close
    }
}

XQL Not present:

Generic, PostgreSQL, Oracle