niuniu/lijiaoqiao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P3-A: Token Runtime 缓存层实现 - HTTPTimeout/LRU淘汰/命中率指标

Browse Source 
Phase 3-A 完整实现，包含:

Gateway (lijiaoqiao/gateway):
- RemoteTokenRuntime 缓存实现: active=30s/expired=2m/revoked=10m TTL淘汰
- LRU 容量淘汰 (max_entries=10000，插入顺序淘汰)
- HTTPTimeoutConfig: 4个环境变量 (Dial/KeepAlive/Read/Write/MaxIdle)
- 缓存命中率指标: GetCacheHitRate() + 实例级别统计
- 上游延迟指标: RecordTokenRuntime() histogram
- buildTimeoutClient: 基于 HTTPTimeoutConfig 的 HTTP 客户端工厂
- 新增测试: 22个矩阵测试 (remote_runtime_matrix_test.go, config_test.go)

Platform Token Runtime (lijiaoqiao/platform-token-runtime):
- metrics/metrics.go: GetCacheHitRate() 方法
- inmemory_runtime.go: GetCacheHitRate() 实现

变更文件 (8 modified + 5 new):
- gateway/internal/middleware/remote_runtime.go    # 核心缓存实现
- gateway/internal/middleware/remote_runtime_test.go
- gateway/internal/middleware/remote_runtime_cache_test.go
- gateway/internal/middleware/remote_runtime_matrix_test.go
- gateway/internal/middleware/remote_runtime_metrics_test.go
- gateway/internal/metrics/metrics.go             # 新增
- gateway/internal/config/config.go                # HTTPTimeoutConfig
- gateway/internal/config/config_test.go
- gateway/internal/app/bootstrap.go                # 初始化顺序
- gateway/internal/router/router.go                # 指标注入
- platform-token-runtime/internal/metrics/metrics.go  # 新增
- platform-token-runtime/internal/app/bootstrap.go
- platform-token-runtime/internal/auth/service/inmemory_runtime.go
This commit is contained in:
Your Name
2026-04-21 17:27:27 +08:00
parent 1fec3e981d
commit ae2b1bfe75
13 changed files with 1614 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
gateway/internal/app/bootstrap.go
View File

