jason810496 commented on code in PR #76359:
URL: https://github.com/apache/airflow/pull/76359#discussion_r4469002210
##########
go-sdk/pkg/execution/task_runner.go:
##########
@@ -107,6 +109,37 @@ func RunTask(
}
}
+ timeoutSeconds, err :=
resolveExecutionTimeout(details.TIContext.ExecutionTimeout)
+ if err != nil {
+ logger.Error("Invalid execution_timeout",
+ "dag_id", details.TI.DagID,
+ "task_id", details.TI.TaskID,
+ "error", err,
+ )
+ return failTask(details.TIContext.ShouldRetry, err)
+ }
+ // Same rules as the Python task runner: zero means no timeout,
negative has already elapsed.
+ switch {
+ case timeoutSeconds == nil || *timeoutSeconds == 0:
+ case *timeoutSeconds < 0:
Review Comment:
How about move these cases into `resolveExecutionTimeout` then we could drop
the switch here entirely?
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -564,8 +562,18 @@ def _prepare_ti_context(self, ti: TaskInstance,
ti_context: TIRunContext) -> TIR
backend's storage.
If an XCom cannot be deleted, this try fails and the task does not run.
+
+ Also fills in the worker's default ``execution_timeout`` when the task
sets none.
"""
- self._execution_timeout_seconds = ti_context.execution_timeout
+ # A task that sets no execution_timeout gets the worker's default, as
a Python task does when the
+ # worker parses its Dag. A zero means no timeout, so it does not fall
back.
+ execution_timeout = ti_context.execution_timeout
Review Comment:
How about mentioning the `execution_timeout` here comes from the API server
and from the serialized Dag?
##########
go-sdk/pkg/execution/execution_timeout.go:
##########
@@ -0,0 +1,49 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package execution
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/apache/airflow/go-sdk/pkg/execution/genmodels"
+)
+
+// resolveExecutionTimeout returns the task's execution_timeout in seconds
from ti_context, or nil
+// when it has none. The coordinator has already filled in the worker's
default.
+func resolveExecutionTimeout(fromContext any) (*float64, error) {
+ if fromContext == nil {
+ return nil, nil
+ }
+ seconds, ok := ifaceFloat(fromContext)
+ if !ok {
+ return nil, fmt.Errorf(
+ "ti_context.execution_timeout has unexpected wire shape
%T",
+ fromContext,
+ )
+ }
+ return &seconds, nil
+}
Review Comment:
Goes with my comment on `RunTask`. With zero and negative handled here:
```suggestion
// resolveExecutionTimeout returns the task's execution_timeout in seconds
from ti_context, or nil
// when the task has no timeout. The coordinator has already filled in the
worker's default.
// As in the Python task runner, zero means no timeout and a negative value
has already elapsed.
func resolveExecutionTimeout(fromContext any) (*float64, error) {
if fromContext == nil {
return nil, nil
}
seconds, ok := ifaceFloat(fromContext)
switch {
case !ok:
return nil, fmt.Errorf(
"ti_context.execution_timeout has unexpected wire shape
%T",
fromContext,
)
case seconds == 0:
return nil, nil
case seconds < 0:
return nil, fmt.Errorf("execution_timeout of %g seconds has
already elapsed", seconds)
}
return &seconds, nil
}
```
the block in `RunTask` shrinks to:
```go
timeoutSeconds, err :=
resolveExecutionTimeout(details.TIContext.ExecutionTimeout)
if err == nil && timeoutSeconds != nil {
err = client.setExecutionTimeout(ctx, *timeoutSeconds)
}
if err != nil {
logger.Error("Unable to apply execution_timeout",
"dag_id", details.TI.DagID,
"task_id", details.TI.TaskID,
"error", err,
)
return failTask(details.TIContext.ShouldRetry, err)
}
```
##########
go-sdk/pkg/execution/iface.go:
##########
@@ -0,0 +1,99 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package execution
+
+import "time"
+
+// ifaceString returns the string carried by a nullable schema field decoded as
+// any, or "" when the value is nil or not a string.
+func ifaceString(v any) string {
+ s, _ := v.(string)
+ return s
+}
+
+// ifaceStringPtr preserves the null-vs-empty distinction for a nullable string
+// field decoded as any: nil when missing/null, else a pointer to the string.
+// Used for connection credentials, where an explicit "" must stay distinct
from
+// "not set".
+func ifaceStringPtr(v any) *string {
+ s, ok := v.(string)
+ if !ok {
+ return nil
+ }
+ return &s
+}
+
+// ifaceTimePtr returns a pointer to the time carried by a nullable date-time
+// field decoded as any, or nil when missing/null. msgpack decodes the
+// supervisor's timestamp extension straight into time.Time.
+func ifaceTimePtr(v any) *time.Time {
+ t, ok := v.(time.Time)
+ if !ok {
+ return nil
+ }
+ return &t
+}
+
+// ifaceInt converts a numeric value decoded as any into int, returning def
when
+// nil or non-numeric.
Review Comment:
nit: `ifaceNumber[int]` truncates a float, worth keeping that in the comment.
```suggestion
// ifaceInt converts a numeric value decoded as any into int, truncating a
float,
// and returns def when the value is nil or non-numeric.
```
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -562,8 +562,18 @@ def _prepare_ti_context(self, ti: TaskInstance,
ti_context: TIRunContext) -> TIR
backend's storage.
If an XCom cannot be deleted, this try fails and the task does not run.
+
+ Also fills in the worker's default ``execution_timeout`` when the task
sets none.
"""
- self._execution_timeout_seconds = ti_context.execution_timeout
+ # A task that sets no execution_timeout gets the worker's default, as
a Python task does when the
+ # worker parses its Dag. A zero means no timeout, so it does not fall
back.
+ execution_timeout = ti_context.execution_timeout
+ if (
+ execution_timeout is None
+ and (default := conf.gettimedelta("core",
"default_task_execution_timeout")) is not None
Review Comment:
`gettimedelta` raises `AirflowConfigException` for a value like `abc`. This
runs after `task_instances.start()`, so the tracker kills the runtime but the
TI stays `running` with nothing in its task log until it is reaped as a zombie.
How about resolving the default before the TI is started and raising a
`TaskLaunchError` on a bad value? Then `_fail_task_before_runtime` fails the
task with the message in its log. A `worker_default="abc"` case in the test
would cover it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
