Introduce pre-/post-snapshot commands (#2941)

Introduce commands which are run before and after a snapshot is being
taken. This is useful for e.g. for database add-ons to suspend write
operations during snapshot.

supervisor/addons/addon.py

@@ -678,6 +678,21 @@ class Addon(AddonModel):
         except DockerError as err:
             raise AddonsError() from err
 
+    async def _snapshot_command(self, command: str) -> None:
+        try:
+            command_return = await self.instance.run_inside(command)
+            if command_return.exit_code != 0:
+                _LOGGER.error(
+                    "Pre-/Post-Snapshot command returned error code: %s",
+                    command_return.exit_code,
+                )
+                raise AddonsError()
+        except DockerError as err:
+            _LOGGER.error(
+                "Failed running pre-/post-snapshot command %s: %s", command, err
+            )
+            raise AddonsError() from err
+
     async def snapshot(self, tar_file: tarfile.TarFile) -> None:
         """Snapshot state of an add-on."""
         with TemporaryDirectory(dir=self.sys_config.path_tmp) as temp:
@@ -729,12 +744,18 @@ class Addon(AddonModel):
                         arcname="data",
                     )
 
+            if self.snapshot_pre is not None:
+                await self._snapshot_command(self.snapshot_pre)
+
             try:
                 _LOGGER.info("Building snapshot for add-on %s", self.slug)
                 await self.sys_run_in_executor(_write_tarfile)
             except (tarfile.TarError, OSError) as err:
                 _LOGGER.error("Can't write tarfile %s: %s", tar_file, err)
                 raise AddonsError() from err
+            finally:
+                if self.snapshot_post is not None:
+                    await self._snapshot_command(self.snapshot_post)
 
         _LOGGER.info("Finish snapshot for addon %s", self.slug)
 

supervisor/addons/model.py

@@ -51,6 +51,8 @@ from ..const import (
     ATTR_SERVICES,
     ATTR_SLUG,
     ATTR_SNAPSHOT_EXCLUDE,
+    ATTR_SNAPSHOT_POST,
+    ATTR_SNAPSHOT_PRE,
     ATTR_STAGE,
     ATTR_STARTUP,
     ATTR_STDIN,
@@ -358,6 +360,16 @@ class AddonModel(CoreSysAttributes, ABC):
         """Return Exclude list for snapshot."""
         return self.data.get(ATTR_SNAPSHOT_EXCLUDE, [])
 
+    @property
+    def snapshot_pre(self) -> Optional[str]:
+        """Return pre-snapshot command."""
+        return self.data.get(ATTR_SNAPSHOT_PRE)
+
+    @property
+    def snapshot_post(self) -> Optional[str]:
+        """Return post-snapshot command."""
+        return self.data.get(ATTR_SNAPSHOT_POST)
+
     @property
     def default_init(self) -> bool:
         """Return True if the add-on have no own init."""

supervisor/addons/validate.py

@@ -68,6 +68,8 @@ from ..const import (
     ATTR_SERVICES,
     ATTR_SLUG,
     ATTR_SNAPSHOT_EXCLUDE,
+    ATTR_SNAPSHOT_POST,
+    ATTR_SNAPSHOT_PRE,
     ATTR_SQUASH,
     ATTR_STAGE,
     ATTR_STARTUP,
@@ -280,6 +282,8 @@ _SCHEMA_ADDON_CONFIG = vol.Schema(
         vol.Optional(ATTR_SERVICES): [vol.Match(RE_SERVICE)],
         vol.Optional(ATTR_DISCOVERY): [valid_discovery_service],
         vol.Optional(ATTR_SNAPSHOT_EXCLUDE): [str],
+        vol.Optional(ATTR_SNAPSHOT_PRE): str,
+        vol.Optional(ATTR_SNAPSHOT_POST): str,
         vol.Optional(ATTR_OPTIONS, default={}): dict,
         vol.Optional(ATTR_SCHEMA, default={}): vol.Any(
             vol.Schema(

supervisor/const.py

@@ -270,6 +270,8 @@ ATTR_SIGNAL = "signal"
 ATTR_SIZE = "size"
 ATTR_SLUG = "slug"
 ATTR_SNAPSHOT_EXCLUDE = "snapshot_exclude"
+ATTR_SNAPSHOT_PRE = "snapshot_pre"
+ATTR_SNAPSHOT_POST = "snapshot_post"
 ATTR_SNAPSHOTS = "snapshots"
 ATTR_SOURCE = "source"
 ATTR_SQUASH = "squash"