Store diffbase/progress in a shared sqlite database

Signed-off-by: Sergey Vasilyev <nolar@nolar.info>

Author: nolar

docs/configuration.rst

@@ -490,6 +490,25 @@ Usually, a mounted volume or a shared filesystem work fine,
 but exercise caution with locally running operators on developer machines
 with no access to the same directory/filesystem.
 
+Storing progress in sqlite
+--------------------------
+
+To store the state in a SQLite database:
+
+.. code-block:: python
+
+    import kopf
+
+    @kopf.on.startup()
+    def configure(settings: kopf.OperatorSettings, **_):
+        settings.persistence.progress_storage = kopf.SQLiteProgressStorage(path='/var/kopf/state.db')
+
+Each handler's progress record is stored as a separate row in a ``progress``
+table, keyed by the resource's namespace, name, uid, and handler id. The table
+is created automatically on first use. A small touch annotation is still written
+to the Kubernetes object to trigger watch events for delayed handler retries.
+Multiple storage types can share the same database file.
+
 Storing progress in multiple places
 -----------------------------------
 
@@ -656,6 +675,25 @@ Usually, a mounted volume or a shared filesystem work fine,
 but exercise caution with locally running operators on developer machines
 with no access to the same directory/filesystem.
 
+Storing diff base in sqlite
+---------------------------
+
+To store the last-handled configuration in a SQLite database:
+
+
+.. code-block:: python
+
+    import kopf
+
+    @kopf.on.startup()
+    def configure(settings: kopf.OperatorSettings, **_):
+        settings.persistence.diffbase_storage = kopf.SQLiteDiffBaseStorage(path='/var/kopf/state.db')
+
+Each resource's body essence is stored as a single row in a ``diffbase`` table,
+keyed by the resource's namespace, name, and uid. The table is created
+automatically on first use. Multiple storage types can share the same
+database file.
+
 Storing diff base in multiple places
 ------------------------------------
 

kopf/__init__.py

@@ -25,6 +25,7 @@
     AnnotationsDiffBaseStorage,
     StatusDiffBaseStorage,
     FileDiffBaseStorage,
+    SQLiteDiffBaseStorage,
     MultiDiffBaseStorage,
 )
 from kopf._cogs.configs.progress import (
@@ -33,6 +34,7 @@
     AnnotationsProgressStorage,
     StatusProgressStorage,
     FileProgressStorage,
+    SQLiteProgressStorage,
     MultiProgressStorage,
     SmartProgressStorage,
 )
@@ -238,12 +240,14 @@
     'AnnotationsDiffBaseStorage',
     'StatusDiffBaseStorage',
     'FileDiffBaseStorage',
+    'SQLiteDiffBaseStorage',
     'MultiDiffBaseStorage',
     'ProgressRecord',
     'ProgressStorage',
     'AnnotationsProgressStorage',
     'StatusProgressStorage',
     'FileProgressStorage',
+    'SQLiteProgressStorage',
     'MultiProgressStorage',
     'SmartProgressStorage',
     'RawEventType',

kopf/_cogs/configs/conventions.py

@@ -38,7 +38,7 @@
 import uuid
 import warnings
 from collections.abc import Collection, Iterable, Iterator
-from typing import Any
+from typing import Any, Protocol, cast, runtime_checkable
 
 from kopf._cogs.structs import bodies, patches
 
@@ -347,3 +347,104 @@ def _temp_filename(self, path: pathlib.Path) -> Iterator[pathlib.Path]:
             raise
         else:
             temp.rename(path)  # atomic overwrite
