Python API

This page documents the triliositerecovery Python client library for Trilio Site Recovery for OpenStack. The library provides programmatic access to all DR operations — site registration, protection group lifecycle, failover, failback, test failover, replication health checks, and metadata synchronization — through a manager-based interface backed by the service's REST API on port 8788. Use it when you need to integrate DR workflows into automation scripts, CI/CD pipelines, or operational tooling that cannot use the openstack dr CLI directly. The client is an OpenStackClient plugin: it is imported from triliositerecovery.v1.client and requires a keystoneauth1 Adapter session rather than raw credentials.

Before using the Python client you need:

• Python 3.10 or later (Ubuntu Jammy or Noble)
• The python3-triliositerecovery Debian package installed on the machine running your script (see Installation)
• A working OpenStack environment with Keystone token authentication configured
• keystoneauth1 installed (it is a dependency of python3-triliositerecovery and will be present after package installation)
• The Trilio Site Recovery service registered in the Keystone catalogue — the client discovers the endpoint via the protector service type
• At minimum the dr_operator role granted to the project your script authenticates as, or dr_site_admin for site-wide operations such as site registration and tenant mapping
• Network access from the machine running your script to the protector API on port 8788

The Python client is shipped in the python3-triliositerecovery Debian package from the Trilio apt repository. Install it on any machine that needs programmatic access to the service.

1. Configure the Trilio apt repository (if not already present):

echo 'deb [trusted=yes] https://apt.fury.io/trilio-os-site-recovery-master/ /' \
  | sudo tee /etc/apt/sources.list.d/trilio-site-recovery.list
sudo apt-get update

2. Install the client package:

sudo apt-get install python3-triliositerecovery

This installs the triliositerecovery Python package under /usr/lib/python3/dist-packages/triliositerecovery and registers the openstack dr CLI plugin.

3. Verify the installation:

openstack dr site list

If the command returns a table (or an empty table) without error, the package and its OpenStack plugin are correctly installed.

On Juju-managed nodes: The charm installs python3-triliositerecovery on the service unit automatically. On any other machine (a bastion, a CI runner, or a client workstation) you must install the package separately with the steps above. The upgrade-packages charm action upgrades only the charm's own unit; run sudo apt-get install --only-upgrade python3-triliositerecovery on each additional machine after a service upgrade.

The Python client is configured entirely at construction time — there is no client-side configuration file. The three parameters that matter are:

Endpoint discovery

The OSC plugin (osc_lib) discovers the endpoint automatically from the Keystone catalogue using the protector service type and the configured interface and region. When constructing the client directly (outside OSC), you must supply the endpoint yourself. Use keystoneauth1's get_endpoint to discover it:

from keystoneauth1 import identity, session as ks_session
from keystoneauth1.adapter import Adapter
from triliositerecovery.v1.client import Client

auth = identity.v3.Password(
    auth_url='https://keystone.primary.example.com:5000/v3',
    username='druser',
    password='secret',
    project_name='my-project',
    user_domain_name='Default',
    project_domain_name='Default',
)
ksession = ks_session.Session(auth=auth)

# Discover the endpoint from the service catalogue
endpoint = ksession.get_endpoint(
    service_type='protector',
    interface='internal',
    region_name='RegionOne',
)

adapter = Adapter(
    session=ksession,
    endpoint_override=endpoint,
    region_name='RegionOne',
)

client = Client(session=adapter, endpoint=endpoint, region_name='RegionOne')

