::nx::EnsembleObject ::ns::stats::SQLiteHistory::slot::_

Methods: All Methods Documented Methods Hide Methods
Source: Display Source Hide Source
Variables: Show Variables Hide Variables

Class Relations

class: ::nx::EnsembleObject

::nx::EnsembleObject create ::ns::stats::SQLiteHistory::slot::__history

Methods (to be applied on the object)

configure (scripted)

#
#   Validate the database pool and retention limits, initialize the
#   schema and start a separate persistence callback. Repeated
#   configuration with identical settings is harmless; changing
#   active settings requires disabling persistence first.
#   Return the resulting history status. Does not start sampling.
#   Normalize retention to fractional seconds and maxBytes to bytes
#   with ns_baseunit before validating the bounds.
#
set retention [ns_baseunit -time $retention]
set maxBytes [ns_baseunit -size $maxBytes]
if {$retention < 60 || $retention > 604800 || $maxBytes < 4194304} {
    error "retention must be 60..604800 seconds and maxBytes at least 4 MiB"
}
if {[info commands ns_db] eq ""} {
    error "history requires nsdb/nsdbsqlite"
}
set config [dict create  enabled true pool $pool retention $retention maxBytes $maxBytes  owner [ns_info server]  busy false scheduler {}  generation [clock microseconds] lastSaved {}  failures 0 lastFailure {} lastLoggedAt 0 gaps {}  mixins [:info object mixins] topology {} topologyAt 0]
set old [nsv_get ns_stats history]
if {$old ne "" && [dict get $old enabled]} {
    foreach k {pool retention maxBytes owner} {
        if {[dict get $old $k] ne [dict get $config $k]} {
            error "history already configured differently; disable before reconfiguring"
        }
    }
    return [:history status]
}
set db [:historyConnect $config]
try {
    :historySchema $db
} finally {
    ns_db releasehandle $db
}
:locked {
    set old [nsv_get ns_stats history]
    if {$old ne "" && [dict get $old enabled]} {
        error "history was concurrently configured; retry"
    }
    dict set config scheduler [ns_schedule_proc -thread 1 [self] history persist]
    nsv_set ns_stats history $config
}
return [:history status]

disable (scripted)

#
#   Stop future persistence callbacks without deleting stored history
#   or stopping sampling. An outstanding write may finish.
#   Return the resulting history status.
#
:locked {
    set config [nsv_get ns_stats history]
    if {$config ne ""} {
        if {[dict get $config enabled] && [dict get $config scheduler] ne ""} {
            ns_unschedule_proc [dict get $config scheduler]
        }
        dict set config enabled false
        dict set config scheduler {}
        nsv_set ns_stats history $config
    }
}
return [:history status]

query (scripted)