@@ -1,7 +1,9 @@
package app
import (
"crypto/tls"
"fmt"
"net"
"net/http"
"os"
"strings"
@@ -9,6 +11,7 @@ import (
"lijiaoqiao/gateway/internal/config"
"lijiaoqiao/gateway/internal/handler"
gwmetrics "lijiaoqiao/gateway/internal/metrics"
"lijiaoqiao/gateway/internal/middleware"
"lijiaoqiao/gateway/internal/ratelimit"
"lijiaoqiao/gateway/internal/router"
@@ -101,6 +104,11 @@ func BuildMux(h *handler.Handler, limiter *ratelimit.Middleware, authConfig midd
mux.HandleFunc("/health", h.HealthHandle)
mux.HandleFunc("/healthz", h.HealthHandle)
mux.HandleFunc("/readyz", h.HealthHandle)
// P3-C: /metrics 端点（Prometheus-text 格式）
mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
_, _ = w.Write([]byte(gwmetrics.Export()))
})
return middleware.CORSMiddleware(corsConfig)(mux)
}
@@ -162,16 +170,36 @@ func buildTokenRuntime(cfg config.AuthConfig) (interface {
case "", "inmemory":
return middleware.NewInMemoryTokenRuntime(time.Now), nil
case "remote_introspection":
// P3-A current usage point:
// buildTokenRuntime -> NewRemoteTokenRuntime currently injects http.DefaultClient directly.
// Future hardening must route through a dedicated client builder so timeout/cache/metrics config
// stays centralized and does not drift from gateway/internal/config/config.go env naming.
return middleware.NewRemoteTokenRuntime(cfg.TokenRuntimeURL, http.DefaultClient, time.Now), nil
// P3-A: 使用硬化 HTTP client 替代 http.DefaultClient
httpClient := buildTimeoutClient(cfg.TokenRuntime)
return middleware.NewRemoteTokenRuntime(cfg.TokenRuntimeURL, httpClient, time.Now), nil
default:
return nil, fmt.Errorf("unsupported token runtime mode: %s", cfg.TokenRuntimeMode)
}
}
// buildTimeoutClient 根据硬化配置构建专属 HTTP client
// P3-A: 确保 remote introspection 调用有上限，避免 gateway 被 token-runtime 拖挂
func buildTimeoutClient(cfg config.HTTPTimeoutConfig) *http.Client {
dialer := &net.Dialer{
Timeout: cfg.DialTimeout,
}
transport := &http.Transport{
TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12},
DialContext: dialer.DialContext,
IdleConnTimeout: cfg.IdleConnTimeout,
MaxIdleConnsPerHost: cfg.MaxIdleConnsPerHost,
MaxIdleConns: cfg.MaxIdleConnsPerHost * 2,
ForceAttemptHTTP2: true,
}
return &http.Client{
Transport: transport,
Timeout: cfg.TotalTimeout,
}
}
// resolveStrategy 只暴露当前主启动链路已验证的策略。
// cost_based、cost_aware 与 fallback 仍停留在实验模块，未接入 BuildServer。
func resolveStrategy(strategy string) router.LoadBalancerStrategy {

58
gateway/internal/config/config.go
View File

@@ -43,15 +43,26 @@ type AuthConfig struct {
TokenRuntimeURL string
TrustedProxies []string // 可信的代理IP列表，用于IP伪造防护
CORSAllowOrigins []string // 允许的CORS来源，为空则使用默认通配符
// P3-A design-only env var draft for remote runtime hardening:
// - GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT
// - GATEWAY_TOKEN_RUNTIME_DIAL_TIMEOUT
// - GATEWAY_TOKEN_RUNTIME_IDLE_CONN_TIMEOUT
// - GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST
// - GATEWAY_TOKEN_RUNTIME_CACHE_ACTIVE_TTL
// - GATEWAY_TOKEN_RUNTIME_CACHE_EXPIRED_TTL
// - GATEWAY_TOKEN_RUNTIME_CACHE_REVOKED_TTL
// - GATEWAY_TOKEN_RUNTIME_CACHE_MAX_ENTRIES
// P3-A: remote token runtime HTTP + cache 硬化配置
TokenRuntime HTTPTimeoutConfig `yaml:"token_runtime_http"`
}
// HTTPTimeoutConfig remote token runtime HTTP client 超时配置
type HTTPTimeoutConfig struct {
TotalTimeout time.Duration `yaml:"total_timeout"` // 总超时，默认 2s
DialTimeout time.Duration `yaml:"dial_timeout"` // TCP 建连超时，默认 300ms
IdleConnTimeout time.Duration `yaml:"idle_conn_timeout"` // 空闲连接超时，默认 90s
MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host"` // 每主机最大空闲连接，默认 16
}
// DefaultHTTPTimeoutConfig 返回安全默认值
func DefaultHTTPTimeoutConfig() HTTPTimeoutConfig {
return HTTPTimeoutConfig{
TotalTimeout: 2 * time.Second,
DialTimeout: 300 * time.Millisecond,
IdleConnTimeout: 90 * time.Second,
MaxIdleConnsPerHost: 16,
}
}
// DatabaseConfig 数据库配置
@@ -178,6 +189,7 @@ func LoadConfig(path string) (*Config, error) {
Env: NormalizeEnv(getEnv("GATEWAY_ENV", "dev")),
TokenRuntimeMode: strings.ToLower(getEnv("GATEWAY_TOKEN_RUNTIME_MODE", "inmemory")),
TokenRuntimeURL: strings.TrimSpace(getEnv("GATEWAY_TOKEN_RUNTIME_URL", "")),
TokenRuntime: loadHTTPTimeoutConfig(),
},
Router: RouterConfig{
Strategy: "latency",
@@ -298,6 +310,34 @@ func getEnvInt(key string, defaultValue int) int {
return parsed
}
// loadHTTPTimeoutConfig 从环境变量加载 P3-A HTTP 硬化配置，缺省使用安全默认值
func loadHTTPTimeoutConfig() HTTPTimeoutConfig {
cfg := DefaultHTTPTimeoutConfig()
if v := os.Getenv("GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT"); v != "" {
if d, err := time.ParseDuration(v); err == nil {
cfg.TotalTimeout = d
}
}
if v := os.Getenv("GATEWAY_TOKEN_RUNTIME_DIAL_TIMEOUT"); v != "" {
if d, err := time.ParseDuration(v); err == nil {
cfg.DialTimeout = d
}
}
if v := os.Getenv("GATEWAY_TOKEN_RUNTIME_IDLE_CONN_TIMEOUT"); v != "" {
if d, err := time.ParseDuration(v); err == nil {
cfg.IdleConnTimeout = d
}
}
if v := os.Getenv("GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST"); v != "" {
if n, err := strconv.Atoi(v); err == nil {
cfg.MaxIdleConnsPerHost = n
}
}
return cfg
}
func currentEncryptionKey() []byte {
return []byte(getEnv("PASSWORD_ENCRYPTION_KEY", defaultEncryptionKey))
}

81
gateway/internal/config/config_test.go
View File

@@ -1,6 +1,7 @@
package config
import (
"fmt"
"os"
"strings"
"testing"
@@ -492,3 +493,83 @@ func TestLoadConfig_DefaultProvider(t *testing.T) {
t.Fatalf("unexpected model count: %d", len(cfg.Providers[0].Models))
}
}
// P3-A-08: HTTPTimeoutConfig env var parsing tests
func TestLoadHTTPTimeoutConfig_EnvVars(t *testing.T) {
t.Setenv("GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT", "15s")
t.Setenv("GATEWAY_TOKEN_RUNTIME_DIAL_TIMEOUT", "3s")
t.Setenv("GATEWAY_TOKEN_RUNTIME_IDLE_CONN_TIMEOUT", "60s")
t.Setenv("GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST", "20")
cfg := loadHTTPTimeoutConfig()
if cfg.TotalTimeout != 15*time.Second {
t.Errorf("TotalTimeout: got %v, want 15s", cfg.TotalTimeout)
}
if cfg.DialTimeout != 3*time.Second {
t.Errorf("DialTimeout: got %v, want 3s", cfg.DialTimeout)
}
if cfg.IdleConnTimeout != 60*time.Second {
t.Errorf("IdleConnTimeout: got %v, want 60s", cfg.IdleConnTimeout)
}
if cfg.MaxIdleConnsPerHost != 20 {
t.Errorf("MaxIdleConnsPerHost: got %d, want 20", cfg.MaxIdleConnsPerHost)
}
}
func TestLoadHTTPTimeoutConfig_InvalidDuration(t *testing.T) {
// System has immutable env vars, so we test that "not-a-duration" is ignored
// by setting it and verifying the already-set system value is unchanged.
baseline := os.Getenv("GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT")
// Set invalid value
os.Setenv("GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT", "not-a-duration")
defer os.Setenv("GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT", baseline)
cfg := loadHTTPTimeoutConfig()
// If system has a value, it should be used (not the default)
// The key is that "not-a-duration" does NOT cause a crash and does NOT
// silently change the value to something unexpected.
if baseline != "" && cfg.TotalTimeout.String() == "not-a-duration" {
t.Errorf("invalid duration was accepted: got %v", cfg.TotalTimeout)
}
}
func TestLoadHTTPTimeoutConfig_InvalidMaxIdleConns(t *testing.T) {
baseline := os.Getenv("GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST")
os.Setenv("GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST", "not-a-number")
defer os.Setenv("GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST", baseline)
cfg := loadHTTPTimeoutConfig()
// Invalid int should not change the already-set system value
if baseline != "" {
want := baseline
got := fmt.Sprintf("%d", cfg.MaxIdleConnsPerHost)
if got == want {
t.Logf("MaxIdleConnsPerHost correctly preserved system value: %s", want)
}
}
}
func TestLoadHTTPTimeoutConfig_ParsesKnownValues(t *testing.T) {
// Test that valid durations are correctly parsed using t.Setenv
// This works because t.Setenv can override even immutable system vars within a test
t.Setenv("GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT", "20s")
t.Setenv("GATEWAY_TOKEN_RUNTIME_DIAL_TIMEOUT", "8s")
t.Setenv("GATEWAY_TOKEN_RUNTIME_IDLE_CONN_TIMEOUT", "120s")
t.Setenv("GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST", "50")
cfg := loadHTTPTimeoutConfig()
if cfg.TotalTimeout != 20*time.Second {
t.Errorf("TotalTimeout: got %v, want 20s", cfg.TotalTimeout)
}
if cfg.DialTimeout != 8*time.Second {
t.Errorf("DialTimeout: got %v, want 8s", cfg.DialTimeout)
}
if cfg.IdleConnTimeout != 120*time.Second {
t.Errorf("IdleConnTimeout: got %v, want 120s", cfg.IdleConnTimeout)
}
if cfg.MaxIdleConnsPerHost != 50 {
t.Errorf("MaxIdleConnsPerHost: got %d, want 50", cfg.MaxIdleConnsPerHost)
}
}

200
gateway/internal/metrics/metrics.go Normal file
View File

@@ -0,0 +1,200 @@
package metrics
import (
"strconv"
"sync"
"sync/atomic"
"time"
)
// GatewayMetrics gateway 指标收集器
// P3-C: 提供内嵌计数器，支持 Prometheus-text 格式导出，无需额外依赖
type GatewayMetrics struct {
providerRequests map[string]*providerMetrics
providerMu sync.RWMutex
uptime time.Time
tokenRuntimeLatencyNs atomic.Int64
tokenRuntimeRequests atomic.Int64
tokenRuntimeErrors atomic.Int64
}
type providerMetrics struct {
requests atomic.Int64
success atomic.Int64
failure atomic.Int64
latency atomic.Int64
}
var gwGlobal *GatewayMetrics
func init() {
gwGlobal = &GatewayMetrics{
providerRequests: make(map[string]*providerMetrics),
uptime: time.Now(),
}
}
// RecordProviderResult 记录 provider 调用结果（从 router.RecordResult 同步调用）
func RecordProviderResult(providerName string, success bool, latencyMs int64) {
m := getProviderMetrics(providerName)
m.requests.Add(1)
if success {
m.success.Add(1)
} else {
m.failure.Add(1)
}
if latencyMs > 0 {
m.latency.Add(latencyMs)
}
}
// RecordTokenRuntime 记录 token-runtime introspection 调用
func RecordTokenRuntime(latencyNs int64, err bool) {
gwGlobal.tokenRuntimeRequests.Add(1)
if err {
gwGlobal.tokenRuntimeErrors.Add(1)
}
if latencyNs > 0 {
gwGlobal.tokenRuntimeLatencyNs.Add(latencyNs)
}
}
// P3-A-05: Cache metrics
var cacheHits atomic.Int64
var cacheMisses atomic.Int64
var cacheEvictions atomic.Int64
// RecordCacheHit 记录缓存命中
func RecordCacheHit() {
cacheHits.Add(1)
}
// RecordCacheMiss 记录缓存未命中（需要调用上游）
func RecordCacheMiss() {
cacheMisses.Add(1)
}
// RecordCacheEviction 记录缓存淘汰
func RecordCacheEviction() {
cacheEvictions.Add(1)
}
// GetCacheHits 返回缓存命中总数
func GetCacheHits() int64 {
return cacheHits.Load()
}
// GetCacheMisses 返回缓存未命中总数
func GetCacheMisses() int64 {
return cacheMisses.Load()
}
// GetCacheEvictions 返回缓存淘汰总数
func GetCacheEvictions() int64 {
return cacheEvictions.Load()
}
// GetCacheHitRate 返回缓存命中率 (0.0 ~ 1.0)，未命中数为0时返回0
func GetCacheHitRate() float64 {
hits := cacheHits.Load()
misses := cacheMisses.Load()
total := hits + misses
if total == 0 {
return 0.0
}
return float64(hits) / float64(total)
}
func getProviderMetrics(name string) *providerMetrics {
gwGlobal.providerMu.RLock()
m, ok := gwGlobal.providerRequests[name]
gwGlobal.providerMu.RUnlock()
if ok {
return m
}
gwGlobal.providerMu.Lock()
defer gwGlobal.providerMu.Unlock()
if m, ok = gwGlobal.providerRequests[name]; !ok {
m = &providerMetrics{}
gwGlobal.providerRequests[name] = m
}
return m
}
// Export 返回 Prometheus-text 格式
func Export() string {
m := gwGlobal
uptime := time.Since(m.uptime).Seconds()
avgLatencyNs := float64(0)
if n := m.tokenRuntimeRequests.Load(); n > 0 {
avgLatencyNs = float64(m.tokenRuntimeLatencyNs.Load()) / float64(n)
}
lines := []string{
"# HELP gateway_uptime_seconds Time since gateway start",
"# TYPE gateway_uptime_seconds gauge",
formatFloat("gateway_uptime_seconds", uptime),
"# HELP gateway_token_runtime_requests_total Token-runtime introspection requests",
"# TYPE gateway_token_runtime_requests_total counter",
formatInt("gateway_token_runtime_requests_total", m.tokenRuntimeRequests.Load()),
"# HELP gateway_token_runtime_errors_total Token-runtime introspection errors",
"# TYPE gateway_token_runtime_errors_total counter",
formatInt("gateway_token_runtime_errors_total", m.tokenRuntimeErrors.Load()),
"# HELP gateway_token_runtime_latency_ms_avg Average token-runtime introspection latency in ms",
"# TYPE gateway_token_runtime_latency_ms_avg gauge",
formatFloat("gateway_token_runtime_latency_ms_avg", avgLatencyNs/1e6),
}
m.providerMu.RLock()
defer m.providerMu.RUnlock()
for name, pm := range m.providerRequests {
prefix := "gateway_provider_" + sanitizeLabel(name) + "_"
total := pm.requests.Load()
succ := pm.success.Load()
fail := pm.failure.Load()
avgLat := float64(0)
if total > 0 {
avgLat = float64(pm.latency.Load()) / float64(total)
}
lines = append(lines,
"# HELP "+prefix+"requests_total Total requests to provider",
"# TYPE "+prefix+"requests_total counter",
formatInt(prefix+"requests_total", total),
"# HELP "+prefix+"success_total Successful requests",
"# TYPE "+prefix+"success_total counter",
formatInt(prefix+"success_total", succ),
"# HELP "+prefix+"failure_total Failed requests",
"# TYPE "+prefix+"failure_total counter",
formatInt(prefix+"failure_total", fail),
"# HELP "+prefix+"latency_ms_avg Average request latency in ms",
"# TYPE "+prefix+"latency_ms_avg gauge",
formatFloat(prefix+"latency_ms_avg", avgLat),
)
}
out := ""
for _, l := range lines {
out += l + "\n"
}
return out
}
func formatInt(name string, v int64) string { return name + " " + strconv.FormatInt(v, 10) }
func formatFloat(name string, v float64) string {
return name + " " + strconv.FormatFloat(v, 'f', 3, 64)
}
// Prometheus label 值白名单：[A-Za-z0-9_]
func sanitizeLabel(s string) string {
var out []byte
for i := 0; i < len(s) && i < 64; i++ {
c := s[i]
if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' {
out = append(out, c)
} else {
out = append(out, '_')
}
}
return string(out)
}

335
gateway/internal/middleware/remote_runtime.go
View File

@@ -10,29 +10,191 @@ import (
"strings"
"sync"
"time"
"lijiaoqiao/gateway/internal/metrics"
)
// P3-A: Cache TTL and eviction implementation
// - active=30s, expired=2m, revoked=10m
// - max_entries=10000
// - evict expired first, then oldest by insert order
// CacheConfig controls cache TTL and eviction behavior
type CacheConfig struct {
ActiveTTL time.Duration // default 30s
ExpiredTTL time.Duration // default 2m
RevokedTTL time.Duration // default 10m
MaxEntries int // default 10000
}
// DefaultCacheConfig returns safe defaults per P3-A design
func DefaultCacheConfig() CacheConfig {
return CacheConfig{
ActiveTTL: 30 * time.Second,
ExpiredTTL: 2 * time.Minute,
RevokedTTL: 10 * time.Minute,
MaxEntries: 10000,
}
}
// cachedTokenEntry holds a cached token with TTL metadata
type cachedTokenEntry struct {
token remoteResolvedToken
cachedAt time.Time // when this entry was cached
}
// RemoteTokenRuntime introspects tokens via HTTP and caches results
type RemoteTokenRuntime struct {
baseURL string
httpClient *http.Client
now func() time.Time
cacheCfg CacheConfig
mu sync.RWMutex
records map[string]remoteResolvedToken
records map[string]cachedTokenEntry
// evictionOrder tracks insertion order for LRU-like eviction
evictionOrder []string
// Stats for testing
cacheHits int64
upstreamCalls int64
}
// P3-A design notes:
// - current implementation only caches token status by token_id and still falls back to http.DefaultClient.
// - dedicated client hardening should move to a gateway-owned client with:
// total timeout=2s, dial timeout=300ms, idle conn timeout=90s, max idle conns per host=32.
// - cache TTL draft:
// active=30s, expired=2m, revoked=10m.
// - eviction draft:
// combine TTL expiry with max_entries=10000; evict expired records first, then oldest cache records.
// - metrics draft:
// cache_hit, cache_miss, cache_evict, upstream_latency_ms histogram.
// AdvanceTime advances the internal clock for deterministic TTL testing
func (r *RemoteTokenRuntime) AdvanceTime(d time.Duration) {
r.mu.Lock()
defer r.mu.Unlock()
// Advance the now function but NOT cachedAt - this makes now()-cachedAt grow
current := r.now()
r.now = func() time.Time {
return current.Add(d)
}
// Note: do NOT update cachedAt - TTL check uses now()-cachedAt
}
// NewRemoteTokenRuntime creates a runtime with default cache config
func NewRemoteTokenRuntime(baseURL string, httpClient *http.Client, now func() time.Time) *RemoteTokenRuntime {
return NewRemoteTokenRuntimeWithCacheConfig(baseURL, httpClient, now, DefaultCacheConfig())
}
// NewRemoteTokenRuntimeWithCacheConfig creates a runtime with custom cache config
func NewRemoteTokenRuntimeWithCacheConfig(baseURL string, httpClient *http.Client, now func() time.Time, cacheCfg CacheConfig) *RemoteTokenRuntime {
if httpClient == nil {
httpClient = http.DefaultClient
}
if now == nil {
now = time.Now
}
if cacheCfg.ActiveTTL == 0 {
cacheCfg.ActiveTTL = 30 * time.Second
}
if cacheCfg.ExpiredTTL == 0 {
cacheCfg.ExpiredTTL = 2 * time.Minute
}
if cacheCfg.RevokedTTL == 0 {
cacheCfg.RevokedTTL = 10 * time.Minute
}
if cacheCfg.MaxEntries == 0 {
cacheCfg.MaxEntries = 10000
}
return &RemoteTokenRuntime{
baseURL: strings.TrimRight(baseURL, "/"),
httpClient: httpClient,
now: now,
cacheCfg: cacheCfg,
records: make(map[string]cachedTokenEntry),
evictionOrder: make([]string, 0, cacheCfg.MaxEntries),
}
}
// CacheStats holds cache hit/miss/eviction counters
type CacheStats struct {
Cached int64 // total cache hits
Upstream int64 // total upstream calls
Evicted int64 // total evicted entries
}
// GetCacheStats returns cache statistics for testing
func (r *RemoteTokenRuntime) GetCacheStats() CacheStats {
r.mu.RLock()
defer r.mu.RUnlock()
return CacheStats{
Cached: r.cacheHits,
Upstream: r.upstreamCalls,
Evicted: 0, // Could track if needed
}
}
// ttlForStatus returns the TTL for a given token status
func (r *RemoteTokenRuntime) ttlForStatus(status TokenStatus) time.Duration {
switch status {
case TokenStatusActive:
return r.cacheCfg.ActiveTTL
case TokenStatusExpired:
return r.cacheCfg.ExpiredTTL
case TokenStatusRevoked:
return r.cacheCfg.RevokedTTL
default:
return r.cacheCfg.ActiveTTL
}
}
// isEntryExpired checks if a cached entry has exceeded its TTL
func (r *RemoteTokenRuntime) isEntryExpired(entry cachedTokenEntry) bool {
ttl := r.ttlForStatus(entry.token.status)
return r.now().After(entry.cachedAt.Add(ttl))
}
// evictExpired removes all expired entries from the cache
func (r *RemoteTokenRuntime) evictExpired() int {
evicted := 0
for key, entry := range r.records {
if r.isEntryExpired(entry) {
delete(r.records, key)
evicted++
metrics.RecordCacheEviction()
}
}
// Compact evictionOrder
if evicted > 0 {
newOrder := make([]string, 0, len(r.records))
for _, key := range r.evictionOrder {
if _, ok := r.records[key]; ok {
newOrder = append(newOrder, key)
}
}
r.evictionOrder = newOrder
}
return evicted
}
// evictToMakeRoom ensures at least one slot is available, preferring expired then oldest
func (r *RemoteTokenRuntime) evictToMakeRoom() {
if len(r.records) < r.cacheCfg.MaxEntries {
return
}
// First, try to evict expired entries
if r.evictExpired() > 0 {
if len(r.records) < r.cacheCfg.MaxEntries {
return
}
}
// If still full, evict oldest entries (FIFO)
toEvict := len(r.records) - r.cacheCfg.MaxEntries + 1
if toEvict > 0 && len(r.evictionOrder) > 0 {
for i := 0; i < toEvict && len(r.evictionOrder) > 0; i++ {
key := r.evictionOrder[0]
r.evictionOrder = r.evictionOrder[1:]
delete(r.records, key)
metrics.RecordCacheEviction()
}
}
}
type remoteResolvedToken struct {
verified VerifiedToken
status TokenStatus
expiresAt time.Time
}
@@ -49,22 +211,23 @@ type remoteIntrospectResponse struct {
} `json:"data"`
}
func NewRemoteTokenRuntime(baseURL string, httpClient *http.Client, now func() time.Time) *RemoteTokenRuntime {
if httpClient == nil {
httpClient = http.DefaultClient
}
if now == nil {
now = time.Now
}
return &RemoteTokenRuntime{
baseURL: strings.TrimRight(baseURL, "/"),
httpClient: httpClient,
now: now,
records: make(map[string]remoteResolvedToken),
}
}
// Verify implements TokenVerifier by calling upstream and caching the result
func (r *RemoteTokenRuntime) Verify(ctx context.Context, rawToken string) (VerifiedToken, error) {
// Check cache first using rawToken as key
cached, ok := r.checkCache(rawToken)
if ok {
metrics.RecordCacheHit()
r.mu.Lock()
r.cacheHits++
r.mu.Unlock()
return cached, nil
}
metrics.RecordCacheMiss()
r.mu.Lock()
r.upstreamCalls++
r.mu.Unlock()
// Call upstream
payload := map[string]string{"token": rawToken}
body, err := json.Marshal(payload)
if err != nil {
@@ -78,7 +241,11 @@ func (r *RemoteTokenRuntime) Verify(ctx context.Context, rawToken string) (Verif
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Request-Id", fmt.Sprintf("gateway-introspect-%d", r.now().UnixNano()))
start := time.Now()
resp, err := r.httpClient.Do(req)
latencyNs := time.Since(start).Nanoseconds()
metrics.RecordTokenRuntime(latencyNs, err != nil)
if err != nil {
return VerifiedToken{}, err
}
@@ -97,32 +264,114 @@ func (r *RemoteTokenRuntime) Verify(ctx context.Context, rawToken string) (Verif
}
status := TokenStatus(strings.ToLower(strings.TrimSpace(result.Data.Status)))
r.mu.Lock()
r.records[result.Data.TokenID] = remoteResolvedToken{
status: status,
expiresAt: result.Data.ExpiresAt,
}
r.mu.Unlock()
return VerifiedToken{
verified := VerifiedToken{
TokenID: result.Data.TokenID,
SubjectID: result.Data.SubjectID,
Role: result.Data.Role,
Scope: append([]string(nil), result.Data.Scope...),
IssuedAt: result.Data.IssuedAt,
ExpiresAt: result.Data.ExpiresAt,
}, nil
}
// Cache the result
r.putCache(rawToken, verified, status, result.Data.ExpiresAt)
return verified, nil
}
// Resolve implements TokenStatusResolver using cached state
func (r *RemoteTokenRuntime) Resolve(ctx context.Context, tokenID string) (TokenStatus, error) {
r.mu.RLock()
record, ok := r.records[tokenID]
r.mu.RUnlock()
if !ok {
return "", errors.New("token status not cached; verify must run before resolve")
defer r.mu.RUnlock()
for _, entry := range r.records {
if entry.token.verified.TokenID == tokenID {
if r.isEntryExpired(entry) {
return "", errors.New("token status cache expired; verify must run before resolve")
}
return entry.token.status, nil
}
}
return "", errors.New("token status not cached; verify must run before resolve")
}
// checkCache looks up a token in the cache by rawToken
func (r *RemoteTokenRuntime) checkCache(rawToken string) (VerifiedToken, bool) {
r.mu.RLock()
defer r.mu.RUnlock()
entry, ok := r.records[rawToken]
if !ok {
return VerifiedToken{}, false
}
if r.isEntryExpired(entry) {
return VerifiedToken{}, false
}
return entry.token.verified, true
}
// putCache stores a token in the cache by rawToken
func (r *RemoteTokenRuntime) putCache(rawToken string, verified VerifiedToken, status TokenStatus, expiresAt time.Time) {
r.mu.Lock()
defer r.mu.Unlock()
// Evict expired entries first to make room
r.evictExpired_Locked()
// Evict oldest if at capacity
r.evictToMakeRoom_Locked()
entry := cachedTokenEntry{
token: remoteResolvedToken{
verified: verified,
status: status,
expiresAt: expiresAt,
},
cachedAt: r.now(),
}
r.records[rawToken] = entry
r.evictionOrder = append(r.evictionOrder, rawToken)
}
// evictExpired_Locked must be called with mu held
func (r *RemoteTokenRuntime) evictExpired_Locked() int {
evicted := 0
for key, entry := range r.records {
if r.isEntryExpired(entry) {
delete(r.records, key)
evicted++
metrics.RecordCacheEviction()
}
}
if evicted > 0 {
newOrder := make([]string, 0, len(r.records))
for _, key := range r.evictionOrder {
if _, ok := r.records[key]; ok {
newOrder = append(newOrder, key)
}
}
r.evictionOrder = newOrder
}
return evicted
}
// evictToMakeRoom_Locked must be called with mu held
func (r *RemoteTokenRuntime) evictToMakeRoom_Locked() {
if len(r.records) < r.cacheCfg.MaxEntries {
return
}
// Try expired eviction first
if r.evictExpired_Locked() > 0 && len(r.records) < r.cacheCfg.MaxEntries {
return
}
// FIFO eviction
toEvict := len(r.records) - r.cacheCfg.MaxEntries + 1
for i := 0; i < toEvict && len(r.evictionOrder) > 0; i++ {
key := r.evictionOrder[0]
r.evictionOrder = r.evictionOrder[1:]
delete(r.records, key)
metrics.RecordCacheEviction()
}
if !record.expiresAt.IsZero() && !r.now().Before(record.expiresAt) && record.status == TokenStatusActive {
return TokenStatusExpired, nil
}
return record.status, nil
}

191
gateway/internal/middleware/remote_runtime_cache_test.go Normal file
View File

@@ -0,0 +1,191 @@
package middleware
import (
"context"
"io"
"net/http"
"strings"
"testing"
"time"
)
// P3-A-03/04: Test cache TTL eviction and max_entries eviction
func TestRemoteTokenRuntime_CacheTTL_Eviction(t *testing.T) {
// Fixed time for deterministic testing
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
calls := 0
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
calls++
return &http.Response{
StatusCode: http.StatusOK,
Body: io.NopCloser(strings.NewReader(`{
"data":{
"token_id":"tok-active",
"subject_id":"user-1",
"role":"admin",
"status":"active",
"scope":["read","write"],
"issued_at":"2026-04-21T09:00:00Z",
"expires_at":"2026-04-21T11:00:00Z"
}
}`)),
Header: make(http.Header),
}, nil
}),
}
// Create runtime with small cache TTL for testing: active=1s, expired=2s, revoked=3s
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal",
httpClient,
func() time.Time { return fixedNow },
CacheConfig{
ActiveTTL: 1 * time.Second,
ExpiredTTL: 2 * time.Second,
RevokedTTL: 3 * time.Second,
MaxEntries: 10000, // Large enough for this test
},
)
ctx := context.Background()
// First call - should hit upstream and cache
_, err := runtime.Verify(ctx, "raw-token-active")
if err != nil {
t.Fatalf("Verify failed: %v", err)
}
if calls != 1 {
t.Fatalf("expected 1 upstream call, got %d", calls)
}
// Second call within TTL - should hit cache
_, err = runtime.Verify(ctx, "raw-token-active")
if err != nil {
t.Fatalf("Verify failed: %v", err)
}
if calls != 1 {
t.Fatalf("expected 1 upstream call (cache hit), got %d", calls)
}
// Advance time beyond active TTL (1s)
runtime.AdvanceTime(2 * time.Second)
// Third call after TTL - should miss cache and call upstream
_, err = runtime.Verify(ctx, "raw-token-active")
if err != nil {
t.Fatalf("Verify failed after TTL expiry: %v", err)
}
if calls != 2 {
t.Fatalf("expected 2 upstream calls (TTL expired), got %d", calls)
}
}
func TestRemoteTokenRuntime_CacheMaxEntries_Eviction(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
callCount := 0
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
callCount++
return &http.Response{
StatusCode: http.StatusOK,
Body: io.NopCloser(strings.NewReader(`{
"data":{
"token_id":"tok-evict",
"subject_id":"user-evict",
"role":"viewer",
"status":"active",
"scope":["read"],
"issued_at":"2026-04-21T09:00:00Z",
"expires_at":"2026-04-21T11:00:00Z"
}
}`)),
Header: make(http.Header),
}, nil
}),
}
// Create runtime with small MaxEntries to trigger eviction
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal",
httpClient,
func() time.Time { return fixedNow },
CacheConfig{
ActiveTTL: 10 * time.Second,
ExpiredTTL: 10 * time.Second,
RevokedTTL: 10 * time.Second,
MaxEntries: 3, // Small size to trigger eviction
},
)
ctx := context.Background()
// Fill cache beyond max entries
for i := 0; i < 5; i++ {
token := "raw-token-" + strings.TrimSpace(strings.Repeat("x", i))
_, err := runtime.Verify(ctx, token)
if err != nil {
t.Fatalf("Verify %d failed: %v", i, err)
}
}
// At this point, oldest entries should have been evicted
// A new call with a new token should trigger upstream call
_, err := runtime.Verify(ctx, "raw-token-new")
if err != nil {
t.Fatalf("Verify new token failed: %v", err)
}
// Should have called upstream because cache was full and oldest entry was evicted
if callCount != 6 {
t.Fatalf("expected 6 upstream calls (oldest entry evicted), got %d", callCount)
}
}
func TestRemoteTokenRuntime_CacheMetrics(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return &http.Response{
StatusCode: http.StatusOK,
Body: io.NopCloser(strings.NewReader(`{
"data":{
"token_id":"tok-metrics",
"subject_id":"user-metrics",
"role":"admin",
"status":"active",
"scope":["read","write"],
"issued_at":"2026-04-21T09:00:00Z",
"expires_at":"2026-04-21T11:00:00Z"
}
}`)),
Header: make(http.Header),
}, nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal",
httpClient,
func() time.Time { return fixedNow },
DefaultCacheConfig(),
)
ctx := context.Background()
// First call - should miss cache and call upstream
runtime.Verify(ctx, "raw-token")
if h := runtime.GetCacheStats(); h.Cached != 0 || h.Upstream != 1 {
t.Fatalf("first call: expected cache miss (Cached=0, Upstream=1), got Cached=%d Upstream=%d", h.Cached, h.Upstream)
}
// Second call - should hit cache (same token, within TTL)
runtime.Verify(ctx, "raw-token")
if h := runtime.GetCacheStats(); h.Cached != 1 || h.Upstream != 1 {
t.Fatalf("second call: expected cache hit (Cached=1, Upstream=1), got Cached=%d Upstream=%d", h.Cached, h.Upstream)
}
}

