112 lines
8.8 KiB
Go
112 lines
8.8 KiB
Go
package main
|
||
|
||
import (
|
||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||
"math"
|
||
"runtime"
|
||
"sort"
|
||
"time"
|
||
)
|
||
|
||
// distribution 仅统计已逐项核验的任务；空集合百分位为null，单位毫秒。
|
||
type distribution struct {
|
||
Count int `json:"count"` // Count 统计分母。
|
||
P50 *float64 `json:"p50_ms"` // P50 最近秩中位数。
|
||
P95 *float64 `json:"p95_ms"` // P95 最近秩95百分位。
|
||
P99 *float64 `json:"p99_ms"` // P99 最近秩99百分位。
|
||
Max *float64 `json:"max_ms"` // Max 最大耗时。
|
||
}
|
||
|
||
// planReport 保存原操作身份；请求结果不明时只能按原键和原请求核查。
|
||
type planReport struct {
|
||
CalendarVersion string `json:"calendar_version,omitempty"` // CalendarVersion 创建回执中的时区快照标识，后续读取不得变化。
|
||
EffectiveAt time.Time `json:"effective_at"` // EffectiveAt 固定计划版本生效时间。
|
||
Index int `json:"index"` // Index 从零开始的计划序号。
|
||
ID string `json:"id,omitempty"` // ID 已取得的计划身份；空不证明创建未提交。
|
||
CreateKey string `json:"create_key"` // CreateKey 创建键，同时作为计划名称。
|
||
ResumeKey string `json:"resume_key,omitempty"` // ResumeKey 原恢复键。
|
||
ResumeRevision uint64 `json:"resume_revision,string"` // ResumeRevision 原恢复请求版本，零表示尚未发送。
|
||
Resumed bool `json:"resumed"` // Resumed 已收到且核对恢复回执。
|
||
Exhausted bool `json:"exhausted"` // Exhausted 已观察到固定版本无下个触发点和扫描工作。
|
||
PauseKey string `json:"pause_key,omitempty"` // PauseKey 最后发送的暂停键，仅明确版本冲突才换键。
|
||
PauseRevision uint64 `json:"pause_revision,string"` // PauseRevision 最后暂停请求的预期版本。
|
||
PauseConfirmed bool `json:"pause_confirmed"` // PauseConfirmed 已确认计划暂停；不取消已生成任务。
|
||
Occurrences int `json:"occurrences"` // Occurrences 已观察唯一触发点数。
|
||
Fired int `json:"fired"` // Fired 已关联唯一任务的单点触发数。
|
||
VerifiedSuccess int `json:"verified_success"` // VerifiedSuccess 完成任务/尝试/来源核对数。
|
||
}
|
||
|
||
// scheduleReport 是隔离环境有界样本，不作为生产容量或高可用承诺。
|
||
type scheduleReport struct {
|
||
RunID string `json:"run_id"` // RunID 计划、处理器和操作键的隔离身份。
|
||
GoVersion string `json:"go_version"` // GoVersion 实际编译工具链。
|
||
GOOS string `json:"goos"` // GOOS 运行系统。
|
||
GOARCH string `json:"goarch"` // GOARCH 运行架构。
|
||
StartedAt time.Time `json:"started_at"` // StartedAt 预检开始本机时间。
|
||
FinishedAt time.Time `json:"finished_at"` // FinishedAt 全部有界收尾完成本机时间。
|
||
Requests int `json:"requests_per_schedule"` // Requests 每计划预期触发数。
|
||
Schedules int `json:"schedules"` // Schedules 预期计划数。
|
||
Expected int `json:"expected_occurrences"` // Expected 总计划触发数，包含未观察的触发点。
|
||
Workers int `json:"workers"` // Workers 本进程独立WSS Worker数。
|
||
Capacity int `json:"capacity_per_worker"` // Capacity 每Worker并行槽数。
|
||
IntervalMS int64 `json:"interval_ms"` // IntervalMS 固定间隔毫秒。
|
||
StartDelayMS int64 `json:"start_delay_ms"` // StartDelayMS 创建/恢复的预留时间毫秒。
|
||
RunTimeoutMS int64 `json:"run_timeout_ms"` // RunTimeoutMS 整个测量截止毫秒。
|
||
PollIntervalMS int64 `json:"poll_interval_ms"` // PollIntervalMS 历史查询间隔毫秒。
|
||
HTTPTimeoutMS int64 `json:"http_timeout_ms"` // HTTPTimeoutMS 每HTTP请求截止毫秒。
|
||
Queue *orbit.Queue `json:"queue,omitempty"` // Queue 预检队列快照，只读配置。
|
||
ScheduleStart time.Time `json:"schedule_start"` // ScheduleStart 所有计划共同起点，包含边界。
|
||
ScheduleEnd time.Time `json:"schedule_end"` // ScheduleEnd 所有计划共同截止，排除边界。
|
||
Plans []planReport `json:"plans"` // Plans 已尝试创建的固定身份和收尾证据。
|
||
Occurrences int `json:"occurrences"` // Occurrences 已观察唯一触发数。
|
||
Fired int `json:"fired"` // Fired 单点生成且关联唯一任务数，不代表成功。
|
||
VerifiedSuccess int `json:"verified_success"` // VerifiedSuccess 持久化结果/唯一尝试/来源全部通过数。
|
||
Unobserved int `json:"unobserved"` // Unobserved 预期但未观察到的触发点数。
|
||
Unverified int `json:"unverified"` // Unverified 已触发但未核验成功的任务数。
|
||
HandlerCalls int `json:"handler_calls"` // HandlerCalls 通过输入核对的处理器调用总数。
|
||
DuplicateInvocations int `json:"duplicate_invocations"` // DuplicateInvocations 同一任务的额外调用数。
|
||
UnconfirmedInvocations int `json:"unconfirmed_invocations"` // UnconfirmedInvocations 未对应到已核验任务及尝试的调用数。
|
||
WorkerStarted uint64 `json:"worker_started"` // WorkerStarted SDK实际调用处理器数。
|
||
Reconnects uint64 `json:"worker_reconnects"` // Reconnects SDK连接重建次数。
|
||
Renewals uint64 `json:"worker_renewals_confirmed"` // Renewals 已确认租约续期次数。
|
||
TriggerLag distribution `json:"trigger_lag"` // TriggerLag 计划时间至任务受理数据库耗时。
|
||
QueueWait distribution `json:"queue_wait"` // QueueWait 任务受理至开始授权数据库耗时。
|
||
Execution distribution `json:"authorized_execution"` // Execution 授权至结果落库耗时，包含网络和事务。
|
||
Completion distribution `json:"completion"` // Completion 计划时间至结果落库耗时。
|
||
Errors map[string]int `json:"errors"` // Errors 有限分类，不含凭据或原异常正文。
|
||
Complete bool `json:"complete"` // Complete 全部计数、证据和收尾满足时为真。
|
||
}
|
||
|
||
func newReport(c config, runID string) scheduleReport {
|
||
return scheduleReport{RunID: runID, GoVersion: runtime.Version(), GOOS: runtime.GOOS, GOARCH: runtime.GOARCH, StartedAt: time.Now().UTC(), Requests: c.Requests, Schedules: c.Schedules, Expected: c.Requests * c.Schedules, Workers: c.Workers, Capacity: c.Capacity, IntervalMS: c.Interval.Milliseconds(), StartDelayMS: c.StartDelay.Milliseconds(), RunTimeoutMS: c.RunTimeout.Milliseconds(), PollIntervalMS: c.PollInterval.Milliseconds(), HTTPTimeoutMS: c.Timeout.Milliseconds(), Plans: []planReport{}, Errors: map[string]int{}}
|
||
}
|
||
func (r scheduleReport) finish() scheduleReport {
|
||
for code, n := range r.Errors {
|
||
if n == 0 {
|
||
delete(r.Errors, code)
|
||
}
|
||
}
|
||
r.FinishedAt = time.Now().UTC()
|
||
r.Unobserved = r.Expected - r.Occurrences
|
||
r.Unverified = r.Fired - r.VerifiedSuccess
|
||
plansReady := len(r.Plans) == r.Schedules
|
||
for _, p := range r.Plans {
|
||
plansReady = plansReady && p.Resumed && p.Exhausted && p.PauseConfirmed && p.VerifiedSuccess == r.Requests
|
||
}
|
||
r.Complete = r.Expected > 0 && plansReady && r.Occurrences == r.Expected && r.Fired == r.Expected && r.VerifiedSuccess == r.Expected && r.HandlerCalls == r.Expected && r.WorkerStarted == uint64(r.Expected) && r.DuplicateInvocations == 0 && r.UnconfirmedInvocations == 0 && len(r.Errors) == 0
|
||
return r
|
||
}
|
||
func distributionOf(values []time.Duration) distribution {
|
||
d := distribution{Count: len(values)}
|
||
if len(values) == 0 {
|
||
return d
|
||
}
|
||
sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
|
||
percentile := func(q float64) *float64 {
|
||
v := float64(values[int(math.Ceil(float64(len(values))*q))-1]) / float64(time.Millisecond)
|
||
return &v
|
||
}
|
||
d.P50, d.P95, d.P99, d.Max = percentile(.5), percentile(.95), percentile(.99), percentile(1)
|
||
return d
|
||
}
|