chore(git): 同步主分支发布与业务变更
# Conflicts: # apps/api/go.mod # apps/api/go.sum
This commit is contained in:
@@ -44,6 +44,20 @@ type Metrics struct {
|
||||
billingEstimateFailed atomic.Uint64
|
||||
billingIdempotentReplay atomic.Uint64
|
||||
billingPricingUnavailable atomic.Uint64
|
||||
asyncWorkerCapacity atomic.Int64
|
||||
asyncWorkerDesiredCapacity atomic.Int64
|
||||
asyncWorkerHardLimit atomic.Int64
|
||||
asyncWorkerCapacityCapped atomic.Int64
|
||||
asyncWorkerResizeSuccess atomic.Uint64
|
||||
asyncWorkerRefreshFailed atomic.Uint64
|
||||
asyncWorkerCreateFailed atomic.Uint64
|
||||
asyncWorkerStartFailed atomic.Uint64
|
||||
leaseRenewalSuccess atomic.Uint64
|
||||
leaseRenewalFailure atomic.Uint64
|
||||
leaseRenewalLost atomic.Uint64
|
||||
taskEventDuplicate atomic.Uint64
|
||||
taskEventUnknownType atomic.Uint64
|
||||
taskEventBudgetExceeded atomic.Uint64
|
||||
}
|
||||
|
||||
var processingDurationBounds = [...]time.Duration{
|
||||
@@ -123,6 +137,52 @@ func (m *Metrics) ObserveBillingEvent(event string) {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Metrics) SetAsyncWorkerCapacity(current, desired, hardLimit int, capped bool) {
|
||||
m.asyncWorkerCapacity.Store(int64(current))
|
||||
m.asyncWorkerDesiredCapacity.Store(int64(desired))
|
||||
m.asyncWorkerHardLimit.Store(int64(hardLimit))
|
||||
cappedValue := int64(0)
|
||||
if capped {
|
||||
cappedValue = 1
|
||||
}
|
||||
m.asyncWorkerCapacityCapped.Store(cappedValue)
|
||||
}
|
||||
|
||||
func (m *Metrics) ObserveAsyncWorkerResize(outcome string) {
|
||||
switch outcome {
|
||||
case "success":
|
||||
m.asyncWorkerResizeSuccess.Add(1)
|
||||
case "refresh_failed":
|
||||
m.asyncWorkerRefreshFailed.Add(1)
|
||||
case "create_failed":
|
||||
m.asyncWorkerCreateFailed.Add(1)
|
||||
case "start_failed":
|
||||
m.asyncWorkerStartFailed.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Metrics) ObserveConcurrencyLeaseRenewal(outcome string) {
|
||||
switch outcome {
|
||||
case "success":
|
||||
m.leaseRenewalSuccess.Add(1)
|
||||
case "lost":
|
||||
m.leaseRenewalLost.Add(1)
|
||||
default:
|
||||
m.leaseRenewalFailure.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Metrics) ObserveTaskEventSkip(reason string) {
|
||||
switch reason {
|
||||
case "duplicate":
|
||||
m.taskEventDuplicate.Add(1)
|
||||
case "budget_exceeded":
|
||||
m.taskEventBudgetExceeded.Add(1)
|
||||
default:
|
||||
m.taskEventUnknownType.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience string, enabled bool) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
mode := "disabled"
|
||||
@@ -212,6 +272,26 @@ func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience str
|
||||
plainCounter(w, "easyai_gateway_billing_estimate_failures_total", "Pricing estimate requests that failed.", m.billingEstimateFailed.Load())
|
||||
plainCounter(w, "easyai_gateway_billing_idempotent_replays_total", "Generation requests replayed idempotently.", m.billingIdempotentReplay.Load())
|
||||
plainCounter(w, "easyai_gateway_billing_pricing_unavailable_total", "Pricing requests rejected because no effective price was available.", m.billingPricingUnavailable.Load())
|
||||
plainGauge(w, "easyai_gateway_async_worker_capacity", "Current River asynchronous worker capacity.", m.asyncWorkerCapacity.Load())
|
||||
plainGauge(w, "easyai_gateway_async_worker_desired_capacity", "Uncapped asynchronous worker capacity derived from model and user-group policies.", m.asyncWorkerDesiredCapacity.Load())
|
||||
plainGauge(w, "easyai_gateway_async_worker_hard_limit", "Per-process asynchronous worker safety limit.", m.asyncWorkerHardLimit.Load())
|
||||
plainGauge(w, "easyai_gateway_async_worker_capacity_capped", "Whether desired asynchronous worker capacity is capped by the hard limit.", m.asyncWorkerCapacityCapped.Load())
|
||||
outcomeCounters(w, "easyai_gateway_async_worker_resizes_total", "Asynchronous worker resize attempts by bounded outcome.", []outcomeValue{
|
||||
{"success", m.asyncWorkerResizeSuccess.Load()},
|
||||
{"refresh_failed", m.asyncWorkerRefreshFailed.Load()},
|
||||
{"create_failed", m.asyncWorkerCreateFailed.Load()},
|
||||
{"start_failed", m.asyncWorkerStartFailed.Load()},
|
||||
})
|
||||
outcomeCounters(w, "easyai_gateway_concurrency_lease_renewals_total", "Concurrency lease renewals by bounded outcome.", []outcomeValue{
|
||||
{"success", m.leaseRenewalSuccess.Load()},
|
||||
{"failure", m.leaseRenewalFailure.Load()},
|
||||
{"lost", m.leaseRenewalLost.Load()},
|
||||
})
|
||||
outcomeCounters(w, "easyai_gateway_task_events_skipped_total", "Task events skipped before persistence by bounded reason.", []outcomeValue{
|
||||
{"duplicate", m.taskEventDuplicate.Load()},
|
||||
{"unknown_type", m.taskEventUnknownType.Load()},
|
||||
{"budget_exceeded", m.taskEventBudgetExceeded.Load()},
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
@@ -249,3 +329,7 @@ func outcomeCounters(w http.ResponseWriter, name, help string, values []outcomeV
|
||||
func plainCounter(w http.ResponseWriter, name, help string, value uint64) {
|
||||
fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s counter\n%s %d\n", name, help, name, name, value)
|
||||
}
|
||||
|
||||
func plainGauge(w http.ResponseWriter, name, help string, value int64) {
|
||||
fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s gauge\n%s %d\n", name, help, name, name, value)
|
||||
}
|
||||
|
||||
@@ -26,6 +26,12 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) {
|
||||
metrics.ObserveHeartbeat("accepted")
|
||||
metrics.ObserveIntrospection("failed")
|
||||
metrics.ObserveJWKSRefreshFailure("ssf")
|
||||
metrics.SetAsyncWorkerCapacity(96, 128, 96, true)
|
||||
metrics.ObserveAsyncWorkerResize("success")
|
||||
metrics.ObserveConcurrencyLeaseRenewal("success")
|
||||
metrics.ObserveConcurrencyLeaseRenewal("lost")
|
||||
metrics.ObserveTaskEventSkip("duplicate")
|
||||
metrics.ObserveTaskEventSkip("budget_exceeded")
|
||||
|
||||
recorder := httptest.NewRecorder()
|
||||
metrics.Handler(metricsSnapshot{mode: "push_healthy", last: time.Now().Add(-time.Minute)}, "issuer", "audience", true).
|
||||
@@ -42,6 +48,14 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) {
|
||||
`easyai_gateway_ssf_mode{mode="push_healthy"} 1`,
|
||||
`easyai_gateway_oidc_introspection_total{outcome="failed"} 1`,
|
||||
`easyai_gateway_jwks_refresh_failures_total{outcome="ssf"} 1`,
|
||||
`easyai_gateway_async_worker_capacity 96`,
|
||||
`easyai_gateway_async_worker_desired_capacity 128`,
|
||||
`easyai_gateway_async_worker_capacity_capped 1`,
|
||||
`easyai_gateway_async_worker_resizes_total{outcome="success"} 1`,
|
||||
`easyai_gateway_concurrency_lease_renewals_total{outcome="success"} 1`,
|
||||
`easyai_gateway_concurrency_lease_renewals_total{outcome="lost"} 1`,
|
||||
`easyai_gateway_task_events_skipped_total{outcome="duplicate"} 1`,
|
||||
`easyai_gateway_task_events_skipped_total{outcome="budget_exceeded"} 1`,
|
||||
} {
|
||||
if !strings.Contains(recorder.Body.String(), expected) {
|
||||
t.Fatalf("missing metric %q in:\n%s", expected, recorder.Body.String())
|
||||
|
||||
Reference in New Issue
Block a user