Crate ytsaurus_client

Expand description

A thin YTsaurus client: enough of the HTTP API v4 to run a Rust worker without a Python installation.

It is deliberately small. It does what launching a job needs — create a node, upload the worker, write and read tables, start an operation and wait for it — and nothing else. For everything beyond that, the yt CLI remains the right tool.

§Launching a job

use ytsaurus_client::{Client, MapSpec};

let client = Client::from_env()?;

// Upload the worker, marked executable so the node can run it.
client.upload_worker("target/.../my_job", "//tmp/my_job")?;

let spec = MapSpec::new("./my_job", ["//tmp/input"], ["//tmp/output"])
    .with_local_file("//tmp/my_job")
    .with_memory_limit(512 * 1024 * 1024);

let id = client.start_map(&spec)?;
client.wait_for_operation(&id)?;

§Configuration

Client::from_env reads YT_PROXY for the cluster address, and finds a token the way the yt CLI does: YT_TOKEN, then the file named by YT_TOKEN_PATH, then ~/.yt/token. A machine where the CLI already works needs nothing else. A bare host is assumed to be HTTPS; a local cluster is reached as http://localhost:8000.

YT_CA_BUNDLE names a PEM file of root certificates, for an installation whose certificate chains to a CA the Mozilla bundle has never heard of. It is read by any build with the tls feature — which is the default, and the only kind that has a handshake to configure — and the platform-verifier feature is the same answer without a variable to set. Every block in the file must be an X.509 certificate: one that is not, a .p7b re-armoured under a BEGIN CERTIFICATE label being the usual case, refuses the whole file rather than becoming a root store quietly shorter than the caller wrote down. Without it, and without that feature, a cluster behind a private CA fails its very first request with invalid peer certificate: UnknownIssuer — the refusal names both ways out, because on a machine where curl reaches the same cluster nothing else about it suggests whose roots were consulted.

An installation differs from a local cluster in ways a caller of Client::from_env cannot otherwise reach, so it reads four more: YT_PROXY_SUFFIX completes a bare cluster name, YT_HEAVY_PROXY_DOMAINS names another domain its heavy proxies live in, YT_HEAVY_PROXIES_ANYWHERE removes that rule outright, and YT_FILE_CACHE moves the worker cache. Each is inert when unset, and each but the first has a builder method beside it — see Client::from_env for the table.

§When an operation fails

Client::wait_for_operation does not stop at the state. It asks the cluster which jobs failed and what they wrote to stderr, and carries both in ClientError::OperationFailed, so a failure explains itself without a trip to the web UI:

operation 1ba94195-… finished as failed: Failed jobs limit exceeded: Process terminated by signal 6
  job 24c164af-… on localhost:24403: User job failed: Process terminated by signal 6
  stderr:
    thread 'main' panicked at examples/src/bin/boom.rs:37:17:
    boom: this job fails on purpose (row 1, 23 bytes)

That costs one Client::list_jobs and a few Client::get_job_stderr calls per failed operation; Client::with_job_diagnostics turns it off.

§After it has started

An operation can be paused, given more of its pool, finished early, found by the alias its spec gave it, and — the one that matters for a pipeline that restarts — picked up again by a process that did not start it:

let op = client.attach_operation(std::fs::read_to_string("run.id")?);

op.suspend(false)?;
op.update_parameters(&OperationParameters::new().with_weight(2.0))?;
op.resume()?;
op.wait()?;

Everything on Operation is also on Client, taking the id. See the operation module for what the cluster does and does not promise about each of those commands — some of it is surprising, and all of it was measured.

§All at once, or not at all

Each step above can fail halfway and leave something behind — an empty table, a stale worker, an output table holding neither the old result nor the new one. Client::start_transaction makes the whole sequence one event: nothing it does is visible until Transaction::commit, and dropping the handle aborts it, so a ? on any line leaves the cluster as it was.

A transaction can also outlive its handle: Transaction::detach stops the keep-alive and leaves it running, Client::attach_transaction turns the id back into a handle elsewhere, and Client::ping_transaction, Client::commit_transaction and Client::abort_transaction finish one from a process that holds nothing but the id.

§Seeing what it did

The cluster traces itself, so joining its trace costs a header and no dependency: Client::with_trace_context puts every request into the trace a TraceContext names, and the proxy’s own span for that request is placed inside it rather than starting an orphan.

This process’s own side is the tracing feature, off by default: with it, each attempt runs in a span carrying the command, the attempt number and the elapsed time, and the message a retry prints on stderr becomes a WARN event instead. It is off because this crate is linked into worker binaries cross-compiled to musl — the same reason tls is.

§Heavy commands go where the cluster says

Table and file data — Client::write_table, Client::read_table, Client::write_file, Client::read_file, Client::upload_worker and the streaming forms of each — is what YTsaurus calls a heavy command, and a large installation serves those on a separate set of proxies. This client asks /hosts the first time it sends a heavy command, keeps the whole answer as a pool, and sends each heavy command to a member picked at random — the way both official SDKs pick, because /hosts is ordered by load and a client that keeps one pick for its lifetime never rebalances: a draining host keeps every client that ever picked it. The answer is refreshed when it outlives Client::with_host_list_refresh_interval — a minute by default, the documentation’s own advice — lazily, by the heavy command that finds it stale; there is no background thread, and a client that stops uploading stops asking. Light commands stay on the address it was configured with.

