Class ::ns::stats::Sampler (public)

 ::nx::Class ::ns::stats::Sampler

Defined in

Testcases:

No testcase defined.

Source code:

namespace eval ::ns::stats {}
::nsf::object::alloc ::nx::Class ::ns::stats::Sampler {}
::ns::stats::Sampler public method calculate {current previous interval instance} {
        #
        # Derive interval diagnostics from current and previous snapshots.
        # Return the current observations augmented with counter deltas,
        # rates, average request phase times and thread CPU percentages.
        # Rates use measured elapsed time; interval records the configured
        # sampling cadence, and instance identifies the sampler lifetime.
        #
        # Interval results require a previous snapshot from the same process
        # lifetime and positive elapsed time. Missing baselines or decreased
        # counters yield empty derived values. CPU comparisons match thread
        # identifiers and creation times; percentages use one core as 100%.
        #
        # Does not collect observations or modify shared state.
        #
        # @param current Current snapshot dictionary, including its epoch-microsecond timestamp
        #     and process identity.
        # @param previous Previous snapshot dictionary, or empty when no baseline exists.
        # @param interval Configured sampling cadence in seconds; rates use actual elapsed time
        #     instead.
        # @param instance Sampler lifetime identifier attached to the calculated sample.
        # @return Current observations augmented with sampling metadata, counter deltas, rates,
        #     average timings and CPU percentages; unavailable derived values remain empty.
        #
        set stamp [dict get $current timestamp]
        set elapsed {}
        set ready false
        set reset false
        if {$previous ne ""
            && [dict get $previous process pid] eq [dict get $current process pid]
            && [dict get $previous process bootTime] eq [dict get $current process bootTime]
        } {
            set elapsed [expr {($stamp-[dict get $previous timestamp])/1000000.0}]
            if {$elapsed > 0} {
                set ready true
            } else {
                set reset true
            }
        }
        set result $current
        dict set result instance $instance
        dict set result sampling [dict create  ready $ready  from {}  to [expr {$stamp/1000000.0}]  elapsedSeconds $elapsed  ageSeconds 0  reset $reset  intervalSeconds $interval]
        if {$ready} {
            dict set result sampling from [expr {[dict get $previous timestamp]/1000000.0}]
        }

        dict for {s server} [dict get $current servers] {
            dict for {p entry} [dict get $server pools] {
                set deltas {}
                set rates {}
                set invalid {}
                dict for {k v} [dict get $entry counters] {
                    if {$k ni {requests queued spools spooled dropped queuetime filtertime runtime tracetime}} {
                        continue
                    }
                    set d {}
                    if {$ready && [dict exists $previous servers $s pools $p counters $k]} {
                        set d [:delta $v [dict get $previous servers $s pools $p counters $k]]
                        if {$d eq ""} {
                            lappend invalid $k
                        }
                    }
                    dict set deltas $k $d
                    dict set rates $k [expr {$d eq "" ? "" : $d/$elapsed}]
                }
                dict set result servers $s pools $p deltas $deltas
                dict set result servers $s pools $p rates $rates
                dict set result servers $s pools $p resetCounters $invalid
                set timings {}
                set n [expr {[dict exists $deltas requests] ? [dict get $deltas requests] : ""}]

                foreach k {queuetime filtertime runtime tracetime} {
                    set d [expr {[dict exists $deltas $k] ? [dict get $deltas $k] : ""}]
                    dict set timings $k [expr {$n eq "" || $n == 0 || $d eq "" ? "" : $d/$n}]
                }
                dict set result servers $s pools $p timingAverageSeconds $timings
            }
        }

        dict for {thread entry} [dict get $current drivers] {
            set deltas {}
            set rates {}
            set invalid {}
            foreach k {received partial spooled errors} {
                set d {}
                if {$ready && [dict exists $entry counters $k] && [dict exists $previous drivers $thread counters $k]} {
                    set d [:delta [dict get $entry counters $k] [dict get $previous drivers $thread counters $k]]
                    if {$d eq ""} {
                        lappend invalid $k
                    }
                }
                dict set deltas $k $d
                dict set rates $k [expr {$d eq "" ? "" : $d/$elapsed}]
            }
            dict set result drivers $thread deltas $deltas
            dict set result drivers $thread rates $rates
            dict set result drivers $thread resetCounters $invalid
        }

        set percentages {}
        set totals {}
        set matched 0

        dict for {id entry} [dict get $current cpu] {
            set percent {}
            if {$ready && [dict exists $previous cpu $id]} {
                set old [dict get $previous cpu $id]
                set user [:delta [dict get $entry user] [dict get $old user]]
                set system [:delta [dict get $entry system] [dict get $old system]]
                if {$user ne "" && $system ne ""} {
                    set percent [expr {100.0*($user+$system)/($elapsed*1000000)}]
                    incr matched
                }
            }
            set name [string trim [dict get $entry name] -]
            set group other
            foreach {pattern kind} {driver:* driver conn:* connection writer:* writer spooler:* spooler} {
                if {[string match $pattern $name]} {
                    set group $kind
                    break
                }
            }
            dict set percentages $id [dict create name $name group $group percent $percent]
            if {$percent ne ""} {
                set total [expr {[dict exists $totals $group] ? [dict get $totals $group] : 0.0}]
                dict set totals $group [expr {$total+$percent}]
            }
        }
        if {[dict exists $current process cpuCapacity throttling]} {
            dict for {path row} [dict get $current process cpuCapacity throttling] {
                foreach field {periods throttledPeriods throttledSeconds} {
                    set delta {}
                    if {$ready && [dict exists $row $field] && [dict exists $previous process cpuCapacity throttling $path $field]} {
                        set old [dict get $previous process cpuCapacity throttling $path $field]
                        if {[dict get $row $field]>=$old} {set delta [expr {[dict get $row $field]-$old}]}
                    }
                    dict set row delta[string toupper [string index $field 0]][string range $field 1 end] $delta
                }
                dict set row throttledPeriodPercent {}
                if {[dict get $row deltaPeriods] ne "" && [dict get $row deltaPeriods]>0 && [dict get $row deltaThrottledPeriods] ne ""} {
                    dict set row throttledPeriodPercent [expr {100.0*[dict get $row deltaThrottledPeriods]/[dict get $row deltaPeriods]}]
                }
                dict set result process cpuCapacity throttling $path $row
            }
        }
        dict set result cpuPercent [dict create  scope process  matchedThreads $matched  observedThreads [dict size [dict get $current cpu]]  groups $totals  threads $percentages]
        return $result
    }
