--- /dev/null
+# -*- coding: utf-8 -*-
+# SPDX-License-Identifier: EPL-1.0
+##############################################################################
+# Copyright (c) 2025 The Linux Foundation and others.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Eclipse Public License v1.0
+# which accompanies this distribution, and is available at
+# http://www.eclipse.org/legal/epl-v10.html
+##############################################################################
+"""COE cluster related sub-commands for openstack command."""
+
+__author__ = "Anil Belur"
+
+import logging
+import sys
+from datetime import datetime, timedelta
+
+import openstack
+import openstack.config
+from openstack.cloud.exc import OpenStackCloudException
+
+log = logging.getLogger(__name__)
+
+
+def _filter_clusters(clusters, days=0):
+ """Filter cluster data and return list.
+
+ :arg list clusters: List of cluster objects
+ :arg int days: Filter clusters older than number of days
+ :returns: Filtered list of clusters
+ """
+ filtered = []
+ for cluster in clusters:
+ # Handle different timestamp formats
+ try:
+ created_time = datetime.strptime(cluster.created_at, "%Y-%m-%dT%H:%M:%SZ")
+ except (ValueError, AttributeError):
+ try:
+ created_time = datetime.strptime(cluster.created_at, "%Y-%m-%dT%H:%M:%S.%f")
+ except (ValueError, AttributeError):
+ # Skip if we can't parse the date
+ continue
+
+ if days and (created_time >= datetime.now() - timedelta(days=days)):
+ continue
+
+ filtered.append(cluster)
+ return filtered
+
+
+def list(os_cloud, days=0):
+ """List COE clusters found according to parameters.
+
+ :arg str os_cloud: Cloud name as defined in OpenStack clouds.yaml
+ :arg int days: Filter clusters older than number of days (default: 0 = all)
+ """
+ cloud = openstack.connection.from_config(cloud=os_cloud)
+ clusters = cloud.list_coe_clusters()
+
+ filtered_clusters = _filter_clusters(clusters, days)
+ for cluster in filtered_clusters:
+ log.info(cluster.name)
+
+
+def cleanup(os_cloud, days=0):
+ """Remove clusters from cloud.
+
+ :arg str os_cloud: Cloud name as defined in OpenStack clouds.yaml
+ :arg int days: Filter clusters that are older than number of days
+ """
+
+ def _remove_clusters_from_cloud(clusters, cloud):
+ log.info(f"Removing {len(clusters)} clusters from {cloud.cloud_config.name}.")
+ for cluster in clusters:
+ try:
+ # COE clusters use UUID, delete by ID
+ result = cloud.delete_coe_cluster(cluster.uuid)
+ except OpenStackCloudException as e:
+ if str(e).startswith("Multiple matches found for"):
+ log.warning(f"{str(e)}. Skipping cluster...")
+ continue
+ else:
+ log.error(f"Unexpected exception: {str(e)}")
+ raise
+
+ if not result:
+ log.warning(
+ f'Failed to remove "{cluster.name}" from {cloud.cloud_config.name}. ' f"Possibly already deleted."
+ )
+ else:
+ log.info(f'Removed "{cluster.name}" from {cloud.cloud_config.name}.')
+
+ cloud = openstack.connection.from_config(cloud=os_cloud)
+ clusters = cloud.list_coe_clusters()
+ filtered_clusters = _filter_clusters(clusters, days)
+ _remove_clusters_from_cloud(filtered_clusters, cloud)
+
+
+def remove(os_cloud, cluster_name, minutes=0):
+ """Remove a cluster from cloud.
+
+ :arg str os_cloud: Cloud name as defined in OpenStack clouds.yaml
+ :arg str cluster_name: Name or UUID of cluster to remove
+ :arg int minutes: Only delete cluster if it is older than number of minutes
+ """
+ cloud = openstack.connection.from_config(cloud=os_cloud)
+ cluster = cloud.get_coe_cluster(cluster_name)
+
+ if not cluster:
+ log.error("Cluster not found.")
+ sys.exit(1)
+
+ # Parse created_at timestamp
+ try:
+ created_time = datetime.strptime(cluster.created_at, "%Y-%m-%dT%H:%M:%SZ")
+ except ValueError:
+ try:
+ created_time = datetime.strptime(cluster.created_at, "%Y-%m-%dT%H:%M:%S.%f")
+ except ValueError:
+ log.error("Unable to parse cluster creation time.")
+ sys.exit(1)
+
+ if minutes > 0 and created_time >= datetime.utcnow() - timedelta(minutes=minutes):
+ log.warning(f'Cluster "{cluster.name}" is not older than {minutes} minutes.')
+ else:
+ cloud.delete_coe_cluster(cluster.uuid)
+ log.info(f'Deleted cluster "{cluster.name}".')
+
+
+def show(os_cloud, cluster_name):
+ """Show details of a specific cluster.
+
+ :arg str os_cloud: Cloud name as defined in OpenStack clouds.yaml
+ :arg str cluster_name: Name or UUID of cluster
+ """
+ cloud = openstack.connection.from_config(cloud=os_cloud)
+ cluster = cloud.get_coe_cluster(cluster_name)
+
+ if not cluster:
+ log.error("Cluster not found.")
+ sys.exit(1)
+
+ # Pretty print cluster details
+ cloud.pprint(cluster)
import click
+from lftools.openstack import cluster as os_cluster
from lftools.openstack import image as os_image
from lftools.openstack import object as os_object
from lftools.openstack import server as os_server
volume.add_command(volume_cleanup, "cleanup")
volume.add_command(volume_list, "list")
volume.add_command(volume_remove, "remove")
+
+
+@openstack.group()
+@click.pass_context
+def cluster(ctx):
+ """Command for manipulating COE clusters."""
+ pass
+
+
+@click.command(name="cleanup")
+@click.option("--days", type=int, default=0, help="Find clusters older than or equal to days.")
+@click.pass_context
+def cluster_cleanup(ctx, days):
+ """Cleanup old COE clusters."""
+ os_cluster.cleanup(ctx.obj["os_cloud"], days=days)
+
+
+@click.command()
+@click.option("--days", type=int, default=0, help="Find clusters older than or equal to days.")
+@click.pass_context
+def cluster_list(ctx, days):
+ """List COE clusters."""
+ os_cluster.list(ctx.obj["os_cloud"], days=days)
+
+
+@click.command()
+@click.argument("cluster_name")
+@click.option("--minutes", type=int, default=0, help="Delete cluster if older than x minutes.")
+@click.pass_context
+def cluster_remove(ctx, cluster_name, minutes):
+ """Remove COE cluster."""
+ os_cluster.remove(ctx.obj["os_cloud"], cluster_name=cluster_name, minutes=minutes)
+
+
+@click.command()
+@click.argument("cluster_name")
+@click.pass_context
+def cluster_show(ctx, cluster_name):
+ """Show COE cluster details."""
+ os_cluster.show(ctx.obj["os_cloud"], cluster_name)
+
+
+cluster.add_command(cluster_cleanup, "cleanup")
+cluster.add_command(cluster_list, "list")
+cluster.add_command(cluster_remove, "remove")
+cluster.add_command(cluster_show, "show")
--- /dev/null
+---
+features:
+ - |
+ Add OpenStack COE cluster management commands to lftools.
+
+ New commands:
+
+ * ``lftools openstack cluster list`` - List COE clusters
+ * ``lftools openstack cluster show`` - Show cluster details
+ * ``lftools openstack cluster remove`` - Remove a specific cluster
+ * ``lftools openstack cluster cleanup`` - Cleanup old clusters
+
+ All commands use the OpenStack SDK Python library for direct API
+ interaction, following the same patterns as other lftools openstack
+ subcommands (server, volume, stack, image).
+
+ Example usage::
+
+ # List all clusters
+ lftools openstack --os-cloud mycloud cluster list
+
+ # List clusters older than 30 days
+ lftools openstack --os-cloud mycloud cluster list --days 30
+
+ # Show cluster details
+ lftools openstack --os-cloud mycloud cluster show my-cluster-name
+
+ # Remove a specific cluster (only if older than 60 minutes)
+ lftools openstack --os-cloud mycloud cluster remove my-cluster --minutes 60
+
+ # Cleanup clusters older than 90 days
+ lftools openstack --os-cloud mycloud cluster cleanup --days 90
--- /dev/null
+# -*- code: utf-8 -*-
+# SPDX-License-Identifier: EPL-1.0
+##############################################################################
+# Copyright (c) 2025 The Linux Foundation and others.
+##############################################################################
+"""Unit tests for openstack cluster commands."""
+
+from datetime import datetime, timedelta
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from lftools.openstack import cluster
+
+
+@pytest.fixture
+def mock_cloud():
+ """Mock OpenStack cloud connection."""
+ with patch("openstack.connection.from_config") as mock:
+ cloud = MagicMock()
+ mock.return_value = cloud
+ yield cloud
+
+
+@pytest.fixture
+def sample_clusters():
+ """Sample cluster data for testing."""
+ clusters = []
+
+ # Old cluster (30 days old)
+ old_cluster = MagicMock()
+ old_cluster.name = "old-cluster-001"
+ old_cluster.uuid = "uuid-old-001"
+ old_cluster.created_at = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%dT%H:%M:%SZ")
+ clusters.append(old_cluster)
+
+ # Recent cluster (1 day old)
+ new_cluster = MagicMock()
+ new_cluster.name = "new-cluster-001"
+ new_cluster.uuid = "uuid-new-001"
+ new_cluster.created_at = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
+ clusters.append(new_cluster)
+
+ return clusters
+
+
+class TestClusterFilter:
+ """Test cluster filtering functionality."""
+
+ def test_filter_clusters_no_filter(self, sample_clusters):
+ """Test filtering with no days filter returns all clusters."""
+ result = cluster._filter_clusters(sample_clusters, days=0)
+ assert len(result) == 2
+
+ def test_filter_clusters_with_days(self, sample_clusters):
+ """Test filtering clusters older than specified days."""
+ result = cluster._filter_clusters(sample_clusters, days=7)
+ assert len(result) == 1
+ assert result[0].name == "old-cluster-001"
+
+
+class TestClusterList:
+ """Test cluster list command."""
+
+ def test_list_clusters(self, mock_cloud, sample_clusters, caplog):
+ """Test listing clusters."""
+ mock_cloud.list_coe_clusters.return_value = sample_clusters
+ cluster.list("test-cloud", days=0)
+ assert "old-cluster-001" in caplog.text
+ assert "new-cluster-001" in caplog.text
+
+ def test_list_clusters_with_filter(self, mock_cloud, sample_clusters, caplog):
+ """Test listing clusters with days filter."""
+ mock_cloud.list_coe_clusters.return_value = sample_clusters
+ cluster.list("test-cloud", days=7)
+ assert "old-cluster-001" in caplog.text
+ assert "new-cluster-001" not in caplog.text
+
+
+class TestClusterCleanup:
+ """Test cluster cleanup command."""
+
+ def test_cleanup_clusters(self, mock_cloud, sample_clusters, caplog):
+ """Test cleanup of old clusters."""
+ import logging
+
+ caplog.set_level(logging.INFO)
+ mock_cloud.list_coe_clusters.return_value = sample_clusters
+ mock_cloud.cloud_config.name = "test-cloud"
+ mock_cloud.delete_coe_cluster.return_value = True
+ cluster.cleanup("test-cloud", days=7)
+ mock_cloud.delete_coe_cluster.assert_called_once_with("uuid-old-001")
+ assert "Removing 1 clusters" in caplog.text
+
+
+class TestClusterRemove:
+ """Test cluster remove command."""
+
+ def test_remove_cluster_found(self, mock_cloud, sample_clusters):
+ """Test removing an existing cluster."""
+ mock_cloud.get_coe_cluster.return_value = sample_clusters[0]
+ mock_cloud.delete_coe_cluster.return_value = True
+ cluster.remove("test-cloud", "old-cluster-001", minutes=0)
+ mock_cloud.delete_coe_cluster.assert_called_once_with("uuid-old-001")
+
+ def test_remove_cluster_not_found(self, mock_cloud):
+ """Test removing non-existent cluster."""
+ mock_cloud.get_coe_cluster.return_value = None
+ with pytest.raises(SystemExit) as exc_info:
+ cluster.remove("test-cloud", "nonexistent-cluster", minutes=0)
+ assert exc_info.value.code == 1
+
+
+class TestClusterShow:
+ """Test cluster show command."""
+
+ def test_show_cluster_found(self, mock_cloud, sample_clusters):
+ """Test showing cluster details."""
+ mock_cloud.get_coe_cluster.return_value = sample_clusters[0]
+ cluster.show("test-cloud", "old-cluster-001")
+ mock_cloud.pprint.assert_called_once()
+
+ def test_show_cluster_not_found(self, mock_cloud):
+ """Test showing non-existent cluster."""
+ mock_cloud.get_coe_cluster.return_value = None
+ with pytest.raises(SystemExit) as exc_info:
+ cluster.show("test-cloud", "nonexistent-cluster")
+ assert exc_info.value.code == 1
Code Review / releng / lftools.git / commitdiff