fix: migrate cluster, constraint for k8s_objects (#971)

Adds a missing migration to the k8s_objects table for the watcher. It
also modifies the uniqueness constraint to accomodate multiple clusters.

Author: olevski

components/renku_data_services/k8s_watcher/orm.py

@@ -6,7 +6,7 @@
 from typing import Any
 
 from box import Box
-from sqlalchemy import ColumnElement, DateTime, MetaData, String, func, text
+from sqlalchemy import ColumnElement, DateTime, MetaData, String, UniqueConstraint, func, text
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.hybrid import Comparator, hybrid_property
 from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, mapped_column
@@ -47,6 +47,17 @@ class K8sObjectORM(BaseORM):
     """Representation of a k8s resource."""
 
     __tablename__ = "k8s_objects"
+    __table_args__ = (
+        UniqueConstraint(
+            "group",
+            "version",
+            "kind",
+            "cluster",
+            "namespace",
+            "name",
+            name="_unique_common_k8s_objects_gvk_cluster_namespace_name",
+        ),
+    )
 
     id: Mapped[ULID] = mapped_column(
         "id",
@@ -56,7 +67,7 @@ class K8sObjectORM(BaseORM):
         default_factory=lambda: str(ULID()),
         server_default=text("generate_ulid()"),
     )
-    name: Mapped[str] = mapped_column("name", String(), index=True, unique=True)
+    name: Mapped[str] = mapped_column("name", String(), index=True)
     namespace: Mapped[str] = mapped_column("namespace", String(), index=True)
     creation_date: Mapped[datetime] = mapped_column(
         "creation_date",

components/renku_data_services/migrations/versions/c8061499b966_change_cluster_name_and_uniuqe_.py

@@ -0,0 +1,44 @@
+"""change cluster name and uniuqe constraint
+
+Revision ID: c8061499b966
+Revises: e117405fed51
+Create Date: 2025-08-14 07:40:10.492620
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c8061499b966"
+down_revision = "e117405fed51"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        sa.text(
+            "UPDATE common.k8s_objects SET cluster='0RENK1RENK2RENK3RENK4RENK5' where k8s_objects.cluster='renkulab'"
+        )
+    )
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index("ix_common_k8s_objects_name", table_name="k8s_objects", schema="common")
+    op.create_index(op.f("ix_common_k8s_objects_name"), "k8s_objects", ["name"], unique=False, schema="common")
+    op.create_unique_constraint(
+        "_unique_common_k8s_objects_gvk_cluster_namespace_name",
+        "k8s_objects",
+        ["group", "version", "kind", "cluster", "namespace", "name"],
+        schema="common",
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(
+        "_unique_common_k8s_objects_gvk_cluster_namespace_name", "k8s_objects", schema="common", type_="unique"
+    )
+    op.drop_index(op.f("ix_common_k8s_objects_name"), table_name="k8s_objects", schema="common")
+    op.create_index("ix_common_k8s_objects_name", "k8s_objects", ["name"], unique=True, schema="common")
+    # ### end Alembic commands ###

test/bases/renku_data_services/data_api/test_migrations.py

@@ -609,3 +609,43 @@ async def test_migration_to_dcb9648c3c15(app_manager_instance: DependencyManager
     assert k8s_objs[1].tuple()[0] == "amalthea.dev"
     assert k8s_objs[1].tuple()[1] == "v1alpha1"
     assert k8s_objs[1].tuple()[2] == "jupyterserver"
+
+
+@pytest.mark.asyncio
+async def test_migration_to_c8061499b966(app_manager_instance: DependencyManager, admin_user: UserInfo) -> None:
+    run_migrations_for_app("common", "e117405fed51")
+    async with app_manager_instance.config.db.async_session_maker() as session, session.begin():
+        await session.execute(
+            sa.text(
+                "INSERT into "
+                "common.k8s_objects(name, namespace, manifest, deleted, kind, version, cluster, user_id) "
+                "VALUES ('name_pod', 'ns', '{}', FALSE, 'pod', 'v1', 'renkulab', 'user_id')"
+            )
+        )
+        await session.execute(
+            sa.text(
+                "INSERT into "
+                "common.k8s_objects(name, namespace, manifest, deleted, kind, version, cluster, user_id) "
+                "VALUES ('name_js', 'ns', '{}', FALSE, 'jupyterserver', 'amalthea.dev/v1alpha1', 'renkulab', 'user_id')"
+            )
+        )
+    run_migrations_for_app("common", "c8061499b966")
+    async with app_manager_instance.config.db.async_session_maker() as session, session.begin():
+        k8s_objs = (await session.execute(sa.text("SELECT name, cluster FROM common.k8s_objects"))).all()
+    assert len(k8s_objs) == 2
+    # Check that the cluster name was changed
+    assert k8s_objs[0].tuple()[1] == "0RENK1RENK2RENK3RENK4RENK5"
+    assert k8s_objs[1].tuple()[1] == "0RENK1RENK2RENK3RENK4RENK5"
+    id = ULID()
+    async with app_manager_instance.config.db.async_session_maker() as session, session.begin():
+        await session.execute(
+            sa.text(
+                "INSERT into "
+                "common.k8s_objects(name, namespace, manifest, deleted, kind, version, cluster, user_id) "
+                f"VALUES ('name_pod', 'ns', '{{}}', FALSE, 'pod', 'v1', '{id}', 'user_id')"
+            )
+        )
+    async with app_manager_instance.config.db.async_session_maker() as session, session.begin():
+        k8s_objs = (await session.execute(sa.text("SELECT name, cluster FROM common.k8s_objects"))).all()
+    # Check that we can insert another object with the same name, gvk, namespace, but a different cluster
+    assert len(k8s_objs) == 3