mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(workbench): bound session summary runtime query (#1952)

Browse Source
This commit is contained in:
Lyon
2026-06-23 05:58:00 +08:00
committed by GitHub
parent 5e1cc8ade5
commit 6e68cacd9b
4 changed files with 71 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/deploy.schema.json
+10
View File
@@ -392,6 +392,7 @@
"type": "object",
"properties": {
"client": { "$ref": "#/$defs/workbenchRuntimeClient" },
"sessionsSummary": { "$ref": "#/$defs/workbenchRuntimeSessionsSummary" },
"factsQuery": { "$ref": "#/$defs/workbenchRuntimeFactsQuery" },
"postgres": { "$ref": "#/$defs/workbenchRuntimePostgres" },
"cache": { "$ref": "#/$defs/workbenchRuntimeCache" }
@@ -414,6 +415,15 @@
},
"additionalProperties": false
},
"workbenchRuntimeSessionsSummary": {
"type": "object",
"properties": {
"maxAttempts": { "type": "integer", "minimum": 1 },
"attemptTimeoutMs": { "type": "integer", "minimum": 1 },
"backoffMs": { "type": "integer", "minimum": 0 }
},
"additionalProperties": false
},
"workbenchRuntimePostgres": {
"type": "object",
"properties": {
deploy/deploy.yaml
+5 -1
View File
@@ -309,12 +309,16 @@ lanes:
workbenchRuntime:
client:
timeoutMs: 3000
sessionsSummary:
maxAttempts: 1
attemptTimeoutMs: 2500
backoffMs: 0
factsQuery:
maxAttempts: 1
backoffBaseMs: 0
backoffMaxMs: 1
postgres:
poolMax: 2
poolMax: 4
queryTimeoutMs: 3000
readyTimeoutMs: 2000
cache:
internal/workbenchruntime/service.go
+43 -21
View File
@@ -71,12 +71,15 @@ type sessionsSummaryRevision struct {
}
type Config struct {
Addr string
DatabaseURL string
QueryTimeout time.Duration
ReadinessDelay time.Duration
PoolMax int
Cache cacheConfig
Addr string
DatabaseURL string
QueryTimeout time.Duration
ReadinessDelay time.Duration
PoolMax int
SessionSummaryMaxAttempts int
SessionSummaryAttemptTimeout time.Duration
SessionSummaryRetryBackoff time.Duration
Cache cacheConfig
}
type Server struct {
@@ -1442,16 +1445,17 @@ func (s *Server) dbQueryAttrs(table string, argCount int, extra map[string]any)
func (s *Server) querySessionSummaries(ctx context.Context, stage string, sqlText string, args []any) ([]any, error) {
var lastErr error
for attempt := 1; attempt <= sessionSummaryQueryMaxAttempts; attempt++ {
attemptResult, err := s.querySessionSummariesAttempt(ctx, stage, sqlText, args, attempt, sessionSummaryQueryMaxAttempts)
maxAttempts := positiveInt(s.config.SessionSummaryMaxAttempts, sessionSummaryQueryMaxAttempts)
for attempt := 1; attempt <= maxAttempts; attempt++ {
attemptResult, err := s.querySessionSummariesAttempt(ctx, stage, sqlText, args, attempt, maxAttempts)
if err == nil {
return attemptResult, nil
}
lastErr = err
if attempt == sessionSummaryQueryMaxAttempts || !retryableSessionSummaryQueryError(err) || ctx.Err() != nil {
if attempt == maxAttempts || !retryableSessionSummaryQueryError(err) || ctx.Err() != nil {
break
}
backoff := sessionSummaryRetryBackoff(attempt)
backoff := s.sessionSummaryRetryBackoff(attempt)
timer := time.NewTimer(backoff)
select {
case <-ctx.Done():
@@ -1465,7 +1469,7 @@ func (s *Server) querySessionSummaries(ctx context.Context, stage string, sqlTex
func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string, sqlText string, args []any, attempt int, maxAttempts int) ([]any, error) {
result := []any{}
attemptCtx, cancel := context.WithTimeout(ctx, sessionSummaryQueryAttemptTimeout)
attemptCtx, cancel := context.WithTimeout(ctx, positiveDuration(s.config.SessionSummaryAttemptTimeout, sessionSummaryQueryAttemptTimeout))
defer cancel()
attrs := s.dbQueryAttrs("workbench_sessions", len(args), map[string]any{
"db.index.expected": "idx_workbench_sessions_owner_updated|idx_workbench_sessions_updated",
@@ -1473,7 +1477,7 @@ func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string,
"retryMax": int64(maxAttempts),
})
if attempt > 1 {
attrs["retryBackoffMs"] = sessionSummaryRetryBackoff(attempt - 1).Milliseconds()
attrs["retryBackoffMs"] = s.sessionSummaryRetryBackoff(attempt - 1).Milliseconds()
}
err := s.withOtelInternalSpan(attemptCtx, stage, attrs, func(spanCtx context.Context) error {
rows, err := s.db.QueryContext(spanCtx, sqlText, args...)
@@ -1513,11 +1517,26 @@ func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string,
return result, err
}
func sessionSummaryRetryBackoff(attempt int) time.Duration {
func (s *Server) sessionSummaryRetryBackoff(attempt int) time.Duration {
base := positiveDuration(s.config.SessionSummaryRetryBackoff, sessionSummaryQueryRetryBackoff)
if attempt <= 1 {
return sessionSummaryQueryRetryBackoff
return base
}
return time.Duration(attempt) * sessionSummaryQueryRetryBackoff
return time.Duration(attempt) * base
}
func positiveDuration(value time.Duration, fallback time.Duration) time.Duration {
if value > 0 {
return value
}
return fallback
}
func positiveInt(value int, fallback int) int {
if value > 0 {
return value
}
return fallback
}
func retryableSessionSummaryQueryError(err error) bool {
@@ -2310,12 +2329,15 @@ func normalizeProjectionHealth(value string, projectionStatus string) string {
func configFromEnv() Config {
port := first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_PORT"), os.Getenv("PORT"), "6671")
return Config{
Addr: ":" + port,
DatabaseURL: first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_DB_URL"), os.Getenv("HWLAB_CLOUD_DB_URL")),
QueryTimeout: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_QUERY_TIMEOUT_MS", 10000)) * time.Millisecond,
ReadinessDelay: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_READY_TIMEOUT_MS", 2000)) * time.Millisecond,
PoolMax: envInt("HWLAB_WORKBENCH_RUNTIME_DB_POOL_MAX", 8),
Cache: cacheConfigFromEnv(),
Addr: ":" + port,
DatabaseURL: first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_DB_URL"), os.Getenv("HWLAB_CLOUD_DB_URL")),
QueryTimeout: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_QUERY_TIMEOUT_MS", 10000)) * time.Millisecond,
ReadinessDelay: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_READY_TIMEOUT_MS", 2000)) * time.Millisecond,
PoolMax: envInt("HWLAB_WORKBENCH_RUNTIME_DB_POOL_MAX", 8),
SessionSummaryMaxAttempts: envInt("HWLAB_WORKBENCH_SESSIONS_SUMMARY_QUERY_MAX_ATTEMPTS", sessionSummaryQueryMaxAttempts),
SessionSummaryAttemptTimeout: time.Duration(envInt("HWLAB_WORKBENCH_SESSIONS_SUMMARY_QUERY_ATTEMPT_TIMEOUT_MS", int(sessionSummaryQueryAttemptTimeout/time.Millisecond))) * time.Millisecond,
SessionSummaryRetryBackoff: time.Duration(envInt("HWLAB_WORKBENCH_SESSIONS_SUMMARY_QUERY_BACKOFF_MS", int(sessionSummaryQueryRetryBackoff/time.Millisecond))) * time.Millisecond,
Cache: cacheConfigFromEnv(),
}
}
scripts/gitops-render.mjs
+13
View File
@@ -941,6 +941,7 @@ function deployServicesForProfile(deploy, profile) {
runtimeStoreEnvForProfile(deploy, profile, service.serviceId),
workbenchRuntimeClientEnvForProfile(deploy, profile, service.serviceId),
workbenchRuntimeFactsQueryEnvForProfile(deploy, profile, service.serviceId),
workbenchRuntimeSessionsSummaryEnvForProfile(deploy, profile, service.serviceId),
workbenchRuntimePostgresEnvForProfile(deploy, profile, service.serviceId),
workbenchRuntimeRedisEnvForProfile(deploy, profile, service.serviceId)
);
@@ -957,6 +958,7 @@ function deployServicesForProfile(deploy, profile) {
runtimeStoreEnvForProfile(deploy, profile, override.serviceId),
workbenchRuntimeClientEnvForProfile(deploy, profile, override.serviceId),
workbenchRuntimeFactsQueryEnvForProfile(deploy, profile, override.serviceId),
workbenchRuntimeSessionsSummaryEnvForProfile(deploy, profile, override.serviceId),
workbenchRuntimePostgresEnvForProfile(deploy, profile, override.serviceId),
workbenchRuntimeRedisEnvForProfile(deploy, profile, override.serviceId)
)
@@ -1075,6 +1077,17 @@ function workbenchRuntimeFactsQueryEnvForProfile(deploy, profile, serviceId) {
return env;
}
function workbenchRuntimeSessionsSummaryEnvForProfile(deploy, profile, serviceId) {
if (!isRuntimeLane(profile) || serviceId !== "hwlab-workbench-runtime") return {};
const sessionsSummary = runtimeLaneConfig(deploy, profile)?.workbenchRuntime?.sessionsSummary;
if (!sessionsSummary || typeof sessionsSummary !== "object" || Array.isArray(sessionsSummary)) return {};
const env = {};
if (Number.isInteger(sessionsSummary.maxAttempts)) env.HWLAB_WORKBENCH_SESSIONS_SUMMARY_QUERY_MAX_ATTEMPTS = String(sessionsSummary.maxAttempts);
if (Number.isInteger(sessionsSummary.attemptTimeoutMs)) env.HWLAB_WORKBENCH_SESSIONS_SUMMARY_QUERY_ATTEMPT_TIMEOUT_MS = String(sessionsSummary.attemptTimeoutMs);
if (Number.isInteger(sessionsSummary.backoffMs)) env.HWLAB_WORKBENCH_SESSIONS_SUMMARY_QUERY_BACKOFF_MS = String(sessionsSummary.backoffMs);
return env;
}
function workbenchRuntimePostgresEnvForProfile(deploy, profile, serviceId) {
if (!isRuntimeLane(profile) || serviceId !== "hwlab-workbench-runtime") return {};
const postgres = runtimeLaneConfig(deploy, profile)?.workbenchRuntime?.postgres;