Workers, Fleets and Certificates#

Author: Vlad Poberezhny

July 9, 2025

Create a new Fleet#

Create a new Fleet in an organization you’re currently logged in to.

[ ]:

import os

from air_sdk import AirApi

api = AirApi(api_url=os.getenv('API_URL'))

# Check if a fleet already exists in your organization
fleet = next(api.fleets.list(), None)
if fleet is None:
    fleet = api.fleets.create(name='my-fleet')

# Display fleet details
fleet

Fleet(name='Free Tier')

List workers which are part of your Fleet#

[11]:

import os

from air_sdk import AirApi

api = AirApi(api_url=os.getenv('API_URL'))

# Fetch the fleet and list workers which are part of it
fleet = next(api.fleets.list())
workers = list(fleet.workers.list())

# Display the workers
workers

[11]:

[Worker(fqdn='worker01.my-domain.nvidia.com', available=False, cpu_arch='x86', ip_address='1.2.3.4')]

Add a new Worker to your Fleet#

Each Worker must reside in a specific Fleet.

[ ]:

import os

from air_sdk import AirApi

api = AirApi(api_url=os.getenv('API_URL'))

# Fetch the fleet and list workers which are part of it
fleet = next(api.fleets.list())
worker = fleet.create_worker(
    fqdn='worker02.my-domain.nvidia.com',
    ip_address='4.3.2.1',
    cpu_arch='x86',
)

# Each newly created worker instance will contain a registration token
# This token should be used on the worker machine itself with the `nvair-worker`
# binary in order to complete registration
#
# Display the registration token
worker.registration_token

'eyJ3b3JrZXJfaWQiOiJhN2IxNjg0YS1hNjFhLTQ5ZTItYmIyYi04NzBjM2UyYmJmODEiLCJhcGlfdXJsIjoiaHR0cHM6Ly8xNzIuMjcuNjguMTcxOjg0NDMifQ:1uZTHt:sPC-f-yAIXNVRHt3yXHyYL60Fw2JGuTBiWFqxtjkXow'

Update the properties of your Worker#

[14]:

import os

from air_sdk import AirApi

api = AirApi(api_url=os.getenv('API_URL'))

# Fetch the worker and update its properties
worker = next(api.workers.list(fqdn='worker02.my-domain.nvidia.com'))
worker.update(
    fqdn='worker03.my-domain.nvidia.com',
    cpu=12,
    memory=4096,
    storage=128,
    ip_address='1.2.3.5',
)

# You may also control the worker's availability
# Unavailable workers will not get simulations scheduled on them
worker.update(available=False)

worker

[14]:

Worker(fqdn='worker03.my-domain.nvidia.com', available=False, cpu_arch='x86', ip_address='1.2.3.5')

View Worker client certificates#

Worker authenticates with the API by means of a client certificate initially provided to it as part of the registration process.

[ ]:

import os

from air_sdk import AirApi

api = AirApi(api_url=os.getenv('API_URL'))

# Fetch the worker, then fetch its certificates
worker = next(api.workers.list(fqdn='worker03.my-domain.nvidia.com'))
certificates = list(api.worker_client_certificates.list(worker=worker))

# Display the certificates
certificates

[WorkerClientCertificate(worker_fqdn='worker03.my-domain.nvidia.com', expires=datetime.datetime(2026, 7, 9, 11, 49, 53, tzinfo=datetime.timezone.utc), revoked=False)]

Issue a new Worker certificate#

New worker certificate can be issued at any given time. If an existing client certificate is soon to be expired or has been revoked for any reason, a newly issued certificate can then be placed on the worker for it to be used instead of the current one.

[ ]:

import os

from air_sdk import AirApi

api = AirApi(api_url=os.getenv('API_URL'))

# Fetch the worker, then fetch its certificates
worker = next(api.workers.list(fqdn='worker03.my-domain.nvidia.com'))
certificate_data = worker.issue_certificate()
certificate, private_key = (
    certificate_data['certificate'],
    certificate_data['private_key'],
)

# Display the certificate
certificate, private_key

('<certificate PEM redacted for example>',
 '<private key PEM redacted for example>')

Revoke an existing certificate#

In case an existing certificate has been compromised, a new one should be issued and replaced, followed by revocation of the compromised certificate. You will require a certificate’s unique SHA-256 fingerprint in order to identify it at API level. If you have the certificate file, you can retrieve the fingerprint using the following command:

> openssl x509 -in </path/to/certificate.pem> -noout -fingerprint -sha256
SHA256 Fingerprint=CA:33:DE:02:66:03:59:47:BB:62:6F:DF:9A:49:30:32:F3:5B:AB:4D:23:31:6E:02:3C:4F:48:8D:DF:E0:32:C3

Once a certificate is revoked, it can no longer be used for worker authentication.

[ ]:

# ruff: noqa: E501
import os

from air_sdk import AirApi

# Colons have to be stripped from the fingerprint
FINGERPRINT = 'CA:33:DE:02:66:03:59:47:BB:62:6F:DF:9A:49:30:32:F3:5B:AB:4D:23:31:6E:02:3C:4F:48:8D:DF:E0:32:C3'.replace(
    ':', ''
)

api = AirApi(api_url=os.getenv('API_URL'))

# Fetch the worker, then fetch its certificates
worker = next(api.workers.list(fqdn='worker03.my-domain.nvidia.com'))
certificates = api.worker_client_certificates.list(worker=worker)
for certificate in certificates:
    # Use a case-insensitive comparison to match the fingerprint
    if certificate.fingerprint.casefold() == FINGERPRINT.casefold():
        certificate.revoke()

# Display the certificate post-revocation
certificate

WorkerClientCertificate(worker_fqdn='worker03.my-domain.nvidia.com', expires=datetime.datetime(2026, 7, 9, 11, 53, 22, tzinfo=datetime.timezone.utc), revoked=True)