From 2354a218c9a510fd2e37b3cbef95665cf8988745 Mon Sep 17 00:00:00 2001 From: imakunin Date: Mon, 10 Aug 2026 12:00:39 +0300 Subject: [PATCH 1/3] feat(timeouts): configure task route timeouts --- README.md | 40 +++++ chart/templates/deployment.yaml | 3 + chart/values.yaml | 8 + .../2026-08-10-configurable-route-timeouts.md | 113 ++++++++++++++ server/main.go | 53 +++++-- server/pkg/discovery.go | 140 ++++++++++++------ server/pkg/discovery_test.go | 76 +++++++--- server/pkg/task.go | 19 ++- server/pkg/task_test.go | 9 ++ server/pkg/timeout.go | 77 ++++++++++ server/pkg/timeout_test.go | 44 ++++++ server/pkg/updater.go | 11 +- server/pkg/updater_test.go | 2 +- server/pkg/xds.go | 12 +- server/pkg/xds_test.go | 69 ++++++++- 15 files changed, 586 insertions(+), 90 deletions(-) create mode 100644 docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md create mode 100644 server/pkg/timeout.go create mode 100644 server/pkg/timeout_test.go diff --git a/README.md b/README.md index 77c4461..f4dd6b0 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,46 @@ For more information, refer to: - [Spark UI](https://ytsaurus.tech/docs/user-guide/data-processing/spyt/spark-ui) to learn how to open UI of [SPYT](https://ytsaurus.tech/docs/en/user-guide/data-processing/spyt/overview) clusters and jobs, - [Admin docs](https://ytsaurus.tech/docs/admin-guide/install-task-proxy) for installation instructions. +## Annotating an operation + +To publish services from a regular YTsaurus operation, add the `task_proxy` annotation to its specification. `enabled` is required; `tasks_info` describes services by task name, service name, protocol, and zero-based job port index. + +```yson +<"task_proxy"={ + "enabled"=%true; + "tasks_info"={ + "worker"={ + "api"={ + "protocol"="http"; + "port_index"=0; + }; + "grpc"={ + "protocol"="grpc"; + "port_index"=1; + }; + }; + }; +}> +``` + +`protocol` must be `http` or `grpc`. If `tasks_info` is omitted, task-proxy publishes every job port as an HTTP service named `port`. + +The annotation can also override request timeouts for every service in that operation: + +```yson +<"task_proxy"={ + "enabled"=%true; + "route_timeout_seconds"=600; + "stream_idle_timeout_seconds"=120; +}> +``` + +- `route_timeout_seconds` is the maximum time to receive a complete upstream response after Envoy has received the full request. +- `stream_idle_timeout_seconds` is the maximum period without request or response traffic. +- Both values are non-negative integer seconds. `0` explicitly disables the corresponding timeout; an omitted value inherits the global Helm setting. + +The Helm defaults are 2 seconds for connecting to a job, 15 seconds for a complete response, and 300 seconds for a stream idle period. Configure them through `timeouts.connectTimeoutSeconds`, `timeouts.routeTimeoutSeconds`, and `timeouts.streamIdleTimeoutSeconds`. + ## Development Install chart to cluster from local directory using: diff --git a/chart/templates/deployment.yaml b/chart/templates/deployment.yaml index 525a633..8b7ef28 100644 --- a/chart/templates/deployment.yaml +++ b/chart/templates/deployment.yaml @@ -54,6 +54,9 @@ spec: - "-base-domain={{ .Values.baseDomain }}" - "-dir-path={{ .Values.dirPath }}" - "-discovery-period-seconds={{ .Values.discoveryPeriodSeconds }}" + - "-connect-timeout-seconds={{ .Values.timeouts.connectTimeoutSeconds }}" + - "-route-timeout-seconds={{ .Values.timeouts.routeTimeoutSeconds }}" + - "-stream-idle-timeout-seconds={{ .Values.timeouts.streamIdleTimeoutSeconds }}" - "-auth-enabled={{ .Values.auth.enabled }}" - "-auth-cookie-name={{ .Values.auth.cookieName }}" - "-auth-cache-enabled={{ .Values.auth.cache.enabled }}" diff --git a/chart/values.yaml b/chart/values.yaml index 895f780..a0092b3 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -12,6 +12,14 @@ dirPath: //sys/task_proxies discoveryPeriodSeconds: 60 +timeouts: + # Maximum time to establish a TCP connection to a task job. + connectTimeoutSeconds: 2 + # Maximum time to wait for a complete upstream response. Set to 0 to disable. + routeTimeoutSeconds: 15 + # Maximum period without request or response traffic. Set to 0 to disable. + streamIdleTimeoutSeconds: 300 + auth: enabled: true cookieName: YTCypressCookie diff --git a/docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md b/docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md new file mode 100644 index 0000000..265c776 --- /dev/null +++ b/docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md @@ -0,0 +1,113 @@ +# Configurable Route Timeouts Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make task-proxy's upstream connection, response, and stream-idle timeouts configurable globally, with route and stream-idle overrides in a task-proxy operation annotation. + +**Architecture:** `main` owns validated global defaults and passes a `TaskProxyTimeoutConfig` to the updater and xDS builder. Discovery stores optional operation-level overrides on `Task`; the xDS builder resolves each task's effective timeout values when constructing every equivalent route action. + +**Tech Stack:** Go, Envoy xDS v3 protobufs, Helm, testify. + +## Global Constraints + +- Defaults preserve existing Envoy behavior: 2s connect, 15s route, 5m stream-idle. +- A zero route or stream-idle timeout explicitly disables that limit; connect timeout must be positive. +- New Go configuration fields require comments explaining behavior and zero-value semantics. +- Timeout overrides must influence snapshot versioning. + +--- + +### Task 1: Timeout model and annotation parsing + +**Files:** +- Modify: `server/pkg/task.go` +- Modify: `server/pkg/discovery.go` +- Test: `server/pkg/discovery_test.go` + +**Interfaces:** +- Produces `TaskProxyTimeoutConfig` and optional timeout fields on `Task`. +- Produces `parseTaskProxyAnnotation(any) ([]taskServiceInfo, *TaskTimeoutOverrides)`. + +- [ ] **Step 1: Write failing annotation tests** + +Add a valid `task_proxy` annotation with `route_timeout_seconds: 600` and `stream_idle_timeout_seconds: 120`, and assert the returned operation overrides preserve both durations. Add invalid negative and non-integer values and assert parsing rejects the annotation. + +- [ ] **Step 2: Run the focused test and verify it fails** + +Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestParseTaskProxyAnnotation` + +Expected: FAIL because no override values are returned or validated. + +- [ ] **Step 3: Add the minimal model and parser changes** + +Define documented `TaskProxyTimeoutConfig` defaults and optional per-operation `TaskTimeoutOverrides`. Parse the two optional annotation fields after `enabled`, reject malformed values, and copy the resulting overrides to every discovered task. + +- [ ] **Step 4: Run the focused test and verify it passes** + +Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestParseTaskProxyAnnotation` + +Expected: PASS. + +### Task 2: xDS effective timeouts + +**Files:** +- Modify: `server/pkg/xds.go` +- Modify: `server/pkg/updater.go` +- Test: `server/pkg/xds_test.go` +- Test: `server/pkg/updater_test.go` + +**Interfaces:** +- Consumes `TaskProxyTimeoutConfig`, `TaskTimeoutOverrides`. +- Produces xDS `RouteAction.Timeout`, `RouteAction.IdleTimeout`, and task upstream `Cluster.ConnectTimeout`. + +- [ ] **Step 1: Write failing xDS tests** + +Build a snapshot with defaults and a task override. Assert task clusters use the configured connect timeout, all routes for that task share the expected route timeout and idle timeout, and a zero override serializes as an explicit zero duration. + +- [ ] **Step 2: Run the focused test and verify it fails** + +Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestMakeSnapshotTimeouts` + +Expected: FAIL because `makeSnapshot` has no timeout configuration parameter and route actions omit both fields. + +- [ ] **Step 3: Add xDS resolution and updater plumbing** + +Pass the global config from updater to `makeSnapshot`; resolve per-task overrides while building the shared route action. Apply global connect timeout only to discovered task clusters, preserving the authz cluster's internal timeout. Include timeout overrides in task snapshot identity. + +- [ ] **Step 4: Run focused xDS and updater tests** + +Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run 'TestMakeSnapshotTimeouts|TestTaskUpdater'` + +Expected: PASS. + +### Task 3: Runtime and Helm configuration + +**Files:** +- Modify: `server/main.go` +- Modify: `chart/values.yaml` +- Modify: `chart/templates/deployment.yaml` +- Test: `server/pkg/xds_test.go` + +**Interfaces:** +- Consumes Helm `timeouts.connectTimeoutSeconds`, `timeouts.routeTimeoutSeconds`, `timeouts.streamIdleTimeoutSeconds`. +- Produces validated `TaskProxyTimeoutConfig` passed to `CreateTaskUpdater`. + +- [ ] **Step 1: Add a failing configuration validation test where feasible** + +Exercise the public timeout-validation helper with a zero connect timeout and negative route or stream-idle timeout, expecting errors; verify zero route and stream-idle are accepted. + +- [ ] **Step 2: Run the focused validation test and verify it fails** + +Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestValidateTaskProxyTimeoutConfig` + +Expected: FAIL because the validation helper does not exist. + +- [ ] **Step 3: Implement flags, validation, and chart values** + +Add documented timeout fields and validation helper in Go. Add command-line flags and Helm arguments with compatible defaults. Avoid exposing per-operation configuration in Helm because it belongs to the operation annotation. + +- [ ] **Step 4: Run complete verification** + +Run: `make test && helm template task-proxy ./chart >/tmp/task-proxy-rendered.yaml` + +Expected: all Go tests pass and Helm renders successfully. diff --git a/server/main.go b/server/main.go index a4aa5e0..aa45a9f 100644 --- a/server/main.go +++ b/server/main.go @@ -19,20 +19,24 @@ import ( func main() { ctx := context.Background() + defaultTimeoutConfig := pkg.DefaultTaskProxyTimeoutConfig() var args struct { - ytProxy string - ytTokenPath string - baseDomain string - dirPath string - discoveryPeriodSeconds uint - authEnabled bool - authCookieName string - authCacheEnabled bool - authCacheTTLSeconds int - authCacheCapacity int - authCacheMaxConcurrency int - authCacheRefreshBefore int + ytProxy string + ytTokenPath string + baseDomain string + dirPath string + discoveryPeriodSeconds uint + authEnabled bool + authCookieName string + authCacheEnabled bool + authCacheTTLSeconds int + authCacheCapacity int + authCacheMaxConcurrency int + authCacheRefreshBefore int + connectTimeoutSeconds int + routeTimeoutSeconds int + streamIdleTimeoutSeconds int } flag.StringVar(&args.ytProxy, "yt-proxy", "", "YT proxy host") flag.StringVar(&args.ytTokenPath, "yt-token-path", "", "YT token path") @@ -46,6 +50,9 @@ func main() { flag.IntVar(&args.authCacheCapacity, "auth-cache-capacity", 0, "auth cache maximum number of entries (0 means unlimited)") flag.IntVar(&args.authCacheMaxConcurrency, "auth-cache-max-concurrent-backend-requests", 0, "auth cache max concurrent backend requests per key on misses (0 means unlimited)") flag.IntVar(&args.authCacheRefreshBefore, "auth-cache-refresh-before-seconds", 0, "auth cache proactive refresh threshold in seconds before TTL deadline (0 disables proactive refresh)") + flag.IntVar(&args.connectTimeoutSeconds, "connect-timeout-seconds", int(defaultTimeoutConfig.ConnectTimeout/time.Second), "maximum time in seconds to establish an upstream job connection") + flag.IntVar(&args.routeTimeoutSeconds, "route-timeout-seconds", int(defaultTimeoutConfig.RouteTimeout/time.Second), "maximum time in seconds to wait for a complete upstream response (0 disables the timeout)") + flag.IntVar(&args.streamIdleTimeoutSeconds, "stream-idle-timeout-seconds", int(defaultTimeoutConfig.StreamIdleTimeout/time.Second), "maximum idle time in seconds for an upstream request or response stream (0 disables the timeout)") flag.Parse() if args.ytProxy == "" { @@ -75,6 +82,26 @@ func main() { if args.authCacheRefreshBefore < 0 { log.Fatal("'auth-cache-refresh-before-seconds' argument must be non-negative") } + connectTimeout, err := pkg.DurationFromSeconds(args.connectTimeoutSeconds) + if err != nil { + log.Fatalf("invalid connect timeout: %v", err) + } + routeTimeout, err := pkg.DurationFromSeconds(args.routeTimeoutSeconds) + if err != nil { + log.Fatalf("invalid route timeout: %v", err) + } + streamIdleTimeout, err := pkg.DurationFromSeconds(args.streamIdleTimeoutSeconds) + if err != nil { + log.Fatalf("invalid stream idle timeout: %v", err) + } + timeoutConfig := pkg.TaskProxyTimeoutConfig{ + ConnectTimeout: connectTimeout, + RouteTimeout: routeTimeout, + StreamIdleTimeout: streamIdleTimeout, + } + if err := timeoutConfig.Validate(); err != nil { + log.Fatalf("invalid task proxy timeout configuration: %v", err) + } ytTokenBytes, err := os.ReadFile(args.ytTokenPath) if err != nil { @@ -109,7 +136,7 @@ func main() { RefreshBeforeSeconds: args.authCacheRefreshBefore, }) - taskUpdater := pkg.CreateTaskUpdater(args.baseDomain, tls, args.authEnabled, authServer, taskDiscovery, cache) + taskUpdater := pkg.CreateTaskUpdater(args.baseDomain, tls, args.authEnabled, timeoutConfig, authServer, taskDiscovery, cache) go func() { if err := pkg.ServeMetrics(pkg.DefaultGatherer()); err != nil { diff --git a/server/pkg/discovery.go b/server/pkg/discovery.go index f0968a7..4a5dff5 100644 --- a/server/pkg/discovery.go +++ b/server/pkg/discovery.go @@ -3,6 +3,7 @@ package pkg import ( "context" "fmt" + "math" "net" "net/url" "strconv" @@ -14,7 +15,17 @@ import ( ytsdk "go.ytsaurus.tech/yt/go/yt" ) -const servicesTableName = "services" +const ( + servicesTableName = "services" + + taskProxyAnnotationKey = "task_proxy" + taskProxyEnabledKey = "enabled" + taskProxyTasksInfoKey = "tasks_info" + taskProxyProtocolKey = "protocol" + taskProxyPortIndexKey = "port_index" + taskProxyRouteTimeoutSecondsKey = "route_timeout_seconds" + taskProxyStreamIdleTimeoutKey = "stream_idle_timeout_seconds" +) type taskDiscovery struct { baseDomain string @@ -63,7 +74,7 @@ func (d *taskDiscovery) Discovery(ctx context.Context) (TaskList, error) { d.logger.Errorf("unable to process SPYT standalone cluster operation %q: %v", op.ID, err) continue } - } else if _, ok := annotations["task_proxy"]; ok { + } else if _, ok := annotations[taskProxyAnnotationKey]; ok { opTasks, err = d.processTaskProxyAnnotatedOperation(ctx, op) if err != nil { d.logger.Errorf("unable to process task proxy annotated operation %q: %v", op.ID, err) @@ -195,8 +206,8 @@ func (d *taskDiscovery) processSPYTStandaloneClusterOperation(ctx context.Contex } func (d *taskDiscovery) processTaskProxyAnnotatedOperation(ctx context.Context, op ytsdk.OperationStatus) ([]Task, error) { - taskProxyAnnotation := op.RuntimeParameters.Annotations["task_proxy"] - taskServiceInfos := parseTaskProxyAnnotation(taskProxyAnnotation) + taskProxyAnnotation := op.RuntimeParameters.Annotations[taskProxyAnnotationKey] + taskServiceInfos, timeoutOverrides := parseTaskProxyAnnotation(taskProxyAnnotation) if taskServiceInfos == nil { return nil, fmt.Errorf("invalid task_proxy annotation: %v", taskProxyAnnotation) } @@ -250,11 +261,12 @@ func (d *taskDiscovery) processTaskProxyAnnotatedOperation(ctx context.Context, hostParts := strings.Split(job.Address, ":") // job address contains port also taskProto := Task{ - operationID: op.ID.String(), - operationAlias: parseOperationAlias(op), - taskName: job.TaskName, - service: serviceInfo.service, - protocol: serviceInfo.protocol, + operationID: op.ID.String(), + operationAlias: parseOperationAlias(op), + taskName: job.TaskName, + service: serviceInfo.service, + protocol: serviceInfo.protocol, + timeoutOverrides: timeoutOverrides, } if _, ok := idToTask[taskProto.ID()]; !ok { idToTask[taskProto.ID()] = &taskProto @@ -371,32 +383,36 @@ type taskServiceInfo struct { portIndex int } -func parseTaskProxyAnnotation(taskProxyAny any) []taskServiceInfo { +func parseTaskProxyAnnotation(taskProxyAny any) ([]taskServiceInfo, TaskTimeoutOverrides) { taskProxy, ok := taskProxyAny.(map[string]any) if !ok { - return nil + return nil, TaskTimeoutOverrides{} } - enabledAny, ok := taskProxy["enabled"] + enabledAny, ok := taskProxy[taskProxyEnabledKey] if !ok { - return nil + return nil, TaskTimeoutOverrides{} } enabled, ok := enabledAny.(bool) if !ok { - return nil + return nil, TaskTimeoutOverrides{} } if !enabled { - return nil + return nil, TaskTimeoutOverrides{} + } + timeoutOverrides, ok := parseTaskTimeoutOverrides(taskProxy) + if !ok { + return nil, TaskTimeoutOverrides{} } taskServiceInfos := make([]taskServiceInfo, 0) - tasksInfoAny, ok := taskProxy["tasks_info"] + tasksInfoAny, ok := taskProxy[taskProxyTasksInfoKey] if !ok { - return taskServiceInfos + return taskServiceInfos, timeoutOverrides } tasksInfo, ok := tasksInfoAny.(map[string]any) if !ok { - return taskServiceInfos + return taskServiceInfos, timeoutOverrides } for task, infoAny := range tasksInfo { @@ -409,7 +425,7 @@ func parseTaskProxyAnnotation(taskProxyAny any) []taskServiceInfo { if !ok { continue } - protocolAny, ok := info["protocol"] + protocolAny, ok := info[taskProxyProtocolKey] if !ok { continue } @@ -420,31 +436,16 @@ func parseTaskProxyAnnotation(taskProxyAny any) []taskServiceInfo { if protocol != string(HTTP) && protocol != string(GRPC) { continue } - portIndexAny, ok := info["port_index"] + portIndexAny, ok := info[taskProxyPortIndexKey] if !ok { continue } - var portIndex int - switch v := portIndexAny.(type) { - case int: - portIndex = v - case int64: - portIndex = int(v) - case int32: - portIndex = int(v) - case int16: - portIndex = int(v) - case int8: - portIndex = int(v) - case uint64: - portIndex = int(v) - case uint32: - portIndex = int(v) - case uint16: - portIndex = int(v) - case uint8: - portIndex = int(v) - default: + portIndex64, ok := parseInteger(portIndexAny) + if !ok { + continue + } + portIndex := int(portIndex64) + if int64(portIndex) != portIndex64 { continue } taskServiceInfos = append(taskServiceInfos, taskServiceInfo{ @@ -456,7 +457,62 @@ func parseTaskProxyAnnotation(taskProxyAny any) []taskServiceInfo { } } - return taskServiceInfos + return taskServiceInfos, timeoutOverrides +} + +func parseTaskTimeoutOverrides(taskProxy map[string]any) (TaskTimeoutOverrides, bool) { + var overrides TaskTimeoutOverrides + if value, ok := taskProxy[taskProxyRouteTimeoutSecondsKey]; ok { + timeout, ok := parseTimeoutSeconds(value) + if !ok { + return TaskTimeoutOverrides{}, false + } + overrides.routeTimeout = &timeout + } + if value, ok := taskProxy[taskProxyStreamIdleTimeoutKey]; ok { + timeout, ok := parseTimeoutSeconds(value) + if !ok { + return TaskTimeoutOverrides{}, false + } + overrides.streamIdleTimeout = &timeout + } + return overrides, true +} + +func parseTimeoutSeconds(value any) (time.Duration, bool) { + seconds, ok := parseInteger(value) + if !ok || seconds < 0 || seconds > int64(math.MaxInt64/time.Second) { + return 0, false + } + return time.Duration(seconds) * time.Second, true +} + +func parseInteger(value any) (int64, bool) { + switch v := value.(type) { + case int: + return int64(v), true + case int64: + return v, true + case int32: + return int64(v), true + case int16: + return int64(v), true + case int8: + return int64(v), true + case uint64: + if v > uint64(math.MaxInt64) { + return 0, false + } + return int64(v), true + case uint32: + return int64(v), true + case uint16: + return int64(v), true + case uint8: + return int64(v), true + default: + return 0, false + } } func makeHostPortFromNode(node string) (*HostPort, error) { diff --git a/server/pkg/discovery_test.go b/server/pkg/discovery_test.go index 1ca4402..9081ae5 100644 --- a/server/pkg/discovery_test.go +++ b/server/pkg/discovery_test.go @@ -1,9 +1,12 @@ package pkg import ( + "math" "testing" + "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestParseTaskProxyAnnotation(t *testing.T) { @@ -15,12 +18,12 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "full annotation", annotation: map[string]any{ - "enabled": true, - "tasks_info": map[string]any{ + taskProxyEnabledKey: true, + taskProxyTasksInfoKey: map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - "protocol": "grpc", - "port_index": 0, + taskProxyProtocolKey: "grpc", + taskProxyPortIndexKey: 0, }, }, }, @@ -37,14 +40,14 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "minimal annotation", annotation: map[string]any{ - "enabled": true, + taskProxyEnabledKey: true, }, expected: []taskServiceInfo{}, }, { name: "disabled annotation (false)", annotation: map[string]any{ - "enabled": false, + taskProxyEnabledKey: false, }, expected: nil, }, @@ -61,12 +64,12 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "unknown protocol", annotation: map[string]any{ - "enabled": true, - "tasks_info": map[string]any{ + taskProxyEnabledKey: true, + taskProxyTasksInfoKey: map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - "protocol": "dns", - "port_index": 0, + taskProxyProtocolKey: "dns", + taskProxyPortIndexKey: 0, }, }, }, @@ -76,12 +79,12 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "invalid port type", annotation: map[string]any{ - "enabled": true, - "tasks_info": map[string]any{ + taskProxyEnabledKey: true, + taskProxyTasksInfoKey: map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - "protocol": "http", - "port_index": "0", + taskProxyProtocolKey: "http", + taskProxyPortIndexKey: "0", }, }, }, @@ -91,11 +94,11 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "missing service attributes", annotation: map[string]any{ - "enabled": true, - "tasks_info": map[string]any{ + taskProxyEnabledKey: true, + taskProxyTasksInfoKey: map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - "protocol": "http", + taskProxyProtocolKey: "http", }, }, }, @@ -104,8 +107,45 @@ func TestParseTaskProxyAnnotation(t *testing.T) { }, } { t.Run(tt.name, func(t *testing.T) { - taskServiceInfos := parseTaskProxyAnnotation(tt.annotation) + taskServiceInfos, _ := parseTaskProxyAnnotation(tt.annotation) assert.Equal(t, tt.expected, taskServiceInfos) }) } } + +func TestParseInteger(t *testing.T) { + for _, value := range []any{int(1), int64(1), int32(1), int16(1), int8(1), uint64(1), uint32(1), uint16(1), uint8(1)} { + parsed, ok := parseInteger(value) + require.True(t, ok) + require.EqualValues(t, 1, parsed) + } + + _, ok := parseInteger(uint64(math.MaxInt64) + 1) + require.False(t, ok) + _, ok = parseInteger("1") + require.False(t, ok) +} + +func TestParseTaskProxyAnnotationTimeoutOverrides(t *testing.T) { + annotation := map[string]any{ + taskProxyEnabledKey: true, + taskProxyRouteTimeoutSecondsKey: 600, + taskProxyStreamIdleTimeoutKey: 120, + } + + _, overrides := parseTaskProxyAnnotation(annotation) + + require.Equal(t, durationPtr(10*time.Minute), overrides.routeTimeout) + require.Equal(t, durationPtr(2*time.Minute), overrides.streamIdleTimeout) +} + +func TestParseTaskProxyAnnotationRejectsInvalidTimeoutOverrides(t *testing.T) { + annotation := map[string]any{ + taskProxyEnabledKey: true, + taskProxyRouteTimeoutSecondsKey: -1, + } + + services, _ := parseTaskProxyAnnotation(annotation) + + assert.Nil(t, services) +} diff --git a/server/pkg/task.go b/server/pkg/task.go index 929c839..5ce4411 100644 --- a/server/pkg/task.go +++ b/server/pkg/task.go @@ -20,12 +20,13 @@ type HostPort struct { } type Task struct { - operationID string - operationAlias string - taskName string - service string - protocol Protocol - jobs []HostPort + operationID string + operationAlias string + taskName string + service string + protocol Protocol + jobs []HostPort + timeoutOverrides TaskTimeoutOverrides } var valueRegexp = regexp.MustCompile(`^[a-z0-9_]{1,30}$`) @@ -55,6 +56,12 @@ func (t *Task) IDWithHostPort() string { sb.WriteString(job.host) fmt.Fprintf(&sb, "%d", job.port) } + if t.timeoutOverrides.routeTimeout != nil { + fmt.Fprintf(&sb, "route-timeout=%d", *t.timeoutOverrides.routeTimeout) + } + if t.timeoutOverrides.streamIdleTimeout != nil { + fmt.Fprintf(&sb, "stream-idle-timeout=%d", *t.timeoutOverrides.streamIdleTimeout) + } return sb.String() } diff --git a/server/pkg/task_test.go b/server/pkg/task_test.go index aae4ce8..79b440f 100644 --- a/server/pkg/task_test.go +++ b/server/pkg/task_test.go @@ -3,6 +3,7 @@ package pkg import ( "errors" "testing" + "time" "github.com/stretchr/testify/assert" ) @@ -76,3 +77,11 @@ func TestValidateTask(t *testing.T) { }) } } + +func TestTaskIDWithHostPortIncludesTimeoutOverrides(t *testing.T) { + withoutOverrides := Task{operationID: "op", taskName: "task", service: "service"} + withOverride := withoutOverrides + withOverride.timeoutOverrides.routeTimeout = durationPtr(10 * time.Minute) + + assert.NotEqual(t, withoutOverrides.IDWithHostPort(), withOverride.IDWithHostPort()) +} diff --git a/server/pkg/timeout.go b/server/pkg/timeout.go new file mode 100644 index 0000000..4c62d6e --- /dev/null +++ b/server/pkg/timeout.go @@ -0,0 +1,77 @@ +package pkg + +import ( + "fmt" + "math" + "time" +) + +const ( + defaultConnectTimeout = 2 * time.Second + defaultRouteTimeout = 15 * time.Second + defaultStreamIdleTimeout = 5 * time.Minute +) + +// TaskProxyTimeoutConfig controls timeout behavior for every task-proxy route. +type TaskProxyTimeoutConfig struct { + // ConnectTimeout limits how long Envoy waits to establish a TCP connection to a job. + // It must be positive because Envoy requires a connect timeout for every cluster. + ConnectTimeout time.Duration + // RouteTimeout limits the total time Envoy waits for an upstream response after it receives the full request. + // Zero disables this timeout, which is useful for long-lived streaming responses. + RouteTimeout time.Duration + // StreamIdleTimeout limits a request or response stream with no upstream or downstream traffic. + // Zero disables this timeout; active traffic keeps the stream alive regardless of this value. + StreamIdleTimeout time.Duration +} + +func DefaultTaskProxyTimeoutConfig() TaskProxyTimeoutConfig { + return TaskProxyTimeoutConfig{ + ConnectTimeout: defaultConnectTimeout, + RouteTimeout: defaultRouteTimeout, + StreamIdleTimeout: defaultStreamIdleTimeout, + } +} + +func (c TaskProxyTimeoutConfig) Validate() error { + if c.ConnectTimeout <= 0 { + return fmt.Errorf("connect timeout must be positive") + } + if c.RouteTimeout < 0 { + return fmt.Errorf("route timeout must be non-negative") + } + if c.StreamIdleTimeout < 0 { + return fmt.Errorf("stream idle timeout must be non-negative") + } + return nil +} + +// DurationFromSeconds converts a non-negative whole-second setting without allowing time.Duration overflow. +func DurationFromSeconds(seconds int) (time.Duration, error) { + if seconds < 0 { + return 0, fmt.Errorf("timeout seconds must be non-negative") + } + if uint64(seconds) > uint64(math.MaxInt64/int64(time.Second)) { + return 0, fmt.Errorf("timeout seconds value %d is too large", seconds) + } + return time.Duration(seconds) * time.Second, nil +} + +type TaskTimeoutOverrides struct { + routeTimeout *time.Duration + streamIdleTimeout *time.Duration +} + +func (o TaskTimeoutOverrides) routeTimeoutOr(defaultValue time.Duration) time.Duration { + if o.routeTimeout != nil { + return *o.routeTimeout + } + return defaultValue +} + +func (o TaskTimeoutOverrides) streamIdleTimeoutOr(defaultValue time.Duration) time.Duration { + if o.streamIdleTimeout != nil { + return *o.streamIdleTimeout + } + return defaultValue +} diff --git a/server/pkg/timeout_test.go b/server/pkg/timeout_test.go new file mode 100644 index 0000000..7d78674 --- /dev/null +++ b/server/pkg/timeout_test.go @@ -0,0 +1,44 @@ +package pkg + +import ( + "math" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func durationPtr(value time.Duration) *time.Duration { return &value } + +func TestValidateTaskProxyTimeoutConfig(t *testing.T) { + valid := TaskProxyTimeoutConfig{ + ConnectTimeout: 2 * time.Second, + RouteTimeout: 0, + StreamIdleTimeout: 0, + } + require.NoError(t, valid.Validate()) + + invalidConnect := valid + invalidConnect.ConnectTimeout = 0 + require.ErrorContains(t, invalidConnect.Validate(), "connect timeout must be positive") + + invalidRoute := valid + invalidRoute.RouteTimeout = -time.Second + require.ErrorContains(t, invalidRoute.Validate(), "route timeout must be non-negative") + + invalidIdle := valid + invalidIdle.StreamIdleTimeout = -time.Second + require.ErrorContains(t, invalidIdle.Validate(), "stream idle timeout must be non-negative") +} + +func TestDurationFromSeconds(t *testing.T) { + duration, err := DurationFromSeconds(300) + require.NoError(t, err) + require.Equal(t, 5*time.Minute, duration) + + _, err = DurationFromSeconds(-1) + require.ErrorContains(t, err, "must be non-negative") + + _, err = DurationFromSeconds(int(math.MaxInt64/time.Second) + 1) + require.ErrorContains(t, err, "is too large") +} diff --git a/server/pkg/updater.go b/server/pkg/updater.go index f5aa79c..11019aa 100644 --- a/server/pkg/updater.go +++ b/server/pkg/updater.go @@ -12,9 +12,10 @@ type snapshotSetter interface { } type taskUpdater struct { - baseDomain string - tls bool - authEnabled bool + baseDomain string + tls bool + authEnabled bool + timeoutConfig TaskProxyTimeoutConfig authServer *authServer taskDiscovery *taskDiscovery @@ -25,6 +26,7 @@ func CreateTaskUpdater( baseDomain string, tls bool, authEnabled bool, + timeoutConfig TaskProxyTimeoutConfig, authServer *authServer, taskDiscovery *taskDiscovery, cache snapshotSetter, @@ -33,6 +35,7 @@ func CreateTaskUpdater( baseDomain: baseDomain, tls: tls, authEnabled: authEnabled, + timeoutConfig: timeoutConfig, authServer: authServer, taskDiscovery: taskDiscovery, cache: cache, @@ -45,7 +48,7 @@ func (u *taskUpdater) Update( operationAliasToID map[string]string, version string, ) error { - snapshot, err := makeSnapshot(hashToTask, version, u.baseDomain, u.tls, u.authEnabled) + snapshot, err := makeSnapshot(hashToTask, version, u.baseDomain, u.tls, u.authEnabled, u.timeoutConfig) if err != nil { return fmt.Errorf("failed to make snapshot: %v", err) } diff --git a/server/pkg/updater_test.go b/server/pkg/updater_test.go index 67f7c83..b588006 100644 --- a/server/pkg/updater_test.go +++ b/server/pkg/updater_test.go @@ -34,7 +34,7 @@ func TestUpdateDoesNotChangeAuthDataIfSetSnapshotFails(t *testing.T) { ) cache := &failingSnapshotSetter{err: errors.New("set snapshot failed")} - updater := CreateTaskUpdater("example.com", false, true, authServer, &taskDiscovery{}, cache) + updater := CreateTaskUpdater("example.com", false, true, DefaultTaskProxyTimeoutConfig(), authServer, &taskDiscovery{}, cache) newTask := Task{ operationID: "op-new", diff --git a/server/pkg/xds.go b/server/pkg/xds.go index 0b5176f..026956f 100644 --- a/server/pkg/xds.go +++ b/server/pkg/xds.go @@ -65,7 +65,7 @@ func ServeGRPC(s serverv3.Server, authServer *authServer) error { return gs.Serve(lis) } -func makeSnapshot(hashToTask map[string]Task, version string, baseDomain string, tls bool, authEnabled bool) (*cachev3.Snapshot, error) { +func makeSnapshot(hashToTask map[string]Task, version string, baseDomain string, tls bool, authEnabled bool, timeoutConfig TaskProxyTimeoutConfig) (*cachev3.Snapshot, error) { var clusters []cachetypes.Resource var vhosts []*routev3.VirtualHost @@ -78,7 +78,7 @@ func makeSnapshot(hashToTask map[string]Task, version string, baseDomain string, var vhostClusters []*routev3.WeightedCluster_ClusterWeight for i, job := range task.jobs { clusterName := fmt.Sprintf("%s-%d", vhostName, i) - clusters = append(clusters, makeCluster(clusterName, job.host, job.port, grpc, true)) + clusters = append(clusters, makeCluster(clusterName, job.host, job.port, grpc, true, timeoutConfig.ConnectTimeout)) vhostClusters = append(vhostClusters, &routev3.WeightedCluster_ClusterWeight{ Name: clusterName, Weight: &wrapperspb.UInt32Value{Value: 1}, @@ -86,6 +86,8 @@ func makeSnapshot(hashToTask map[string]Task, version string, baseDomain string, } action := &routev3.Route_Route{ Route: &routev3.RouteAction{ + Timeout: durationpb.New(task.timeoutOverrides.routeTimeoutOr(timeoutConfig.RouteTimeout)), + IdleTimeout: durationpb.New(task.timeoutOverrides.streamIdleTimeoutOr(timeoutConfig.StreamIdleTimeout)), ClusterSpecifier: &routev3.RouteAction_WeightedClusters{ WeightedClusters: &routev3.WeightedCluster{ Clusters: vhostClusters, @@ -158,7 +160,7 @@ func makeSnapshot(hashToTask map[string]Task, version string, baseDomain string, }) if authEnabled { - authzCluster := makeCluster(extAuthClusterName, "127.0.0.1", serverPort, true, false) + authzCluster := makeCluster(extAuthClusterName, "127.0.0.1", serverPort, true, false, defaultConnectTimeout) clusters = append(clusters, authzCluster) } @@ -281,7 +283,7 @@ func makeSnapshot(hashToTask map[string]Task, version string, baseDomain string, return snap, snap.Consistent() } -func makeCluster(name string, host string, port uint32, grpc bool, resolveDomain bool) *clusterv3.Cluster { +func makeCluster(name string, host string, port uint32, grpc bool, resolveDomain bool, connectTimeout time.Duration) *clusterv3.Cluster { discoveryType := clusterv3.Cluster_STATIC if resolveDomain { discoveryType = clusterv3.Cluster_STRICT_DNS @@ -289,7 +291,7 @@ func makeCluster(name string, host string, port uint32, grpc bool, resolveDomain cluster := clusterv3.Cluster{ Name: name, - ConnectTimeout: durationpb.New(2 * time.Second), + ConnectTimeout: durationpb.New(connectTimeout), ClusterDiscoveryType: &clusterv3.Cluster_Type{Type: discoveryType}, LbPolicy: clusterv3.Cluster_ROUND_ROBIN, LoadAssignment: &endpointv3.ClusterLoadAssignment{ diff --git a/server/pkg/xds_test.go b/server/pkg/xds_test.go index 2ecb9d5..eb90f3c 100644 --- a/server/pkg/xds_test.go +++ b/server/pkg/xds_test.go @@ -2,8 +2,14 @@ package pkg import ( "sort" + "strings" "testing" + "time" + clusterv3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3" + listenerv3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" + hcmv3 "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3" + cachetypes "github.com/envoyproxy/go-control-plane/pkg/cache/types" resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -26,7 +32,7 @@ func TestMakeSnapshot(t *testing.T) { }, } - snapshot, err := makeSnapshot(hashToTask, "v1", "example.com", false, true) + snapshot, err := makeSnapshot(hashToTask, "v1", "example.com", false, true, DefaultTaskProxyTimeoutConfig()) require.NoError(t, err) // Convert snapshot to a structured map for YAML comparison @@ -231,5 +237,66 @@ listener: name: listener_0 ` + expectedYAML = strings.ReplaceAll( + expectedYAML, + "route:\n weightedClusters:", + "route:\n idleTimeout: 300s\n timeout: 15s\n weightedClusters:", + ) assert.YAMLEq(t, expectedYAML, string(resultYAML)) } + +func TestMakeSnapshotTimeouts(t *testing.T) { + zero := time.Duration(0) + task := Task{ + operationID: "op123", + taskName: "worker", + service: "api", + protocol: HTTP, + jobs: []HostPort{{host: "10.0.0.1", port: 8080}}, + timeoutOverrides: TaskTimeoutOverrides{ + routeTimeout: durationPtr(10 * time.Minute), + streamIdleTimeout: &zero, + }, + } + config := TaskProxyTimeoutConfig{ + ConnectTimeout: 3 * time.Second, + RouteTimeout: 15 * time.Second, + StreamIdleTimeout: 5 * time.Minute, + } + + snapshot, err := makeSnapshot(map[string]Task{"abc12345": task}, "v1", "example.com", false, false, config) + require.NoError(t, err) + + cluster := snapshot.GetResources(resourcev3.ClusterType)["op123-worker-api-0"].(*clusterv3.Cluster) + require.Equal(t, 3*time.Second, cluster.ConnectTimeout.AsDuration()) + + listener := onlyListener(t, snapshot.GetResources(resourcev3.ListenerType)) + hcm := httpConnectionManager(t, listener) + for _, vhost := range hcm.GetRouteConfig().GetVirtualHosts() { + for _, route := range vhost.GetRoutes() { + action := route.GetRoute() + if action == nil || action.GetWeightedClusters() == nil { + continue + } + require.Equal(t, 10*time.Minute, action.GetTimeout().AsDuration()) + require.Equal(t, time.Duration(0), action.GetIdleTimeout().AsDuration()) + } + } +} + +func onlyListener(t *testing.T, resources map[string]cachetypes.Resource) *listenerv3.Listener { + t.Helper() + require.Len(t, resources, 1) + for _, resource := range resources { + return resource.(*listenerv3.Listener) + } + return nil +} + +func httpConnectionManager(t *testing.T, listener *listenerv3.Listener) *hcmv3.HttpConnectionManager { + t.Helper() + var hcm hcmv3.HttpConnectionManager + err := listener.GetFilterChains()[0].GetFilters()[0].GetTypedConfig().UnmarshalTo(&hcm) + require.NoError(t, err) + return &hcm +} From f7c89fed5263fbb4ca99feced5fd93502852e4a1 Mon Sep 17 00:00:00 2001 From: imakunin Date: Mon, 10 Aug 2026 12:02:35 +0300 Subject: [PATCH 2/3] chore: remove implementation plan --- .../2026-08-10-configurable-route-timeouts.md | 113 ------------------ 1 file changed, 113 deletions(-) delete mode 100644 docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md diff --git a/docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md b/docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md deleted file mode 100644 index 265c776..0000000 --- a/docs/superpowers/plans/2026-08-10-configurable-route-timeouts.md +++ /dev/null @@ -1,113 +0,0 @@ -# Configurable Route Timeouts Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Make task-proxy's upstream connection, response, and stream-idle timeouts configurable globally, with route and stream-idle overrides in a task-proxy operation annotation. - -**Architecture:** `main` owns validated global defaults and passes a `TaskProxyTimeoutConfig` to the updater and xDS builder. Discovery stores optional operation-level overrides on `Task`; the xDS builder resolves each task's effective timeout values when constructing every equivalent route action. - -**Tech Stack:** Go, Envoy xDS v3 protobufs, Helm, testify. - -## Global Constraints - -- Defaults preserve existing Envoy behavior: 2s connect, 15s route, 5m stream-idle. -- A zero route or stream-idle timeout explicitly disables that limit; connect timeout must be positive. -- New Go configuration fields require comments explaining behavior and zero-value semantics. -- Timeout overrides must influence snapshot versioning. - ---- - -### Task 1: Timeout model and annotation parsing - -**Files:** -- Modify: `server/pkg/task.go` -- Modify: `server/pkg/discovery.go` -- Test: `server/pkg/discovery_test.go` - -**Interfaces:** -- Produces `TaskProxyTimeoutConfig` and optional timeout fields on `Task`. -- Produces `parseTaskProxyAnnotation(any) ([]taskServiceInfo, *TaskTimeoutOverrides)`. - -- [ ] **Step 1: Write failing annotation tests** - -Add a valid `task_proxy` annotation with `route_timeout_seconds: 600` and `stream_idle_timeout_seconds: 120`, and assert the returned operation overrides preserve both durations. Add invalid negative and non-integer values and assert parsing rejects the annotation. - -- [ ] **Step 2: Run the focused test and verify it fails** - -Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestParseTaskProxyAnnotation` - -Expected: FAIL because no override values are returned or validated. - -- [ ] **Step 3: Add the minimal model and parser changes** - -Define documented `TaskProxyTimeoutConfig` defaults and optional per-operation `TaskTimeoutOverrides`. Parse the two optional annotation fields after `enabled`, reject malformed values, and copy the resulting overrides to every discovered task. - -- [ ] **Step 4: Run the focused test and verify it passes** - -Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestParseTaskProxyAnnotation` - -Expected: PASS. - -### Task 2: xDS effective timeouts - -**Files:** -- Modify: `server/pkg/xds.go` -- Modify: `server/pkg/updater.go` -- Test: `server/pkg/xds_test.go` -- Test: `server/pkg/updater_test.go` - -**Interfaces:** -- Consumes `TaskProxyTimeoutConfig`, `TaskTimeoutOverrides`. -- Produces xDS `RouteAction.Timeout`, `RouteAction.IdleTimeout`, and task upstream `Cluster.ConnectTimeout`. - -- [ ] **Step 1: Write failing xDS tests** - -Build a snapshot with defaults and a task override. Assert task clusters use the configured connect timeout, all routes for that task share the expected route timeout and idle timeout, and a zero override serializes as an explicit zero duration. - -- [ ] **Step 2: Run the focused test and verify it fails** - -Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestMakeSnapshotTimeouts` - -Expected: FAIL because `makeSnapshot` has no timeout configuration parameter and route actions omit both fields. - -- [ ] **Step 3: Add xDS resolution and updater plumbing** - -Pass the global config from updater to `makeSnapshot`; resolve per-task overrides while building the shared route action. Apply global connect timeout only to discovered task clusters, preserving the authz cluster's internal timeout. Include timeout overrides in task snapshot identity. - -- [ ] **Step 4: Run focused xDS and updater tests** - -Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run 'TestMakeSnapshotTimeouts|TestTaskUpdater'` - -Expected: PASS. - -### Task 3: Runtime and Helm configuration - -**Files:** -- Modify: `server/main.go` -- Modify: `chart/values.yaml` -- Modify: `chart/templates/deployment.yaml` -- Test: `server/pkg/xds_test.go` - -**Interfaces:** -- Consumes Helm `timeouts.connectTimeoutSeconds`, `timeouts.routeTimeoutSeconds`, `timeouts.streamIdleTimeoutSeconds`. -- Produces validated `TaskProxyTimeoutConfig` passed to `CreateTaskUpdater`. - -- [ ] **Step 1: Add a failing configuration validation test where feasible** - -Exercise the public timeout-validation helper with a zero connect timeout and negative route or stream-idle timeout, expecting errors; verify zero route and stream-idle are accepted. - -- [ ] **Step 2: Run the focused validation test and verify it fails** - -Run: `cd server/pkg && GOCACHE=/tmp/go-build GOTMPDIR=/tmp go test ./... -run TestValidateTaskProxyTimeoutConfig` - -Expected: FAIL because the validation helper does not exist. - -- [ ] **Step 3: Implement flags, validation, and chart values** - -Add documented timeout fields and validation helper in Go. Add command-line flags and Helm arguments with compatible defaults. Avoid exposing per-operation configuration in Helm because it belongs to the operation annotation. - -- [ ] **Step 4: Run complete verification** - -Run: `make test && helm template task-proxy ./chart >/tmp/task-proxy-rendered.yaml` - -Expected: all Go tests pass and Helm renders successfully. From b5bc6f79149d17cb426f4ed67704e2f51fe3a526 Mon Sep 17 00:00:00 2001 From: imakunin Date: Mon, 10 Aug 2026 12:07:45 +0300 Subject: [PATCH 3/3] test: keep annotation keys literal --- server/pkg/discovery_test.go | 44 ++++++++++++++++++------------------ 1 file changed, 22 insertions(+), 22 deletions(-) diff --git a/server/pkg/discovery_test.go b/server/pkg/discovery_test.go index 9081ae5..b37fc83 100644 --- a/server/pkg/discovery_test.go +++ b/server/pkg/discovery_test.go @@ -18,12 +18,12 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "full annotation", annotation: map[string]any{ - taskProxyEnabledKey: true, - taskProxyTasksInfoKey: map[string]any{ + "enabled": true, + "tasks_info": map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - taskProxyProtocolKey: "grpc", - taskProxyPortIndexKey: 0, + "protocol": "grpc", + "port_index": 0, }, }, }, @@ -40,14 +40,14 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "minimal annotation", annotation: map[string]any{ - taskProxyEnabledKey: true, + "enabled": true, }, expected: []taskServiceInfo{}, }, { name: "disabled annotation (false)", annotation: map[string]any{ - taskProxyEnabledKey: false, + "enabled": false, }, expected: nil, }, @@ -64,12 +64,12 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "unknown protocol", annotation: map[string]any{ - taskProxyEnabledKey: true, - taskProxyTasksInfoKey: map[string]any{ + "enabled": true, + "tasks_info": map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - taskProxyProtocolKey: "dns", - taskProxyPortIndexKey: 0, + "protocol": "dns", + "port_index": 0, }, }, }, @@ -79,12 +79,12 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "invalid port type", annotation: map[string]any{ - taskProxyEnabledKey: true, - taskProxyTasksInfoKey: map[string]any{ + "enabled": true, + "tasks_info": map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - taskProxyProtocolKey: "http", - taskProxyPortIndexKey: "0", + "protocol": "http", + "port_index": "0", }, }, }, @@ -94,11 +94,11 @@ func TestParseTaskProxyAnnotation(t *testing.T) { { name: "missing service attributes", annotation: map[string]any{ - taskProxyEnabledKey: true, - taskProxyTasksInfoKey: map[string]any{ + "enabled": true, + "tasks_info": map[string]any{ "example_grpc_server": map[string]any{ "server": map[string]any{ - taskProxyProtocolKey: "http", + "protocol": "http", }, }, }, @@ -128,9 +128,9 @@ func TestParseInteger(t *testing.T) { func TestParseTaskProxyAnnotationTimeoutOverrides(t *testing.T) { annotation := map[string]any{ - taskProxyEnabledKey: true, - taskProxyRouteTimeoutSecondsKey: 600, - taskProxyStreamIdleTimeoutKey: 120, + "enabled": true, + "route_timeout_seconds": 600, + "stream_idle_timeout_seconds": 120, } _, overrides := parseTaskProxyAnnotation(annotation) @@ -141,8 +141,8 @@ func TestParseTaskProxyAnnotationTimeoutOverrides(t *testing.T) { func TestParseTaskProxyAnnotationRejectsInvalidTimeoutOverrides(t *testing.T) { annotation := map[string]any{ - taskProxyEnabledKey: true, - taskProxyRouteTimeoutSecondsKey: -1, + "enabled": true, + "route_timeout_seconds": -1, } services, _ := parseTaskProxyAnnotation(annotation)