nsmcp::AccessRecordStoreSQLite method cRead (public)

 <instance of nsmcp::AccessRecordStoreSQLite> cRead fid file \
    offset stop budget limits redactor ingestOnly [ batched ]

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

Read a bounded source slice through the C access-log scanner. Use a connection-local temporary table to materialize raw scanner fields, then cook and redact valid requests. Raw fields are never written to the canonical store. Leave an incomplete tail uncommitted.

Parameters:

fid (required)

Stored source-file identifier.

file (required)

Discovered source descriptor.

offset (required)

Starting source byte offset.

stop (required)

Planned exclusive source boundary.

budget (required)

Maximum source bytes to scan in this invocation.

limits (required)

Scanner limits, including the maximum line size.

redactor (required)

Redaction object used when cooking requests.

ingestOnly (required)

True for ingestion; false allows time-dependent records to request the Tcl scanner fallback.

batched (optional, defaults to "false")

True to construct prepared rows; false to construct SQL reference tuples.

Returns:

A list of end offset, cooked tuples, pending flag, fallback flag and parsing-loss counts.

Testcases:

No testcase defined.

Source code:

:accessIngestionObserve batch-capabilities
if {![llength [info commands ns_accesslog]]} {
    error "C scanner ingestion requires nslog's C scan API"
}
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 prepare-raw-stage
${: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 open-c-scanner
set scanner [ns_accesslog scan open {*}$scanOptions]
set begin $offset
set tuples {}
set pending false
set counts {malformed 0 oversized 0 skipped 0}
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 scan-c
    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 stage-raw-batch
    ${:db} transaction {
        ${:db} batchInsert -mapping $mapping {temp nsmcp_scan_batch} $batch
    }
    :accessIngestionObserve cook-records
    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 {
            # Consume decoded C fields and typed timing values directly.
            set timing [::nsmcp::accessRecordParser timingMicroseconds [dict get $item stamp]  [dict get $item start] [dict get $item a] [dict get $item q]  [dict get $item f] [dict get $item r]]
            set bytes [dict get $item bytes]
            if {$bytes eq {}} {
                set bytes 0
            }
            if {$timing eq {}} {
                set row {}
            } else {
                set row [::nsmcp::accessRecordParser requestFields [dict get $item peer]  [dict get $item identifier] [dict get $item usr] [dict get $item stamp]  [dict get $item request] [dict get $item status] $bytes  [dict get $item combined] [dict get $item agent] $timing]
            }
            if {$row ne {} && [dict exists $row timeDependent] && !$ingestOnly} {
                return [list $begin {} false true {malformed 0 oversized 0 skipped 0}]
            }
            if {$row eq {} || [dict exists $row timeDependent]} {
                set row {}
                set kind 1
            }
        }
        if {$kind==0} {
            lappend tuples [:tuple $fid $offset $next 0 $row $redactor $file $batched]
        } elseif {$kind==2} {
            dict incr counts oversized
        } else {
            dict incr counts malformed
        }
        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 false $counts]

XQL Not present:

Generic, PostgreSQL, Oracle