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/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..b37fc83 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) { @@ -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{ + "enabled": true, + "route_timeout_seconds": 600, + "stream_idle_timeout_seconds": 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{ + "enabled": true, + "route_timeout_seconds": -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 +}