Apply Core

chaos.lib.apply

Core orchestration logic for the apply command, handling data gathering, state computation, and execution.

execute_plans(payload)

Executes all computed state operations for the roles on the target hosts using pyinfra, and gathers the results of the execution.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the pyinfra state with the computed plans for each role and host as well as any flags for telemetry.	required

gather_apply(payload)

Gather necessary data for applying roles, such as sudo password and secrets if needed.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the initial data and flags for the apply operation.	required

Returns:

Type	Description
DataGatherRequest | None	A DataGatherRequest if additional data needs to be gathered from the user, or None
ResultPayload[GatherApplyResultData | None]	A ResultPayload indicating the success or failure of the data gathering process, or None if a DataGatherRequest is returned. The ResultPayload.data["loaded_roles"] field will contain the loaded role classes based on the tags in the payload.

gather_fleet(payload, chobolo_config, chobolo_path)

Gather necessary data for fleet configuration, such as host information and parallelism settings.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the initial data and flags for the apply operation.	required
chobolo_config	DictConfig | ListConfig	the loaded chobolo configuration as a DictConfig object.	required
chobolo_path	str	the file path to the chobolo configuration file, used for error messages.	required

Returns:

Type	Description
DataGatherRequest | None	A DataGatherRequest if additional data needs to be gathered from the user, or None
ResultPayload[GatherFleetResultData | None]	A ResultPayload indicating the success or failure of the data gathering process.

Notes

Expected format in chobolo file:

fleet:
    parallelism: int (optional, default 0 for no parallelism)
    hosts:
        host1:
            param1: value1
            param2: value2

    # OPTIONAL
    boats:
        - provider: boat_provider_name
          config:
              param1: value1
              param2: value2

    # OPTIONAL
    restrictions:
        black_list:
            host1:
                role1: true
                role2: true
                # host1 wont be able to run role1 nor role2
            host2:
                role3: true
                # host2 wont be able ot run role3
        allow_list:
            host1:
                role3: true
                # host1 will ONLY be able to run role3
            host3:
                role1: true
                role4: true
                # host3 will ONLY be able to run role

get_configs(payload)

Loads global configuration from a chobolo file and validates paths for chobolo, secrets file, and sops file based on the payload and global configuration.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing any overrides for configuration paths, such as chobolo file path, secrets file path, and sops file path.	required

Returns:

Type	Description
DictConfig	A DictConfig object representing the loaded global configuration, which may include overrides from the payload.
ResultPayload[GetConfigsResultData | None]	A ResultPayload indicating the success or failure of the configuration loading and validation process, with any error messages in the error field, and the relevant configuration data (such as validated paths) in the data field if successful.

resolve_aliases(payload)

Resolves any aliases in the payload tags based on the plugin aliases and user configuration aliases, while also checking for circular references and conflicts.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the initial tags and global configuration for resolving aliases.	required

Returns:

Type	Description
ResultPayload[list[str]]	A ResultPayload indicating the success or failure of the alias resolution process, with any error messages in the error field,

resolve_allowlist_blacklist(restrictions, role_name, host)

This function resolves all blacklist and allowlist restrictions given for a role and host, and determines if the role should be ran in said host or if there are any conflicts in the configuration that should be reported as errors.

Parameters:

Name	Type	Description	Default
role_name	str	the name of the role for which to check restrictions, used for error messages.	required
host	Host	the Host object representing the target host for which to check restrictions, used for error messages.	required
restrictions	dict[str, dict[str, dict[str, bool]]]	a dictionary containing any allowlist or blacklist restrictions for roles and hosts, used to determine if the role should be applied to the host or if there are any conflicts in the configuration.	required

Returns:

Type	Description
ResultPayload[None] | None	ResultPayload[None] | None: A ResultPayload indicating the success or failure of the resolution process.

Notes

It should be called before running the get_context for a role on a host.

This dictionary should be inside of the chobolo file, and should have the following structure:

    fleet:
        restrictions:
            black_list:
                host1:
                    role1: true
                    role2: true
                host2:
                    role3: true

            allow_list:
                host1:
                    role3: true
                host3:
                    role1: true
                    role4: true

The function will check the restrictions in the following order: 1. A conflict check to see if the role is both blacklisted and allowlisted for the host, which will result in an error. 2. A check to see if the role is blacklisted for the host, which will skip the role for that host. 3. A check to see if the host is completely blacklisted (if it is host: {} and not in an allow_list), which will skip all roles for that host. 4. A check to see if the host is allowlisted but the role is not on the allowlist, which will skip it.

