Class ::nsmcp::CachedLogSearch (public)

 ::nx::Class ::nsmcp::CachedLogSearch

Defined in /usr/local/ns/tcl/nsmcp/log-summary.tcl

Testcases:

No testcase defined.

Source code:

    :property cacheDirectory:required
    :property {maxSearches:integer 16}
    :property {maxMatchesPerCall:integer 1000}
    :property {maxCacheBytes:integer 67108864}

    :variable recordBuffer {}
    :variable recordBufferFrom 0
    :variable recordBufferEnd 0
    :variable scanStarted 0
    :variable scanPhases {}
    :variable scanIO {}

    :method beginScanMetrics {} {
        set :recordBuffer {}
        set :recordBufferFrom 0
        set :recordBufferEnd 0
        set :scanStarted [clock clicks -microseconds]
        set :scanPhases {sourceRead 0 candidateSearch 0 recordFraming 0 metadata 0 resultMaterialization 0}
        set :scanIO {sourceReadBytes 0 sourceReadCalls 0 resultReadBytes 0 resultReadCalls 0 resultWriteBytes 0 temporaryWriteBytes 0}
    }

    :method readSource {channel count} {
        set started [clock clicks -microseconds]
        set bytes [read $channel $count]
        dict incr :scanPhases sourceRead [expr {max(0,[clock clicks -microseconds]-$started)}]
        dict incr :scanIO sourceReadBytes [string length $bytes]
        dict incr :scanIO sourceReadCalls
        return $bytes
    }

    :method finishScanMetrics {scope {previous {}}} {
        set elapsed [expr {max(0,[clock clicks -microseconds]-${:scanStarted})}]
        set accounted 0
        foreach value [dict values ${:scanPhases}] {
            incr accounted $value
        }
        dict set :scanPhases other [expr {max(0,$elapsed-$accounted)}]
        if {$previous ne ""} {
            dict for {key value} [dict get $previous phaseMicroseconds] {
                dict incr :scanPhases $key $value
            }
            dict for {key value} [dict get $previous io] {
                dict incr :scanIO $key $value
            }
            incr elapsed [dict get $previous totalMicroseconds]
        }
        return [dict create scope $scope totalMicroseconds $elapsed  phaseMicroseconds ${:scanPhases} io ${:scanIO}]
    }

    :method init {} {
        next
        if {${:maxSearches} < 1 || ${:maxMatchesPerCall} < 1 || ${:maxCacheBytes} < 1} {
            error "search cache limits must be positive"
        }
        set :cacheDirectory [file normalize ${:cacheDirectory}]
        if {![file exists ${:cacheDirectory}]} {
            file mkdir ${:cacheDirectory}
            file attributes ${:cacheDirectory} -permissions 0700
        }
    }

    :method removeQueries {queries} {
        dict for {key query} $queries {
            file delete -force [dict get $query resultFile] [dict get $query pendingFile]
        }
    }

    :public method flush {} {
        #
        # Discard cached search state for this log source.
        #
        # Serialize the update and clean up retained result resources owned by the
        # source.
        #
        # @return The cache cleanup result.
        #

        ns_mutex lock ${:mutex}
        try {
            set key ${:cacheKey}:search
            if {[nsv_exists ${:store} $key]} {
                set meta [nsv_get ${:store} $key]
                :removeQueries [dict get $meta queries]
                dict set meta queries {}
                dict incr meta generation
                nsv_set ${:store} $key $meta
            }
        } finally {
            ns_mutex unlock ${:mutex}
        }
    }

    # Cursors are byte offsets in the completed JSONL cache, never source paths.
    :public method results {handle {cursor 0} {limit 20}} {
        #
        # Retrieve a bounded page from an unexpired cached search.
        #
        # Validate the handle and page bounds under the mutex and retain coverage
        # and truncation metadata.
        #
        # @param handle Retained job or conversation handle; ownership and expiry
        #        checks still apply.
        # @param cursor Zero-based cursor into retained results. Defaults to '0'.
        # @param limit Maximum number of records or entries to return. Defaults to
        #        '20'.
        #
        # @return The cached page dictionary.
        #

        if {![regexp {^[0-9a-f]{64}$} $handle]} {
            error "invalid result handle"
        }
        if {![string is entier -strict $cursor] || $cursor < 0 ||
            ![string is integer -strict $limit] || $limit < 1 || $limit > 100} {
            error "expected a nonnegative cursor and a limit between 1 and 100"
        }
        ns_mutex lock ${:mutex}
        try {
            :beginScanMetrics
            set key ${:cacheKey}:search
            set query {}
            if {[nsv_exists ${:store} $key]} {
                set meta [nsv_get ${:store} $key]
                dict for {q candidate} [dict get $meta queries] {
                    if {[dict exists $candidate resultHandle] &&
                        [dict get $candidate resultHandle] eq $handle} {
                        set query $candidate
                        break
                    }
                }
            }
            if {$query eq ""} {
                return -code error -errorcode {NSMCP RESULT UNKNOWN} "unknown or expired result handle"
            }
            # Result handles created before a live reload have no cap flag.
            set query [dict merge {limitReached false} $query]
            file stat ${:path} current
            set probe [open ${:path} rb]
            try {
                set prefix [:readSource $probe [string length [dict get $meta prefix]]]
            } finally {
                close $probe
            }
            if {[list $current(dev) $current(ino)] ne [dict get $meta identity] ||
                $current(size) < [dict get $meta observedSize] || $prefix ne [dict get $meta prefix]} {
                error "result handle expired after log rotation; repeat the search"
            }
            set end [dict get $query fileBytes]
            if {$cursor > $end} {
                error "cursor is beyond the published results"
            }
            set f [open [dict get $query resultFile] rb]
            try {
                if {$cursor > 0} {
                    seek $f [expr {$cursor-1}]
                    if {[read $f 1] ne "\n"} {
                        error "cursor must be a record boundary"
                    }
                }
                set reader [::nsmcp::BufferedResultReader new -channel $f -limit $end -offset $cursor]
                try {
                    set records {}
                    set bytes 0
                    set next $cursor
                    set started [clock clicks -microseconds]
                    while {$next<$end && [llength $records]<$limit} {
                        set line [$reader next]
                        set length [expr {[dict get $line end]-[dict get $line start]}]
                        if {$bytes+$length>2097152} {
                            break
                        }
                        incr bytes $length
                        lappend records [ns_json parse [encoding convertfrom utf-8 [dict get $line raw]]]
                        set next [dict get $line end]
                    }
                    set stats [$reader statistics]
                    dict set :scanPhases resultMaterialization [expr {
                        max(0,[clock clicks -microseconds]-$started-[dict get $stats readMicroseconds])
                    }]
                    dict incr :scanPhases sourceRead [dict get $stats readMicroseconds]
                    dict set :scanIO resultReadBytes [dict get $stats readBytes]
                    dict set :scanIO resultReadCalls [dict get $stats readCalls]
                } finally {
                    $reader destroy
                }
            } finally {
                close $f
            }
            set more [expr {$next < $end}]
            return [dict create resultHandle $handle records $records scanMetrics [:finishScanMetrics page]  nextCursor [expr {$more ? $next : ""}] more $more  resultCount [dict get $query resultCount]  scanIncomplete [dict get $query scanIncomplete] pending [dict get $query pending]  cacheFull [dict get $query cacheFull] limitReached [dict get $query limitReached] generation [dict get $meta generation]]
        } finally {
            ns_mutex unlock ${:mutex}
        }
    }

    :method temporaryFile {} {
        set channel [file tempfile name [file join ${:cacheDirectory} nsmcp-]]
        close $channel
        return $name
    }

    :method lineStart {channel offset} {
        set position $offset
        if {$offset>${:recordBufferFrom} && $offset<=${:recordBufferEnd}} {
            set newline [string last \n ${:recordBuffer} [expr {$offset-${:recordBufferFrom}-1}]]
            if {$newline>=0} {
                return [expr {${:recordBufferFrom}+$newline+1}]
            }
            set position ${:recordBufferFrom}
        }
        while {$position > 0} {
            set start [expr {max(0, $position-32768)}]
            seek $channel $start
            set block [:readSource $channel [expr {$position-$start}]]
            set :recordBuffer $block
            set :recordBufferFrom $start
            set :recordBufferEnd $position
            set newline [string last \n $block]
            if {$newline >= 0} {
                return [expr {$start+$newline+1}]
            }
            set position $start
        }
        return 0
    }

    :method physicalLine {channel limit} {
        set start [tell $channel]
        set position $start
        set sample {}
        set clipped false
        set terminated false
        while {$position<$limit} {
            if {$position<${:recordBufferFrom} || $position>=${:recordBufferEnd}} {
                seek $channel $position
                set :recordBuffer [:readSource $channel [expr {min(65536,$limit-$position)}]]
                set :recordBufferFrom $position
                set :recordBufferEnd [expr {$position+[string length ${:recordBuffer}]}]
                if {${:recordBuffer} eq {}} {
                    break
                }
            }
            set index [expr {$position-${:recordBufferFrom}}]
            set newline [string first \n ${:recordBuffer} $index]
            set end [expr {$newline<0 ? [string length ${:recordBuffer}] : $newline}]
            set count [expr {$end-$index}]
            set room [expr {max(0,${:maxMessageBytes}-[string length $sample])}]
            if {$count>$room} {
                set clipped true
            }
            if {$room>0} {
                append sample [string range ${:recordBuffer} $index [expr {min($end-1,$index+$room-1)}]]
            }
            incr position [expr {$count+($newline>=0)}]
            if {$newline>=0} {
                set terminated true
                break
            }
        }
        # Retain read-ahead while preserving the channel-position contract used
        # by record joining and reverse continuation lookup.
        seek $channel $position
        set text [encoding convertfrom utf-8 [string trimright $sample \r]]
        regsub -all {\x1b\[[0-9;]*m} $text {} text
        return [dict create start $start end $position text $text  truncated $clipped terminated $terminated]
    }

    :method recordStart {channel offset limit} {
        set start [:lineStart $channel $offset]
        while {1} {
            seek $channel $start
            set line [:physicalLine $channel $limit]
            if {$start == 0 || [string index [dict get $line text] 0] ne ":"} {
                return $start
            }
            set start [:lineStart $channel [expr {$start-1}]]
        }
    }

    :method record {channel start limit} {
        seek $channel $start
        set event [dict create offset $start endOffset $start message {}  complete false truncated false orphanContinuation false]
        set first true
        while {[tell $channel] < $limit} {
            set line [:physicalLine $channel $limit]
            set text [dict get $line text]
            if {!$first && ![${:recordParser} continuation [dict get $event message] $text]} {
                dict set event complete true
                seek $channel [dict get $line start]
                return $event
            }
            if {$first} {
                dict set event orphanContinuation [expr {[string index $text 0] eq ":"}]
            } else {
                set text \n$text
            }
            set event [:appendText $event $text]
            if {[dict get $line truncated]} {
                dict set event truncated true
            }
            dict set event endOffset [dict get $line end]
            set first false
        }
        return $event
    }

    :method encodeEvent {event} {
        set event [dict merge $event [${:recordParser} metadata [dict get $event message]]]
        set triples {}
        foreach key {timestamp processThread logThread severity requestIdentifier peerAddress method url protocol recordType} {
            lappend triples $key string [dict get $event $key]
        }
        foreach key {mutexName mutexOperation mutexMetric} {
            if {[dict exists $event $key]} {
                lappend triples $key string [dict get $event $key]
            }
        }
        if {[dict exists $event mutexDurationSeconds]} {
            lappend triples mutexDurationSeconds number [dict get $event mutexDurationSeconds]
        }
        foreach key {status responseBytes} {
            set value [dict get $event $key]
            lappend triples $key [expr {$value eq "" ? "string" : "number"}] $value
        }
        foreach key {offset endOffset} {
            lappend triples $key number [dict get $event $key]
        }
        lappend triples message string [dict get $event message]
        foreach key {complete truncated orphanContinuation} {
            lappend triples $key boolean [dict get $event $key]
        }
        return [ns_json value -type object -- $triples]
    }

    :public method search {literal {severity {}} {correlationField {}} {maxResults 0}} {
        #
        # Search raw records for literal text or exact request identifiers.
        #
        # Use one literal key for ordinary searches or a bounded identifier batch
        # for the systemMany/accessMany correlation modes. Retain partial-scan and
        # result-limit evidence.
        #
        # @param literal Literal search text; an empty value disables this
        #        additional text filter.
        # @param severity System-log severity name to match.
        # @param correlationField Exact header-field correlation mode, including
        #        systemMany/accessMany batches.
        # @param maxResults Retained record cap; zero keeps the legacy slice
        #        behavior. Defaults to '0'.
        #
        # @return The cached snapshot with explicit partial-scan and result-limit
        #         flags.
        #

        if {![string is integer -strict $maxResults] || $maxResults < 0 || $maxResults > 100} {
            error "result limit must be 0..100"
        }
        set many [expr {$correlationField in {systemMany accessMany}}]
        if {$many} {
            if {[llength $literal] < 1 || [llength $literal] > 20} {
                error "expected 1..20 request identifiers"
            }
            set literal [lsort -unique $literal]
            foreach id $literal {
                set check [expr {$correlationField eq "accessMany" ? "validConnectionIdentifier" : "validRequestIdentifier"}]
                if {[string length [encoding convertto utf-8 $id]] > 512 || ![${:recordParser} $check $id]} {
                    error "correlation requires complete request identifiers"
                }
            }
        }

        if {$correlationField ni {{} system access systemMany accessMany}} {
            error "expected an empty, system or access correlation field"
        }
        set identifierCheck [expr {$correlationField eq "access" ? "validConnectionIdentifier" : "validRequestIdentifier"}]
        if {!$many && $correlationField ne "" && ![${:recordParser} $identifierCheck $literal]} {
            error "correlation requires a complete request identifier"
        }
        if {$severity ne "" && ![regexp {^[A-Za-z][A-Za-z0-9_()-]*$} $severity]} {
            error "expected a log severity name"
        }
        if {!$many && ($literal eq "" || [string length [encoding convertto utf-8 $literal]] > 4096 ||
            [string first \n $literal] >= 0 || [string first \r $literal] >= 0 ||
            [string first \x00 $literal] >= 0)} {
            error "expected a nonempty, single-line literal search key of at most 4096 bytes"
        }
        ns_mutex lock ${:mutex}
        try {
            :beginScanMetrics
            set metaKey ${:cacheKey}:search
            set queryKey [ns_crypto::md string -digest sha256 -- [list log-metadata-v8 $literal $severity $correlationField $maxResults]]
            file stat ${:path} before
            set source [open ${:path} rb]
            set scanner [open ${:path} rb]
            set bounded {}
            try {
                fconfigure $source -translation binary -encoding binary
                fconfigure $scanner -translation binary -encoding binary
                file stat ${:path} opened
                set identity [list $before(dev) $before(ino)]
                if {$identity ne [list $opened(dev) $opened(ino)]} {
                    error "log rotated while opening; retry"
                }
                set keys [expr {$many ? $literal : [list $literal]}]
                set bounded [::nsmcp::BoundedLogScanner new -channel $scanner -limit $before(size)  -literals $keys -cacheDirectory ${:cacheDirectory}]
                set reset false
                set generation 1
                if {[nsv_exists ${:store} $metaKey]} {
                    set meta [nsv_get ${:store} $metaKey]
                    set generation [dict get $meta generation]
                    set prefix [dict get $meta prefix]
                    if {$identity ne [dict get $meta identity] ||
                        $before(size) < [dict get $meta observedSize] ||
                        [:readSource $source [string length $prefix]] ne $prefix} {
                        :removeQueries [dict get $meta queries]
                        set reset true
                        incr generation
                    }
                } else {
                    set reset true
                }
                if {$reset} {
                    seek $source 0
                    set meta [dict create identity $identity generation $generation  observedSize $before(size) queries {}  prefix [:readSource $source [expr {min(256,$before(size))}]]]
                }
                if {[dict get $meta prefix] eq "" && $before(size) > 0} {
                    seek $source 0
                    dict set meta prefix [:readSource $source [expr {min(256,$before(size))}]]
                }
                set queries [dict get $meta queries]
                if {![dict exists $queries $queryKey]} {
                    if {[dict size $queries] >= ${:maxSearches}} {
                        set oldest {}
                        set oldestTime {}
                        dict for {key query} $queries {
                            if {$oldest eq "" || [dict get $query used] < $oldestTime} {
                                set oldest $key
                                set oldestTime [dict get $query used]
                            }
                        }
                        :removeQueries [dict create $oldest [dict get $queries $oldest]]
                        dict unset queries $oldest
                    }
                    dict set queries $queryKey [dict create resultFile [:temporaryFile]  pendingFile [:temporaryFile] resultHandle [ns_crypto::randombytes -encoding hex 32] cursor 0 resultCount 0 fileBytes 0  pending false used [clock clicks -milliseconds] cacheFull false limitReached false  sourceBytes -1 scanIncomplete true]
                }
                set query [dict get $queries $queryKey]
                # Publish ownership before processing, so an interrupted scan's
                # files remain discoverable for invalidation/eviction.
                dict set meta queries $queries
                nsv_set ${:store} $metaKey $meta
                if {[dict get $query limitReached] || ([dict get $query sourceBytes] == $before(size) &&
                    ![dict get $query scanIncomplete])} {
                    dict set query used [clock clicks -milliseconds]
                    dict set meta queries $queryKey $query
                    dict set meta observedSize $before(size)
                    nsv_set ${:store} $metaKey $meta
                    return [dict merge $query [dict create generation $generation reset $reset]]
                }
                set output [open [dict get $query resultFile] r+]
                set pending [open [dict get $query pendingFile] w]
                try {
                    fconfigure $output -encoding utf-8 -translation lf
                    fconfigure $pending -encoding utf-8 -translation lf
                    # Discard writes beyond the last published checkpoint after
                    # a failed scan, rather than appending duplicate results.
                    ns_ftruncate $output [dict get $query fileBytes]
                    seek $output 0 end
                    dict set query pending false
                    set cursor [dict get $query cursor]
                    set matched 0
                    set byteBudgetReached false
                    set scanEnd [expr {$many ? min($before(size),$cursor+8388608) : $before(size)}]
                    while {$cursor < $scanEnd && $matched < ${:maxMatchesPerCall} &&
                           ![dict get $query cacheFull] && ($maxResults == 0 || [dict get $query resultCount] < $maxResults)} {
                        # Shared bounded candidate search; record framing and exact
                        # metadata checks below retain their format-specific behavior.
                        set started [clock clicks -microseconds]
                        set priorRead [dict get [$bounded statistics] readMicroseconds]
                        set found [$bounded find $cursor $scanEnd]
                        set scannerStats [$bounded statistics]
                        dict incr :scanPhases candidateSearch [expr {
                            max(0,[clock clicks -microseconds]-$started-[dict get $scannerStats readMicroseconds]+$priorRead)
                        }]
                        if {$found < 0 && $many && $scanEnd < $before(size)} {
                            set overlap 0
                            foreach id $literal {
                                set overlap [expr {max($overlap,[string length $id]-1)}]
                            }
                            set cursor [expr {$scanEnd-$overlap}]
                            set byteBudgetReached true
                            break
                        }
                        if {$found < 0 || $found >= $before(size)} {
                            set cursor [:recordStart $source [expr {$before(size)-1}] $before(size)]
                            break
                        }
                        set started [clock clicks -microseconds]
                        set priorRead [dict get ${:scanPhases} sourceRead]
                        set start [:recordStart $source $found $before(size)]
                        set event [:record $source $start $before(size)]
                        dict incr :scanPhases recordFraming [expr {
                            max(0,[clock clicks -microseconds]-$started-[dict get ${:scanPhases} sourceRead]+$priorRead)
                        }]
                        incr matched
                        set started [clock clicks -microseconds]
                        set metadata [${:recordParser} metadata [dict get $event message]]
                        dict incr :scanPhases metadata [expr {max(0,[clock clicks -microseconds]-$started)}]
                        set accepted [expr {$severity eq "" || [dict get $metadata severity] eq $severity}]
                        if {$correlationField eq "system"} {
                            set accepted [expr {$accepted && [dict get $metadata requestIdentifier] eq $literal}]
                        } elseif {$correlationField eq "access"} {
                            set accepted [expr {$accepted && [${:recordParser} accessRequestIdentifier [dict get $event message]] eq $literal}]
                        } elseif {$correlationField eq "systemMany"} {
                            set accepted [expr {$accepted && [dict get $metadata requestIdentifier] in $literal}]
                        } elseif {$correlationField eq "accessMany"} {
                            set accepted [expr {$accepted && [${:recordParser} accessRequestIdentifier [dict get $event message]] in $literal}]
                        }
                        if {!$accepted} {
                            if {![dict get $event complete]} {
                                set cursor $start
                                break
                            }
                            set cursor [dict get $event endOffset]
                            continue
                        }
                        set started [clock clicks -microseconds]
                        set json [:encodeEvent $event]
                        dict incr :scanPhases resultMaterialization [expr {max(0,[clock clicks -microseconds]-$started)}]
                        if {![dict get $event complete]} {
                            set started [clock clicks -microseconds]
                            puts $pending $json
                            dict incr :scanPhases resultMaterialization [expr {max(0,[clock clicks -microseconds]-$started)}]
                            dict incr :scanIO resultWriteBytes [string length [encoding convertto utf-8 $json\n]]
                            dict set query pending true
                            set cursor $start
                            break
                        }
                        set bytes [string length [encoding convertto utf-8 $json\n]]
                        if {[tell $output]+$bytes > ${:maxCacheBytes}} {
                            dict set query cacheFull true
                            set cursor $start
                            break
                        }
                        set started [clock clicks -microseconds]
                        puts $output $json
                        dict incr :scanPhases resultMaterialization [expr {max(0,[clock clicks -microseconds]-$started)}]
                        dict incr :scanIO resultWriteBytes $bytes
                        dict incr query resultCount
                        set cursor [dict get $event endOffset]
                    }
                    set started [clock clicks -microseconds]
                    flush $output
                    flush $pending
                    dict incr :scanPhases resultMaterialization [expr {max(0,[clock clicks -microseconds]-$started)}]
                    dict set query fileBytes [tell $output]
                    dict set query cursor $cursor
                    dict set query used [clock clicks -milliseconds]
                } finally {
                    close $output
                    close $pending
                }
                file stat ${:path} after
                seek $source 0
                set prefix [dict get $meta prefix]
                if {$identity ne [list $after(dev) $after(ino)] ||
                    $after(size) < $before(size) || [:readSource $source [string length $prefix]] ne $prefix} {
                    error "log changed during scan; retry"
                }
                # cursor may point to the pending final message. Distinguish that
                # deliberate recheck from a match-budget backlog.
                if {$many && $cursor >= $scanEnd && $scanEnd < $before(size)} {
                    set byteBudgetReached true
                }
                dict set query limitReached [expr {$maxResults > 0 && [dict get $query resultCount] >= $maxResults}]
                dict set query scanIncomplete [expr {$matched >= ${:maxMatchesPerCall} || [dict get $query cacheFull] || [dict get $query limitReached] || $byteBudgetReached}]
                set stats [$bounded statistics]
                dict incr :scanPhases sourceRead [dict get $stats readMicroseconds]
                dict incr :scanIO sourceReadBytes [dict get $stats readBytes]
                dict incr :scanIO sourceReadCalls [dict get $stats readCalls]
                dict set query scanMetrics [:finishScanMetrics query [:dict getdef $query scanMetrics {}]]
                dict set query sourceBytes $before(size)
                dict set meta observedSize $before(size)
                dict set meta queries $queryKey $query
                nsv_set ${:store} $metaKey $meta
                return [dict merge $query [dict create generation $generation reset $reset]]
            } finally {
                if {$bounded ne ""} {
                    $bounded destroy
                }
                close $source
                close $scanner
            }
        } finally {
            ns_mutex unlock ${:mutex}
        }
    }

XQL Not present:

Generic, PostgreSQL, Oracle