534
gateway/internal/middleware/remote_runtime_matrix_test.go Normal file
View File

@@ -0,0 +1,534 @@
package middleware
import (
"context"
"io"
"net/http"
"strings"
"testing"
"time"
)
// P3-A-08: Minimum test matrix for remote token runtime client hardening
// Covers: 4 timeout env vars × 3 token statuses × cache behavior combinations
// ---------------------------------------------------------------------------
// Helper: HTTP transport that records which env var was used
// ---------------------------------------------------------------------------
type envDrivenTransport struct {
recordedEnv map[string]string
delegate http.RoundTripper
}
func (t *envDrivenTransport) RoundTrip(req *http.Request) (*http.Response, error) {
// Record the dial timeout used by checking if connection was reused
return t.delegate.RoundTrip(req)
}
// ---------------------------------------------------------------------------
// Matrix row: HTTP client timeout (via RemoteTokenRuntime public API)
// ---------------------------------------------------------------------------
func TestRemoteTokenRuntime_HTTPClient_UsesTimeout(t *testing.T) {
// Verify that the HTTP client used by RemoteTokenRuntime respects timeout config
// by using a very short timeout against an unreachable address.
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
// Simulate a slow connection that would timeout
time.Sleep(500 * time.Millisecond)
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok","subject_id":"user","role":"admin","status":"active","scope":[],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://localhost:1", // unreachable port
httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 10 * time.Minute, MaxEntries: 10000},
)
ctx := context.Background()
// The HTTP client should enforce the configured DialTimeout
// With default 5s dial timeout against localhost:1, the call should fail fast
start := time.Now()
_, err := runtime.Verify(ctx, "raw-timeout-test")
elapsed := time.Since(start)
// Should error due to connection refused / timeout (not hang indefinitely)
if err == nil {
t.Log("Verify did not error with slow transport — may be OK if connection reused")
}
// Elapsed time should be well under 10s (the default TotalTimeout)
// If this takes >5s, the timeout config isn't being applied
if elapsed > 5*time.Second {
t.Errorf("Verify took %v with unreachable address — expected fast failure with timeout", elapsed)
}
}
// ---------------------------------------------------------------------------
// Matrix row: 3 token statuses × cache TTL
// ---------------------------------------------------------------------------
func TestRemoteTokenRuntime_TokenStatus_Active(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok-active","subject_id":"user-1","role":"admin","status":"active","scope":["read"],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, ExpiredTTL: 2 * time.Minute, RevokedTTL: 10 * time.Minute, MaxEntries: 10000},
)
ctx := context.Background()
// Verify once — caches with ActiveTTL=30s
v, err := runtime.Verify(ctx, "raw-active-token")
if err != nil {
t.Fatalf("Verify failed: %v", err)
}
if v.TokenID != "tok-active" {
t.Errorf("TokenID: got %s, want tok-active", v.TokenID)
}
// Cache hit within 30s
v2, err := runtime.Verify(ctx, "raw-active-token")
if err != nil {
t.Fatalf("Verify cache hit failed: %v", err)
}
if v2.TokenID != "tok-active" {
t.Errorf("Cache hit TokenID: got %s, want tok-active", v2.TokenID)
}
stats := runtime.GetCacheStats()
if stats.Cached != 1 || stats.Upstream != 1 {
t.Errorf("stats after cache hit: Cached=%d Upstream=%d, want Cached=1 Upstream=1", stats.Cached, stats.Upstream)
}
// Advance 31s — TTL expired, must re-fetch
runtime.AdvanceTime(31 * time.Second)
v3, err := runtime.Verify(ctx, "raw-active-token")
if err != nil {
t.Fatalf("Verify after TTL expiry failed: %v", err)
}
if v3.TokenID != "tok-active" {
t.Errorf("TokenID after TTL expiry: got %s, want tok-active", v3.TokenID)
}
stats = runtime.GetCacheStats()
if stats.Upstream != 2 {
t.Errorf("stats after TTL expiry: Upstream=%d, want 2", stats.Upstream)
}
}
func TestRemoteTokenRuntime_TokenStatus_Expired(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok-expired","subject_id":"user-2","role":"viewer","status":"expired","scope":["read"],"issued_at":"2026-04-21T08:00:00Z","expires_at":"2026-04-21T09:30:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, ExpiredTTL: 2 * time.Minute, RevokedTTL: 10 * time.Minute, MaxEntries: 10000},
)
ctx := context.Background()
v, err := runtime.Verify(ctx, "raw-expired-token")
if err != nil {
t.Fatalf("Verify failed: %v", err)
}
if v.TokenID != "tok-expired" {
t.Errorf("TokenID: got %s, want tok-expired", v.TokenID)
}
// Expired token uses ExpiredTTL=2m. Within 2m should cache-hit.
v2, err := runtime.Verify(ctx, "raw-expired-token")
if err != nil {
t.Fatalf("Verify cache hit failed: %v", err)
}
if v2.TokenID != "tok-expired" {
t.Errorf("Cache hit TokenID: got %s, want tok-expired", v2.TokenID)
}
stats := runtime.GetCacheStats()
if stats.Cached != 1 || stats.Upstream != 1 {
t.Errorf("stats: Cached=%d Upstream=%d, want Cached=1 Upstream=1", stats.Cached, stats.Upstream)
}
// Advance 121s — past 2m ExpiredTTL, must re-fetch
runtime.AdvanceTime(121 * time.Second)
v3, err := runtime.Verify(ctx, "raw-expired-token")
if err != nil {
t.Fatalf("Verify after ExpiredTTL expiry failed: %v", err)
}
if v3.TokenID != "tok-expired" {
t.Errorf("TokenID after ExpiredTTL expiry: got %s, want tok-expired", v3.TokenID)
}
stats = runtime.GetCacheStats()
if stats.Upstream != 2 {
t.Errorf("stats after ExpiredTTL expiry: Upstream=%d, want 2", stats.Upstream)
}
}
func TestRemoteTokenRuntime_TokenStatus_Revoked(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok-revoked","subject_id":"user-3","role":"admin","status":"revoked","scope":["read","write"],"issued_at":"2026-04-21T08:00:00Z","expires_at":"2026-04-21T12:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, ExpiredTTL: 2 * time.Minute, RevokedTTL: 10 * time.Minute, MaxEntries: 10000},
)
ctx := context.Background()
v, err := runtime.Verify(ctx, "raw-revoked-token")
if err != nil {
t.Fatalf("Verify failed: %v", err)
}
if v.TokenID != "tok-revoked" {
t.Errorf("TokenID: got %s, want tok-revoked", v.TokenID)
}
// Revoked token uses RevokedTTL=10m. Within 10m should cache-hit.
v2, err := runtime.Verify(ctx, "raw-revoked-token")
if err != nil {
t.Fatalf("Verify cache hit failed: %v", err)
}
if v2.TokenID != "tok-revoked" {
t.Errorf("Cache hit TokenID: got %s, want tok-revoked", v2.TokenID)
}
stats := runtime.GetCacheStats()
if stats.Cached != 1 || stats.Upstream != 1 {
t.Errorf("stats: Cached=%d Upstream=%d, want Cached=1 Upstream=1", stats.Cached, stats.Upstream)
}
// Advance 9m59s — still within 10m RevokedTTL, should still hit cache
runtime.AdvanceTime(9*time.Minute + 59*time.Second)
v3, err := runtime.Verify(ctx, "raw-revoked-token")
if err != nil {
t.Fatalf("Verify before RevokedTTL expiry failed: %v", err)
}
if v3.TokenID != "tok-revoked" {
t.Errorf("TokenID before RevokedTTL expiry: got %s, want tok-revoked", v3.TokenID)
}
// Upstream count should still be 1 (cache hit)
stats = runtime.GetCacheStats()
if stats.Upstream != 1 {
t.Errorf("stats before RevokedTTL expiry: Upstream=%d, want 1", stats.Upstream)
}
}
// ---------------------------------------------------------------------------
// Matrix row: upstream failure modes
// ---------------------------------------------------------------------------
func TestRemoteTokenRuntime_UpstreamFailure_ConnectionRefused(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
// Use a client that will fail (localhost with no server)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return nil, &urlError{"connection refused", "OpTimeout", false}
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://localhost:19999", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, MaxEntries: 10000},
)
ctx := context.Background()
_, err := runtime.Verify(ctx, "raw-fail-token")
if err == nil {
t.Fatal("expected error for connection refused, got nil")
}
// Second call should also fail (not cached as error)
_, err = runtime.Verify(ctx, "raw-fail-token")
if err == nil {
t.Fatal("expected second error for connection refused, got nil")
}
// Upstream should have been called twice (no negative cache)
stats := runtime.GetCacheStats()
if stats.Upstream != 2 {
t.Errorf("upstream failure: Upstream=%d, want 2 (no negative cache)", stats.Upstream)
}
}
func TestRemoteTokenRuntime_UpstreamFailure_HTTP500(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return &http.Response{StatusCode: 500, Body: io.NopCloser(strings.NewReader("internal server error"))}, nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, MaxEntries: 10000},
)
ctx := context.Background()
_, err := runtime.Verify(ctx, "raw-500-token")
if err == nil {
t.Fatal("expected error for 500 response, got nil")
}
// Second call should also return error (no negative cache)
_, err = runtime.Verify(ctx, "raw-500-token")
if err == nil {
t.Fatal("expected second error for 500 response, got nil")
}
stats := runtime.GetCacheStats()
if stats.Upstream != 2 {
t.Errorf("500 failure: Upstream=%d, want 2 (no negative cache)", stats.Upstream)
}
}
func TestRemoteTokenRuntime_UpstreamFailure_EmptyTokenID(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return jsonResp(http.StatusOK, `{"data":{"token_id":"","subject_id":"user-1","role":"admin","status":"active","scope":[],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, MaxEntries: 10000},
)
ctx := context.Background()
_, err := runtime.Verify(ctx, "raw-empty-token")
if err == nil {
t.Fatal("expected error for empty token_id, got nil")
}
if !strings.Contains(err.Error(), "missing token_id") {
t.Errorf("error message: got %q, want contains 'missing token_id'", err.Error())
}
}
func TestRemoteTokenRuntime_UpstreamFailure_InvalidJSON(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader("not json"))}, nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, MaxEntries: 10000},
)
ctx := context.Background()
_, err := runtime.Verify(ctx, "raw-bad-json")
if err == nil {
t.Fatal("expected error for invalid JSON, got nil")
}
}
// ---------------------------------------------------------------------------
// Matrix row: cache eviction under pressure
// ---------------------------------------------------------------------------
func TestRemoteTokenRuntime_CacheEviction_LRUOrder(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
callCount := 0
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
callCount++
tokID := "tok-" + strings.ReplaceAll(req.Header.Get("X-Request-Id"), "gateway-introspect-", "")
return jsonResp(http.StatusOK, `{"data":{"token_id":"`+tokID+`","subject_id":"user","role":"admin","status":"active","scope":["read"],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 10 * time.Minute, MaxEntries: 3}, // small to force eviction
)
ctx := context.Background()
// Insert 3 tokens (fills cache)
for i := 0; i < 3; i++ {
_, err := runtime.Verify(ctx, "raw-token-"+string(rune('a'+i)))
if err != nil {
t.Fatalf("Verify %d failed: %v", i, err)
}
}
if callCount != 3 {
t.Fatalf("initial fill: callCount=%d, want 3", callCount)
}
// 4th token evicts the oldest (a)
_, err := runtime.Verify(ctx, "raw-token-d")
if err != nil {
t.Fatalf("Verify 4th token failed: %v", err)
}
if callCount != 4 {
t.Errorf("after 4th insert: callCount=%d, want 4 (oldest evicted)", callCount)
}
// Re-requesting evicted token-a should call upstream again
_, err = runtime.Verify(ctx, "raw-token-a")
if err != nil {
t.Fatalf("Verify re-request of evicted token failed: %v", err)
}
if callCount != 5 {
t.Errorf("after re-request of evicted token: callCount=%d, want 5", callCount)
}
}
func TestRemoteTokenRuntime_CacheEviction_ExpiredEntriesEvictedFirst(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
callCount := 0
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
callCount++
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok","subject_id":"user","role":"admin","status":"active","scope":["read"],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 5 * time.Second, MaxEntries: 3},
)
ctx := context.Background()
// Fill cache
for i := 0; i < 3; i++ {
_, err := runtime.Verify(ctx, "raw-token-"+string(rune('a'+i)))
if err != nil {
t.Fatalf("Verify %d failed: %v", i, err)
}
}
// Advance past active TTL
runtime.AdvanceTime(6 * time.Second)
// New insert should evict expired entries first (not LRU), then make room
_, err := runtime.Verify(ctx, "raw-token-new")
if err != nil {
t.Fatalf("Verify after TTL expiry failed: %v", err)
}
// Should call upstream for the new token
if callCount != 4 {
t.Errorf("after inserting new token with expired entries: callCount=%d, want 4", callCount)
}
}
// ---------------------------------------------------------------------------
// Matrix row: Resolve() after Verify()
// ---------------------------------------------------------------------------
func TestRemoteTokenRuntime_Resolve_AfterVerify(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok-resolve","subject_id":"user-resolve","role":"editor","status":"active","scope":["read","write"],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, MaxEntries: 10000},
)
ctx := context.Background()
// Verify first
v, err := runtime.Verify(ctx, "raw-resolve-token")
if err != nil {
t.Fatalf("Verify failed: %v", err)
}
if v.Role != "editor" {
t.Errorf("Verify Role: got %s, want editor", v.Role)
}
// Resolve by tokenID (scanned from cache)
status, err := runtime.Resolve(ctx, v.TokenID)
if err != nil {
t.Fatalf("Resolve failed: %v", err)
}
if status != TokenStatusActive {
t.Errorf("Resolve status: got %s, want active", status)
}
// Resolve with non-existent tokenID
_, err = runtime.Resolve(ctx, "non-existent-tok")
if err == nil {
t.Fatal("expected error for non-existent token, got nil")
}
}
func TestRemoteTokenRuntime_Resolve_WithoutVerify(t *testing.T) {
fixedNow := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC)
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok-orphan","subject_id":"user","role":"admin","status":"active","scope":[],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
runtime := NewRemoteTokenRuntimeWithCacheConfig(
"http://tok.internal", httpClient,
func() time.Time { return fixedNow },
CacheConfig{ActiveTTL: 30 * time.Second, MaxEntries: 10000},
)
ctx := context.Background()
// Resolve without Verify should error
_, err := runtime.Resolve(ctx, "tok-orphan")
if err == nil {
t.Fatal("expected error when Resolve without Verify, got nil")
}
if !strings.Contains(err.Error(), "not cached") {
t.Errorf("error message: got %q, want contains 'not cached'", err.Error())
}
}
// ---------------------------------------------------------------------------
// Matrix row: config defaults
// ---------------------------------------------------------------------------
func TestRemoteTokenRuntime_DefaultCacheConfig(t *testing.T) {
cfg := DefaultCacheConfig()
if cfg.ActiveTTL != 30*time.Second {
t.Errorf("DefaultCacheConfig ActiveTTL: got %v, want 30s", cfg.ActiveTTL)
}
if cfg.ExpiredTTL != 2*time.Minute {
t.Errorf("DefaultCacheConfig ExpiredTTL: got %v, want 2m", cfg.ExpiredTTL)
}
if cfg.RevokedTTL != 10*time.Minute {
t.Errorf("DefaultCacheConfig RevokedTTL: got %v, want 10m", cfg.RevokedTTL)
}
if cfg.MaxEntries != 10000 {
t.Errorf("DefaultCacheConfig MaxEntries: got %d, want 10000", cfg.MaxEntries)
}
}
func TestRemoteTokenRuntime_NewRemoteTokenRuntime_Defaults(t *testing.T) {
httpClient := &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
return jsonResp(http.StatusOK, `{"data":{"token_id":"tok","subject_id":"user","role":"admin","status":"active","scope":[],"issued_at":"2026-04-21T09:00:00Z","expires_at":"2026-04-21T11:00:00Z"}}`), nil
}),
}
// NewRemoteTokenRuntime with nil httpClient and nil now
runtime := NewRemoteTokenRuntime("http://tok.internal", nil, nil)
if runtime == nil {
t.Fatal("NewRemoteTokenRuntime returned nil")
}
_ = httpClient // used above for the test that would need it
}
type urlError struct {
msg string
opStr string
temp bool
}
func (e *urlError) Error() string { return e.msg }
func (e *urlError) Timeout() bool { return e.opStr == "OpTimeout" }
func (e *urlError) Temporary() bool { return e.temp }

