Add VM inventory CSV export support

Adds a `PROVIDER_TYPE_ENDPOINT_INVENTORY_EXPORT` capability type that
supports exporting VM inventory as CSV.

Signed-off-by: Mihaela Balutoiu <mbalutoiu@cloudbasesolutions.com>

coriolis/api/v1/endpoint_inventory.py

@@ -0,0 +1,39 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from coriolis.api import wsgi as api_wsgi
+from coriolis.endpoint_resources import api
+from coriolis.policies import endpoints as endpoint_policies
+from coriolis import utils
+
+from oslo_log import log as logging
+
+LOG = logging.getLogger(__name__)
+
+
+class EndpointInventoryController(api_wsgi.Controller):
+    """Returns a VM inventory CSV for endpoints that support it."""
+
+    def __init__(self):
+        self._endpoint_resources_api = api.API()
+        super(EndpointInventoryController, self).__init__()
+
+    def index(self, req, endpoint_id):
+        context = req.environ['coriolis.context']
+        context.can("%s:export_inventory" % (
+            endpoint_policies.ENDPOINTS_POLICY_PREFIX))
+
+        env = req.GET.get("env")
+        if env is not None:
+            env = utils.decode_base64_param(env, is_json=True)
+        else:
+            env = {}
+
+        csv_content = self._endpoint_resources_api.get_endpoint_inventory_csv(
+            context, endpoint_id, env)
+
+        return api_wsgi.ResponseObject(csv_content)
+
+
+def create_resource():
+    return api_wsgi.Resource(EndpointInventoryController())

coriolis/api/v1/router.py

@@ -11,6 +11,7 @@ from coriolis.api.v1 import endpoint_actions
 from coriolis.api.v1 import endpoint_destination_minion_pool_options
 from coriolis.api.v1 import endpoint_destination_options
 from coriolis.api.v1 import endpoint_instances
+from coriolis.api.v1 import endpoint_inventory
 from coriolis.api.v1 import endpoint_networks
 from coriolis.api.v1 import endpoint_source_minion_pool_options
 from coriolis.api.v1 import endpoint_source_options
@@ -109,6 +110,11 @@ class APIRouter(api.APIRouter):
         mapper.resource('instance', 'endpoints/{endpoint_id}/instances',
                         controller=self.resources['endpoint_instances'])
 
+        self.resources['endpoint_inventory'] = \
+            endpoint_inventory.create_resource()
+        mapper.resource('inventory', 'endpoints/{endpoint_id}/inventory',
+                        controller=self.resources['endpoint_inventory'])
+
         self.resources['endpoint_networks'] = \
             endpoint_networks.create_resource()
         mapper.resource('network', 'endpoints/{endpoint_id}/networks',

coriolis/conductor/rpc/client.py

@@ -102,6 +102,13 @@ class ConductorClient(rpc.BaseRPCClient):
             endpoint_id=endpoint_id,
             env=env)
 