#
#   Return stored samples, associated topology, paging information and
#   coverage for an epoch interval. Queries run in the configuring
#   virtual server and use an operation-local database handle.
#   Historical server selection may include servers no longer active.
#
if {$from < 0
    || $to <= $from
    || $to - $from > 86400
    || $to > [clock seconds]
    || $limit < 1
    || $limit > 1000
    || $cursor < 0
} {
    error "expected an epoch window up to 24 hours, limit 1..1000 and nonnegative cursor"
}
if {[string length $server]>512 || [string first \x00 $server] >= 0} {
    error "invalid server name"
}
set config [nsv_get ns_stats history]
if {$config eq ""} {
    error "history is not configured"
}
if {[dict get $config owner] ne [ns_info server]} {
    error "query history from its configured owner server"
}
set db [:historyConnect $config]
set transaction false
try {
    :sql $db {BEGIN}
    set transaction true
    set low [expr {$from * 1000000}]
    set high [expr {$to * 1000000}]
    set count [expr {$limit + 1}]
    set rows [:sql $db "SELECT at,topology,payload FROM ns_stats_samples WHERE at >= $low AND at < $high AND at > $cursor ORDER BY at LIMIT $count"]
    set more [expr {[llength $rows]>$limit}]
    set rows [lrange $rows 0 [expr {$limit - 1}]]
    set records {}
    set gaps {}
    set restarts {}
    set previous {}
    set topology {}
    set resets 0
    set boundary [expr {max($low,$cursor)}]
    set operator [expr {$cursor >= $low ? "<=" : "<"}]
    set prior [:scalar $db "SELECT payload FROM ns_stats_samples WHERE at $operator $boundary ORDER BY at DESC LIMIT 1"]

    if {$prior ne ""} {
        set previous [:unpack $prior]
    }
    foreach row $rows {
        set record [:unpack [dict get $row payload]]
        if {![dict get $record sampling ready]} {
            incr resets
        }
        if {$previous ne ""} {
            if {[dict get $previous instance] ne [dict get $record instance]} {
                lappend restarts [dict get $record sampling to]
            }
            set delta [expr {([dict get $record timestamp]-[dict get $previous timestamp])/1000000.0}]
            if {$delta > 1.5 * max([dict get $previous sampling intervalSeconds],[dict get $record sampling intervalSeconds])} {
                lappend gaps [dict create from [dict get $previous sampling to] to [dict get $record sampling to]]
            }
        }
        set previous $record
        set id [dict get $row topology]
        dict set record topologyId $id
        if {![dict exists $topology $id]} {
            dict set topology $id [:selectServer  [:unpack [:scalar $db "SELECT payload FROM ns_stats_topology WHERE id=[:quote $id]"]]  $server false]
        }
        lappend records [:selectServer $record $server false]
    }

    if {!$more && $previous ne ""} {
        set following [:scalar $db "SELECT payload FROM ns_stats_samples WHERE at >= $high ORDER BY at LIMIT 1"]
        if {$following ne ""} {
            set nextRecord [:unpack $following]
            set delta [expr {([dict get $nextRecord timestamp]-[dict get $previous timestamp])/1000000.0}]
            if {$delta > 1.5 * max([dict get $previous sampling intervalSeconds],[dict get $nextRecord sampling intervalSeconds])} {
                lappend gaps [dict create from [dict get $previous sampling to] to [dict get $nextRecord sampling to]]
            }
        }
    }
    set bounds [lindex [:sql $db {SELECT coalesce(min(at),0) AS first,coalesce(max(at),0) AS last FROM ns_stats_samples}] 0]
    set events [:sql $db {SELECT at,first,last,n,reason FROM ns_stats_events ORDER BY id DESC}]
    set first [dict get $bounds first]
    set last [dict get $bounds last]
    set overlap false
    foreach event $events {
        if {[dict get $event first] < $high && [dict get $event last] >= $low} {
            set overlap true
        }
    }
    set through [:scalar $db {SELECT value FROM ns_stats_meta WHERE key='droppedThrough'}]
    if {$through ne "" && $through >= $low} {
        set overlap true
    }
    set next {}
    if {$more} {
        set next [dict get [lindex $rows end] at]
    }
    set complete [expr {$first > 0 && $first <= $low && $last >= $high && !$more && $cursor == 0
                        && !$overlap && $gaps eq "" && $restarts eq "" && $resets == 0}]
    set result [dict create  schemaVersion 1 from $from to $to records $records  topologies $topology nextCursor $next more $more  coverage [dict create  complete $complete  windowExamined [expr {!$more && $cursor == 0}]  retainedFrom [expr {$first / 1000000.0}]  retainedTo [expr {$last / 1000000.0}]  gaps $gaps  restartBoundaries $restarts  resetSamples $resets  events $events  evictionOrFailureOverlap $overlap  droppedThrough $through  writeFailuresSinceStart [dict get $config failures]]]
    :sql $db {COMMIT}
    set transaction false
    return $result
} finally {
    if {$transaction} {
        catch {:sql $db {ROLLBACK}}
    }
    ns_db releasehandle $db
}

save (scripted)

