Documentation
¶
Overview ¶
Package dataplane provides the transport half of a volume transfer, shared by every d8 command that moves volume bytes: a typed client for the data-exporter API, a classifier that decides which transport failures are worth another try, and the bounded retry-with-resume policy built on top of them.
The data-exporter API has two endpoints: api/v1/block (block volumes served via http.ServeContent with Range support) and api/v1/files (filesystem volumes: trailing-slash paths return a JSON directory listing; other paths stream file bytes).
It deliberately knows nothing about snapshots, chunk geometry, part files or any other caller-specific framing: control-plane concerns (resolving a DataExport, waiting for it to become ready, run-owner annotations) live with their commands, which import this package rather than the other way round.
Index ¶
- Constants
- Variables
- func BlockURL(baseURL string) (string, error)
- func FilesURL(baseURL string) (string, error)
- func IsFatalDataPlaneError(err error) bool
- func IsTransientDataPlaneError(err error) bool
- type Attempt
- type Doer
- type DoerFunc
- type Fetcher
- func (f *Fetcher) GetFile(ctx context.Context, fileURL string) (io.ReadCloser, error)
- func (f *Fetcher) HeadVolume(ctx context.Context, blockURL string) (int64, error)
- func (f *Fetcher) ListDir(ctx context.Context, filesURL string, yield func(Item) error) error
- func (f *Fetcher) OpenStream(ctx context.Context, rawURL string, from int64) (io.ReadCloser, int64, error)
- func (f *Fetcher) RangeGet(ctx context.Context, blockURL string, start, end int64) (io.ReadCloser, error)
- func (f *Fetcher) SourceMD5(ctx context.Context, fileURL string, size int64) (string, error)
- type FetcherOption
- type Item
- type Progress
- type Retrier
- type RetryPolicy
Constants ¶
const DefaultIdleReadTimeout = 2 * time.Minute
DefaultIdleReadTimeout is the conservative default idle window applied to every data-plane response body a Fetcher hands out or consumes. It bounds how long a single Read may block WITHOUT receiving any bytes; it is not an overall transfer deadline, so a slow-but-flowing stream is never aborted. Override per Fetcher with WithIdleReadTimeout (tests use a short window).
const SourceHashTimeoutCeiling = 7 * 24 * time.Hour
SourceHashTimeoutCeiling caps the size-derived budget SourceMD5 allows a producer for hashing one file. Transports dedicated to source-hash requests size their own response-header timeout from it, so it is exported.
Variables ¶
var ErrContentRangeMismatch = errors.New("server Content-Range does not match requested range")
ErrContentRangeMismatch is returned when a 206 response's Content-Range header does not cover the byte range the caller requested, so the body must not be trusted at the caller's intended offset.
var ErrDataPlaneIdle = errors.New("data-plane read stalled: no bytes within idle timeout")
ErrDataPlaneIdle is reported by a Fetcher-issued response body when no bytes arrive for the configured idle window (see idleReadCloser). The data plane deliberately runs without an overall request deadline — volume transfers are long — so a TCP connection that stops delivering bytes WITHOUT erroring (NAT/LB half-close, wedged exporter pod) would otherwise block a Read (and the whole download) forever. It is wrapped with %w so the chunk/file retry+resume machinery treats a silent stall as an ordinary fetch error.
var ErrDataPlaneNotAccepted = errors.New("data-plane endpoint did not accept the request")
ErrDataPlaneNotAccepted is one statement: the endpoint ANSWERED, and the request did not complete. The answer is the whole of what it asserts — it separates a far end that is there and did not take this request from silence, which is the distinction that decides whether trying again can help.
It says nothing about how much the destination now holds, and must not be read as saying nothing arrived. A connection torn down mid-request leaves the importer holding whatever it managed to take, and the carrier for that case moves the durable offset forward onto it before returning this error, so a run reports both on one line: delivered_bytes above zero beside this very failure.
The shapes it is put on today, which is a list of the callers rather than a closed set, all belong to the upload path: an importer that acknowledged a chunk and named back the offset the chunk began at; one that refused a chunk and named that same offset back; one that refused without naming an offset and then could not resolve the question when asked; and a request whose connection died while the importer went on answering probes at an offset short of the whole.
It is a member of the transient set below rather than a judgement a caller makes for itself, and it has to be, because it must survive an attempt that delivered NOTHING. The progress rule in Retrier.Resume cannot retry such an attempt, and RetryPolicy.Fatal can only add to the fatal set.
The distinction it draws is one only a WRITING transfer needs. A download measures its own destination, so a broken body still leaves a durable prefix behind and the progress rule carries it. An upload has no local measurement — bytes pushed into a connection that then died may or may not have been written — so the far end still being there is the only evidence available that the failure was the transport rather than the setup.
Being transient does not make it unbounded: an endpoint that keeps answering and keeps not accepting delivers nothing, so MaxNoProgress ends the loop (TestRetrier_NotAcceptedIsStillBoundedWithoutDelivery).
ErrExportUnauthorized classifies a 401/403 from the data-exporter endpoint. On the public (Ingress) path this is the expected outcome for a certificate-authenticated kubeconfig: Ingress terminates TLS with its own certificate and does not forward the client certificate to the exporter pod, so only bearer-token kubeconfigs authenticate through it.
var ErrRangeIgnored = errors.New("server ignored the Range header and returned the whole object")
ErrRangeIgnored is returned when a request that carried a Range header asking to continue from a non-zero offset is answered with 200 OK — the whole object from offset zero — instead of 206 Partial Content. The body is then not the continuation the caller asked for, so writing it at the caller's resume offset would corrupt the destination.
Functions ¶
func BlockURL ¶
BlockURL returns the block-volume endpoint for a DataExport base URL. The block volume is served at api/v1/block.
func FilesURL ¶
FilesURL returns the filesystem root-listing endpoint for a DataExport base URL. The trailing slash instructs the server to return a directory listing.
func IsFatalDataPlaneError ¶
IsFatalDataPlaneError reports whether err must end a transfer immediately, no matter how many bytes the attempt delivered before it. It is the ONE place this set is written down; Retrier.Resume consults it BEFORE it consults progress, and IsTransientDataPlaneError below is defined against it so the two can never disagree.
The set is deliberately small, and every member is a statement about the request rather than about the link:
- context cancellation or deadline: the caller asked to stop, and an aborted request surfaces through the HTTP transport looking exactly like an ordinary broken connection;
- ErrExportUnauthorized: credentials the exporter rejected once it will reject again, so retrying only multiplies the rejection;
- ErrContentRangeMismatch: the body cannot be trusted at the offset the caller intended to write it, so continuing from it would corrupt the destination rather than merely waste time.
Everything else — including errors this package has never seen — is left to IsTransientDataPlaneError and to the progress rule in Retrier.Resume.
func IsTransientDataPlaneError ¶
IsTransientDataPlaneError reports whether err is a RECOGNIZED transient transport failure on the volume data plane — one where re-issuing the Range GET from the caller's durable resume offset is the correct response.
It fails CLOSED: anything not explicitly listed is not recognized here. That is a statement about this function only, not about the transfer: an unrecognized error that arrives AFTER the attempt delivered bytes is still retried, by the progress rule in Retrier.Resume. The reason that rule cannot live here is that the failure it was written for — an HTTP/2 session torn down mid-body — carries a type net/http keeps in its own unexported copy of the http2 package, so no errors.As or errors.Is from outside can reach it. A classifier that could only ever grow by naming types would keep missing it.
Types ¶
type Attempt ¶
Attempt performs exactly one try of a resumable transfer and reports how far it got. It must not retry internally: the retry seam is Resume, and an attempt that looped inside would hide its failures from both bounds of the policy.
The attempt is responsible for resuming from its own durable state; Resume never tells it where to start, which is why the same mechanism serves resuming across attempts within one run and resuming across separate runs.
type Doer ¶
Doer executes a single HTTP request and returns the response. *http.Client satisfies it directly, and so does the pinned per-origin client the snapshot commands build (internal/snapshot/transport.PersistentHTTPClient, which has both Do and HTTPDo).
pkg/libsaferequest/client.SafeClient does NOT satisfy it: its only request method is HTTPDo, and it has no Do at all. A caller holding a SafeClient — the d8 data commands do — therefore needs an adapter, not a direct assignment, and DoerFunc below is that adapter.
type DoerFunc ¶
DoerFunc adapts a plain request function to Doer, so a client whose request method is named something other than Do (SafeClient.HTTPDo) reaches a Fetcher without a bespoke wrapper type per caller.
type Fetcher ¶
type Fetcher struct {
// contains filtered or unexported fields
}
Fetcher wraps a Doer and exposes typed methods for the data-exporter HTTP API.
func NewFetcher ¶
func NewFetcher(doer Doer, opts ...FetcherOption) *Fetcher
NewFetcher creates a Fetcher backed by the given Doer. Unless overridden via WithIdleReadTimeout, response bodies carry an idle-read watchdog with DefaultIdleReadTimeout.
func (*Fetcher) GetFile ¶
GetFile GETs fileURL and returns the response body for streaming. The caller must close the returned ReadCloser. It is OpenStream from offset zero for a caller with no use for the declared length.
func (*Fetcher) HeadVolume ¶
HeadVolume issues a HEAD request to blockURL and returns the total content length in bytes.
func (*Fetcher) ListDir ¶
ListDir GETs filesURL (which must end with a trailing slash for directory semantics), requests only the inexpensive stat attributes, and calls yield once per stream-decoded directory entry. It retains only the current item, so a flat directory's memory use is independent of its entry count. Source hashes are fetched separately after each regular file's declared size is known because the producer computes hash.md5 synchronously before emitting that listing item.
func (*Fetcher) OpenStream ¶
func (f *Fetcher) OpenStream(ctx context.Context, rawURL string, from int64) (io.ReadCloser, int64, error)
OpenStream GETs the byte range [from, end of object] from rawURL and returns the body together with the length the producer declares for the WHOLE object, or -1 when it declares none. The caller must close the returned ReadCloser.
It is the resume primitive of this package, and what separates it from RangeGet is what the caller must already know: RangeGet serves a caller holding the object's size and cutting it into bounded chunks, while a caller resuming ONE stream knows only how far it got, asks for "the rest", and learns the size from the answer.
from == 0 is sent WITHOUT a Range header, and that is deliberate rather than an omission:
- a producer with no range support at all still serves the whole object on a plain GET, so a transfer that never breaks keeps working against it. Only resuming needs ranges, and only a transfer that already broke pays for their absence;
- it also keeps every EMPTY object out of the disagreement about what a byte range even means for a zero-length representation. Go's http.ServeContent, which both exporters serve files and the block device through, carves that case out and answers 200 (net/http/fs.go handles errNoOverlap with size 0 that way, and TestOpenStream_FirstRequestServesAnEmptyObject holds it to it), while a producer reading RFC 9110 literally has nothing to satisfy and answers 416. At offset zero there is nothing to gain by depending on which of the two is in front of us.
from > 0 requires 206 whose Content-Range starts exactly at from. A 200 answer there is ErrRangeIgnored, never a body to append: it is the whole object from offset zero, and appending it to what is already on disk corrupts the destination silently — the one failure of this family that leaves nothing to find later.
func (*Fetcher) RangeGet ¶
func (f *Fetcher) RangeGet(ctx context.Context, blockURL string, start, end int64) (io.ReadCloser, error)
RangeGet issues a GET request with a Range: bytes=start-end header to blockURL and returns the response body. The caller must close the returned ReadCloser. Returns an error unless the server responds with 206 Partial Content AND its Content-Range header confirms the returned body actually covers [start, end]: the block exporter (storage-volume-data-manager images/data-exporter/internal/export_block/ handler.go HandleGetMethod) serves the block device via stdlib http.ServeContent, which always sets Content-Range on a 206 response, so a missing or mismatched header means a misbehaving server/proxy returned bytes from the wrong offset (or the whole object) and must not be trusted at the caller's intended offset.
func (*Fetcher) SourceMD5 ¶
SourceMD5 retrieves the producer-computed plaintext MD5 for one regular file through the filesystem exporter's HEAD attribute contract. The producer must read the complete source file before it can send this response header, so the request uses an overall size-derived deadline rather than the transfer body's progress-based idle watchdog.
The budget assumes the source can be hashed at no less than 1 MiB/s, adds one minute of fixed scheduling slack, floors small or unknown sizes at five minutes, and caps untrusted declared sizes at seven days. It is finite for every int64 size. An empty result means an older exporter returned 200 without the optional hash header.
type FetcherOption ¶
type FetcherOption func(*Fetcher)
FetcherOption customizes a Fetcher at construction time.
func WithIdleReadTimeout ¶
func WithIdleReadTimeout(d time.Duration) FetcherOption
WithIdleReadTimeout sets the idle-read watchdog window for response bodies this Fetcher issues. A value <= 0 disables the watchdog (bodies are returned unwrapped). When unset, DefaultIdleReadTimeout applies.
func WithPublishUnauthorizedHint ¶
func WithPublishUnauthorizedHint() FetcherOption
WithPublishUnauthorizedHint makes this Fetcher append an actionable hint to ErrExportUnauthorized failures. Only the publish path sets it, so the in-cluster path never advertises an Ingress-specific remedy.
func WithSourceHashDoer ¶
func WithSourceHashDoer(doer Doer) FetcherOption
WithSourceHashDoer sets the transport used for source-hash HEAD requests. Production uses a separate transport because source hashing may legitimately take longer than the ordinary data-plane response-header timeout.
type Item ¶
type Item struct {
Name string `json:"name"`
Type string `json:"type"`
URI string `json:"uri"`
TargetPath string `json:"targetPath,omitempty"`
Attributes map[string]any `json:"attributes"`
}
Item is one entry returned by the data-exporter filesystem listing API.
type Progress ¶
Progress is what one Attempt reports about how far the transfer got. Start is the durable offset the attempt resumed from; Durable is the offset durably persisted when the attempt ended. Both are in the caller's own coordinates (an absolute file offset, an offset within one chunk, ...) — Resume only ever subtracts one from the other, never interprets them.
An attempt that fails before it can even establish where it stands should report the zero value, which reads as a zero-delivery attempt like any other.
func (Progress) Delivered ¶
Delivered reports how many bytes this attempt added to the durable prefix. It is the sole progress signal Resume acts on. A report whose Durable ran backwards yields a non-positive value and is read as no delivery, which is the safe reading: an attempt that cannot say it moved forward must not buy another one.
type Retrier ¶
type Retrier struct {
// contains filtered or unexported fields
}
Retrier carries one run's retry policy plus the aggregate count of retries it absorbed. One instance may be shared by several concurrent transfers: Resume keeps all its per-transfer state in locals, and only recovered is mutated concurrently — it is atomic for that reason.
func NewRetrier ¶
func NewRetrier(policy RetryPolicy) *Retrier
NewRetrier returns a Retrier bound to policy, with any bound the caller left unset replaced by the production default (see RetryPolicy.withDefaults).
func (*Retrier) Recovered ¶
Recovered reports how many retries this Retrier has absorbed across every transfer that used it — worth reporting once a run finishes, since a recovered transfer is otherwise indistinguishable from an untroubled one.
func (*Retrier) Resume ¶
func (r *Retrier) Resume(ctx context.Context, log *slog.Logger, subject string, attempt Attempt) error
Resume retries attempt with bounded exponential backoff, each attempt continuing from the durable offset the previous one persisted, until the transfer completes, a fatal error occurs, ctx is cancelled, or the retry budget is exhausted. subject names the transfer in errors and logs (for example "chunk 7").
WHICH FAILURES BUY ANOTHER ATTEMPT. In order, and the order is the point:
ctx cancellation, every member of IsFatalDataPlaneError, and anything the policy's own Fatal predicate names end the loop at once, no matter how many bytes the attempt delivered first. This check stands BEFORE the progress check below, so a rejected credential costs one attempt rather than the whole budget, and a body whose Content-Range cannot be trusted is never re-read from a later offset.
A failure IsTransientDataPlaneError recognizes is retried.
A failure nobody recognizes is retried too, but only if the attempt DELIVERED BYTES before it. An unknown error that arrives without any delivery is fatal on the first attempt, so a genuinely broken setup (a missing destination directory, a refused connection) still fails immediately and loudly.
Rule 3 exists because a transport can break in a way no list can name: an HTTP/2 session torn down mid-body surfaces as a type net/http keeps inside its own unexported copy of the http2 package, unreachable to errors.As from any other package. Delivered bytes are the observable that does not depend on naming the error: a connection that carried data and then stopped is by construction a connection worth re-opening where it stopped.
The price of rule 3 is that a fatal condition this package cannot recognize costs a whole budget instead of a single attempt when it happens to strike mid-stream. Both bounds of the policy keep that price finite and noisy.
type RetryPolicy ¶
type RetryPolicy struct {
Backoff wait.Backoff
MaxNoProgress int
// Fatal, when non-nil, names failures THIS caller knows are not a broken
// transport, and so are not worth another attempt however many bytes
// preceded them. It can only ADD to IsFatalDataPlaneError: Resume checks
// the built-in set first, so a permissive predicate cannot talk a refused
// credential or an untrustworthy Content-Range into being retried.
//
// It exists because the progress rule in Resume is deliberately blind to
// error identity, and a caller that CAN identify a failure — a producer
// that ended a response short of the range it promised, say — should not
// have to spend a whole budget rediscovering that it will keep doing so.
Fatal func(error) bool
}
RetryPolicy bounds one transfer's retry loop: Retrier.Resume stops re-issuing the request once either Backoff's step budget or MaxNoProgress consecutive zero-delivery attempts is exhausted.
The two bounds are not interchangeable, and only one of them survives a link that keeps delivering:
MaxNoProgress counts CONSECUTIVE zero-delivery attempts and a delivering attempt resets it to zero. It catches a far end that accepts the request and sends nothing, and it catches nothing else.
Backoff's step budget is spent by every attempt the loop makes, delivering or not, and nothing refreshes it. It is therefore the only thing standing between the loop and the nastiest shape of broken link: one that delivers some bytes and then breaks, over and over, so that MaxNoProgress is reset each time round.
func DefaultRetryPolicy ¶
func DefaultRetryPolicy() RetryPolicy
DefaultRetryPolicy returns the production retry policy applied to every resumable data-plane transfer.
Source Files