buildgrid.server.cas.logstream.stream_storage.stream_storage_abc module

StreamStorageABC

The abstract base class for stream storage providers.

class buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamHandle(name, write_resource_name)

Bases: tuple

property name

Alias for field number 0

property write_resource_name

Alias for field number 1

class buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamLength(length, finished)

Bases: tuple

property length

Alias for field number 0

property finished

Alias for field number 1

class buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamChunk(data, chunk_length, finished)

Bases: tuple

property data

Alias for field number 0

property chunk_length

Alias for field number 1

property finished

Alias for field number 2

class buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamStorageABC

Bases: abc.ABC

create_stream(parent: str, prefix: str)buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamHandle

Creates a stream for a specific parent with a specific prefix and a random UUID, and returns a StreamHandle namedtuple containing:

name (str): The name of the string for read access write_resource_name (str): The secret name used only for writing

to the stream

abstract classmethod stream_exists(stream_name: str) → bool

Returns True when there is a stream with the given instance/stream_name.

abstract classmethod stream_finished(stream_name: str) → bool

Returns True/False depending on whether the stream has been marked

as completed.

Raises:

NotFoundError when a stream with the given instance/stream_name

does not exist.

abstract classmethod stream_length(stream_name: str)buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamLength

Returns a namedtuple of type StreamLength for the stream with the given instance/stream_name.

length (int): The length of the stream in bytes finished (bool): A boolean indicating whether the stream has finished

Raises

NotFoundError when a stream with the given instance/stream_name – does not exist.

writeable_stream_length(write_resource_name: str)buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamLength

Returns a namedtuple of type StreamLength for the stream with the given instance/write_resource_name.

length (int): The length of the stream in bytes finished (bool): A boolean indicating whether the stream has finished

Raises

NotFoundError when a stream with the given instance/stream_name – does not exist.

append_to_stream(write_resource_name: str, message: Optional[bytes] = None, *, mark_finished: Optional[bool] = False) → None

Appends the message to the stream with the passed write_resource_name

as long as the stream has not been marked as finished yet and marks the stream as finished when the relevant arg is set to True.

Raises:

NotFoundError when a stream with the given instance/write_resource_name

does not exist.

StreamFinishedError when the stream was already marked as finished and

no further writes are allowed.

StorageFullError when the backing storage is full. WriteError when the write_resource_name is correct but

the write failed for other reasons.

abstract classmethod read_stream_chunk_blocking(stream_name: str, read_offset_bytes: int, read_limit_bytes: Optional[int] = None)buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamChunk
Returns a commited chunk of the stream message or waits until it is available as follows:

chunk = message[read_offset_bytes, min(read_limit_bytes, finished_stream_length) )

When the read_limit_bytes argument is not set, this method will return the whole message starting from the specified offset, blocking until the stream finishes.

In cases in which the stream has not finished and the chunk offset and size requested was not commited yet, this method will block and wait until the chunk is commited and/or the stream is finished, returning the appropriate chunk.

Raises

NotFoundError when a stream with the given instance/stream_name

does not exist.

OutOfRangeError when a finished stream does not contain a chunk of data

at the specified offset.

abstract classmethod read_stream_chunk(stream_name: str, read_offset_bytes: int, read_limit_bytes: Optional[int] = None)buildgrid.server.cas.logstream.stream_storage.stream_storage_abc.StreamChunk
Returns a commited chunk of the stream message or raises if it is not available as follows:

chunk = message[read_offset_bytes, min(read_limit_bytes, finished_stream_length) )

When the read_limit_bytes argument is not set, this method will return the whole message starting from the specified offest, raising if the stream hasn’t finished.

Raises

NotFoundError when a stream with the given instance/stream_name

does not exist.

StreamWritePendingError when the requested chunk is not fully commited. OutOfRangeError when a finished stream does not contain a chunk of data

at the specified offset.

read_stream_bytes_blocking_iterator(stream_name: str, max_chunk_size: int, offset: int = 0) → Iterator[bytes]

An iterator returning commited chunks of up to chunk_size until the end of the stream. Blocks and waits until the stream finishes.

Can optionally start at a specific chunk_offset, which starts streaming data from

(stream_offset = chunk_offset * chunk_size)

Raises

NotFoundError when a stream with the given instance/stream_name

does not exist.

OutOfRangeError when a finished stream does not contain a chunk of data

at the specified offset.

new_client_streaming(stream_name: str)

Inform the StreamStorage backend that a new client is streaming the stream stream_name in case it cares about the number of clients streaming a specific stream. This can be useful for e.g. cleaning up old streams. No-op by default. :raises NotFoundError when a stream with the given instance/stream_name:

does not exist.

streaming_client_left(stream_name: str)

Inform the StreamStorage backend that a client streaming the stream stream_name has left, in case it cares about the number of clients streaming a specific stream. This can be useful for e.g. cleaning up old streams. No-op by default. :raises NotFoundError when a stream with the given instance/stream_name:

does not/no longer exist(s).

wait_for_streaming_clients(write_resource_name: str, timeout: int = 300) → bool

Returns a boolean specifying if at least one client is streaming this stream within the specified timeout.

This allows the writers to only start writing when there are readers interested in the stream. Note: Accepts write_resource_name and calls

_wait_for_streaming_clients(read_name) on the backend.

Raises

NotFoundError when a stream with the given instance/stream_name – does not exist.

abstract classmethod delete_stream(write_resource_name)
set_instance_name(instance_name: str) → None