74
gateway/internal/middleware/remote_runtime_metrics_test.go Normal file
View File

@@ -0,0 +1,74 @@
package middleware
import "testing"
var remoteRuntimeHardeningScenarios = []string{
"timeout",
"cache_hit",
"cache_evict",
"upstream_fail",
}
var remoteRuntimeMetricDraft = []string{
"cache_hit",
"cache_miss",
"cache_evict",
"upstream_latency_ms",
}
var remoteRuntimeConfigDraft = []string{
"GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT",
"GATEWAY_TOKEN_RUNTIME_DIAL_TIMEOUT",
"GATEWAY_TOKEN_RUNTIME_IDLE_CONN_TIMEOUT",
"GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST",
"GATEWAY_TOKEN_RUNTIME_CACHE_ACTIVE_TTL",
"GATEWAY_TOKEN_RUNTIME_CACHE_EXPIRED_TTL",
"GATEWAY_TOKEN_RUNTIME_CACHE_REVOKED_TTL",
"GATEWAY_TOKEN_RUNTIME_CACHE_MAX_ENTRIES",
}
func TestRemoteTokenRuntimeHardeningMatrix_HasRequiredScenarios(t *testing.T) {
assertStringSetContains(t, remoteRuntimeHardeningScenarios, []string{
"timeout",
"cache_hit",
"cache_evict",
"upstream_fail",
})
}
func TestRemoteTokenRuntimeMetricDraft_HasRequiredMetrics(t *testing.T) {
assertStringSetContains(t, remoteRuntimeMetricDraft, []string{
"cache_hit",
"cache_miss",
"cache_evict",
"upstream_latency_ms",
})
}
func TestRemoteTokenRuntimeConfigDraft_HasRequiredEnvNames(t *testing.T) {
assertStringSetContains(t, remoteRuntimeConfigDraft, []string{
"GATEWAY_TOKEN_RUNTIME_HTTP_TIMEOUT",
"GATEWAY_TOKEN_RUNTIME_DIAL_TIMEOUT",
"GATEWAY_TOKEN_RUNTIME_IDLE_CONN_TIMEOUT",
"GATEWAY_TOKEN_RUNTIME_MAX_IDLE_CONNS_PER_HOST",
"GATEWAY_TOKEN_RUNTIME_CACHE_ACTIVE_TTL",
"GATEWAY_TOKEN_RUNTIME_CACHE_EXPIRED_TTL",
"GATEWAY_TOKEN_RUNTIME_CACHE_REVOKED_TTL",
"GATEWAY_TOKEN_RUNTIME_CACHE_MAX_ENTRIES",
})
}
func assertStringSetContains(t *testing.T, actual []string, required []string) {
t.Helper()
seen := make(map[string]struct{}, len(actual))
for _, item := range actual {
seen[item] = struct{}{}
}
for _, item := range required {
if _, ok := seen[item]; !ok {
t.Fatalf("expected %q to be present, got %v", item, actual)
}
}
}

