chengcheng
5b2b94b1bd
统一任务成功、Attempt 与结算 Outbox 的事务边界，增加多实例安全结算、人工复核、请求幂等与执行租约。钱包决策使用九位精确金额，并通过审计保护约束保留流水事实；同时补充管理接口、指标与 PostgreSQL 集成测试。
252 lines
11 KiB
Go
252 lines
11 KiB
Go
package securityevents
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
|
|
)
|
|
|
|
type MetricsSnapshotProvider interface {
|
|
SecurityEventMetrics(context.Context, string, string) (string, *time.Time, error)
|
|
}
|
|
|
|
type DynamicMetricsSnapshotProvider interface {
|
|
MetricsSnapshotProvider
|
|
SecurityEventConnection(context.Context) (store.SecurityEventConnection, error)
|
|
BillingMetrics(context.Context) (store.BillingMetricsSnapshot, error)
|
|
}
|
|
|
|
type Metrics struct {
|
|
accepted atomic.Uint64
|
|
rejected atomic.Uint64
|
|
duplicate atomic.Uint64
|
|
sessionsDeleted atomic.Uint64
|
|
watermarkRejected atomic.Uint64
|
|
verificationAccepted atomic.Uint64
|
|
heartbeatAccepted atomic.Uint64
|
|
heartbeatFailed atomic.Uint64
|
|
introspectionActive atomic.Uint64
|
|
introspectionInactive atomic.Uint64
|
|
introspectionFailed atomic.Uint64
|
|
jwksSSFFailed atomic.Uint64
|
|
jwksOIDCFailed atomic.Uint64
|
|
processingCount atomic.Uint64
|
|
processingNanos atomic.Uint64
|
|
processingBuckets [6]atomic.Uint64
|
|
billingSettlementCompleted atomic.Uint64
|
|
billingSettlementRetry atomic.Uint64
|
|
billingManualReview atomic.Uint64
|
|
billingEstimateFailed atomic.Uint64
|
|
billingIdempotentReplay atomic.Uint64
|
|
billingPricingUnavailable atomic.Uint64
|
|
}
|
|
|
|
var processingDurationBounds = [...]time.Duration{
|
|
10 * time.Millisecond, 50 * time.Millisecond, 100 * time.Millisecond,
|
|
500 * time.Millisecond, time.Second, 3 * time.Second,
|
|
}
|
|
|
|
func (m *Metrics) ObserveReceipt(outcome string, duration time.Duration, sessionsDeleted int64) {
|
|
switch outcome {
|
|
case "accepted":
|
|
m.accepted.Add(1)
|
|
case "duplicate":
|
|
m.duplicate.Add(1)
|
|
default:
|
|
m.rejected.Add(1)
|
|
}
|
|
if sessionsDeleted > 0 {
|
|
m.sessionsDeleted.Add(uint64(sessionsDeleted))
|
|
}
|
|
if duration < 0 {
|
|
duration = 0
|
|
}
|
|
m.processingCount.Add(1)
|
|
m.processingNanos.Add(uint64(duration))
|
|
for index, bound := range processingDurationBounds {
|
|
if duration <= bound {
|
|
m.processingBuckets[index].Add(1)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *Metrics) ObserveWatermarkRejection() { m.watermarkRejected.Add(1) }
|
|
|
|
func (m *Metrics) ObserveVerificationAccepted() { m.verificationAccepted.Add(1) }
|
|
|
|
func (m *Metrics) ObserveHeartbeat(outcome string) {
|
|
if outcome == "accepted" {
|
|
m.heartbeatAccepted.Add(1)
|
|
return
|
|
}
|
|
m.heartbeatFailed.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) ObserveIntrospection(outcome string) {
|
|
switch outcome {
|
|
case "active":
|
|
m.introspectionActive.Add(1)
|
|
case "inactive":
|
|
m.introspectionInactive.Add(1)
|
|
default:
|
|
m.introspectionFailed.Add(1)
|
|
}
|
|
}
|
|
|
|
func (m *Metrics) ObserveJWKSRefreshFailure(source string) {
|
|
if source == "ssf" {
|
|
m.jwksSSFFailed.Add(1)
|
|
return
|
|
}
|
|
m.jwksOIDCFailed.Add(1)
|
|
}
|
|
|
|
func (m *Metrics) ObserveBillingEvent(event string) {
|
|
switch event {
|
|
case "settlement_completed":
|
|
m.billingSettlementCompleted.Add(1)
|
|
case "settlement_retry":
|
|
m.billingSettlementRetry.Add(1)
|
|
case "manual_review":
|
|
m.billingManualReview.Add(1)
|
|
case "estimate_failed":
|
|
m.billingEstimateFailed.Add(1)
|
|
case "idempotent_replay":
|
|
m.billingIdempotentReplay.Add(1)
|
|
case "pricing_unavailable":
|
|
m.billingPricingUnavailable.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"
|
|
var lastVerificationAt *time.Time
|
|
if enabled {
|
|
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
|
|
defer cancel()
|
|
var err error
|
|
mode, lastVerificationAt, err = provider.SecurityEventMetrics(ctx, issuer, audience)
|
|
if err != nil {
|
|
http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
}
|
|
billing := store.BillingMetricsSnapshot{}
|
|
if billingProvider, ok := provider.(interface {
|
|
BillingMetrics(context.Context) (store.BillingMetricsSnapshot, error)
|
|
}); ok {
|
|
var err error
|
|
billing, err = billingProvider.BillingMetrics(r.Context())
|
|
if err != nil {
|
|
http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
}
|
|
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
|
outcomeCounters(w, "easyai_gateway_ssf_receipts_total", "Received SETs by bounded outcome.", []outcomeValue{
|
|
{"accepted", m.accepted.Load()}, {"rejected", m.rejected.Load()}, {"duplicate", m.duplicate.Load()},
|
|
})
|
|
plainCounter(w, "easyai_gateway_ssf_sessions_deleted_total", "OIDC browser sessions deleted by accepted session-revoked events.", m.sessionsDeleted.Load())
|
|
plainCounter(w, "easyai_gateway_ssf_watermark_rejections_total", "OIDC access tokens rejected by a local revocation watermark.", m.watermarkRejected.Load())
|
|
plainCounter(w, "easyai_gateway_ssf_verifications_total", "Valid verification SETs accepted by the receiver.", m.verificationAccepted.Load())
|
|
outcomeCounters(w, "easyai_gateway_ssf_heartbeat_requests_total", "Verification requests by bounded outcome.", []outcomeValue{
|
|
{"accepted", m.heartbeatAccepted.Load()}, {"failed", m.heartbeatFailed.Load()},
|
|
})
|
|
outcomeCounters(w, "easyai_gateway_oidc_introspection_total", "OIDC introspection requests by bounded outcome.", []outcomeValue{
|
|
{"active", m.introspectionActive.Load()}, {"inactive", m.introspectionInactive.Load()}, {"failed", m.introspectionFailed.Load()},
|
|
})
|
|
outcomeCounters(w, "easyai_gateway_jwks_refresh_failures_total", "JWKS refresh failures by bounded source.", []outcomeValue{
|
|
{"ssf", m.jwksSSFFailed.Load()}, {"oidc", m.jwksOIDCFailed.Load()},
|
|
})
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_ssf_processing_duration_seconds SSF receiver transaction processing time.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_processing_duration_seconds histogram")
|
|
for index, bound := range processingDurationBounds {
|
|
fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_bucket{le=\"%.2f\"} %d\n", bound.Seconds(), m.processingBuckets[index].Load())
|
|
}
|
|
fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_bucket{le=\"+Inf\"} %d\n", m.processingCount.Load())
|
|
fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_sum %.6f\n", float64(m.processingNanos.Load())/float64(time.Second))
|
|
fmt.Fprintf(w, "easyai_gateway_ssf_processing_duration_seconds_count %d\n", m.processingCount.Load())
|
|
verificationAge := float64(0)
|
|
if lastVerificationAt != nil {
|
|
verificationAge = time.Since(*lastVerificationAt).Seconds()
|
|
if verificationAge < 0 {
|
|
verificationAge = 0
|
|
}
|
|
}
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_ssf_verification_age_seconds Seconds since the last matched verification SET.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_verification_age_seconds gauge")
|
|
fmt.Fprintf(w, "easyai_gateway_ssf_verification_age_seconds %.6f\n", verificationAge)
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_ssf_mode Current SSF receiver mode as a one-hot gauge.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_ssf_mode gauge")
|
|
for _, modeName := range []string{"disabled", "bootstrap", "push_healthy", "introspection_fallback"} {
|
|
value := 0
|
|
if mode == modeName {
|
|
value = 1
|
|
}
|
|
fmt.Fprintf(w, "easyai_gateway_ssf_mode{mode=\"%s\"} %d\n", modeName, value)
|
|
}
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_billing_settlement_backlog Current unsettled Outbox records.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_billing_settlement_backlog gauge")
|
|
fmt.Fprintf(w, "easyai_gateway_billing_settlement_backlog %d\n", billing.SettlementBacklog)
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_billing_settlement_delay_seconds Age of the oldest unsettled Outbox record.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_billing_settlement_delay_seconds gauge")
|
|
fmt.Fprintf(w, "easyai_gateway_billing_settlement_delay_seconds %.6f\n", billing.SettlementDelaySecs)
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_billing_manual_review Current billing records requiring manual review.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_billing_manual_review gauge")
|
|
fmt.Fprintf(w, "easyai_gateway_billing_manual_review %d\n", billing.ManualReview)
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_billing_orphan_frozen Wallets whose frozen amount differs from active reservations.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_billing_orphan_frozen gauge")
|
|
fmt.Fprintf(w, "easyai_gateway_billing_orphan_frozen %d\n", billing.OrphanFrozen)
|
|
fmt.Fprintln(w, "# HELP easyai_gateway_billing_pricing_unavailable_tasks Current tasks rejected because no effective price was available.")
|
|
fmt.Fprintln(w, "# TYPE easyai_gateway_billing_pricing_unavailable_tasks gauge")
|
|
fmt.Fprintf(w, "easyai_gateway_billing_pricing_unavailable_tasks %d\n", billing.PricingUnavailable)
|
|
plainCounter(w, "easyai_gateway_billing_settlements_completed_total", "Billing settlements completed by this process.", m.billingSettlementCompleted.Load())
|
|
plainCounter(w, "easyai_gateway_billing_settlement_retries_total", "Billing settlement retries scheduled by this process.", m.billingSettlementRetry.Load())
|
|
plainCounter(w, "easyai_gateway_billing_manual_review_transitions_total", "Billing records transitioned to manual review by this process.", m.billingManualReview.Load())
|
|
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())
|
|
})
|
|
}
|
|
|
|
func (m *Metrics) DynamicHandler(provider DynamicMetricsSnapshotProvider) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
connection, err := provider.SecurityEventConnection(r.Context())
|
|
if errors.Is(err, store.ErrSecurityEventConnectionNotFound) {
|
|
m.Handler(provider, "", "", false).ServeHTTP(w, r)
|
|
return
|
|
}
|
|
if err != nil {
|
|
http.Error(w, "metrics unavailable", http.StatusServiceUnavailable)
|
|
return
|
|
}
|
|
if connection.Audience == nil {
|
|
m.Handler(provider, "", "", false).ServeHTTP(w, r)
|
|
return
|
|
}
|
|
m.Handler(provider, connection.TransmitterIssuer, *connection.Audience, true).ServeHTTP(w, r)
|
|
})
|
|
}
|
|
|
|
type outcomeValue struct {
|
|
name string
|
|
value uint64
|
|
}
|
|
|
|
func outcomeCounters(w http.ResponseWriter, name, help string, values []outcomeValue) {
|
|
fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s counter\n", name, help, name)
|
|
for _, value := range values {
|
|
fmt.Fprintf(w, "%s{outcome=\"%s\"} %d\n", name, value.name, value.value)
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|