andygrove opened a new pull request, #2225:
URL: https://github.com/apache/datafusion-ballista/pull/2225
# Which issue does this PR close?
Closes #2224.
# Rationale for this change
In push-based scheduling the executor tells the scheduler about itself
before its own gRPC server is listening, and the scheduler dials that port back
to verify connectivity. When the callback wins the race the executor dies at
startup.
The ordering in `executor_server::startup` was:
1. `tokio::spawn` the tonic server. `Server::serve` binds *inside* the
future, so the socket does not exist yet when `spawn` returns — even though the
`"... Grpc Server listening on ..."` line is already in the log.
2. `register_executor`, immediately.
On the scheduler side `ExecutorManager::register_executor` calls
`test_connectivity`, a single `ExecutorGrpcClient::connect(...)` with no retry
and no backoff. If the spawned task has not reached its `bind` yet, that
connect gets `ECONNREFUSED`, registration returns an error, and
`executor_process` treats a registration error as fatal, so the executor exits
instead of joining the cluster.
The opposite direction is already tolerant: the executor retries its
connection *to* the scheduler in a loop. Only the scheduler's callback to the
executor is one-shot, which is why the executor's own startup ordering has to
be right.
It is timing-dependent and so shows up rarely and on loaded machines. The
instance that prompted this was CI, where one of two executors lost the race
and the HA chaos harness timed out waiting for it:
```
timed out waiting for 2 executors to register
# executor-1
INFO Ballista v54.0.0 Rust Executor Grpc Server listening on 127.0.0.1:42635
ERROR Executor registration failed due to: ... Failed to register executor
at 127.0.0.1:42635,
could not connect: ... Os { code: 111, kind: ConnectionRefused }
# scheduler, same second
ERROR Fail to do executor registration due to: ... ConnectionRefused ...
```
The consequence outside of tests is the same shape: an executor that is
otherwise healthy fails to join, and in a restart loop it can keep failing to
join.
# What changes are included in this PR?
- `ballista/core/src/utils.rs` — new `create_grpc_server_incoming(addr,
&GrpcServerConfig)`, which binds the listening socket eagerly and returns
tonic's `TcpIncoming`. tonic ignores the server builder's `tcp_nodelay` /
`tcp_keepalive` when serving from a pre-bound listener, so the helper applies
the same values `create_grpc_server` sets and keeps the two in one place. The
remaining settings (timeout, HTTP/2 keep-alive) still come from the builder as
before, so the socket is configured exactly as it was.
- `ballista/executor/src/executor_server.rs` — `startup` binds via that
helper on the current task, before spawning, and the spawned task now serves
with `serve_with_incoming_shutdown`. Registration therefore cannot run before
the port is accepting connections. This retires the standing `// TODO the
executor registration should happen only after the executor grpc server
started.` The `"listening on"` log line is now true when it is printed.
One incidental improvement falls out of binding eagerly: a port conflict now
fails `startup` directly, with the bind error, rather than being discovered
later through the spawned server task.
No change to the scheduler's `test_connectivity`. The issue also floats
giving it a bounded retry; that is worth doing on its own merits for genuinely
remote executors, but it is a separate resilience change and is not needed to
fix this race, which is entirely local to the executor's startup ordering.
New tests in `ballista-core`:
- `grpc_server_incoming_accepts_connections_before_it_is_served` — connects
to the bound address while nothing is serving on it. This is the property the
fix depends on, and it fails against a lazily-bound socket.
- `test_create_grpc_server_incoming_port_in_use` — binding an address twice
is an error rather than a panic, so a port conflict still surfaces as a normal
startup failure.
# Are there any user-facing changes?
No behaviour changes and no API breakage. `create_grpc_server_incoming` is a
new public function in `ballista-core`, additive alongside
`create_grpc_server`. Note that it must be called from within a Tokio runtime,
since the listener registers with the reactor; this is documented on the
function.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