+    def get_endpoint_inventory_csv(
+            self, ctxt, endpoint_id, source_environment):
+        return self._call(
+            ctxt, 'get_endpoint_inventory_csv',
+            endpoint_id=endpoint_id,
+            source_environment=source_environment)
+
     def validate_endpoint_connection(self, ctxt, endpoint_id):
         return self._call(
             ctxt, 'validate_endpoint_connection',

coriolis/conductor/rpc/server.py

@@ -572,6 +572,21 @@ class ConductorServerEndpoint(object):
         return worker_rpc.get_endpoint_storage(
             ctxt, endpoint.type, endpoint.connection_info, env)
 
+    def get_endpoint_inventory_csv(
+            self, ctxt, endpoint_id, source_environment):
+        endpoint = self.get_endpoint(ctxt, endpoint_id)
+
+        worker_rpc = self._get_worker_service_rpc_for_specs(
+            ctxt, enabled=True,
+            region_sets=[[reg.id for reg in endpoint.mapped_regions]],
+            provider_requirements={
+                endpoint.type: [
+                    constants.PROVIDER_TYPE_ENDPOINT_INVENTORY_EXPORT]})
+
+        return worker_rpc.get_endpoint_inventory_csv(
+            ctxt, endpoint.type, endpoint.connection_info,
+            source_environment)
+
     def validate_endpoint_connection(self, ctxt, endpoint_id):
         endpoint = self.get_endpoint(ctxt, endpoint_id)
 

coriolis/constants.py

@@ -205,6 +205,7 @@ PROVIDER_TYPE_SOURCE_ENDPOINT_OPTIONS = 131072
 PROVIDER_TYPE_DESTINATION_TRANSFER_UPDATE = 262144
 PROVIDER_TYPE_SOURCE_MINION_POOL = 524288
 PROVIDER_TYPE_DESTINATION_MINION_POOL = 1048576
+PROVIDER_TYPE_ENDPOINT_INVENTORY_EXPORT = 2097152
 # NOTE(dvincze): These are deprecated, we should remove them,
 # and de-increment the rest
 PROVIDER_TYPE_VALIDATE_MIGRATION_EXPORT = 2048

coriolis/endpoint_resources/api.py

@@ -27,3 +27,8 @@ class API(object):
     def get_endpoint_storage(self, ctxt, endpoint_id, env):
         return self._rpc_client.get_endpoint_storage(
             ctxt, endpoint_id, env)
+
+    def get_endpoint_inventory_csv(
+            self, ctxt, endpoint_id, source_environment):
+        return self._rpc_client.get_endpoint_inventory_csv(
+            ctxt, endpoint_id, source_environment)

coriolis/policies/endpoints.py

@@ -173,6 +173,17 @@ ENDPOINTS_POLICY_DEFAULT_RULES = [
             }
         ]
     ),
+    policy.DocumentedRuleDefault(
+        get_endpoints_policy_label('export_inventory'),
+        ENDPOINTS_POLICY_DEFAULT_RULE,
+        "Export VM inventory as CSV for a supported endpoint",
+        [
+            {
+                "path": "/endpoints/{endpoint_id}/inventory",
+                "method": "GET"
+            }
+        ]
+    ),
 ]
 
 

coriolis/providers/base.py

@@ -1385,6 +1385,27 @@ class _BaseMinionPoolProvider(
         pass
 
 
+class BaseEndpointInventoryExportProvider(
+        object, with_metaclass(abc.ABCMeta)):
+    """Capability class for providers that support VM inventory CSV export.
+
+    Providers that implement this class will be offered in the UI as
+    supporting the inventory export action. Providers that do not implement
+    this class cleanly do not support it — no additional error handling is
+    required at the provider level.
+    """
+
+    @abc.abstractmethod
+    def export_instance_inventory(
+            self, ctxt, connection_info, source_environment):
+        """Export the full VM inventory as a CSV-formatted string.
+
+        Returns a standards-compliant CSV string with a header row and one
+        row per VM, sorted deterministically by VM ID.
+        """
+        raise NotImplementedError()
+
+
 class BaseSourceMinionPoolProvider(_BaseMinionPoolProvider):
     pass
 

coriolis/providers/factory.py

@@ -45,7 +45,9 @@ PROVIDER_TYPE_MAP = {
     constants.PROVIDER_TYPE_SOURCE_MINION_POOL: (
         base.BaseSourceMinionPoolProvider),
     constants.PROVIDER_TYPE_DESTINATION_MINION_POOL: (
-        base.BaseDestinationMinionPoolProvider)
+        base.BaseDestinationMinionPoolProvider),
+    constants.PROVIDER_TYPE_ENDPOINT_INVENTORY_EXPORT: (
+        base.BaseEndpointInventoryExportProvider),
 }
 
 

coriolis/tests/api/v1/test_endpoint_inventory.py