run_context(payload, role, host, chobolo_data)

Run the context method for a given role and host, gathering necessary configuration and secrets.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the initial data and flags for the apply operation.	required
role	Role	the Role class for which to run the context method.	required
host	Host	the Host object representing the target host for which to gather context.	required

Returns:

Type	Description
ResultPayload[dict[str, Any]]	A ResultPayload indicating the success or failure of the context gathering process, with the gathered context data if successful. The context data is inside of the ResultPayload.data field, and any error messages are in the error field.

run_delta(context, role, role_name)

Run the delta method for a given role and context, computing the necessary changes to apply the role.

Parameters:

Name	Type	Description	Default
context	dict[str, Any]	the context data gathered for the role, which should contain all necessary information for computing the delta. this should be the data returned in the ResultPayload.data field from the run_context function.	required
role	Role	the Role class for which to run the delta method.	required
role_name	str	the name of the role, used for error messages.	required

Returns:

Type	Description
ResultPayload[Delta]	A ResultPayload indicating the success or failure of the delta computation process, with any error messages in the error field.
ResultPayload[Delta]	A Delta object representing the changes that need to be applied for the role, if the computation was successful. If there was an error, this will be an empty Delta with no changes.

run_filtered_context(host, roles, payload, chobolo_config, restrictions)

run_context implementation integrated with resolve_allowlist_blacklist to filter out roles that should not be applied to the host based on the restrictions specified in the chobolo configuration.

Parameters:

Name	Type	Description	Default
host	Host	The Host object representing the target host for which to gather context.	required
roles	list[Role]	A list of Role classes that are applicable to the host based on the fleet configuration.	required
payload	ApplyPayload	The ApplyPayload containing the initial data and flags for the apply operation.	required
chobolo_config	dict[str, Any]	The entire chobolo configuration data, used for passing to the context method of the roles.	required
restrictions	dict[str, dict[str, dict[str, bool]]]	A dictionary containing any allowlist or blacklist restrictions for roles and hosts, used to determine if each role should be applied to the host or if there are any conflicts in the configuration.	required

Returns:

Type	Description
ResultPayload[FilteredContextResultData]	ResultPayload[FilteredContextResultData]: A ResultPayload indicating the success or failure of the context gathering process for the host, with the gathered context data for all applicable roles in the data field if successful, and any error messages in the error field.

Notes

It should be called before running the get_context for a role on a host.

This dictionary should be inside of the chobolo file, and should have the following structure:

    fleet:
        restrictions:
            black_list:
                host1:
                    role1: true
                    role2: true
                host2:
                    role3: true

            allow_list:
                host1:
                    role3: true
                host3:
                    role1: true
                    role4: true

run_plan(payload, delta, role, role_name, host)

Runs the plan method for a given role and host, computing the necessary operations to apply the role, while also checking against any allowlist or blacklist restrictions for the role and host.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the initial data and flags for the apply operation.	required
delta	Delta	the Delta object representing the changes that need to be applied for the role, which should be the output from the run_delta function. this should be the data returned in the ResultPayload.data field from the run_delta function.	required
role	Role	the Role class for which to run the plan method.	required
role_name	str	the name of the role, used for error messages and checking restrictions.	required
host	Host	the Host object representing the target host for which to compute the plan.	required

Returns:

Type

Description

ResultPayload[dict[Literal['plan'], ResultPayload[Any]]]

A ResultPayload indicating the success or failure of the plan computation process, with any error messages in the error field, and the computed plan in the data field if successful. The plan should be the data returned from the role.plan() method if all checks pass and the plan is computed successfully. If there are any errors or if the role is skipped due to restrictions, the plan will not be included in the data field.

setup_pyinfra(payload)

Set up the pyinfra state and inventory based on the gathered fleet configuration, and establish connections to the target hosts.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the gathered data for the apply operation, including fleet configuration and sudo password.	required

Returns:

Type	Description
ResultPayload[State | None]	ResultPayload indicating the success or failure of the pyinfra setup process, with any error messages in the error field. Additionally, if successful, the ResultPayload.data field will contain the initialized pyinfra State object that is ready for executing plans.

teardown_pyinfra(payload, run_status)

Teardown the pyinfra state and connections after the apply operation is complete.

Parameters:

Name	Type	Description	Default
payload	ApplyPayload	the ApplyPayload containing the pyinfra state to be torn down.	required

Returns:

Type	Description
ResultPayload[None]	ResultPayload indicating the success or failure of the pyinfra teardown process, with any error messages in the error field.

Notes

Should be used inside of a finally block to ensure all connections will be properly closed.