bmanan7 opened a new issue, #70867:
URL: https://github.com/apache/airflow/issues/70867
### Description
`WasbRemoteLogIO.write()` appends to a task log by rewriting the entire
object. It checks whether the
blob exists, downloads the complete existing log, joins it with the new
content in memory, and uploads
the whole result again as a block blob with `overwrite=True`. The whole
retained log therefore crosses
the network in both directions on every upload.
One task attempt can be uploaded to many times, because the remote key comes
from `try_number` and
some lifecycles reuse a single attempt. A reschedule-mode sensor is the
clearest case: every poke is a
separate worker process, and `UP_FOR_RESCHEDULE` does not change
`try_number`, so all of them write to
the same `attempt=N.log`. The poke count is bounded by the sensor's timeout,
but that bound is high:
`[sensors] default_timeout` is 604800 seconds, so at the default
`poke_interval` of 60 seconds a
single attempt can reach around 10,000 uploads before it times out.
What I would like is for a new segment of log to be sent as a new segment,
without reading back and
rewriting everything already stored.
### Preferred approach: Azure AppendBlob
Append blobs are Azure's native answer to this. They keep one logical log
object per attempt, so
nothing on the read path or in the UI has to change in the normal case, and
the diff is contained
inside this provider.
The local-segment behavior proposed in #70860 makes the flow below possible.
With that change, the
local log file holds exactly the current lifecycle's output: on the same
worker the previous
successful upload truncated it, and on a different worker it starts empty.
So the local file is the
segment to append, with no offset bookkeeping needed.
1. Read the local file. That is the new segment.
2. If the blob does not exist, create an empty append blob.
3. If it does exist, read its current length, which is the append position.
4. Append the segment with `append_block()`, passing an append-position
condition.
5. Truncate the local file only after the append succeeds.
6. If the object already exists as a block blob, use the current write path
for it and never convert
it in place.
Step 4 has to use explicit append calls rather than
`BlobClient.upload_blob(blob_type="AppendBlob")`. In
`_upload_helpers.upload_append_blob` the
condition object is built as
`AppendPositionAccessConditions(append_position=None)` and a
caller-supplied `appendpos_condition` is ignored, so the first block of
every call carries no
position guard. Since `azure-core` retries connection errors by default, a
block that the service
committed but whose response was lost would be appended twice.
`append_block()` accepts the condition
directly.
#### Rollout
Worth making this opt-in to begin with, with block blob remaining the
default, so the change can land
and be exercised before it becomes the behaviour everyone gets:
```
remote_wasb_log_write_mode = block_blob # default, current behaviour
remote_wasb_log_write_mode = append_blob # opt-in
```
The name needs checking against current conventions in the
`[azure_remote_logging]` section, which
today holds `remote_wasb_log_container`, and against the `from_config()`
work in #70265 and #70268,
since that is where the option would be read.
#### Constraints to handle and document
- An append-position mismatch has to fail safely rather than silently write
at the wrong offset.
- Segments larger than the service block limit need chunking.
- A partial failure part way through a multi-block append needs explicit
recovery, so the next attempt
neither duplicates a committed block nor skips one.
- An append blob holds at most 50,000 blocks, so a very long-lived attempt
needs a rollover to a
sibling key.
- Rollover naming and read ordering both need validating. The reader already
lists by prefix and
concatenates in sorted order, so the mechanism exists, but the naming has
to sort correctly
alongside the existing `.trigger.<job_id>.log` suffix, and the extra
objects will show up in the
log source list.
- Append blobs cannot be moved to cool or archive by a lifecycle policy.
Lifecycle delete rules still
work, but anyone tiering old task logs would need to stay on block blobs.
- Objects already written as block blobs stay readable and keep using the
current write path. Azure
does not support changing an existing blob's type in place; versioned
blobs make replacement even
more restrictive. So conversion is not an option anyway.
- UTF-8 and newline handling across a segment boundary needs to be right,
since the reader decodes
the assembled object as text.
### Alternatives considered
**Immutable segment objects.** Each lifecycle writes a new
`...part<NNNNN>.log` and nothing is ever
modified. Attractive because it is not Azure-specific and would fix S3, GCS,
OSS and HDFS too, none of
which have a native append. But it is a cross-provider architecture change
and it alters the
on-storage layout for anything reading these logs outside Airflow, so it is
a much larger discussion
than this provider.
**Block blob block lists.** `Put Block` plus `Put Block List`, re-committing
the previous block ids.
Keeps block blob semantics, so lifecycle tiering still works. Against it:
the client has to track the
block id list across separate worker processes, concurrent writers have to
be handled explicitly, and
`Put Block List` is a version-creating operation, so accounts with
versioning enabled would still
accumulate a version per upload. It also still caps at 50,000 blocks.
### Use case/motivation
I want a long-running sensor to cost roughly its own log size in traffic,
rather than a multiple of it
that grows with how long the sensor ran.
The growth is quadratic. With N uploads of `s` new bytes each, the
cumulative upload is `s*N(N+1)/2`
while the log actually retained is only `N*s`. Measured against the current
class with an instrumented
fake blob store, using a fresh local log directory per lifecycle so only the
rewrite behaviour is in
play:
| uploads | log retained | uploaded | downloaded | uploaded / retained |
|---|---|---|---|---|
| 1 | 20 B | 20 B | 0 B | 1.0x |
| 10 | 210 B | 1,146 B | 936 B | 5.5x |
| 20 | 430 B | 4,456 B | 4,026 B | 10.4x |
| 40 | 870 B | 17,676 B | 16,806 B | 20.3x |
| 80 | 1,750 B | 70,516 B | 68,766 B | 40.3x |
Projecting that, on the illustrative assumption of 1 KiB of new log per poke
(I measured 758 bytes for
a deliberately minimal reschedule sensor on Airflow 3.3; other Dags may
generate more):
| | 24 hours at the default 60 s interval | default 7 day sensor timeout |
|---|---|---|
| uploads | 1,440 | 10,080 |
| log retained | 1.4 MiB | 9.8 MiB |
| uploaded | 1,013 MiB | 48.5 GiB |
| downloaded | 1,012 MiB | 48.4 GiB |
| total moved | 2.0 GiB | 96.9 GiB |
| storage operations | 4,319 | 30,239 |
Total bytes moved works out to roughly N times the size of the log being
stored.
Two things follow from rewriting the whole object, beyond the byte count.
Memory. Each upload holds the entire log in memory to concatenate it, so
peak usage tracks the log
size rather than the size of the new segment. The same problem on the S3
handler was reported as a
worker OOM in #67144.
Latency at the end of every lifecycle. The upload happens as the task
finishes and gets slower as the
log grows, so a sensor's teardown time increases over the life of the
attempt.
On storage accounts with blob versioning enabled it is worse again, because
`Put Blob` creates a new
version each time. One attempt can leave thousands of full versions of a
single blob, against
Microsoft's guidance to stay under 1,000 versions per blob.
### Testing this would need
- Both configuration modes, including that the default is unchanged.
- Successive lifecycles on the same worker, and on different workers.
- Retry of a failed append, confirming no duplicated or skipped block.
- An existing block blob, confirming it stays readable and keeps the current
write path.
- A segment larger than one block, and a partial failure part way through a
multi-block append.
- The 50,000 block limit and whatever rollover is chosen, including that
reads come back in order.
- UTF-8 across segment boundaries, and newline handling where one segment
ends and the next begins.
- Reading the assembled log through the UI and the API.
- Bytes and operations before and after, so the improvement is measured
rather than asserted.
The core flow can be tested with Azurite. Conditional retry, append-position
conflict, and block-limit
behavior should be validated against Azurite's supported behavior and
supplemented with mocks or a real
Azure Storage account where necessary.
### Related issues
- #70860 proposes a fix for a separate content-duplication defect in the
same code path, where a local
log file that survived between lifecycles was re-uploaded in full. This
issue is the remaining half:
even with every line stored exactly once, each upload still rewrites the
whole object. That proposal
is also what makes the design above straightforward, since the local file
becomes a clean segment.
- #22496 proposed append blobs for this handler in 2022. It was closed by
the stale bot without
review and its author noted they had not tested it. It went through
`load_string(blob_type='AppendBlob')`, so it inherits the unguarded first
block described above, and
it does not deal with existing block blobs or the block limit.
- #67144 fixed the duplication half for S3 and is where the OOM symptom was
first reported.
- #68304, a closed follow-up draft to that fix, noted that "the Google and
Azure object store logging
may have a similar memory bug due to a download, append, upload process".
- #45079 is the read side of the same shape, running out of memory serving
large logs.
- #70265 and #70268 cover the `from_config()` migration for this handler,
which is where a new option
would be wired in.
- #15907 and the PRs around it are the history of how append behaviour
reached its current form after
the move to azure-storage-blob v12.
### Are you willing to submit a PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
