feat: support k8s-backup-target interface for generic backup operations

charmcraft.yaml

@@ -83,7 +83,7 @@ actions:
     - command
     additionalProperties: false
   create-backup:
-    description: Create a Velero backup using the velero-backups relation.
+    description: Create a Velero backup using the velero-backups or k8s-backup-target relation.
     params:
       target:
         description: |
@@ -210,6 +210,9 @@ requires:
   velero-backups:
     interface: velero_backup_config
     optional: true
+  k8s-backup-target:
+    interface: k8s_backup_target
+    optional: true
 
 provides:
   metrics-endpoint:

pyproject.toml

@@ -13,6 +13,7 @@ lightkube = "^0.17.1"
 pydantic = "^2.11.3"
 tenacity = "^9.0.0"
 object-storage-charmlib = "^0.1.0"
+charmlibs-interfaces-k8s-backup-target = "~=0.1.0"
 
 [tool.coverage.run]
 branch = true

requirements.txt

@@ -1,6 +1,7 @@
 annotated-types==0.7.0 ; python_version >= "3.12" and python_version < "4.0"
 anyio==4.12.1 ; python_version >= "3.12" and python_version < "4.0"
 certifi==2026.1.4 ; python_version >= "3.12" and python_version < "4.0"
+charmlibs-interfaces-k8s-backup-target==0.1.0.post0 ; python_version >= "3.12" and python_version < "4.0"
 cosl==1.4.0 ; python_version >= "3.12" and python_version < "4.0"
 h11==0.16.0 ; python_version >= "3.12" and python_version < "4.0"
 httpcore==1.0.9 ; python_version >= "3.12" and python_version < "4.0"

src/charm.py

@@ -11,6 +11,7 @@
 from typing import Dict, List, Optional, Union, cast
 
 import ops
+from charmlibs.interfaces.k8s_backup_target import K8sBackupTargetRequirer
 from charms.data_platform_libs.v0.azure_storage import AzureStorageRequires
 from charms.data_platform_libs.v0.data_models import TypedCharmBase
 from charms.data_platform_libs.v0.s3 import S3Requirer
@@ -28,6 +29,7 @@
 from config import CharmConfig
 from constants import (
     AZURE_SERVICE_PRINCIPAL_RELATION_NAME,
+    K8S_BACKUP_TARGET_ENDPOINT,
     VELERO_ALLOWED_SUBCOMMANDS,
     VELERO_BACKUPS_ENDPOINT,
     VELERO_BINARY_PATH,
@@ -100,6 +102,7 @@ def __init__(self, framework: ops.Framework):
 
         self._grafana_dashboard = GrafanaDashboardProvider(self)
         self._backup_configs = VeleroBackupRequier(self, VELERO_BACKUPS_ENDPOINT)
+        self._k8s_backup_targets = K8sBackupTargetRequirer(self, K8S_BACKUP_TARGET_ENDPOINT)
 
         self.framework.observe(self.on.install, self._reconcile)
         self.framework.observe(self.on.update_status, self._reconcile)
@@ -110,6 +113,7 @@ def __init__(self, framework: ops.Framework):
         for relation in [r.value for r in StorageRelation] + [
             AZURE_SERVICE_PRINCIPAL_RELATION_NAME,
             VELERO_BACKUPS_ENDPOINT,
+            K8S_BACKUP_TARGET_ENDPOINT,
         ]:
             self.framework.observe(self.on[relation].relation_changed, self._reconcile)
             self.framework.observe(self.on[relation].relation_broken, self._reconcile)
@@ -263,7 +267,12 @@ def _on_create_backup_action(self, event: ops.ActionEvent) -> None:
             event.fail("Invalid target format. Use 'app:endpoint'")
             return
 
-        backup_spec = self._backup_configs.get_backup_spec(app, endpoint, model)
+        target_relation = self._find_backup_relation(app)
+        if not target_relation:
+            event.fail(f"No relation found for target '{target}'")
+            return
+
+        backup_spec = self._resolve_backup_spec(target_relation, app, endpoint, model)
         if not backup_spec:
             event.fail(f"No backup spec found for target '{target}' in model '{model}'")
             return
@@ -525,6 +534,45 @@ def _backup_list_to_dict(self, backups: List[BackupInfo]) -> dict:
             }
         return result
 