endpoint_override vs endpoint: Pass the discovered URL as endpoint_override on the Adapter (which pins the base URL for that adapter's requests) and as endpoint on Client (which Client.request uses to build f"{self.endpoint}/{url}"). The Adapter constructor does not accept endpoint_url — use endpoint_override.

Return shapes

Every manager method returns parsed JSON — plain Python dicts and lists, not resource objects. Use dict-style access (site['id'], not site.id); attribute access raises AttributeError.

Mapping arguments

At the CLI layer, network and flavor mappings use source-id=target-id syntax because argparse parses that string into a dict. At the library layer, pass a dict directly — there is no separator:

# Correct
client.operations.failover(pg_id, network_mapping={'src-net-id': 'dst-net-id'})

# Wrong — the API receives a list and raises TypeError
client.operations.failover(pg_id, network_mapping=['src-net-id=dst-net-id'])

Positional resource IDs

Manager methods that act on a specific resource take the resource ID positionally. The signatures use **kwargs for optional parameters, so passing the ID as a keyword argument fails with "missing 1 required positional argument":

# Correct
client.operations.failover(pg_id, failover_type='planned')

# Wrong — KeyError at runtime
client.operations.failover(protection_group_id=pg_id, failover_type='planned')

Constructing the client

Import from triliositerecovery.v1.client. The OSC plugin uses triliositerecovery.v1.client.Client internally; you use the same class directly:

from keystoneauth1 import identity, session as ks_session
from keystoneauth1.adapter import Adapter
from triliositerecovery.v1.client import Client

auth = identity.v3.Password(
    auth_url='https://keystone.primary.example.com:5000/v3',
    username='druser',
    password='secret',
    project_name='my-project',
    user_domain_name='Default',
    project_domain_name='Default',
)
ksession = ks_session.Session(auth=auth)
endpoint = ksession.get_endpoint(
    service_type='protector',
    interface='internal',
    region_name='RegionOne',
)
adapter = Adapter(
    session=ksession,
    endpoint_override=endpoint,
    region_name='RegionOne',
)
client = Client(session=adapter, endpoint=endpoint, region_name='RegionOne')

Available managers

The Client object exposes these managers:

Listing sites

sites = client.sites.list()
for site in sites:
    print(site['id'], site['name'])

Returns a list of dicts. Use site['id'], not site.id.

Creating a protection group

Both --primary-site and --secondary-site are required. The API never infers the secondary site:

pg = client.protection_groups.create(
    name='web-tier-pg',
    primary_site_id='<primary-site-uuid>',
    secondary_site_id='<secondary-site-uuid>',
    replication_type='async',
    volume_type='ceph-replicated',
)
pg_id = pg['id']

volume_type is required when replication_type is async (the default). Omitting it returns HTTP 400.

Adding a member

add_member takes pg_id and instance_id positionally:

member = client.protection_groups.add_member(
    pg_id,
    '<nova-instance-uuid>',
    boot_order=1,
)

Checking replication health

health = client.protection_groups.replication_health(pg_id)
print(health)

Pass probe=True to also query the recovery site's Neutron and Nova, verifying that the networks, security groups, and flavors the protection group needs actually resolve there as the project it recovers into. This is not a storage check and does not report replication lag — it is a resource-resolution check that catches resources belonging to the wrong project on the recovery site:

health = client.protection_groups.replication_health(pg_id, probe=True)

probe=True is slower because it contacts the peer site; run it before a planned failover window, not in a tight polling loop.

Listing recovery points

result = client.protection_groups.recovery_points(pg_id, limit=10)

Executing a planned failover

Mappings are dicts, not strings:

operation = client.operations.failover(
    pg_id,
    failover_type='planned',
    network_mapping={'primary-net-id': 'secondary-net-id'},
    flavor_mapping={'m1.large': 'r5.large'},
)
op_id = operation['id']

Polling an operation to completion

The service does not provide estimated completion times. Poll the progress field (0–100) to track progress:

import time

while True:
    op = client.operations.get(op_id)
    print(f"status={op['status']} progress={op.get('progress', 0)}%")
    if op['status'] in ('completed', 'failed', 'error'):
        break
    time.sleep(15)

Executing a test failover and cleaning up

# Start test failover
operation = client.operations.test_failover(
    pg_id,
    recovery_point='<rp-name>',
    network_mapping={'primary-net-id': 'isolated-test-net-id'},
)

# ... validate test VMs ...

# Clean up — method is cleanup_test_failover, not test_failover_cleanup
client.operations.cleanup_test_failover(pg_id)

Executing a failback

operation = client.operations.failback(
    pg_id,
    failback_type='planned',
)

Failback waits up to 120 seconds per volume for Nova to release the Cinder attachment before unmanaging the volume. If a volume is still attached at the deadline it is unmanaged anyway and logged — failback is not aborted by a slow detach, but a site with lagging detaches will show a visibly slower failback.

Checking replication audit

The replication audit reports volumes that Pure Storage is replicating but that belong to no protection group. These volumes are not protected — they have no recovery points and cannot be failed over — but they consume replication capacity:

audit = client.sites.replication_audit()
print(audit)

Site-level failover (dr_site_admin only)

Fails over all protection groups at a site simultaneously. Requires the dr_site_admin role — grant this only to cloud operators on the admin project, never to tenant DR users:

site_op = client.sites.failover(
    site_id='<site-uuid>',
    failover_type='unplanned',
    force=True,
)

The request body key is failover (not failover_to_secondary). The API endpoint is POST /v1/{tenant_id}/sites/{id}/action.

Observability

The service emits no oslo.messaging notifications. Build observability by polling:

• client.protection_groups.replication_health(pg_id) — replication status and RPO compliance
• client.protection_groups.replication_health(pg_id, probe=True) — resource resolution on the recovery site
• client.sites.replication_audit() — untracked replicated volumes
• client.protection_groups.recovery_points(pg_id) — available recovery points
• client.operations.list(protection_group_id=pg_id) — operation history and current status

Example 1 — Discover the local site and list all protection groups

from keystoneauth1 import identity, session as ks_session
from keystoneauth1.adapter import Adapter
from triliositerecovery.v1.client import Client

auth = identity.v3.Password(
    auth_url='https://keystone.primary.example.com:5000/v3',
    username='druser',
    password='secret',
    project_name='my-project',
    user_domain_name='Default',
    project_domain_name='Default',
)
ksession = ks_session.Session(auth=auth)
endpoint = ksession.get_endpoint(
    service_type='protector',
    interface='internal',
    region_name='RegionOne',
)
adapter = Adapter(session=ksession, endpoint_override=endpoint, region_name='RegionOne')
client = Client(session=adapter, endpoint=endpoint, region_name='RegionOne')

local_site = client.sites.local()
print('Local site:', local_site['name'] if local_site else 'not configured')

groups = client.protection_groups.list()
for pg in groups:
    print(pg['id'], pg['name'], pg['status'])

Example 2 — Create a protection group and add members with boot order

# Identify sites
sites = client.sites.list()
site_map = {s['name']: s['id'] for s in sites}

primary_site_id = site_map['primary-site']
secondary_site_id = site_map['secondary-site']

# Create the protection group
pg = client.protection_groups.create(
    name='web-tier-pg',
    primary_site_id=primary_site_id,
    secondary_site_id=secondary_site_id,
    replication_type='async',
    volume_type='ceph-replicated',  # required for async
)
pg_id = pg['id']
print('Created PG:', pg_id)

# Add a database VM first (boots first, stops last)
db_member = client.protection_groups.add_member(
    pg_id,
    '<db-instance-uuid>',
    boot_order=1,
)

# Add an application VM (boots after DB)
app_member = client.protection_groups.add_member(
    pg_id,
    '<app-instance-uuid>',
    boot_order=2,
)
print('Members added:', db_member['id'], app_member['id'])

Example 3 — Validate a member before adding it

Use validate_member to run all pre-add checks without committing the change. This surfaces issues such as LUKS-encrypted volumes (which the service rejects) or missing Cinder boot volumes before you attempt add_member:

result = client.protection_groups.validate_member(pg_id, '<instance-uuid>')
if result.get('valid'):
    client.protection_groups.add_member(pg_id, '<instance-uuid>')
else:
    for err in result.get('errors', []):
        print('Validation error:', err)

Example 4 — Check replication health and list recovery points before a failover window

# Basic health (local data only, fast)
health = client.protection_groups.replication_health(pg_id)
print('Replication health:', health)

# Probe health (queries the recovery site — slower, but validates resource resolution)
probed = client.protection_groups.replication_health(pg_id, probe=True)
print('Probed health:', probed)

# List the 5 most recent recovery points
rp_result = client.protection_groups.recovery_points(pg_id, limit=5)
print('Recovery points:', rp_result)

Example 5 — Execute a planned failover with resource mappings and poll to completion

import time

# Mappings are dicts at the library layer — no '=' separator
operation = client.operations.failover(
    pg_id,
    failover_type='planned',
    recovery_point='<rp-name>',
    network_mapping={'primary-net-uuid': 'secondary-net-uuid'},
    subnet_mapping={'primary-subnet-uuid': 'secondary-subnet-uuid'},
    flavor_mapping={'m1.large': 'r5.large'},
)
op_id = operation['id']
print('Failover started, operation:', op_id)

# Poll — no estimated completion time is available; use progress (0-100)
while True:
    op = client.operations.get(op_id)
    status = op['status']
    progress = op.get('progress', 0)
    source_site = op.get('source_site', op.get('source_site_id', ''))
    target_site = op.get('target_site', op.get('target_site_id', ''))
    print(f'  [{source_site} -> {target_site}] status={status} progress={progress}%')
    if status in ('completed', 'failed', 'error'):
        break
    time.sleep(15)

if status == 'completed':
    print('Failover complete. Recovered instances have the suffix "-failover".')
else:
    print('Failover did not complete cleanly, status:', status)

Instance naming after failover: The engine creates recovered instances as <original-name>-failover. After a subsequent failback, instances are named <original-name>-restored. The original name is preserved in the member record but the running instance name changes — update any DNS, monitoring, or CMDB records that key on the instance name.

Example 6 — Run a test failover and clean up

# Start test failover into an isolated network
test_op = client.operations.test_failover(
    pg_id,
    recovery_point='<rp-name>',
    network_mapping={'primary-net-uuid': 'isolated-test-net-uuid'},
)
test_op_id = test_op['id']
print('Test failover started:', test_op_id)

# Poll until complete (same pattern as above)
while True:
    op = client.operations.get(test_op_id)
    if op['status'] in ('completed', 'failed', 'error'):
        break
    time.sleep(15)

print('Test result:', op['status'])

# Clean up test resources
# The method is named cleanup_test_failover, not test_failover_cleanup
client.operations.cleanup_test_failover(pg_id)
print('Test failover cleaned up')

Example 7 — Create and manage a tenant mapping

Tenant mappings are required when primary and secondary sites have independent Keystone deployments (different project UUIDs for the same logical tenant). Without a mapping the engine assumes the project UUID is identical on both sites — this holds for shared or federated Keystone but essentially never for two independent clusters.

# Create a tenant mapping (requires dr_site_admin)
mapping = client.tenant_mappings.create(
    local_tenant_id='<project-uuid-on-primary>',
    remote_site_id='<secondary-site-uuid>',
    remote_tenant_id='<project-uuid-on-secondary>',
)
print('Mapping created:', mapping['id'])

# Validate it (checks connectivity and role grants on the remote side)
validation = client.tenant_mappings.validate(mapping['id'])
print('Validation result:', validation)

Example 8 — Site-level failover (big red button, dr_site_admin only)

# Fails over ALL protection groups at the named site
site_op = client.sites.failover(
    site_id='<primary-site-uuid>',
    failover_type='unplanned',
    force=True,
)
print('Site operation:', site_op['id'])

# Monitor child operations
site_op_detail = client.site_operations.get(site_op['id'])
for child in site_op_detail.get('child_operations', []):
    print(' PG:', child.get('protection_group_id'), 'status:', child.get('status'))

Example 9 — Replication audit (spot untracked replicated volumes)

The audit reports volumes Pure Storage is replicating that no protection group accounts for. These volumes have no recovery points and cannot be failed over, but they consume replication capacity:

audit = client.sites.replication_audit()
print('Replication audit:', audit)

Use a consistent approach for each issue: identify the symptom, diagnose the cause, then apply the fix.

AttributeError: 'dict' object has no attribute 'id'

Symptom: Your script raises AttributeError on the first iteration when accessing a field on a returned object.

Cause: All manager methods return plain dicts, not resource objects. The library does not wrap responses in model classes.

Fix: Use dict-style access throughout:

# Wrong
for pg in client.protection_groups.list():
    print(pg.id)  # AttributeError

# Correct
for pg in client.protection_groups.list():
    print(pg['id'])

TypeError: failover() missing 1 required positional argument: 'pg_id'

Symptom: Calling a manager method with the resource ID as a keyword argument raises TypeError.

Cause: Manager methods take the resource ID positionally. **kwargs swallows keyword arguments silently before the type error surfaces at dispatch.

Fix: Pass the ID as the first positional argument:

# Wrong
client.operations.failover(protection_group_id=pg_id)

# Correct
client.operations.failover(pg_id, failover_type='planned')

Requests fail immediately with a URL starting None/

Symptom: Every API call raises a connection error and the URL in the traceback begins with None/path.

Cause: The endpoint parameter was not passed to Client(). Client.request builds f"{self.endpoint}/{url}", so a missing endpoint produces literal None/ URLs.

Fix: Always supply both endpoint_override on the Adapter and endpoint on Client:

endpoint = ksession.get_endpoint(
    service_type='protector',
    interface='internal',
    region_name='RegionOne',
)
adapter = Adapter(session=ksession, endpoint_override=endpoint, region_name='RegionOne')
client = Client(session=adapter, endpoint=endpoint, region_name='RegionOne')

HTTP 400 when creating a protection group

Symptom: client.protection_groups.create(...) returns or raises HTTP 400.

Cause: The volume_type argument is required when replication_type is async (the default). Omitting it produces HTTP 400 from the server.

Fix: Always supply volume_type for async protection groups:

pg = client.protection_groups.create(
    name='my-pg',
    primary_site_id=primary_id,
    secondary_site_id=secondary_id,
    replication_type='async',
    volume_type='ceph-replicated',  # required
)

Network mapping is silently ignored during failover

Symptom: The failover completes but VMs are attached to the wrong network. The mapping you supplied appears to have had no effect.

Cause: You passed the mapping as a string ('src-id=dst-id') or a list instead of a dict. The CLI parses src=dst into a dict before calling the library; the library itself passes kwargs directly to the wire. A string or list causes a TypeError in the API's mapping logic or is silently dropped.

Fix: Always pass mappings as dicts:

# Wrong — the '=' separator is argparse syntax, not library syntax
client.operations.failover(pg_id, network_mapping='src-net-id=dst-net-id')

# Correct
client.operations.failover(pg_id, network_mapping={'src-net-id': 'dst-net-id'})

The supported mapping types are network, subnet, security_group, and flavor. floating_ip_pool is not a supported mapping type.

HTTP 403 Policy doesn't allow volume_extension:volume_manage

Symptom: Failover fails partway through — storage work succeeds but volume import fails with HTTP 403 from Cinder.

Cause: The Cinder policy on the recovery site has not been relaxed to allow tenant-scoped access to the volume manage/unmanage APIs. The service runs as a tenant through a Keystone trust during cross-site operations. The default Cinder policy restricts these to admin.

Fix: Apply the required Cinder policy relaxation on both sites. The five rules that must be present are:

"volume_extension:volume_manage": "rule:admin_or_owner"
"volume_extension:volume_unmanage": "rule:admin_or_owner"
"volume_extension:services:index": "rule:admin_or_owner"
"volume_extension:type_get": "rule:admin_or_owner"
"volume_extension:type_get_all": "rule:admin_or_owner"

On Juju deployments, apply this as a policyd override using the resource name policyd-override (not policyd).

Trust creation fails with a role error

Symptom: The first DR operation after deployment fails with an error naming a missing Keystone role.

Cause: The default_trust_roles setting (default: member,dr_operator) names roles that are delegated by every per-tenant Keystone trust. If either role does not exist in Keystone on the site where the trust is being created, trust creation fails immediately naming the missing role.

Fix: Ensure both dr_operator and member roles exist on both sites before any DR operation is attempted. The charm creates them at deployment time via keystone-requested-roles; on manually deployed sites create them with:

openstack role create dr_operator
openstack role create member  # usually already exists

Do not modify default_trust_roles in protector.conf on charm-managed deployments — the charm manages the configuration file and will overwrite manual edits.

cleanup_test_failover raises AttributeError: 'OperationManager' object has no attribute 'test_failover_cleanup'

Symptom: Your cleanup call raises AttributeError.

Cause: The method is named cleanup_test_failover, not test_failover_cleanup.

Fix:

# Wrong
client.operations.test_failover_cleanup(pg_id)

# Correct
client.operations.cleanup_test_failover(pg_id)

No events received from oslo.messaging consumer

Symptom: You built a notification consumer subscribing to a protector-notifications topic but receive nothing through failovers and operations.

Cause: Trilio Site Recovery emits no oslo.messaging notifications. The service constructs a Notifier object internally but get_notifier() has no callers and no events are ever published.

Fix: Replace notification-based observability with polling:

• client.protection_groups.replication_health(pg_id) — replication status
• client.protection_groups.replication_health(pg_id, probe=True) — resource resolution on the recovery site
• client.protection_groups.recovery_points(pg_id) — available recovery points
• client.operations.list(protection_group_id=pg_id) — operation history and progress
• client.sites.replication_audit() — untracked replicated volumes

HTTP 409 InvalidState on a protection group action

Symptom: A failover, failback, or test failover call returns HTTP 409 with an InvalidState message.

Cause: The protection group is in a state that does not permit the requested operation (for example, attempting failover on a group that is already in failed_over state, or attempting failback on a group still in active state).

Fix: Check the current state with client.protection_groups.get(pg_id) and confirm it is valid for the intended operation before calling. HTTP 409 is returned for InvalidState on every action endpoint — it is not 400.