#
#   Persist the latest published sample if history is enabled, no write
#   is in progress and its timestamp has not already been saved.
#   Claim the write under the shared coordination lock, then perform
#   database operations outside that lock.
#
#   Store the compressed sample and associated topology in one
#   transaction, deduplicating samples by timestamp. Apply age and
#   storage-budget eviction, retain coverage events and remove topology
#   no longer referenced by retained samples.
#
#   On failure, roll back the transaction, retain the pending sample
#   for retry and update failure status. Repeated identical warnings
#   are throttled. A subsequent successful write records the failed
#   interval for coverage reporting.
#
#   Release the database handle and clear the write claim on completion.
#   Does not collect observations or advance the sampling baseline;
#   callers inspect history status for the persistence outcome.
#
set config {}
set generation {}
set stamp {}
:locked {
    set candidate [nsv_get ns_stats history]
    set sample [nsv_get ns_stats pair]
    if {$candidate ne "" && [dict get $candidate enabled] && ![dict get $candidate busy] && $sample ne ""} {
        set stamp [dict get $sample timestamp]
        if {$stamp ne [dict get $candidate lastSaved]} {
            set config $candidate
            set generation [dict get $config generation]
            dict set candidate busy true
            nsv_set ns_stats history $candidate
        }
    }
}
if {$config eq ""} {
    return
}
set db {}
set transaction false
set ok false
set errorMessage {}
try {
    # Collection and SQLite work happen outside the shared control mutex.
    set now [clock seconds]
    set topology [dict get $config topology]
    if {$topology eq "" || $now-[dict get $config topologyAt]>=60} {
        set topology [:describe]
        foreach k {pid bootTime uptimeSeconds} {dict unset topology process $k}
        dict set config topology $topology
        dict set config topologyAt $now
    }
    # Stable Tcl serialization is enough for a private cache key; store
    # the full topology in the table to detect the unlikely CRC collision.
    set topologyId [format %08x [zlib crc32 [encoding convertto utf-8 $topology]]]
    set record $sample
    dict unset record raw
    dict unset record cpu
    dict set record topologyId $topologyId
    set payload [encoding convertto utf-8 $record]
    set rawSize [string length $payload]
    if {$rawSize > 1048576} {
        error "history sample exceeds 1 MiB"
    }
    set packed [:pack $record]
    set packedTopology [:pack $topology]
    set size [expr {[string length $packed]+[string length $packedTopology]}]
    set db [:historyConnect $config]
    :historySchema $db
    :sql $db {BEGIN IMMEDIATE}
    set transaction true
    set cutoff [expr {([clock seconds]-[dict get $config retention])*1000000}]
    :historyDrop $db "at < $cutoff" age
    set page [:scalar $db {PRAGMA page_size}]
    set max [:scalar $db {PRAGMA max_page_count}]
    while {1} {
        set used [expr {([:scalar $db {PRAGMA page_count}] - [:scalar $db {PRAGMA freelist_count}]) * $page}]
        if {$used+3*$size < $max*$page*0.8} {
            break
        }
        if {![:scalar $db {SELECT count(*) FROM ns_stats_samples}]} {
            error "history metadata or sample exceeds storage budget"
        }
        :historyDrop $db {at IN (SELECT at FROM ns_stats_samples ORDER BY at LIMIT 100)} size
    }
    set stored [:scalar $db "SELECT payload FROM ns_stats_topology WHERE id=[:quote $topologyId]"]
    if {$stored ne "" && [:unpack $stored] ne $topology} {
        error "history topology fingerprint collision"
    }
    :sql $db "INSERT OR IGNORE INTO ns_stats_topology VALUES([:quote $topologyId],[:quote $packedTopology])"
    :sql $db "INSERT OR IGNORE INTO ns_stats_samples VALUES($stamp,[:quote [dict get $record instance]],[:quote $topologyId],[:quote $packed])"
    foreach gap [dict get $config gaps] {
        lassign $gap first last n
        :sql $db "INSERT INTO ns_stats_events(at,first,last,n,reason) VALUES($stamp,$first,$last,$n,'write-failure')"
    }
    :sql $db {DELETE FROM ns_stats_events WHERE id NOT IN (SELECT id FROM ns_stats_events ORDER BY id DESC LIMIT 64)}
    :sql $db {DELETE FROM ns_stats_topology WHERE id NOT IN (SELECT DISTINCT topology FROM ns_stats_samples)}
    :sql $db {COMMIT}
    set transaction false
    set ok true

} on error {message options} {
    set errorMessage $message
    if {$transaction} {
        catch {:sql $db {ROLLBACK}}
    }
    if {$message ne [dict get $config lastFailure] || [clock seconds]-[dict get $config lastLoggedAt] >= 60} {
        ns_log warning "ns_stats: history write failed: $message"
        dict set config lastLoggedAt [clock seconds]
    }
} finally {
    if {$db ne ""} {
        ns_db releasehandle $db
    }
    :locked {
        set current [nsv_get ns_stats history]
        if {$generation eq [dict get $current generation]} {
            dict set current busy false
            dict set current lastLoggedAt [dict get $config lastLoggedAt]
            if {$ok} {
                dict set current lastSaved $stamp
                dict set current lastFailure {}
                dict set current gaps {}
                dict set current topology [dict get $config topology]
                dict set current topologyAt [dict get $config topologyAt]
            } else {
                dict incr current failures
                dict set current lastFailure $errorMessage
                set gaps [dict get $current gaps]
                if {$gaps eq ""} {
                    set gaps [list [list $stamp $stamp 1]]
                } else {
                    lassign [lindex $gaps 0] first last n
                    set gaps [list [list $first $stamp [expr {$n+1}]]]
                }
                dict set current gaps $gaps
            }
            nsv_set ns_stats history $current
        }
    }
}

status (scripted)

#
#   Return shared history configuration and persistence status without
#   opening the database. Internal coordination and cached topology
#   fields are omitted.
#
set config [nsv_get ns_stats history]
if {$config eq ""} {
    return {enabled false}
}
dict unset config busy
dict unset config mixins
dict unset config topology
dict unset config gaps
return $config

Methods: All Methods Documented Methods Hide Methods
Source: Display Source Hide Source
Variables: Show Variables Hide Variables