@@ -0,0 +1,69 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.api.v1 import endpoint_inventory as endpoint
+from coriolis.api import wsgi as api_wsgi
+from coriolis.endpoint_resources import api
+from coriolis.tests import test_base
+from coriolis import utils
+
+
+class EndpointInventoryControllerTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis Endpoint Inventory v1 API"""
+
+    def setUp(self):
+        super(EndpointInventoryControllerTestCase, self).setUp()
+        self.endpoint_api = endpoint.EndpointInventoryController()
+
+    @mock.patch.object(utils, 'decode_base64_param')
+    @mock.patch.object(api.API, 'get_endpoint_inventory_csv')
+    def test_index(
+        self,
+        mock_get_endpoint_inventory_csv,
+        mock_decode_base64_param,
+    ):
+        mock_req = mock.Mock()
+        mock_context = mock.Mock()
+        endpoint_id = mock.sentinel.endpoint_id
+        mock_req.environ = {'coriolis.context': mock_context}
+        env = mock.sentinel.env
+        mock_req.GET = {'env': env}
+        mock_get_endpoint_inventory_csv.return_value = 'vm_id,vm_name\n'
+
+        response = self.endpoint_api.index(mock_req, endpoint_id)
+
+        mock_context.can.assert_called_once_with(
+            'migration:endpoints:export_inventory')
+        mock_decode_base64_param.assert_called_once_with(env, is_json=True)
+        mock_get_endpoint_inventory_csv.assert_called_once_with(
+            mock_context, endpoint_id,
+            mock_decode_base64_param.return_value)
+        self.assertIsInstance(response, api_wsgi.ResponseObject)
+        self.assertEqual(response.code, 200)
+        self.assertEqual(response.obj, 'vm_id,vm_name\n')
+
+    @mock.patch.object(utils, 'decode_base64_param')
+    @mock.patch.object(api.API, 'get_endpoint_inventory_csv')
+    def test_index_no_env(
+        self,
+        mock_get_endpoint_inventory_csv,
+        mock_decode_base64_param,
+    ):
+        mock_req = mock.Mock()
+        mock_context = mock.Mock()
+        endpoint_id = mock.sentinel.endpoint_id
+        mock_req.environ = {'coriolis.context': mock_context}
+        mock_req.GET = {}
+        mock_get_endpoint_inventory_csv.return_value = 'vm_id,vm_name\n'
+
+        response = self.endpoint_api.index(mock_req, endpoint_id)
+
+        mock_decode_base64_param.assert_not_called()
+        mock_get_endpoint_inventory_csv.assert_called_once_with(
+            mock_context, endpoint_id, {})
+        self.assertIsInstance(response, api_wsgi.ResponseObject)
+        self.assertEqual(response.code, 200)
+        self.assertEqual(
+            mock_req.environ['coriolis.best_content_type'], 'text/csv')

coriolis/tests/conductor/rpc/test_server.py

@@ -525,6 +525,47 @@ class ConductorServerEndpointTestCase(test_base.CoriolisBaseTestCase):
             storage_pools, rpc_return_value.get_endpoint_storage.return_value
         )
 
+    @mock.patch.object(
+        server.ConductorServerEndpoint, "_get_worker_service_rpc_for_specs"
+    )
+    @mock.patch.object(server.ConductorServerEndpoint, "get_endpoint")
+    def test_get_endpoint_inventory_csv(
+            self, mock_get_endpoint, mock_get_worker_service_rpc_for_specs
+    ):
+        result = self.server.get_endpoint_inventory_csv(
+            mock.sentinel.context,
+            mock.sentinel.endpoint_id,
+            mock.sentinel.source_environment,
+        )
+
+        mock_get_endpoint.assert_called_once_with(
+            mock.sentinel.context, mock.sentinel.endpoint_id
+        )
+
+        mock_get_worker_service_rpc_for_specs.assert_called_once_with(
+            mock.sentinel.context,
+            enabled=True,
+            region_sets=[[]],
+            provider_requirements={
+                mock_get_endpoint.return_value.type: [
+                    constants.PROVIDER_TYPE_ENDPOINT_INVENTORY_EXPORT
+                ]
+            },
+        )
+
+        rpc_return_value = mock_get_worker_service_rpc_for_specs.return_value
+        rpc_return_value.get_endpoint_inventory_csv.assert_called_once_with(
+            mock.sentinel.context,
+            mock_get_endpoint.return_value.type,
+            mock_get_endpoint.return_value.connection_info,
+            mock.sentinel.source_environment,
+        )
+
+        self.assertEqual(
+            result,
+            rpc_return_value.get_endpoint_inventory_csv.return_value
+        )
+
     @mock.patch.object(
         server.ConductorServerEndpoint, "_get_worker_service_rpc_for_specs"
     )

coriolis/tests/worker/rpc/test_server.py

@@ -942,6 +942,55 @@ class WorkerServerEndpointTestCase(test_base.CoriolisBaseTestCase):
             storage, mock_get_provider.return_value.get_storage.return_value
         )
 
+    @mock.patch.object(utils, "get_secret_connection_info")
+    @mock.patch.object(providers_factory, "get_provider")
+    def test_get_endpoint_inventory_csv(
+        self, mock_get_provider, mock_get_secret
+    ):
+        result = self.server.get_endpoint_inventory_csv(
+            mock.sentinel.context,
+            mock.sentinel.platform_name,
+            mock.sentinel.connection_info,
+            mock.sentinel.source_environment,
+        )
+
+        mock_get_provider.assert_called_once_with(
+            mock.sentinel.platform_name,
+            constants.PROVIDER_TYPE_ENDPOINT_INVENTORY_EXPORT,
+            None,
+        )
+        mock_get_secret.assert_called_once_with(
+            mock.sentinel.context, mock.sentinel.connection_info
+        )
+        mock_get_provider.return_value.export_instance_inventory\
+            .assert_called_once_with(
+                mock.sentinel.context,
+                mock_get_secret.return_value,
+                mock.sentinel.source_environment,
+            )
+        self.assertEqual(
+            result,
+            mock_get_provider.return_value.export_instance_inventory
+            .return_value
+        )
+
+    @mock.patch.object(utils, "get_secret_connection_info")
+    @mock.patch.object(providers_factory, "get_provider")
+    def test_get_endpoint_inventory_csv_unsupported_provider(
+        self, mock_get_provider, mock_get_secret
+    ):
+        mock_get_provider.return_value = None
+
+        self.assertRaises(
+            exception.InvalidInput,
+            self.server.get_endpoint_inventory_csv,
+            mock.sentinel.context,
+            mock.sentinel.platform_name,
+            mock.sentinel.connection_info,
+            mock.sentinel.source_environment,
+        )
+        mock_get_secret.assert_not_called()
+
     @mock.patch.object(providers_factory, "get_available_providers")
     def test_get_available_providers(self, mock_get_available_providers):
         result = self.server.get_available_providers(mock.sentinel.context)

coriolis/worker/rpc/client.py

@@ -146,6 +146,14 @@ class WorkerClient(rpc.BaseRPCClient):
             connection_info=connection_info,
             env=env)
 
+    def get_endpoint_inventory_csv(
+            self, ctxt, platform_name, connection_info, source_environment):
+        return self._call(
+            ctxt, 'get_endpoint_inventory_csv',
+            platform_name=platform_name,
+            connection_info=connection_info,
+            source_environment=source_environment)
+
     def get_available_providers(self, ctxt):
         return self._call(
             ctxt, 'get_available_providers')

coriolis/worker/rpc/server.py

@@ -505,6 +505,22 @@ class WorkerServerEndpoint(object):
 
         return storage
 
+    def get_endpoint_inventory_csv(
+            self, ctxt, platform_name, connection_info, source_environment):
+        provider = providers_factory.get_provider(
+            platform_name,
+            constants.PROVIDER_TYPE_ENDPOINT_INVENTORY_EXPORT, None)
+        if not provider:
+            raise exception.InvalidInput(
+                "Provider plugin for platform '%s' does not support "
+                "VM inventory CSV export." % platform_name)
+
+        secret_connection_info = utils.get_secret_connection_info(
+            ctxt, connection_info)
+
+        return provider.export_instance_inventory(
+            ctxt, secret_connection_info, source_environment)
+
     def get_available_providers(self, ctxt):
         return providers_factory.get_available_providers()