+
+
+# Sqlite3 is sometimes broken, so we import only on demand, so we cannot use it in annotations.
+# Add more methods as needed. Only actually used methods are listed here. There can be more.
+class sqlite3_Cursor(Protocol):
+    def fetchone(self) -> list[Any]: ...
+
+
+@runtime_checkable
+class sqlite3_Connection(Protocol):
+    def execute(self, sql: str, parameters: tuple[Any, ...] = ()) -> sqlite3_Cursor: ...
+    def close(self) -> None: ...
+    def __enter__(self) -> Any: ...
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: ...
+
+
+class SQLiteConvention:
+    """
+    A mixin for SQLite-based storages with optimistic table creation.
+
+    Operations are tried first; if the table does not exist, it is created
+    and the operation is retried. This avoids upfront schema management
+    and lets both storage types share the same database file when pointed
+    to the same path.
+    """
+
+    _create_sql: str  # to be defined by subclasses
+
+    # If path is None, we do not own the connection, someone else does, we just use it.
+    # If path is set, we own the connection, create and close it as needed.
+    _path: pathlib.Path | None
+    _conn: sqlite3_Connection | None
+
+    def __init__(
+            self,
+            path_or_conn: sqlite3_Connection | str | pathlib.Path,
+            /,
+            **kwargs: Any,
+    ) -> None:
+        super().__init__(**kwargs)
+        import sqlite3  # sometimes broken, so import only on demand
+        if isinstance(path_or_conn, (sqlite3_Connection, sqlite3.Connection)):
+            self._conn = path_or_conn
+            self._path = None
+        else:
+            self._conn = None
+            self._path = pathlib.Path(path_or_conn)
+
+    async def __aenter__(self) -> None:
+        if self._path is None:
+            return
+        import sqlite3  # sometimes broken, so import only on demand
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        self._conn = cast(sqlite3_Connection, sqlite3.connect(str(self._path)))
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        if self._path is not None and self._conn is not None:
+            self._conn.close()
+            self._conn = None
+
+    @staticmethod
+    def _extract_keys(body: bodies.Body) -> tuple[str, str, str]:
+        # Primary keys cannot store nulls, so we use empty strings as markers.
+        # K8s does not allow empty names/uids, so we are safe from collisions.
+        namespace = body.get('metadata', {}).get('namespace', '')
+        name = body.get('metadata', {}).get('name', '')
+        uid = body.get('metadata', {}).get('uid', '')
+        return namespace, name, uid
+
+    def _execute(
+            self,
+            sql: str,
+            params: tuple[Any, ...] = (),
+    ) -> sqlite3_Cursor:
+        """Execute SQL, creating the table optimistically if absent."""
+        import sqlite3  # sometimes broken, so import only on demand
+
+        assert self._conn is not None
+        try:
+            return self._conn.execute(sql, params)
+        except sqlite3.OperationalError as e:
+            if 'no such table' not in str(e):
+                raise
+            self._conn.execute(self._create_sql)
+            return self._conn.execute(sql, params)
+
+    def _try_execute(
+            self,
+            sql: str,
+            params: tuple[Any, ...] = (),
+    ) -> sqlite3_Cursor | None:
+        """Execute SQL, returning None if the table does not exist."""
+        import sqlite3  # sometimes broken, so import only on demand
+
+        assert self._conn is not None
+        try:
+            return self._conn.execute(sql, params)
+        except sqlite3.OperationalError as e:
+            if 'no such table' not in str(e):
+                raise
+            return None

kopf/_cogs/configs/diffbase.py

@@ -329,6 +329,91 @@ async def erase(self, *, body: bodies.Body) -> None:
             filepath.unlink(missing_ok=True)
 
 
+class SQLiteDiffBaseStorage(conventions.SQLiteConvention, DiffBaseStorage):
+    """
+    Diff-base storage in a SQLite database file.
+
+    Each resource's body essence is stored as a single row, keyed by the
+    resource's namespace, name, and uid. The essence is stored as a JSON
+    string.
+
+    An example of the ``diffbase`` table contents:
+
+    .. code-block:: text
+
+        namespace | name   | uid   | essence
+        ----------+--------+-------+---------------------------------------
+        default   | my-app | uid1  | {"spec":{"replicas":3,"image":"..."}}
+
+    This storage does not write anything to the Kubernetes object itself.
+    Both the file and SQLite diff-base storages can share the same database
+    file when pointed to the same path.
+    """
+
+    # We do not plan any migrations yet. We pray that this simple schema is sufficient forever.
+    _create_sql = (
+        'CREATE TABLE IF NOT EXISTS diffbase ('
+        'namespace TEXT NOT NULL, '
+        'name TEXT NOT NULL, '
+        'uid TEXT NOT NULL, '
+        'essence TEXT NOT NULL, '
+        'PRIMARY KEY (namespace, name, uid))'
+    )
+
+    def __init__(
+            self,
+            path_or_conn: conventions.sqlite3_Connection | str | pathlib.Path,
+            /,
+            *,
+            ignored_fields: Iterable[dicts.FieldSpec] | None = None,
+    ) -> None:
+        super().__init__(path_or_conn, ignored_fields=ignored_fields)
+
+    async def fetch(
+            self,
+            *,
+            body: bodies.Body,
+    ) -> bodies.BodyEssence | None:
+        namespace, name, uid = self._extract_keys(body)
+        assert self._conn is not None
+        with self._conn:
+            cursor = self._try_execute(
+                'SELECT essence FROM diffbase'
+                ' WHERE namespace=? AND name=? AND uid=?',
+                (namespace, name, uid))
+            if cursor is None:
+                return None
+            row = cursor.fetchone()
+        if row is None:
+            return None
+        return cast(bodies.BodyEssence, json.loads(row[0]))
+
+    async def store(
+            self,
+            *,
+            body: bodies.Body,
+            patch: patches.Patch,
+            essence: bodies.BodyEssence,
+    ) -> None:
+        namespace, name, uid = self._extract_keys(body)
+        encoded = json.dumps(dict(essence), separators=(',', ':'))
+        assert self._conn is not None
+        with self._conn:
+            self._execute(
+                'INSERT OR REPLACE INTO diffbase'
+                ' (namespace, name, uid, essence) VALUES (?, ?, ?, ?)',
+                (namespace, name, uid, encoded))
+
+    async def erase(self, *, body: bodies.Body) -> None:
+        namespace, name, uid = self._extract_keys(body)
+        assert self._conn is not None
+        with self._conn:
+            self._try_execute(
+                'DELETE FROM diffbase'
+                ' WHERE namespace=? AND name=? AND uid=?',
+                (namespace, name, uid))
+
+
 class MultiDiffBaseStorage(DiffBaseStorage):
 
     def __init__(