Class ::nsmcp::LogService (public)

 ::nx::Class ::nsmcp::LogService

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

Testcases:

No testcase defined.

Source code:

    :property {analytics {}}
    :property correlation:required
    :property authorizer:required
    :property redactor:required
    :property peerRedactor:required
    :property store:required
    :property queue:required
    :property {lifetime:integer 900}
    :property {maxJobs:integer 128}
    :property {accessScanMaxBytes 17179869184}
    :variable mutex
    :method init {} {
        if {${:lifetime} < 1 || ${:maxJobs} < 1} {error "positive log service limits required"}
        ::nsmcp::accessScanBudget ${:accessScanMaxBytes}
        set :mutex [ns_mutex create]
        if {${:queue} ni [ns_job queues]} {ns_job create ${:queue} 1}
    }
    :method accessBudget {} {
        # A service retained across definition reload may predate this property.
        if {[info exists :accessScanMaxBytes]} {return [::nsmcp::accessScanBudget ${:accessScanMaxBytes}]}
        return [::nsmcp::accessScanBudget [ns_config ns/server/[ns_info server]/module/nsmcp accessScanMaxBytes 16GiB]]
    }
    :method require {context} {${:authorizer} require $context diagnostics.logs.read}
    :method owned {context record} {
        return [expr {[dict get $record principal] eq [$context principal] &&
            [dict get $record server] eq [$context server] && [dict get $record expires] > [clock seconds]}]
    }
    :public method resolveWindow {context handle} {
        :require $context
        ns_mutex lock ${:mutex}
        try {
            foreach key [nsv_array names ${:store}] {
                set record [nsv_get ${:store} $key]
                if {![:owned $context $record]} continue
                if {[dict exists $record windowHandle] && [dict get $record windowHandle] eq $handle} {return [dict get $record window]}
                if {$key eq $handle} {return -code error -errorcode {NSMCP RPC -32602} "This is a jobHandle; use its windowHandle for access_summary"}
                if {[dict exists $record mappings $handle]} {return -code error -errorcode {NSMCP RPC -32602} "This is a resultHandle; use it with log_results, not access_summary"}
            }
            return -code error -errorcode {NSMCP RPC -32602} "Unknown or expired windowHandle; start a new interval"
        } finally {ns_mutex unlock ${:mutex}}
    }
    :public method start {context operation {identifier {}}} {
        :require $context
        if {$operation ni {errors requestContext requestContexts accessSummary accessFileSummary errorSummary routingSummary}} {error "invalid log operation"}
        if {$operation eq "requestContext" && ![::nsmcp::logRecordParser validRequestIdentifier $identifier]} {
            return -code error -errorcode {NSMCP RPC -32602} "Invalid request identifier"
        }
        if {$operation eq "requestContexts"} {
            set ids [dict get $identifier identifiers]
            if {[llength $ids] < 1 || [llength $ids] > 20} {return -code error -errorcode {NSMCP RPC -32602} "Expected 1..20 identifiers"}
            foreach id $ids {
                if {![::nsmcp::logRecordParser validRequestIdentifier $id]} {return -code error -errorcode {NSMCP RPC -32602} "Expected dash-delimited identifiers"}
            }
        }
        if {$operation eq "accessFileSummary"} {
            dict set identifier scanLimits [dict create bytes [:accessBudget]]
        }
        set requiresRuntime [expr {$operation eq "routingSummary"}]
        if {$requiresRuntime} {${:authorizer} require $context diagnostics.runtime.read}
        set requiresPeers [expr {($operation eq "accessSummary" &&
            ([dict exists $identifier peer] || ([dict exists $identifier includePeerAddresses] && [dict get $identifier includePeerAddresses])))
            || ($operation eq "routingSummary" && [dict get $identifier options includePeerAddresses])}]
        if {$requiresPeers} {${:authorizer} require $context diagnostics.logs.peers.read}
        ns_mutex lock ${:mutex}
        try {
            foreach key [nsv_array names ${:store}] {
                set record [nsv_get ${:store} $key]
                if {[dict get $record expires] <= [clock seconds] && [dict get $record status] in {complete failed}} {
                    if {[dict exists $record identifier scanState] && [dict exists $record operation] && [dict get $record operation] eq "accessFileSummary"} {::nsmcp::accessScanCleanup [dict get $record identifier scanState]}
                    nsv_unset ${:store} $key
                }
            }
            if {[llength [nsv_array names ${:store}]] >= ${:maxJobs}} {error "log job limit reached"}
            set handle [ns_crypto::randombytes -encoding hex 32]
            set record [dict create principal [$context principal] server [$context server]  expires [expr {[clock seconds]+${:lifetime}}] status queued operation $operation identifier $identifier mappings {} requiresPeers $requiresPeers requiresRuntime $requiresRuntime]
            if {$operation eq "accessFileSummary"} {
                dict set record progress [::nsmcp::accessScanProgress {} queued [dict create files {} index 0 examinedBytes 0 limits [dict get $identifier scanLimits]] {}]
            }
            set windowHandle {}
            if {$operation eq "accessSummary"} {
                set windowHandle [ns_crypto::randombytes -encoding hex 32]
                dict set record windowHandle $windowHandle
                dict set record window [dict create from [dict get $identifier from] to [dict get $identifier to]]
            }
            nsv_set ${:store} $handle $record
            try {
                ns_job queue -detached ${:queue} [list [self] run $handle]
            } on error {message options} {
                nsv_unset ${:store} $handle
                return -options $options $message
            }
        } finally {ns_mutex unlock ${:mutex}}
        set data [list jobHandle string $handle status string queued server string [ns_info server]]
        if {$windowHandle ne ""} {lappend data windowHandle string $windowHandle}
        if {$operation eq "accessFileSummary"} {lappend data progress object [dict get $record progress]}
        return [list {} object $data]
    }
    :method summary {snapshot mappingsVar} {
        upvar 1 $mappingsVar mappings
        set handle [ns_crypto::randombytes -encoding hex 32]
        dict set mappings $handle [dict get $snapshot resultHandle]
        set triples [list resultHandle string $handle server string [ns_info server]]
        foreach key {resultCount generation sourceBytes} {lappend triples $key number [dict get $snapshot $key]}
        foreach key {scanIncomplete pending cacheFull} {lappend triples $key boolean [dict get $snapshot $key]}
        if {[dict exists $snapshot limitReached]} {lappend triples limitReached boolean [dict get $snapshot limitReached]}
        if {[dict exists $snapshot cursor]} {lappend triples examinedBytes number [dict get $snapshot cursor]}
        return $triples
    }
    # Progress updates do not copy continuation state or enqueue more work.
    :public method accessProgress {handle fields} {
        ns_mutex lock ${:mutex}
        try {
            if {![nsv_exists ${:store} $handle]} {return}
            set record [nsv_get ${:store} $handle]
            if {[dict get $record status] ne "running"} {return}
            dict set record progress $fields
            nsv_set ${:store} $handle $record
        } finally {ns_mutex unlock ${:mutex}}
    }
    :public method run {handle {operation {}} {identifier {}}} {
        #
        # Execute a scan slice using private state retained under its job handle.
        # The optional arguments accept jobs queued before a live reload.
        # @param handle principal-bound job handle
        # @param operation legacy queued operation
        # @param identifier legacy queued arguments
        #
        ns_mutex lock ${:mutex}
        try {
            if {![nsv_exists ${:store} $handle]} return
            set record [nsv_get ${:store} $handle]
            if {[dict get $record status] in {complete failed}} return
            if {[dict exists $record operation]} {
                set operation [dict get $record operation]
                set identifier [dict get $record identifier]
            }
            dict set record status running
            nsv_set ${:store} $handle $record
        } finally {ns_mutex unlock ${:mutex}}
        set mappings {}
        try {
            if {$operation eq "errors"} {
                if {![catch {dict get $identifier severity} severity]} {
                    set snapshot [${:correlation} errors $severity [dict get $identifier literal] [dict get $identifier limit]]
                } else {
                    # Jobs queued before reload carried only the severity string.
                    set snapshot [${:correlation} errors {*}[expr {$identifier eq "" ? {} : [list $identifier]}]]
                }
                set data [:summary $snapshot mappings]
                set searchSnapshots [list $snapshot]
            } elseif {$operation eq "accessFileSummary"} {
                if {[dict get $record expires]<=[clock seconds]} {
                    if {[dict exists $identifier scanState]} {::nsmcp::accessScanCleanup [dict get $identifier scanState]}
                    error "access-log summary expired before completion"
                }
                set summary [${:correlation} accessFileSummary $identifier ${:redactor} [list [self] accessProgress $handle]]
                ns_mutex lock ${:mutex}
                try {
                    set latest [nsv_get ${:store} $handle]
                    if {[dict exists $latest progress]} {dict set record progress [dict get $latest progress]}
                } finally {ns_mutex unlock ${:mutex}}
                set data [lindex $summary 2]
                set continuation [lindex $summary 3]
                if {$continuation ne ""} {
                    dict set identifier scanState $continuation
                    dict set record identifier $identifier
                    ns_mutex lock ${:mutex}
                    try {nsv_set ${:store} $handle $record} finally {ns_mutex unlock ${:mutex}}
                    ns_job queue -detached ${:queue} [list [self] run $handle]
                    return
                }
            } elseif {$operation eq "errorSummary"} {
                set summary [${:correlation} errorSummary $identifier ${:redactor}]
                set data [lindex $summary 2]
                set continuation [lindex $summary 3]
                if {$continuation ne ""} {
                    if {[dict get $record expires] <= [clock seconds]} {error "summary job expired before completion"}
                    dict set identifier scanState $continuation
                    dict set record operation $operation
                    dict set record identifier $identifier
                    dict set record progress [list examinedBytes number [dict get $continuation examined]  sourceBytes number [dict get $continuation sourceBytes]]
                    ns_mutex lock ${:mutex}
                    try {nsv_set ${:store} $handle $record} finally {ns_mutex unlock ${:mutex}}
                    ns_job queue -detached ${:queue} [list [self] run $handle]
                    return
                }
                set updated {}; set budget 24
                foreach {key type value} $data {
                    if {$key eq "groups"} {
                        set groups {}
                        foreach {index groupType fields} $value {
                            set samples {}; set category {}
                            foreach {field fieldType item} $fields {
                                if {$field eq "category"} {set category $item}
                                if {$field eq "requestSamples"} {
                                    foreach {n sampleType sampleFields} $item {
                                        set sample {}
                                        foreach {k t v} $sampleFields {dict set sample $k $v}
                                        lappend samples $sample
                                    }
                                }
                            }
                            if {$category eq "duplicate-parameter"} {
                                set correlation [list available boolean false reason string {Access analytics is not configured.}]
                                set selected {}
                                if {${:analytics} ne ""} {
                                    if {$budget <= 0} {
                                        set correlation [list available boolean false reason string {The 24-request correlation budget was exhausted.}]
                                    } else {
                                        set selected [lrange $samples 0 [expr {min(11,$budget-1)}]]
                                        incr budget -[llength $selected]
                                        set correlation [list available boolean false reason string {The retained access cache could not be queried.}]
                                        if {![catch {${:analytics} correlateWarnings $selected ${:redactor}} result]} {set correlation [lindex $result 2]}
                                    }
                                }
                                lappend correlation omittedSamples number [expr {[llength $samples]-[llength $selected]}]
                                lappend fields accessCorrelation object $correlation
                            }
                            lappend groups $index $groupType $fields
                        }
                        set value $groups
                    }
                    lappend updated $key $type $value
                }
                set data $updated
            } elseif {$operation eq "routingSummary"} {
                if {[dict get $record expires]<=[clock seconds]} {error "routing scan expired before completion"}
                set permissions {diagnostics.logs.read diagnostics.runtime.read}
                if {[dict get $record requiresPeers]} {lappend permissions diagnostics.logs.peers.read}
                set context [::nsmcp::RequestContext new -principal [dict get $record principal]  -server [dict get $record server] -targets [list [dict get $record server]] -permissions $permissions]
                try {
                    set step [[dict get $identifier tool] scanStep $context $identifier]
                } finally {$context destroy}
                set data [lindex [lindex $step 0] 2]
                dict set record progress [lindex $step 2]
                set continuation [lindex $step 1]
                if {$continuation ne ""} {
                    dict set record operation $operation
                    dict set record identifier $continuation
                    ns_mutex lock ${:mutex}
                    try {nsv_set ${:store} $handle $record} finally {ns_mutex unlock ${:mutex}}
                    ns_job queue -detached ${:queue} [list [self] run $handle]
                    return
                }
            } elseif {$operation eq "accessSummary"} {
                if {${:analytics} eq ""} {error "access analytics is not configured"}
                ${:analytics} ingest [dict get $identifier from]
                set summary [${:analytics} summary $identifier]
                # Jobs queued before a live reload have no interval handle yet.
                if {![dict exists $record windowHandle]} {
                    dict set record windowHandle [ns_crypto::randombytes -encoding hex 32]
                    dict set record window [dict create from [dict get $identifier from] to [dict get $identifier to]]
                }
                dict set summary windowHandle [dict get $record windowHandle]
                set from [clock format [dict get $summary from] -timezone UTC -format {%Y-%m-%d %H:%M:%S}]
                set to [clock format [dict get $summary to] -timezone UTC -format {%Y-%m-%d %H:%M:%S}]
                dict set summary intervalUTC "$from to $to UTC"
                set data [::nsmcp::accessSummaryTriples $summary]
            } else {
                if {$operation eq "requestContexts"} {
                    set context [${:correlation} requestContexts [dict get $identifier identifiers] [dict get $identifier limit]]
                    set ids {}; set n 0
                    foreach id [dict get $context requestIdentifiers] {lappend ids $n string $id; incr n}
                    set data [list server string [ns_info server] requestIdentifiers array $ids]
                } else {
                    set context [${:correlation} requestContext $identifier]
                    set data [list server string [ns_info server] requestIdentifier string $identifier]
                }
                lappend data system object [:summary [dict get $context system] mappings]
                set access {}; set index 0
                dict for {source snapshot} [dict get $context access] {
                    lappend access $index object [:summary $snapshot mappings]
                    incr index
                }
                lappend data access array $access
                set searchSnapshots [list [dict get $context system] {*}[dict values [dict get $context access]]]
            }
            if {[info exists searchSnapshots]} {
                set incomplete false; set examined 0; set total 0
                foreach snapshot $searchSnapshots {
                    incr examined [dict get $snapshot cursor]
                    incr total [dict get $snapshot sourceBytes]
                    if {[dict get $snapshot scanIncomplete] && ![dict get $snapshot cacheFull]
                        && (![dict exists $snapshot limitReached] || ![dict get $snapshot limitReached])} {set incomplete true}
                }
                if {$incomplete} {
                    if {[dict get $record expires] <= [clock seconds]} {error "log search expired before completion"}
                    dict set record progress [list examinedBytes number $examined sourceBytes number $total]
                    ns_mutex lock ${:mutex}
                    try {nsv_set ${:store} $handle $record} finally {ns_mutex unlock ${:mutex}}
                    ns_job queue -detached ${:queue} [list [self] run $handle]
                    return
                }
            }
            dict set record status complete
            dict set record data $data
            dict set record mappings $mappings
        } on error {message options} {
            if {$operation eq "accessFileSummary" && [dict exists $identifier scanState]} {::nsmcp::accessScanCleanup [dict get $identifier scanState]}
            if {$operation eq "accessFileSummary"} {
                ns_mutex lock ${:mutex}
                try {
                    set latest [nsv_get ${:store} $handle]
                    if {[dict exists $latest progress]} {dict set record progress [dict get $latest progress]}
                } finally {ns_mutex unlock ${:mutex}}
            }
            if {$operation eq "accessFileSummary" && [dict exists $record progress]} {
                set fields {}
                foreach {key type value} [dict get $record progress] {
                    if {$key eq "phase"} {set value failed}
                    if {$key eq "updatedAt"} {set value [clock seconds]}
                    lappend fields $key $type $value
                }
                dict set record progress $fields
            }
            dict set record status failed
            # No source paths, log contents, or exception details in client errors.
            ns_log error "nsmcp: background log operation failed: $message"
        }
        # Release potentially large continuation state on success or failure.
        dict unset record identifier
        dict unset record operation
        ns_mutex lock ${:mutex}
        try {nsv_set ${:store} $handle $record} finally {ns_mutex unlock ${:mutex}}
    }
    :public method poll {context handle} {
        :require $context
        ns_mutex lock ${:mutex}
        try {
            if {![nsv_exists ${:store} $handle] || ![:owned $context [nsv_get ${:store} $handle]]} {
                foreach key [nsv_array names ${:store}] {
                    set record [nsv_get ${:store} $key]
                    if {![:owned $context $record]} continue
                    if {[dict exists $record mappings $handle]} {return -code error -errorcode {NSMCP RPC -32602} "This is a resultHandle; pass it as resultHandle to read a page"}
                    if {[dict exists $record windowHandle] && [dict get $record windowHandle] eq $handle} {return -code error -errorcode {NSMCP RPC -32602} "This is a windowHandle; use it with access_summary"}
                }
                return -code error -errorcode {NSMCP RPC -32602} "Unknown or expired job handle"
            }
            set record [nsv_get ${:store} $handle]
            if {[dict exists $record requiresRuntime] && [dict get $record requiresRuntime]} {${:authorizer} require $context diagnostics.runtime.read}
            if {[dict exists $record requiresPeers] && [dict get $record requiresPeers]} {${:authorizer} require $context diagnostics.logs.peers.read}
            set data [list jobHandle string $handle status string [dict get $record status] server string [ns_info server]]
            if {[dict exists $record windowHandle]} {lappend data windowHandle string [dict get $record windowHandle]}
            if {[dict exists $record progress]} {
                set showProgress [expr {[dict get $record status] ne "complete"}]
                foreach {key type value} [dict get $record progress] {
                    if {$key eq "phase"} {set showProgress true}
                }
                if {$showProgress} {lappend data progress object [dict get $record progress]}
            }
            if {[dict get $record status] eq "complete"} {lappend data result object [dict get $record data]}
            if {[dict get $record status] eq "failed"} {lappend data error string {Log search failed; consult the server log.}}
            return [list {} object $data]
        } finally {ns_mutex unlock ${:mutex}}
    }
    :public method results {context handle cursor limit includePeers} {
        :require $context
        if {$includePeers} {${:authorizer} require $context diagnostics.logs.peers.read}
        set internal {}
        ns_mutex lock ${:mutex}
        try {
            foreach key [nsv_array names ${:store}] {
                set record [nsv_get ${:store} $key]
                if {[:owned $context $record] && [dict exists $record mappings $handle]} {
                    set internal [dict get $record mappings $handle]; break
                }
            }
        } finally {ns_mutex unlock ${:mutex}}
        if {$internal eq ""} {
            ns_mutex lock ${:mutex}
            try {
                foreach key [nsv_array names ${:store}] {
                    set record [nsv_get ${:store} $key]
                    if {![:owned $context $record]} continue
                    if {$key eq $handle} {return -code error -errorcode {NSMCP RPC -32602} "This is a jobHandle; pass it as jobHandle to poll the job"}
                    if {[dict exists $record windowHandle] && [dict get $record windowHandle] eq $handle} {return -code error -errorcode {NSMCP RPC -32602} "This is a windowHandle; use it with access_summary"}
                }
            } finally {ns_mutex unlock ${:mutex}}
            return -code error -errorcode {NSMCP RPC -32602} "Unknown or expired result handle"
        }
        set page [${:correlation} results $internal $cursor $limit]
        set policy [expr {$includePeers ? ${:peerRedactor} : ${:redactor}}]
        set page [$policy page $page]
        set records {}; set index 0
        foreach entry [dict get $page records] {
            set fields {}
            foreach key {timestamp processThread logThread severity requestIdentifier peerAddress method url protocol message pool timingSource recordType mutexName mutexOperation mutexMetric} {
                if {[dict exists $entry $key]} {lappend fields $key string [dict get $entry $key]}
            }
            foreach key {offset endOffset status responseBytes} {
                set value [dict get $entry $key]
                lappend fields $key [expr {$value eq "" ? "string" : "number"}] $value
            }
            foreach key {startTime acceptTime queueTime filterTime runTime mutexDurationSeconds} {
                if {[dict exists $entry $key]} {lappend fields $key number [dict get $entry $key]}
            }
            foreach key {complete truncated orphanContinuation redacted} {lappend fields $key boolean [dict get $entry $key]}
            lappend records $index object $fields
            incr index
        }
        set data [list server string [ns_info server] resultHandle string $handle records array $records nextCursor string [dict get $page nextCursor]]
        foreach key {resultCount generation} {lappend data $key number [dict get $page $key]}
        foreach key {more scanIncomplete pending cacheFull} {lappend data $key boolean [dict get $page $key]}
        if {[dict exists $page limitReached]} {lappend data limitReached boolean [dict get $page limitReached]}
        return [list {} object $data]
    }

XQL Not present:

Generic, PostgreSQL, Oracle