::ns::stats::Sampler protected method delta {now:double before:double} {
        # A decreased cumulative counter has no usable interval delta.
        if {$now < $before} {
            return {}
        }
        expr {$now - $before}
    }
::ns::stats::Sampler protected method init {} {
        :locked {
            if {[nsv_get ns_stats control] eq ""} {
                nsv_set ns_stats control [dict create  running false interval 10 scheduler {} generation 0 busy false  instance "[pid]:[ns_info boottime]:[clock microseconds]"  errors 0 lastError {}]
            }
        }
    }
::ns::stats::Sampler protected method locked script {
        # Evaluate shared-state updates in the calling method's frame.
        uplevel 1 [list nsv_array eval ns_stats $script]
    }
::ns::stats::Sampler protected method retainSample sample {
        # Called while holding the existing ns_stats coordination lock.
        # Keep bounded endpoint CPU counters for display-window calculations.
        # Raw baselines and derived per-thread maps are not retained.
        set record [dict remove $sample raw]
        if {[dict exists $record cpuPercent threads]} {
            dict unset record cpuPercent threads
        }
        set records [nsv_get ns_stats recent]
        lappend records $record
        set cutoff [expr {[dict get $record timestamp]-900000000}]
        set dropped [nsv_get ns_stats recentDroppedThrough]
        while {[llength $records]>180
               || ([llength $records]>1 && [dict get [lindex $records 0] timestamp]<$cutoff)} {
            set dropped [dict get [lindex $records 0] timestamp]
            set records [lrange $records 1 end]
        }
        nsv_set ns_stats recent $records
        nsv_set ns_stats recentDroppedThrough $dropped
    }