+    def _find_backup_relation(self, app: str) -> Optional[ops.Relation]:
+        """Find the backup relation for a target app.
+
+        Args:
+            app: The remote application name to find.
+
+        Returns:
+            The matching relation, or None if not found.
+        """
+        for endpoint in (VELERO_BACKUPS_ENDPOINT, K8S_BACKUP_TARGET_ENDPOINT):
+            for relation in self.model.relations.get(endpoint, []):
+                if relation.app and relation.app.name == app:
+                    return relation
+        return None
+
+    def _resolve_backup_spec(
+        self, relation: ops.Relation, app: str, endpoint: str, model: str
+    ) -> Optional[VeleroBackupSpec]:
+        """Resolve a VeleroBackupSpec from the given relation.
+
+        Args:
+            relation: The relation to read the spec from.
+            app: The application name.
+            endpoint: The relation endpoint name.
+            model: The model name.
+
+        Returns:
+            The backup specification if found, otherwise None.
+        """
+        if relation.name == VELERO_BACKUPS_ENDPOINT:
+            return self._backup_configs.get_backup_spec(app, endpoint, model)
+
+        if relation.name == K8S_BACKUP_TARGET_ENDPOINT:
+            k8s_spec = self._k8s_backup_targets.get_backup_spec(app, endpoint, model)
+            if k8s_spec:
+                return VeleroBackupSpec.model_validate(k8s_spec.model_dump())
+
+        return None
+
     def _cleanup_schedule_for_broken_relation(self, relation: ops.Relation) -> None:
         """Clean up schedule when a velero-backups relation is broken.
 

src/constants.py

@@ -33,6 +33,7 @@
 VELERO_ALLOWED_SUBCOMMANDS = {"backup", "restore", "schedule"}
 
 VELERO_BACKUPS_ENDPOINT = "velero-backups"
+K8S_BACKUP_TARGET_ENDPOINT = "k8s-backup-target"
 
 VELERO_BACKUP_LOCATION_RESOURCE = create_namespaced_resource(
     "velero.io", "v1", "BackupStorageLocation", "backupstoragelocations"

tests/integration/helpers.py

@@ -1,9 +1,10 @@
 # Copyright 2026 Canonical Ltd.
 # See LICENSE file for licensing details.
 
+import asyncio
 import subprocess
 from pathlib import Path
-from typing import Dict, Optional, Type
+from typing import Dict, Optional, Type, Union
 
 import yaml
 from juju.action import Action
@@ -13,8 +14,8 @@
 from lightkube import ApiError, Client
 from lightkube.core.resource import GlobalResource, NamespacedResource
 from lightkube.generic_resource import create_namespaced_resource
-from lightkube.resources.apps_v1 import Deployment
-from lightkube.resources.core_v1 import Pod
+from lightkube.resources.apps_v1 import DaemonSet, Deployment
+from lightkube.resources.core_v1 import Pod, Service, ServiceAccount
 from pytest_operator.plugin import OpsTest
 from tenacity import (
     Retrying,
@@ -36,8 +37,10 @@
     "The charm must be deployed with '--trust' flag enabled, run 'juju trust ...'"
 )
 APP_RELATION_NAME = "velero-backups"
+K8S_BACKUP_TARGET_RELATION_NAME = "k8s-backup-target"
 TEST_APP_FIRST_RELATION_NAME = "first-velero-backup-config"
 TEST_APP_SECOND_RELATION_NAME = "second-velero-backup-config"
+TEST_APP_K8S_BACKUP_ENDPOINT = "k8s-backup-endpoint"
 READY_MESSAGE = "Unit is Ready"
 BACKUP_STORAGE_LOCALTION_UNAVAILABLE_MESSAGE = (
     "Velero Storage location is not ready: BackupStorageLocation is unavailable"
@@ -592,3 +595,83 @@ async def get_application_data(
     relation_data = await get_relation_data(ops_test, application_name, endpoint, related_endpoint)
     application_data = relation_data[0]["application-data"]
     return application_data
+
+
+async def deploy_velero_test_charm_and_s3_integrator(
+    ops_test: OpsTest,
+    velero_operator_charm_path: Union[str, Path],
+    test_charm_path: Union[str, Path],
+) -> None:
+    """Deploy velero-operator, test charm, and s3-integrator.
+
+    Args:
+        ops_test: The ops test framework instance.
+        velero_operator_charm_path: Path to the velero-operator charm.
+        test_charm_path: Path to the test charm.
+    """
+    model = get_model(ops_test)
+    await asyncio.gather(
+        model.deploy(
+            velero_operator_charm_path,
+            application_name=APP_NAME,
+            trust=True,
+            config={"use-node-agent": True, "default-volumes-to-fs-backup": True},
+        ),
+        model.deploy(
+            test_charm_path,
+            application_name=TEST_APP_NAME,
+        ),
+        model.deploy(S3_INTEGRATOR, channel=S3_INTEGRATOR_CHANNEL),
+        model.wait_for_idle(apps=[APP_NAME], status="blocked", timeout=TIMEOUT),
+        model.wait_for_idle(apps=[TEST_APP_NAME], status="waiting", timeout=TIMEOUT),
+    )
+
+
+async def configure_s3_integrator(
+    ops_test: OpsTest,
+    s3_cloud_credentials: dict,
+    s3_cloud_configs: dict,
+) -> None:
+    """Configure the S3 integrator charm with credentials and configs.
+
+    Args:
+        ops_test: The ops test framework instance.
+        s3_cloud_credentials: S3 credential parameters.
+        s3_cloud_configs: S3 config parameters.
+    """
+    model = get_model(ops_test)
+    app = model.applications[S3_INTEGRATOR]
+    await app.set_config(s3_cloud_configs)
+    action = await app.units[0].run_action("sync-s3-credentials", **s3_cloud_credentials)
+    result = await action.wait()
+    assert result.results.get("return-code") == 0
+    await model.wait_for_idle(apps=[S3_INTEGRATOR], status="active", timeout=TIMEOUT)
+
+
+async def remove_all_applications(ops_test: OpsTest, lightkube_client: Client) -> None:
+    """Remove velero-operator, s3-integrator, and test charm.
+
+    Also verifies that the core Velero resources created by velero-operator are
+    cleaned up from the model namespace after removal.
+
+    Args:
+        ops_test: The ops test framework instance.
+        lightkube_client: The lightkube client used to verify resource cleanup.
+    """
+    model = get_model(ops_test)
+    namespace = model.name
+    await asyncio.gather(
+        model.remove_application(APP_NAME, block_until_done=True),
+        model.remove_application(S3_INTEGRATOR, block_until_done=True),
+        model.remove_application(TEST_APP_NAME, block_until_done=True),
+    )
+
+    for resource_type, name in (
+        (Deployment, "velero"),
+        (DaemonSet, "node-agent"),
+        (ServiceAccount, "velero"),
+        (Service, "velero-metrics"),
+    ):
+        k8s_assert_resource_not_exists(
+            lightkube_client, resource_type, name=name, namespace=namespace
+        )