nsmcp::AccessRecordStoreSQLite method nativeRead (public)

 <instance of nsmcp::AccessRecordStoreSQLite> nativeRead fid file \
    offset stop budget limits redactor ingestOnly

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

Parameters:

fid (required)

file (required)

offset (required)

stop (required)

budget (required)

limits (required)

redactor (required)

ingestOnly (required)

Testcases:

No testcase defined.

Source code:

accessIngestionObserve ${:observer} batch-capabilities
if {![llength [info commands ns_accesslog]]} {error "access batch ingestion requires nslog's scan API"}
if {![dict get [${:db} batchCapabilities] supported]} {
    error "access batch ingestion requires a batch-capable SQLite driver"
}
set mapping {offset off endOffset stop timestamp stamp peerAddress peer threadName identifier user usr requestLine request status status responseBytes bytes userAgent agent startTime start acceptTime a queueTime q filterTime f runTime r parseStatus parsed diagnostic diagnostic}
accessIngestionObserve ${:observer} stage-schema
${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_scan_batch;
    CREATE TEMP TABLE nsmcp_scan_batch(off INTEGER,stop INTEGER,stamp TEXT,peer TEXT,identifier TEXT,usr TEXT,request TEXT,status INTEGER,bytes INTEGER,agent TEXT,start INTEGER,a INTEGER,q INTEGER,f INTEGER,r INTEGER,parsed TEXT,diagnostic TEXT)}
set timing [::nsmcp::accessRecordParser cget -timingColumn]
set timing [dict get {-2 none -1 auto 0 0 2 2} $timing]
set scanOptions [list -filename [dict get $file path] -offset $offset -endoffset $stop -maxlinebytes [dict get $limits line]]
if {$timing ne "auto"} {lappend scanOptions -timingcolumn $timing}
accessIngestionObserve ${:observer} open-scanner
set scanner [ns_accesslog scan open {*}$scanOptions]
set begin $offset; set tuples {}; set pending false
try {
    # The scanner enforces this invocation's planned end even if the
    # active log grew after discovery or this is a covered-range gap.
    # Keep framing state within this call, but serialize only completed
    # record offsets between worker slices. Never commit a snapshot tail.
    accessIngestionObserve ${:observer} scan-native
    set batch [ns_accesslog scan next $scanner -maxrecords 512  -maxscanbytes [expr {min(1073741824,max(1,$budget))}]  -maxbytes [expr {max(8388608,4*[dict get $limits line]+2048)}] -timeout 100ms]
    set info [ns_db batchinfo $batch]
    accessIngestionObserve ${:observer} batchinsert
    ${:db} transaction {${:db} batchInsert {temp nsmcp_scan_batch} $mapping $batch}
    accessIngestionObserve ${:observer} cook-batch
    set escapes [list \\ \\\\ \" \\\" \n {\n} \r {\r} \t {\t}]
    foreach item [${:db} execute {SELECT *,agent IS NOT NULL AS combined FROM temp.nsmcp_scan_batch ORDER BY off}] {
        set next [dict get $item stop]
        if {$next>$stop} {set pending true; break}
        set kind 0; set row {}
        if {[dict get $item parsed] ne "ok"} {
            set kind [expr {[dict get $item diagnostic] eq "oversized" ? 2 : 1}]
        } else {
            # SQLite converts lexical seconds to integer microseconds.
            # Reuse the existing policy for start-time plausibility,
            # authentication, query stripping and unavailable phases.
            set fields {}
            if {[dict get $item combined]} {
                lappend fields {""} "\"[string map $escapes [dict get $item agent]]\""
            }
            if {[dict get $item start] ne {}} {
                set phases {}
                foreach key {start a q f r} {
                    set us [dict get $item $key]
                    if {$us eq {}} {set us 0}
                    lappend phases [format {%d.%06d} [expr {$us/1000000}] [expr {$us%1000000}]]
                }
                lappend fields "\"[join $phases { }]\""
            }
            set bytes [dict get $item bytes]
            if {$bytes eq {}} {set bytes -}
            set row [::nsmcp::accessRecordParser fields [dict get $item peer] [dict get $item identifier]  [string map $escapes [dict get $item usr]] [dict get $item stamp]  [string map $escapes [dict get $item request]] [dict get $item status] $bytes {} $fields]
            if {$row ne {} && [dict exists $row timeDependent] && !$ingestOnly} {
                return [list $begin {} false true]
            }
            if {$row eq {} || [dict exists $row timeDependent]} {set row {}; set kind 1}
        }
        lappend tuples [:tuple $fid $offset $next $kind $row $redactor $file]
        set offset $next
    }
    if {[dict get $info state] eq "pending"} {set pending true}
} finally {
    ns_accesslog scan close $scanner
    ${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_scan_batch}
}
return [list $offset $tuples $pending]

XQL Not present:

Generic, PostgreSQL, Oracle