::ns::stats::Sampler public method sample {{-server {}} {-window 0}} {
        #
        # Return the latest published sample, updating its reported age and
        # running state. Before any publication, return a dictionary with
        # sampling readiness false and no interval observations.
        #
        # An optional server argument restricts virtual-server data; CPU
        # measurements remain process-wide. Without -window, does not collect
        # observations or advance the baseline. With -window (0..900 seconds, time units
        # accepted), collect a fresh snapshot and compare it with the nearest retained
        # endpoint to that age (within half a sampling interval). Missing
        # history, gaps or sampler restarts leave readiness false. The
        # measured elapsedSeconds reports the actual selected period.
        # The internal raw baseline is omitted
        # from the returned dictionary.
        #
        # @param server Optional virtual-server name; empty returns all servers. CPU metrics
        #     remain process-wide.
        # @param window Requested display period, accepting ns_baseunit time syntax; 0..900
        #     seconds, default 0 for the latest published sample.
        # @return Sample dictionary with sampling readiness and age; a positive window adds
        #     freshly calculated interval metrics when a suitable continuous baseline exists.
        #
        if {$server ne "" && $server ni [ns_info servers]} {
            error "unknown server: $server"
        }
        set window [ns_baseunit -time $window]
        if {$window < 0 || $window > 900} {
            error "display window must be 0..900 seconds"
        }
        :locked {
            set result [nsv_get ns_stats pair]
            set recent [expr {$window > 0 ? [nsv_get ns_stats recent] : {}}]
        }
        if {$result eq ""} {
            return [dict create schemaVersion 1  sampling {ready false from {} to {} elapsedSeconds {} ageSeconds {} reset false}  servers {} availability {}]
        }
        if {$window > 0} {
            set current [:snapshot]
            set target [expr {[dict get $current timestamp]-$window*1000000}]
            set baseline {}
            set distance Inf
            foreach record $recent {
                if {[dict get $record timestamp] >= [dict get $current timestamp]
                    || [dict get $record instance] ne [dict get $result instance]} {
                    continue
                }
                set difference [expr {abs([dict get $record timestamp]-$target)}]
                if {$difference < $distance} {
                    set baseline $record
                    set distance $difference
                }
            }
            set cadence [dict get $result sampling intervalSeconds]
            set reason {}
            if {$baseline eq "" || $distance > 0.5*$cadence*1000000} {
                set baseline {}
                set reason insufficient-history
            } else {
                set previous $baseline
                foreach record $recent {
                    if {[dict get $record timestamp] <= [dict get $baseline timestamp]
                        || [dict get $record timestamp] > [dict get $current timestamp]} {
                        continue
                    }
                    if {[dict get $record instance] ne [dict get $result instance]
                        || ![dict get $record sampling ready]
                        || [dict get $record timestamp]-[dict get $previous timestamp]
                           > 1.5*max($cadence,[dict get $previous sampling intervalSeconds])*1000000} {
                        set reason sampling-gap
                        set baseline {}
                        break
                    }
                    set previous $record
                }
            }
            if {$baseline ne "" && [dict get $current timestamp]-[dict get $previous timestamp]
                > 1.5*$cadence*1000000} {
                set baseline {}
                set reason sampling-gap
            }
            # Older retained records may predate CPU baseline retention.
            if {$baseline ne "" && ![dict exists $baseline cpu]} {
                dict set baseline cpu {}
            }
            set result [:calculate $current $baseline $cadence [dict get $result instance]]
            dict set result sampling requestedSeconds $window
            dict set result sampling incompleteReason $reason
        }
        dict set result sampling running [dict get [nsv_get ns_stats control] running]
        dict set result sampling ageSeconds [expr {max(0,([clock microseconds]-[dict get $result timestamp])/1000000.0)}]
        if {[dict exists $result raw]} {dict unset result raw}
        # CPU is process-wide; filtering server pools never attributes CPU to a vhost.
        return [:selectServer $result $server]
    }
