Configuration API¶

High-Level API¶

High-level API for SKA SDP configuration.

class ska_sdp_config.config.Config(backend=None, global_prefix='', owner=None, **cargs)[source]¶

Connection to SKA SDP configuration.

property backend¶: Get the backend database object.

property client_lease¶

Return the lease associated with the client.

It will be kept alive until the client gets closed.

close()[source]¶: Close the client connection.

lease(ttl=10)[source]¶

Generate a new lease.

Once entered can be associated with keys, which will be kept alive until the end of the lease. At that point a daemon thread will be started automatically to refresh the lease periodically (default seems to be TTL/4).

Parameters: ttl – Time to live for lease
Returns: lease object

txn(max_retries=64)[source]¶

Create a Transaction for atomic configuration query/change.

As we do not use locks, transactions might have to be repeated in order to guarantee atomicity. Suggested usage is as follows:

for txn in config.txn():
    # Use txn to read+write configuration
    # [Possibly call txn.loop()]

As the for loop suggests, the code might get run multiple times even if not forced by calling Transaction.loop(). Any writes using the transaction will be discarded if the transaction fails, but the application must make sure that the loop body has no other observable side effects.

Parameters: max_retries – Number of transaction retries before a RuntimeError gets raised.

class ska_sdp_config.config.Transaction(config, txn)[source]¶

High-level configuration queries and updates to execute atomically.

create_deployment(dpl: ska_sdp_config.entity.deployment.Deployment)[source]¶

Request a change to cluster configuration.

Parameters: dpl – Deployment to add to database

create_processing_block(pb: ska_sdp_config.entity.pb.ProcessingBlock)[source]¶

Add a new ProcessingBlock to the configuration.

Parameters: pb – Processing block to create

create_processing_block_state(pb_id: str, state: dict)[source]¶

Create processing block state.

Parameters

pb_id – Processing block ID
state – Processing block state to create

create_scheduling_block(sb_id: str, state: dict)[source]¶

Create scheduling block.

Parameters

sb_id – scheduling block ID
state – scheduling block state

create_subarray(subarray_id: str, state: dict)[source]¶

Create subarray.

Parameters

subarray_id – subarray ID
state – subarray state

delete_deployment(dpl: ska_sdp_config.entity.deployment.Deployment)[source]¶

Undo a change to cluster configuration.

Parameters: dpl – Deployment to remove

get_deployment(deploy_id: str) → ska_sdp_config.entity.deployment.Deployment[source]¶

Retrieve details about a cluster configuration change.

Parameters: deploy_id – Name of the deployment
Returns: Deployment details

get_processing_block(pb_id: str) → ska_sdp_config.entity.pb.ProcessingBlock[source]¶

Look up processing block data.

Parameters: pb_id – Processing block ID to look up
Returns: Processing block entity, or None if it doesn’t exist

get_processing_block_owner(pb_id: str) → dict[source]¶

Look up the current processing block owner.

Parameters: pb_id – Processing block ID to look up
Returns: Processing block owner data, or None if not claimed

get_processing_block_state(pb_id: str) → dict[source]¶

Get the current processing block state.

Parameters: pb_id – Processing block ID
Returns: Processing block state, or None if not present

get_scheduling_block(sb_id: str) → dict[source]¶

Get scheduling block.

Parameters: sb_id – scheduling block ID
Returns: scheduling block state

get_subarray(subarray_id: str) → dict[source]¶

Get subarray.

Parameters: subarray_id – subarray ID
Returns: subarray state

is_processing_block_owner(pb_id: str) → bool[source]¶

Check whether this client is owner of the processing block.

Parameters: pb_id – Processing block ID to look up
Returns: Whether processing block exists and is claimed

list_deployments(prefix='')[source]¶

List all current deployments.

Returns: Deployment IDs

list_processing_blocks(prefix='')[source]¶

Query processing block IDs from the configuration.

Parameters: prefix – If given, only search for processing block IDs with the given prefix
Returns: Processing block ids, in lexicographical order

list_scheduling_blocks(prefix='')[source]¶

Query scheduling block IDs from the configuration.

Parameters: prefix – if given, only search for scheduling block IDs with the given prefix
Returns: scheduling block IDs, in lexicographical order

list_subarrays(prefix='')[source]¶

Query subarray IDs from the configuration.

Parameters: prefix – if given, only search for subarray IDs with the given prefix
Returns: subarray IDs, in lexicographical order

loop(wait=False, timeout=None)[source]¶

Repeat transaction regardless of whether commit succeeds.

Parameters

wait – If transaction succeeded, wait for any read values to change before repeating it.
timeout – Maximum time to wait, in seconds

new_processing_block_id(generator: str)[source]¶

Generate a new processing block ID that is not yet in use.

Parameters: generator – Name of the generator
Returns: Processing block ID

property raw¶: Return transaction object for accessing database directly.

take_processing_block(pb_id: str, lease)[source]¶

Take ownership of the processing block.

Parameters: pb_id – Processing block ID to take ownership of
Raises: backend.ConfigCollision

update_processing_block(pb: ska_sdp_config.entity.pb.ProcessingBlock)[source]¶

Update a ProcessingBlock in the configuration.

Parameters: pb – Processing block to update

update_processing_block_state(pb_id: str, state: dict)[source]¶

Update processing block state.

Parameters

pb_id – Processing block ID
state – Processing block state to update

update_scheduling_block(sb_id: str, state: dict)[source]¶

Update scheduling block.

Parameters

sb_id – scheduling block ID
state – scheduling block state

update_subarray(subarray_id: str, state: dict)[source]¶

Update subarray.

Parameters

subarray_id – subarray ID
state – subarray state

class ska_sdp_config.config.TransactionFactory(config, txn)[source]¶: Helper object for making transactions.

ska_sdp_config.config.dict_to_json(obj)[source]¶

Format a dictionary for writing it into the database.

Parameters: obj – Dictionary object to format
Returns: String representation

Entities¶

Processing block configuration entities.

class ska_sdp_config.entity.pb.ProcessingBlock(id, sbi_id, workflow, parameters={}, dependencies=[], **kwargs)[source]¶

Processing block entity.

Collects configuration information relating to a processing job for the SDP. This might be either real-time (supporting a running observation) or batch (to process data after the fact).

Actual execution of processing steps will be performed by a (parameterised) workflow interpreting processing block information.

property dependencies¶: Return dependencies on other processing blocks.

property id¶: Return the processing block ID.

property parameters¶: Return workflow-specific parameters.

property sbi_id¶: Return scheduling block instance ID, if associated with one.

to_dict()[source]¶: Return data as dictionary.

property workflow¶: Return information identifying the workflow.

Backend¶

Backends for SKA SDP configuration DB.