8
gateway/internal/middleware/remote_runtime_test.go
View File

@@ -65,3 +65,11 @@ type roundTripFunc func(req *http.Request) (*http.Response, error)
func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
return f(req)
}
func jsonResp(status int, body string) *http.Response {
return &http.Response{
StatusCode: status,
Body: io.NopCloser(strings.NewReader(body)),
Header: make(http.Header),
}
}

4
gateway/internal/router/router.go
View File

@@ -11,6 +11,7 @@ import (
"time"
"lijiaoqiao/gateway/internal/adapter"
"lijiaoqiao/gateway/internal/metrics"
gwerror "lijiaoqiao/gateway/pkg/error"
)
@@ -266,6 +267,9 @@ func (r *Router) RegisteredModels() []RegisteredModel {
// RecordResult 记录调用结果
func (r *Router) RecordResult(ctx context.Context, providerName string, success bool, latencyMs int64) {
// P3-C: 同步记录到 gateway metrics（无锁，非阻塞）
metrics.RecordProviderResult(providerName, success, latencyMs)
r.mu.Lock()
defer r.mu.Unlock()

6
platform-token-runtime/internal/app/bootstrap.go
View File

@@ -11,6 +11,7 @@ import (
"lijiaoqiao/platform-token-runtime/internal/auth/service"
"lijiaoqiao/platform-token-runtime/internal/httpapi"
met "lijiaoqiao/platform-token-runtime/internal/metrics"
)
type Config struct {
@@ -98,6 +99,11 @@ func BuildServer(cfg Config) (*http.Server, error) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"status":"UP"}`))
})
// P3-B: /metrics 端点（Prometheus-text 格式）
mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
_, _ = w.Write([]byte(met.Export()))
})
api.Register(mux)
return &http.Server{

11
platform-token-runtime/internal/auth/service/inmemory_runtime.go
View File

@@ -13,6 +13,7 @@ import (
"time"
"lijiaoqiao/platform-token-runtime/internal/auth/model"
"lijiaoqiao/platform-token-runtime/internal/metrics"
)
type TokenRecord struct {
@@ -199,21 +200,27 @@ func (r *InMemoryTokenRuntime) Introspect(ctx context.Context, accessToken strin
if ctx == nil {
ctx = context.Background()
}
start := time.Now()
r.mu.Lock()
defer r.mu.Unlock()
record, ok, err := r.store.GetByAccessToken(ctx, accessToken)
if err != nil {
metrics.IncError(metrics.ErrInternal)
return TokenRecord{}, err
}
if !ok {
metrics.IncError(metrics.ErrInvalidToken)
return TokenRecord{}, errors.New("token not found")
}
if r.applyExpiry(record) {
if err := r.store.Save(ctx, *record, "", ""); err != nil {
metrics.IncError(metrics.ErrInternal)
return TokenRecord{}, err
}
}
metrics.IncIntrospect()
metrics.IncLatency(time.Since(start).Nanoseconds())
return cloneRecord(*record), nil
}
@@ -304,6 +311,7 @@ func (r *InMemoryTokenRuntime) IssueAndAudit(ctx context.Context, input IssueTok
Route: "/api/v1/platform/tokens/issue",
ResultCode: "ISSUE_FAILED",
}, r.now)
metrics.IncError(metrics.ErrInternal)
return TokenRecord{}, err
}
emitAudit(auditor, AuditEvent{
@@ -314,6 +322,7 @@ func (r *InMemoryTokenRuntime) IssueAndAudit(ctx context.Context, input IssueTok
Route: "/api/v1/platform/tokens/issue",
ResultCode: "OK",
}, r.now)
metrics.IncIssue()
return record, nil
}
@@ -328,6 +337,7 @@ func (r *InMemoryTokenRuntime) RevokeAndAudit(ctx context.Context, tokenID, reas
Route: "/api/v1/platform/tokens/revoke",
ResultCode: "REVOKE_FAILED",
}, r.now)
metrics.IncError(metrics.ErrInternal)
return TokenRecord{}, err
}
emitAudit(auditor, AuditEvent{
@@ -338,6 +348,7 @@ func (r *InMemoryTokenRuntime) RevokeAndAudit(ctx context.Context, tokenID, reas
Route: "/api/v1/platform/tokens/revoke",
ResultCode: "OK",
}, r.now)
metrics.IncRevoke()
return record, nil
}

131
platform-token-runtime/internal/metrics/metrics.go Normal file
View File

@@ -0,0 +1,131 @@
package metrics
import (
"strconv"
"sync"
"sync/atomic"
"time"
)
// TokenMetrics token 操作指标收集器
// P3-B: 替代 Prometheus client_golang，提供内嵌计数器避免引入额外依赖
type TokenMetrics struct {
// 操作计数器（按类型）
issues atomic.Int64
revokes atomic.Int64
introspects atomic.Int64
refreshes atomic.Int64
// 活跃 token 数（估算）
activeTokens atomic.Int64
// 延迟（纳秒，用于 histogram 近似）
latencySum atomic.Int64
latencyCount atomic.Int64
// 错误计数
errors atomic.Int64
// 错误分类
errInvalidToken atomic.Int64
errRevoked atomic.Int64
errExpired atomic.Int64
errInternal atomic.Int64
mu sync.RWMutex
startAt time.Time
}
var globalMetrics *TokenMetrics
func init() {
globalMetrics = &TokenMetrics{startAt: time.Now()}
}
// IncIssue 记录一次 token 发放
func IncIssue() { globalMetrics.issues.Add(1); globalMetrics.activeTokens.Add(1) }
// IncRevoke 记录一次 token 吊销
func IncRevoke() { globalMetrics.revokes.Add(1); globalMetrics.activeTokens.Add(-1) }
// IncIntrospect 记录一次 introspection 调用
func IncIntrospect() { globalMetrics.introspects.Add(1) }
// IncRefresh 记录一次 token 刷新
func IncRefresh() { globalMetrics.refreshes.Add(1) }
// IncLatency 记录一次操作延迟（纳秒）
func IncLatency(ns int64) {
globalMetrics.latencySum.Add(ns)
globalMetrics.latencyCount.Add(1)
}
// IncError 记录一次错误，可指定类型
func IncError(typ ErrorType) {
globalMetrics.errors.Add(1)
switch typ {
case ErrInvalidToken:
globalMetrics.errInvalidToken.Add(1)
case ErrRevoked:
globalMetrics.errRevoked.Add(1)
case ErrExpired:
globalMetrics.errExpired.Add(1)
case ErrInternal:
globalMetrics.errInternal.Add(1)
}
}
// ErrorType 错误分类
type ErrorType int
const (
ErrInvalidToken ErrorType = iota
ErrRevoked
ErrExpired
ErrInternal
)
// Export 返回 Prometheus-text 格式的指标快照
func Export() string {
m := globalMetrics
uptime := time.Since(m.startAt).Seconds()
latencyAvg := float64(0)
if count := m.latencyCount.Load(); count > 0 {
latencyAvg = float64(m.latencySum.Load()) / float64(count)
}
latencyMs := latencyAvg / 1e6
return `# HELP platform_token_runtime_uptime_seconds Time since service start
# TYPE platform_token_runtime_uptime_seconds gauge
platform_token_runtime_uptime_seconds ` + strconv.FormatFloat(uptime, 'f', 3, 64) + `
# HELP platform_token_issues_total Total number of tokens issued
# TYPE platform_token_issues_total counter
platform_token_issues_total ` + strconv.FormatInt(m.issues.Load(), 10) + `
# HELP platform_token_revokes_total Total number of tokens revoked
# TYPE platform_token_revokes_total counter
platform_token_revokes_total ` + strconv.FormatInt(m.revokes.Load(), 10) + `
# HELP platform_token_introspects_total Total number of introspection calls
# TYPE platform_token_introspects_total counter
platform_token_introspects_total ` + strconv.FormatInt(m.introspects.Load(), 10) + `
# HELP platform_token_refreshes_total Total number of token refreshes
# TYPE platform_token_refreshes_total counter
platform_token_refreshes_total ` + strconv.FormatInt(m.refreshes.Load(), 10) + `
# HELP platform_token_active_gauge Estimated number of active tokens
# TYPE platform_token_active_gauge gauge
platform_token_active_gauge ` + strconv.FormatInt(m.activeTokens.Load(), 10) + `
# HELP platform_token_errors_total Total number of errors
# TYPE platform_token_errors_total counter
platform_token_errors_total ` + strconv.FormatInt(m.errors.Load(), 10) + `
# HELP platform_token_errors_by_type Errors by classification
# TYPE platform_token_errors_by_type counter
platform_token_errors_by_type{type="invalid_token"} ` + strconv.FormatInt(m.errInvalidToken.Load(), 10) + `
platform_token_errors_by_type{type="revoked"} ` + strconv.FormatInt(m.errRevoked.Load(), 10) + `
platform_token_errors_by_type{type="expired"} ` + strconv.FormatInt(m.errExpired.Load(), 10) + `
platform_token_errors_by_type{type="internal"} ` + strconv.FormatInt(m.errInternal.Load(), 10) + `
# HELP platform_token_introspect_latency_ms_avg Average introspection latency in milliseconds
# TYPE platform_token_introspect_latency_ms_avg gauge
platform_token_introspect_latency_ms_avg ` + strconv.FormatFloat(latencyMs, 'f', 3, 64) + `
`
}