::ns::stats::Sampler public method {sampling history} {-from:integer,required -to:integer,required {-server {}} {-limit:integer 180} {-cursor:integer 0}} {
        #
        # Read the bounded process-wide recent sample buffer without SQLite.
        # Retains at most 180 compact samples and 15 minutes at publication.
        # Returns interval metrics, coverage and timestamp-based paging.
        # Raw/per-thread CPU counters are omitted. Reading does not collect,
        # start sampling, advance baselines or write persistent history.
        # The buffer survives sampler stop/start but not a process restart.
        #
        # @param from Inclusive lower bound in epoch seconds; must be nonnegative.
        # @param to Exclusive upper bound in epoch seconds; after from, at most 24 hours later
        #     and no later than now.
        # @param server Optional historical virtual-server name; empty returns all stored server
        #     data.
        # @param limit Maximum records per page, 1..180; default 180.
        # @param cursor Nonnegative timestamp cursor in microseconds, default 0; pass nextCursor
        #     unchanged for continuation.
        # @return Dictionary containing records, retained bounds, coverage, more and nextCursor
        #     for the in-memory history buffer.
        #
        if {$from<0 || $to<=$from || $to-$from>86400 || $to>[clock seconds]
            || $limit<1 || $limit>180 || $cursor<0} {
            error "expected an epoch window up to 24 hours, limit 1..180 and nonnegative cursor"
        }
        if {[string length $server]>512 || [string first \x00 $server]>=0} {
            error "invalid server name"
        }
        :locked {
            set retained [nsv_get ns_stats recent]
            set dropped [nsv_get ns_stats recentDroppedThrough]
        }
        set records {}
        set gaps {}
        set restarts {}
        set resets 0
        set previous {}
        set first {}
        set last {}
        set more false
        foreach record $retained {
            set stamp [dict get $record timestamp]
            if {$first eq ""} {
                set first $stamp
            }
            set last $stamp
            if {$stamp<$from*1000000 || $stamp<=$cursor} {
                set previous $record
                continue
            }
            if {$stamp>=$to*1000000} {
                # A gap crossing the upper bound must not certify coverage.
                if {$previous ne ""} {
                    set delta [expr {($stamp-[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]]
                    }
                }
                break
            }
            if {[llength $records]>=$limit} {
                set more true
                break
            }
            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 {($stamp-[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]]
                }
            }
            lappend records [:selectServer [dict remove $record cpu] $server false]
            set previous $record
        }
        # Retained bounds refer to the buffer, independently of the selected page.
        if {$retained ne ""} {
            set first [dict get [lindex $retained 0] timestamp]
            set last [dict get [lindex $retained end] timestamp]
        }
        set overlap [expr {$dropped ne "" && $dropped>=$from*1000000}]
        if {!$more && $cursor>0} {
            set window [:sampling history -from $from -to $to -server $server -limit 180]
            set coverage [dict get $window coverage]
            return [dict create schemaVersion 1 source memory from $from to $to records $records  nextCursor {} more false retentionSeconds 900 maxSamples 180 coverage $coverage]
        }
        set complete [expr {$first ne "" && $first<=$from*1000000 && $last>=$to*1000000
                             && !$more && $cursor==0 && !$overlap && $gaps eq ""
                             && $restarts eq "" && $resets==0}]
        set next {}
        if {$more} {
            set next [dict get [lindex $records end] timestamp]
        }
        return [dict create schemaVersion 1 source memory from $from to $to records $records  nextCursor $next more $more retentionSeconds 900 maxSamples 180  coverage [dict create complete $complete windowExamined [expr {!$more && $cursor==0}]  retainedFrom [expr {$first eq "" ? 0 : $first/1000000.0}]  retainedTo [expr {$last eq "" ? 0 : $last/1000000.0}]  gaps $gaps restartBoundaries $restarts resetSamples $resets  evictionOrFailureOverlap $overlap droppedThrough $dropped]]
    }
::ns::stats::Sampler public method {sampling start} {{-interval 10}} {
        #
        # Start the process-wide sampling callback at the requested interval
        # in seconds (default 10, range 1..3600). Time units and fractional
        # seconds are normalized with ns_baseunit. Repeated calls with the
        # same running interval reuse the scheduler; a different interval
        # requires stopping the sampler first.
        #
        # Clears the previous sample and baseline when starting a stopped
        # sampler. Returns the resulting status dictionary. Persistence
        # must be enabled separately.
        #
        # @param interval Sampling cadence, accepting ns_baseunit time syntax; 1..3600 seconds,
        #     default 10 seconds.
        # @return Resulting sampling status dictionary.
        #

        set interval [ns_baseunit -time $interval]
        if {$interval < 1 || $interval > 3600} {
            error "interval must be 1..3600 seconds"
        }
        :locked {
            set state [nsv_get ns_stats control]
            if {[dict get $state running]} {
                if {[dict get $state interval]!=$interval} {
                    error "sampler already running with a different interval; stop it before reconfiguring"
                }
            } else {
                dict incr state generation
                dict set state running true
                dict set state interval $interval
                dict set state busy false
                dict set state scheduler [ns_schedule_proc -thread $interval [self] sampling tick]
                nsv_set ns_stats pair {}
                nsv_set ns_stats control $state
            }
        }
        return [:sampling status]
    }
