MiFaZhan
288 lines
7.5 KiB
Go
288 lines
7.5 KiB
Go
// Package supply holds bounded, per-channel telemetry independent of card volume.
|
|
package supply
|
|
|
|
import (
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
const SchemaVersion = 1
|
|
|
|
type Wait struct {
|
|
Count int64 `json:"count"`
|
|
TotalMS int64 `json:"totalMs"`
|
|
LongestMS int64 `json:"longestMs"`
|
|
}
|
|
type Channel struct {
|
|
Key string `json:"chanKey"`
|
|
FirstQueryMS *int64 `json:"firstQueryMs"`
|
|
FirstEnqueueMS *int64 `json:"firstEnqueueMs"`
|
|
Startup Wait `json:"startupWait"`
|
|
Confirmed Wait `json:"confirmedWait"`
|
|
Unconfirmed Wait `json:"unconfirmedIdle"`
|
|
Consumer Wait `json:"consumerStartupWait"`
|
|
UnknownMS int64 `json:"unknownMs"`
|
|
EmptyClaims int64 `json:"emptyClaims"`
|
|
ValidSamples int64 `json:"validSamples"`
|
|
MissingSamples int64 `json:"missingSamples"`
|
|
Stages map[string]int64 `json:"producerStageMs"`
|
|
Completeness string `json:"completeness"`
|
|
Termination string `json:"termination,omitempty"`
|
|
CoverageStart time.Time `json:"coverageStart"`
|
|
CoverageEnd time.Time `json:"coverageEnd"`
|
|
}
|
|
type Summary struct {
|
|
Targets int `json:"targetChannels"`
|
|
Valid int `json:"validChannels"`
|
|
Confirmed int `json:"confirmedChannels"`
|
|
Unknown int `json:"unknownChannels"`
|
|
UnknownMS int64 `json:"unknownChannelMs"`
|
|
UnconfirmedMS int64 `json:"unconfirmedChannelMs"`
|
|
ConsumerMS int64 `json:"consumerStartupChannelMs"`
|
|
TotalMS int64 `json:"confirmedChannelMs"`
|
|
LongestMS int64 `json:"longestMs"`
|
|
Worst string `json:"worstChannel,omitempty"`
|
|
}
|
|
type Snapshot struct {
|
|
SchemaVersion int `json:"schemaVersion"`
|
|
Attempt string `json:"attempt"`
|
|
ResolutionMS int `json:"resolutionMs"`
|
|
Partial bool `json:"partial"`
|
|
Truncated bool `json:"truncated"`
|
|
Summary Summary `json:"summary"`
|
|
Channels []Channel `json:"channels"`
|
|
}
|
|
type View struct {
|
|
Version uint64
|
|
Pending int
|
|
Processing int
|
|
Stage string
|
|
Ended bool
|
|
}
|
|
type Sample struct {
|
|
Ready int64
|
|
Leases int64
|
|
Future int64
|
|
Eligible bool
|
|
Known bool
|
|
}
|
|
type state struct {
|
|
Channel
|
|
View
|
|
last time.Time
|
|
kind string
|
|
continuous int64
|
|
open bool
|
|
}
|
|
type Observer struct {
|
|
mu sync.Mutex
|
|
started time.Time
|
|
attempt string
|
|
partial bool
|
|
retained Summary
|
|
truncated bool
|
|
channels map[string]*state
|
|
}
|
|
|
|
func New(attempt string, started time.Time, keys []string) *Observer {
|
|
o := &Observer{attempt: attempt, started: started, channels: make(map[string]*state, len(keys))}
|
|
for _, k := range keys {
|
|
o.channels[k] = &state{Channel: Channel{Key: k, Stages: map[string]int64{}, CoverageStart: started}, View: View{Stage: "initializing"}}
|
|
}
|
|
return o
|
|
}
|
|
func (o *Observer) Event(key, stage string, pending int, added bool, ended bool, now time.Time) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
s := o.channels[key]
|
|
if s == nil {
|
|
return
|
|
}
|
|
s.Version++
|
|
s.Stage = stage
|
|
s.Pending = pending
|
|
if stage == "sql_query" && s.FirstQueryMS == nil {
|
|
v := now.Sub(o.started).Milliseconds()
|
|
s.FirstQueryMS = &v
|
|
}
|
|
if added && s.FirstEnqueueMS == nil {
|
|
v := now.Sub(o.started).Milliseconds()
|
|
s.FirstEnqueueMS = &v
|
|
s.close()
|
|
}
|
|
if ended {
|
|
s.Ended = true
|
|
s.Termination = stage
|
|
s.CoverageEnd = now
|
|
s.close()
|
|
}
|
|
}
|
|
func (o *Observer) Empty(key string) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
if s := o.channels[key]; s != nil && !s.Ended {
|
|
s.EmptyClaims++
|
|
}
|
|
}
|
|
func (o *Observer) Processing(key string, delta int) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
if s := o.channels[key]; s != nil {
|
|
s.Version++
|
|
s.Processing = max(0, s.Processing+delta)
|
|
if delta > 0 {
|
|
s.close()
|
|
}
|
|
}
|
|
}
|
|
func (o *Observer) Views() map[string]View {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
out := make(map[string]View, len(o.channels))
|
|
for k, s := range o.channels {
|
|
out[k] = s.View
|
|
}
|
|
return out
|
|
}
|
|
func (s *state) close() { s.kind = ""; s.continuous = 0; s.open = false; s.last = time.Time{} }
|
|
func (o *Observer) Partial() { o.mu.Lock(); o.partial = true; o.mu.Unlock() }
|
|
func (o *Observer) Apply(key string, before View, sample Sample, now time.Time) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
s := o.channels[key]
|
|
if s == nil || s.Ended {
|
|
return
|
|
}
|
|
kind := "unknown"
|
|
if sample.Known && before.Version == s.Version {
|
|
s.ValidSamples++
|
|
switch {
|
|
case !sample.Eligible:
|
|
s.close()
|
|
s.CoverageEnd = now
|
|
return
|
|
case sample.Ready > 0:
|
|
if s.Processing == 0 && sample.Leases == 0 {
|
|
kind = "consumer"
|
|
} else {
|
|
kind = "stock"
|
|
}
|
|
case s.Processing > 0 || sample.Leases > 0:
|
|
kind = "processing"
|
|
case sample.Future > 0:
|
|
kind = "scheduled"
|
|
case s.Pending > 0:
|
|
if s.FirstEnqueueMS == nil {
|
|
kind = "startup"
|
|
} else {
|
|
kind = "confirmed"
|
|
}
|
|
default:
|
|
kind = "unconfirmed"
|
|
}
|
|
} else {
|
|
s.MissingSamples++
|
|
o.partial = true
|
|
}
|
|
elapsed := now.Sub(s.last)
|
|
if !s.last.IsZero() && elapsed > 2*time.Second {
|
|
s.MissingSamples += int64(elapsed/time.Second) - 1
|
|
o.partial = true
|
|
s.close()
|
|
}
|
|
if s.kind == kind && !s.last.IsZero() {
|
|
ms := elapsed.Milliseconds()
|
|
if ms < 0 {
|
|
ms = 0
|
|
}
|
|
s.continuous += ms
|
|
var w *Wait
|
|
switch kind {
|
|
case "startup":
|
|
w = &s.Startup
|
|
case "confirmed":
|
|
w = &s.Confirmed
|
|
case "unconfirmed":
|
|
w = &s.Unconfirmed
|
|
case "consumer":
|
|
w = &s.Consumer
|
|
case "unknown":
|
|
s.UnknownMS += ms
|
|
}
|
|
if w != nil {
|
|
if !s.open {
|
|
w.Count++
|
|
s.open = true
|
|
}
|
|
w.TotalMS += ms
|
|
if s.continuous > w.LongestMS {
|
|
w.LongestMS = s.continuous
|
|
}
|
|
if kind == "startup" || kind == "confirmed" {
|
|
s.Stages[s.Stage] += ms
|
|
}
|
|
}
|
|
} else {
|
|
s.continuous = 0
|
|
s.open = false
|
|
}
|
|
s.kind = kind
|
|
s.last = now
|
|
s.CoverageEnd = now
|
|
}
|
|
func (o *Observer) Snapshot() Snapshot {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
out := Snapshot{SchemaVersion: 1, Attempt: o.attempt, ResolutionMS: 1000, Partial: o.partial, Truncated: o.truncated, Channels: make([]Channel, 0, len(o.channels))}
|
|
out.Summary.Targets = len(o.channels)
|
|
for _, s := range o.channels {
|
|
c := s.Channel
|
|
c.Completeness = "observing"
|
|
if s.Ended {
|
|
c.Completeness = "complete"
|
|
}
|
|
if c.MissingSamples > 0 || c.UnknownMS > 0 || c.ValidSamples == 0 || c.Termination == "cancelled" {
|
|
c.Completeness = "partial"
|
|
}
|
|
c.Stages = make(map[string]int64, len(s.Stages))
|
|
for k, v := range s.Stages {
|
|
c.Stages[k] = v
|
|
}
|
|
out.Channels = append(out.Channels, c)
|
|
if c.ValidSamples == 0 {
|
|
out.Partial = true
|
|
}
|
|
out.Summary.UnknownMS += c.UnknownMS
|
|
out.Summary.UnconfirmedMS += c.Unconfirmed.TotalMS
|
|
out.Summary.ConsumerMS += c.Consumer.TotalMS
|
|
if c.ValidSamples > 0 {
|
|
out.Summary.Valid++
|
|
}
|
|
total := c.Startup.TotalMS + c.Confirmed.TotalMS
|
|
longest := max(c.Startup.LongestMS, c.Confirmed.LongestMS)
|
|
if total > 0 {
|
|
out.Summary.Confirmed++
|
|
}
|
|
if c.MissingSamples > 0 || c.UnknownMS > 0 || c.ValidSamples == 0 {
|
|
out.Summary.Unknown++
|
|
}
|
|
out.Summary.TotalMS += total
|
|
if longest > out.Summary.LongestMS {
|
|
out.Summary.LongestMS = longest
|
|
out.Summary.Worst = c.Key
|
|
}
|
|
}
|
|
out.Summary.Confirmed = min(out.Summary.Targets, out.Summary.Confirmed+o.retained.Confirmed)
|
|
out.Summary.TotalMS += o.retained.TotalMS
|
|
out.Summary.UnknownMS += o.retained.UnknownMS
|
|
out.Summary.UnconfirmedMS += o.retained.UnconfirmedMS
|
|
out.Summary.ConsumerMS += o.retained.ConsumerMS
|
|
if o.retained.LongestMS > out.Summary.LongestMS {
|
|
out.Summary.LongestMS = o.retained.LongestMS
|
|
out.Summary.Worst = o.retained.Worst
|
|
}
|
|
sort.Slice(out.Channels, func(i, j int) bool { return out.Channels[i].Key < out.Channels[j].Key })
|
|
return out
|
|
}
|