A proxy that fails is dropped from the pool, not committed to. A heavy command that fails for a reason attributable to the host it went to — a refused connection, a 503, a certificate that does not match that host’s own name — takes that host out of the pool, and the next command picks from what remains; a later refresh that still names the host puts it back. Only a pool with nobody left in it sends the client back to the configured address — and then only until it asks the cluster again, a few seconds later (Client::with_hosts_retry_after). That order matters: on a deployment with separate proxy roles the configured address is a control proxy, and going back there on the first hiccup is the failure this feature exists to prevent.

A cluster that names no heavy proxy is answered by using the configured address, which is what leaves a single-node installation working exactly as it did — asked about again one refresh interval later, so a first lookup that landed during a rolling restart is not a verdict for life. Nor is such a cluster asked in the first place when its address is on loopback: localhost is this machine’s own cluster or a tunnel to one, and the address a far-side proxy publishes for itself is not reachable from either. Client::with_proxy_discovery overrides that in both directions, and Client::heavy_proxy answers the question directly.

A discovered host is used only if it shares the configured address’s own domain, and the scheme and port come from that address rather than from the answer. That rule is a guard against a typo in a configuration and against an obviously foreign name — not a promise about where a token can end up. Steering it with a /hosts body means controlling that body, which over https:// means owning the proxy (which has the token already) and over http:// means being a man-in-the-middle (who reads it out of every light command anyway). Where the rule does bite is a proxy registering itself in the cluster’s coordinator under an unintended name, and even there it is coarse: sharing a parent domain on a hosting platform means sharing it with every other tenant of that platform. Client::with_heavy_proxies_in is the version that is a boundary — a list written out on purpose — Client::with_heavy_proxies_under names one more domain for an installation that publishes its heavy proxies in a second zone, and Client::with_heavy_proxies_anywhere removes the rule. When a whole answer is declined the client says so once, naming what it refused and why, rather than leaving it to be deduced from a cluster error later on.

Getting this wrong does not look like a routing problem, which is why it is worth spelling out what it does look like. The refusal arrives as a structured YTsaurus error — cluster error 1: Control proxy may not serve heavy requests with input data — and this crate’s own error rendering does not print the status beside it, which is how the status came to be recorded here as 200. The cluster’s own rule, from TContext::TryRedirectHeavyRequests, turns on whether the request carries input data: a heavy write gets 503 with Retry-After: 60, and a heavy read gets a 307 to a data proxy. And a deployment behind a balancer is the case that breaks, not the case that works: the balancer fronts the control proxies, so every upload arrives at one.

Re-exports§

pub use crate::error::ClientError;
pub use crate::error::RedirectRefusal;
pub use crate::error::Result;
pub use crate::lock::Lock;
pub use crate::lock::LockMode;
pub use crate::operation::Operation;
pub use crate::operation::OperationEvent;
pub use crate::operation::OperationFilter;
pub use crate::operation::OperationInfo;
pub use crate::operation::OperationList;
pub use crate::operation::OperationParameters;
pub use crate::operation::OperationStatus;
pub use crate::path::Key;
pub use crate::path::RowRange;
pub use crate::path::TablePath;
pub use crate::schema::Column;
pub use crate::schema::ColumnType;
pub use crate::schema::SortOrder;
pub use crate::schema::TableRow;
pub use crate::schema::TableSchema;
pub use crate::stream::FileReader;
pub use crate::stream::ResponseReader;
pub use crate::stream::TableReader;
pub use crate::trace::TraceContext;

Modules§

error: Errors. Errors the client can fail with.
lock: Cypress locks. Locks: what Cypress gives a transaction to coordinate with.
operation: The operation handle, and what its commands take and answer. The operation object, and the filters and parameters its commands take.
path: Table paths that carry attributes. Table paths that carry attributes.
schema: Table schemas. Table schemas: what a table promises about its columns.
stream: Streaming table I/O. Table and file I/O that does not go through memory.
trace: The trace a request belongs to. The trace a request belongs to, carried to the cluster in a traceparent header.
yson_build: Constructors for YSON documents, for specs this crate does not model. Small constructors for the YSON documents the API expects.

Structs§

BatchRequest: Commands batched to be sent in one round trip.
CachedFile: A worker binary on the cluster, as Client::upload_worker_cached left it.
Client: A connection to one YTsaurus cluster.
EraseSpec: An erase operation: rows out of one table, in place.
JobFailure: A failed job and what it printed, as carried by ClientError::OperationFailed.
JobInfo: One job of an operation, as list_jobs reports it.
MapReduceSpec: A map-reduce operation.
MapSpec: A map operation.
MergeSpec: A merge operation: several tables into one, with no user job.
MutationId: A GUID the cluster deduplicates a repeated mutation by.
ReduceSpec: A reduce operation over already-sorted input.
RemoteCopySpec: A remote-copy operation: a table from another cluster onto this one.
RetryPolicy: How often, and how patiently, a failed request is repeated.
SkiffFormat: A YTsaurus <...>skiff format declaration.
SkiffSchema: One node in a Skiff schema tree.
SortSpec: A sort operation.
Transaction: A transaction, alive for as long as this handle is.
VanillaSpec: A vanilla operation: jobs with no input tables.
VanillaTask: One task of a vanilla operation: a group of identical jobs.

Enums§

DataFormat: A supported YTsaurus data format.
MergeMode: How a merge combines its inputs.
Method: The HTTP verb a command is sent with.
OperationType: The kind of operation to start.
Repeatable: How a command may be repeated — and, for a heavy one, where it goes.
SkiffSchemaRef: A table schema placed inline in a format or referenced through its registry.
SkiffWireType: A Skiff encoding used by one schema node.

Derive Macros§

TableRow: Derives a YTsaurus table schema from a struct’s fields.