::ns::stats::Sampler public method {sampling status} {} {
        #
        # Return the process-wide sampler configuration and status, including
        # running state, scheduler identifier, interval, process instance,
        # collection errors and readiness of the latest interval sample.
        #
        # latestAt is in epoch seconds; ageSeconds is the elapsed time since
        # the latest publication. Both are empty before the first sample.
        # Does not collect new observations or modify sampler state.
        #
        # @return Status dictionary including running, interval in seconds, scheduler, instance,
        #     errors, lastError, ready, latestAt and ageSeconds.
        #

        set result [nsv_get ns_stats control]
        dict unset result busy
        dict unset result generation
        set sample [nsv_get ns_stats pair]
        dict set result ready [expr {$sample ne "" && [dict get $sample sampling ready]}]
        dict set result latestAt {}
        dict set result ageSeconds {}
        if {$sample ne ""} {
            set stamp [dict get $sample timestamp]
            dict set result latestAt [expr {$stamp/1000000.0}]
            dict set result ageSeconds [expr {max(0,([clock microseconds]-$stamp)/1000000.0)}]
        }
        return $result
    }
::ns::stats::Sampler public method {sampling stop} {} {
        #
        # Unschedule collection and invalidate any collection already in
        # progress so it cannot publish into the stopped sampler.
        #
        # Retains the latest published sample for subsequent reads.
        # Returns the resulting status dictionary. Does not disable the
        # separately configured history persistence callback.
        #
        # @return Resulting sampling status dictionary; the last published sample remains
        #     available.
        #
        :locked {
            set state [nsv_get ns_stats control]
            if {[dict get $state running]} {
                ns_unschedule_proc [dict get $state scheduler]
            }
            dict set state running false
            dict incr state generation
            dict set state busy false
            dict set state scheduler {}
            nsv_set ns_stats control $state
        }
        return [:sampling status]
    }
::ns::stats::Sampler public method {sampling tick} {} {
        #
        # Perform one scheduled collection and publish its interval sample.
        # Skip collection when the sampler is stopped or another collection
        # is in progress. Collect observations outside the coordination lock,
        # then calculate and publish only if the sampler generation is still
        # current.
        #
        # Retain the raw snapshot as the next interval baseline. Collection
        # failures retain the previous publication, update error status and
        # emit a warning. Does not write SQLite history; persistence runs
        # through a separate callback.
        #
        #
        set generation {}
        :locked {
            set state [nsv_get ns_stats control]
            if {![dict get $state running] || [dict get $state busy]} {
                return
            }
            set generation [dict get $state generation]
            dict set state busy true
            nsv_set ns_stats control $state
        }
        if {$generation eq ""} {
            return
        }
        try {
            set current [:snapshot]
            :locked {
                set state [nsv_get ns_stats control]
                if {$generation!=[dict get $state generation]} {
                    return
                }
                set pair [nsv_get ns_stats pair]
                set previous [expr {$pair eq "" ? "" : [dict get $pair raw]}]
                set sample [:calculate $current $previous [dict get $state interval] [dict get $state instance]]
                :retainSample $sample
                dict set sample raw $current
                nsv_set ns_stats pair $sample
                dict set state busy false
                dict set state lastError {}
                nsv_set ns_stats control $state
            }
        } on error {message options} {
            :locked {
                set state [nsv_get ns_stats control]
                if {$generation == [dict get $state generation]} {
                    dict set state busy false
                    dict incr state errors
                    dict set state lastError $message
                    nsv_set ns_stats control $state
                }
            }
            ns_log warning "ns_stats: sampling failed: $message"
        }
    }
::nsf::relation::set ::ns::stats::Sampler superclass ::ns::stats::Collector


::nx::slotObj -container slot ::ns::stats::Sampler

XQL Not present:

Generic, PostgreSQL, Oracle