Merge pull request #7630 from opsmill/pog-release-1.5-into-develop-20251111

Merge 'release-1.5' into 'develop' with resolved conflicts

Author: ogenstad

.github/workflows/version-upgrade.yml

@@ -126,7 +126,7 @@ jobs:
         run: invoke dev.build
 
       - name: Upgrade Infrahub and apply the latest database migrations
-        run: invoke dev.upgrade
+        run: invoke dev.upgrade --rebase-branches
 
       - name: Start Demo
         run: invoke dev.start
@@ -136,8 +136,8 @@ jobs:
           PORT=$(docker compose -p $INFRAHUB_BUILD_NAME port server 8000 | cut -d: -f2)
           echo "INFRAHUB_ADDRESS=http://localhost:${PORT}" >> $GITHUB_ENV
 
-      - name: Test branch rebase
-        run: poetry run invoke dev.test-branch-rebase --branch atl1-delete-upstream --data-to-check $DATA_NAME
+      - name: Verify branch graph version
+        run: poetry run invoke dev.test-branch-graph-version --branch atl1-delete-upstream
 
       - name: Containers after tests
         if: always()

.vale/styles/spelling-exceptions.txt

@@ -177,6 +177,7 @@ pseudocode
 psutil
 PyPI
 query_peers
+quiescing
 rebase
 rebased
 rebasing
@@ -197,6 +198,7 @@ sdk
 SLA
 SLAs
 Slurp'it
+snapshotting
 streamlit
 Streamlit
 subcommand

CHANGELOG.md

@@ -8,7 +8,6 @@
 from deepdiff import DeepDiff
 from infrahub_sdk.async_typer import AsyncTyper
 from prefect.client.orchestration import get_client
-from rich import print as rprint
 
 from infrahub import config
 from infrahub.core.initialization import (
@@ -18,6 +17,7 @@
     initialize_registry,
 )
 from infrahub.core.manager import NodeManager
+from infrahub.core.migrations.shared import get_migration_console
 from infrahub.core.protocols import CoreAccount, CoreObjectPermission
 from infrahub.dependencies.registry import build_component_registry
 from infrahub.lock import initialize_lock
@@ -35,16 +35,19 @@
 from .db import (
     detect_migration_to_run,
     initialize_internal_schema,
+    mark_branches_needing_rebase,
     migrate_database,
     trigger_rebase_branches,
     update_core_schema,
 )
 
 if TYPE_CHECKING:
     from infrahub.cli.context import CliContext
+    from infrahub.core.branch.models import Branch
     from infrahub.database import InfrahubDatabase
 
 app = AsyncTyper()
+console = get_migration_console()
 
 
 @app.command(name="upgrade")
@@ -53,6 +56,9 @@ async def upgrade_cmd(
     config_file: str = typer.Argument("infrahub.toml", envvar="INFRAHUB_CONFIG"),
     check: bool = typer.Option(False, help="Check the state of the system without upgrading."),
     rebase_branches: bool = typer.Option(False, help="Rebase and apply migrations to branches if required."),
+    interactive: bool = typer.Option(
+        False, help="Use interactive prompt to accept or deny rebase of individual branches."
+    ),
 ) -> None:
     """Upgrade Infrahub to the latest version."""
 
@@ -90,7 +96,7 @@ async def upgrade_cmd(
 
     if not await migrate_database(db=dbdriver, initialize=False, migrations=migrations):
         # A migration failed, stop the upgrade process
-        rprint("Upgrade cancelled due to migration failure.")
+        console.log("Upgrade cancelled due to migration failure.")
         await dbdriver.close()
         return
 
@@ -115,8 +121,22 @@ async def upgrade_cmd(
     # -------------------------------------------
     # Perform branch rebase and apply migrations to them
     # -------------------------------------------
+    branches = await mark_branches_needing_rebase(db=dbdriver)
+    plural = len(branches) != 1
+    get_migration_console().log(
+        f"Found {len(branches)} {'branches' if plural else 'branch'} that {'need' if plural else 'needs'} to be rebased"
+    )
+
     if rebase_branches:
-        await trigger_rebase_branches(db=dbdriver)
+        branches_to_rebase: list[Branch] = []
+        if not interactive:
+            branches_to_rebase = branches
+        else:
+            for branch in branches:
+                if typer.confirm(f"Rebase branch {branch.name}?"):
+                    branches_to_rebase.append(branch)
+
+        await trigger_rebase_branches(db=dbdriver, branches=branches_to_rebase)
 
     await dbdriver.close()
 
@@ -134,21 +154,21 @@ async def upgrade_menu(db: InfrahubDatabase) -> None:
     diff_menu = DeepDiff(menu_items.to_rest(), default_menu_dict.to_rest(), ignore_order=True)
 
     if not diff_menu:
-        rprint("Menu Up to date, nothing to update")
+        console.log("Menu Up to date, nothing to update")
         return
 
     await menu_repository.update_menu(existing_menu=menu_items, new_menu=default_menu_dict, menu_nodes=menu_nodes)
-    rprint("Menu has been updated")
+    console.log("Menu has been updated")
 
 
 async def upgrade_permissions(db: InfrahubDatabase) -> None:
     existing_permissions = await NodeManager.query(schema=CoreObjectPermission, db=db, limit=1)
     if existing_permissions:
-        rprint("Permissions Up to date, nothing to update")
+        console.log("Permissions Up to date, nothing to update")
         return
 
     await setup_permissions(db=db)
-    rprint("Permissions have been updated")
+    console.log("Permissions have been updated")
 
 
 async def setup_permissions(db: InfrahubDatabase) -> None:

backend/infrahub/cli/db.py

@@ -3,13 +3,10 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
-from rich.console import Console
-
 from infrahub.constants.database import IndexType
 from infrahub.core.attribute import MAX_STRING_LENGTH
-from infrahub.core.migrations.shared import MigrationResult
+from infrahub.core.migrations.shared import MigrationResult, get_migration_console
 from infrahub.core.query import Query, QueryType
-from infrahub.core.timestamp import Timestamp
 from infrahub.database.index import IndexItem
 from infrahub.database.neo4j import IndexManagerNeo4j
 from infrahub.log import get_logger
@@ -467,24 +464,19 @@ async def validate_migration(self, db: InfrahubDatabase) -> MigrationResult:  #
         return result
 
     async def execute(self, db: InfrahubDatabase) -> MigrationResult:  # noqa: PLR0915
-        console = Console()
+        console = get_migration_console()
         result = MigrationResult()
 
         # find the active schema attributes that have a LARGE_ATTRIBUTE_TYPE kind on all branches
-        console.print(
-            f"{Timestamp().to_string()} Determining schema attribute types and timestamps on all branches...", end=""
-        )
+        console.print("Determining schema attribute types and timestamps on all branches...", end="")
         get_large_attribute_types_query = await GetLargeAttributeTypesQuery.init(db=db)
         await get_large_attribute_types_query.execute(db=db)
         schema_attribute_timeframes = get_large_attribute_types_query.get_large_attribute_type_timeframes()
         console.print("done")
 
         # find which schema attributes are large_types in the default branch, but updated to non-large_type on other branches
         # {(kind, attr_name): SchemaAttributeTimeframe}
-        console.print(
-            f"{Timestamp().to_string()} Determining which schema attributes have been updated to non-large_type on non-default branches...",
-            end="",
-        )
+        console.print("Determining schema attribute updates on non-default branches...", end="")
         main_schema_attribute_timeframes_map: dict[tuple[str, str], SchemaAttributeTimeframe] = {}
         for schema_attr_time in schema_attribute_timeframes:
             if schema_attr_time.is_default_branch:
@@ -508,15 +500,15 @@ async def execute(self, db: InfrahubDatabase) -> MigrationResult:  # noqa: PLR09
         console.print("done")
 
         # drop the index on the AttributeValueNonIndexed vertex, there won't be any at this point anyway
-        console.print(f"{Timestamp().to_string()} Dropping index on AttributeValueIndexed vertices...", end="")
+        console.print("Dropping index on AttributeValueIndexed vertices...", end="")
         index_manager = IndexManagerNeo4j(db=db)
         index_manager.init(nodes=[AV_INDEXED_INDEX], rels=[])
         await index_manager.drop()
         console.print("done")
 
         # create the temporary non-indexed attribute value vertices for LARGE_ATTRIBUTE_TYPE attributes
         # start with default branch
-        console.print(f"{Timestamp().to_string()} Update non-indexed attribute values with temporary label...", end="")
+        console.print("Creating temporary non-indexed attribute values for large attribute types...", end="")
         large_schema_attribute_timeframes = [
             schema_attr_time for schema_attr_time in schema_attribute_timeframes if schema_attr_time.is_large_type
         ]
@@ -528,10 +520,7 @@ async def execute(self, db: InfrahubDatabase) -> MigrationResult:  # noqa: PLR09
         console.print("done")
 
         # re-index attribute values on branches where the type was updated to non-large_type
-        console.print(
-            f"{Timestamp().to_string()} Indexing attribute values on branches where the attribute schema was updated to a non-large_type...",
-            end="",
-        )
+        console.print("Re-indexing attribute values on branches updated to non-large types...", end="")
         for schema_attr_time in large_type_reverts:
             revert_non_index_on_branch_query = await RevertNonIndexOnBranchQuery.init(
                 db=db, schema_attribute_timeframe=schema_attr_time
@@ -540,35 +529,27 @@ async def execute(self, db: InfrahubDatabase) -> MigrationResult:  # noqa: PLR09
         console.print("done")
 
         # set the AttributeValue vertices to be AttributeValueIndexed
-        console.print(
-            f"{Timestamp().to_string()} Update all AttributeValue vertices to add the AttributeValueIndexed label...",
-            end="",
-        )
+        console.print("Adding AttributeValueIndexed label to AttributeValue vertices...", end="")
         set_attribute_value_indexed_query = await SetAttributeValueIndexedQuery.init(db=db)
         await set_attribute_value_indexed_query.execute(db=db)
         console.print("done")
 
         # set AttributeValueNonIndexed vertices to just AttributeValue
-        console.print(
-            f"{Timestamp().to_string()} Update all AttributeValueNonIndexed vertices to be AttributeValue (no index)...",
-            end="",
-        )
+        console.print("Restoring AttributeValue label on AttributeValueNonIndexed vertices...", end="")
         finalize_attribute_value_non_indexed_query = await FinalizeAttributeValueNonIndexedQuery.init(db=db)
         await finalize_attribute_value_non_indexed_query.execute(db=db)
         console.print("done")
 
         # de-index all attribute values too large to be indexed
-        console.print(
-            f"{Timestamp().to_string()} De-index any legacy attribute data that is too large to be indexed...", end=""
-        )
+        console.print("De-indexing legacy attribute data exceeding index limits...", end="")
         de_index_large_attribute_values_query = await DeIndexLargeAttributeValuesQuery.init(
             db=db, max_value_size=MAX_STRING_LENGTH
         )
         await de_index_large_attribute_values_query.execute(db=db)
         console.print("done")
 
         # add the index back to the AttributeValueNonIndexed vertex
-        console.print(f"{Timestamp().to_string()} Add the index back to the AttributeValueIndexed label...", end="")
+        console.print("Adding index back to the AttributeValueIndexed label...", end="")
         index_manager = IndexManagerNeo4j(db=db)
         index_manager.init(nodes=[AV_INDEXED_INDEX], rels=[])
         await index_manager.add()

backend/infrahub/cli/upgrade.py

@@ -4,14 +4,13 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
-from rich.console import Console
 from rich.progress import Progress
 
 from infrahub.core.branch.models import Branch
 from infrahub.core.constants import InfrahubKind
 from infrahub.core.initialization import initialization
 from infrahub.core.ipam.reconciler import IpamReconciler
-from infrahub.core.migrations.shared import MigrationResult
+from infrahub.core.migrations.shared import MigrationResult, get_migration_console
 from infrahub.core.query import Query, QueryType
 from infrahub.lock import initialize_lock
 from infrahub.log import get_logger
@@ -235,13 +234,13 @@ async def validate_migration(self, db: InfrahubDatabase) -> MigrationResult:  #
         return MigrationResult()
 
     async def execute(self, db: InfrahubDatabase) -> MigrationResult:
-        console = Console()
+        console = get_migration_console()
         result = MigrationResult()
         # load schemas from database into registry
         initialize_lock()
         await initialization(db=db)
 
-        console.print("Identifying IP prefixes/addresses to reconcile...", end="")
+        console.print("Identifying IP prefixes and addresses to reconcile...", end="")
         find_nodes_query = await FindNodesToReconcileQuery.init(db=db)
         await find_nodes_query.execute(db=db)
         console.print("done")
@@ -250,16 +249,17 @@ async def execute(self, db: InfrahubDatabase) -> MigrationResult:
         # reconciler cannot correctly handle a prefix that is its own parent
         ip_node_details_list = find_nodes_query.get_nodes_to_reconcile()
         uuids_to_check = {ip_node_details.node_uuid for ip_node_details in ip_node_details_list}
-        console.print(f"{len(ip_node_details_list)} IP prefixes/addresses will be reconciled")
+        console.log(f"{len(ip_node_details_list)} IP prefixes or addresses will be reconciled.")
 
-        console.print("Deleting any self-parent relationships...", end="")
+        console.print("Deleting self-parent relationships prior to reconciliation...", end="")
         delete_self_parent_relationships_query = await DeleteSelfParentRelationshipsQuery.init(
             db=db, uuids_to_check=list(uuids_to_check)
         )
         await delete_self_parent_relationships_query.execute(db=db)
         console.print("done")
 
-        with Progress() as progress:
+        console.log("Reconciling IP prefixes and addresses across branches...")
+        with Progress(console=console) as progress:
             reconcile_task = progress.add_task("Reconciling IP prefixes/addresses...", total=len(ip_node_details_list))
 
             for ip_node_details in ip_node_details_list:
@@ -271,4 +271,6 @@ async def execute(self, db: InfrahubDatabase) -> MigrationResult:
                 )
                 progress.update(reconcile_task, advance=1)
 
+        console.log("IP prefix and address reconciliation complete.")
+
         return result

backend/infrahub/core/migrations/graph/m037_index_attr_vals.py

@@ -2,13 +2,12 @@
 
 from typing import TYPE_CHECKING, Any
 
-from rich.console import Console
 from rich.progress import Progress
 
 from infrahub.core.branch.models import Branch
 from infrahub.core.initialization import get_root_node
 from infrahub.core.manager import NodeManager
-from infrahub.core.migrations.shared import MigrationResult
+from infrahub.core.migrations.shared import MigrationResult, get_migration_console
 from infrahub.core.query import Query, QueryType
 from infrahub.core.timestamp import Timestamp
 from infrahub.log import get_logger
@@ -98,7 +97,7 @@ async def execute_against_branch(self, db: InfrahubDatabase, branch: Branch) ->
         return await self._do_execute_for_branch(db=db, branch=branch)
 
     async def _do_execute_for_branch(self, db: InfrahubDatabase, branch: Branch) -> MigrationResult:
-        console = Console()
+        console = get_migration_console()
         result = MigrationResult()
         await get_or_load_schema_branch(db=db, branch=branch)
 
@@ -111,7 +110,7 @@ async def _do_execute_for_branch(self, db: InfrahubDatabase, branch: Branch) ->
         console.print("done")
 
         node_ids_to_update: set[str] = set()
-        with Progress() as progress:
+        with Progress(console=console) as progress:
             gather_nodes_task = progress.add_task(
                 f"Gathering affected objects for each profile on branch {branch.name}...", total=len(profiles_map)
             )
@@ -121,6 +120,7 @@ async def _do_execute_for_branch(self, db: InfrahubDatabase, branch: Branch) ->
                 node_peers = await node_relationship_manager.get_db_peers(db=db)
                 node_ids_to_update.update(str(peer.peer_id) for peer in node_peers)
                 progress.update(gather_nodes_task, advance=1)
+        console.log(f"Collected nodes impacted by profiles on branch {branch.name}.")
 
         console.print("Identifying nodes with profile updates by branch...", end="")
         get_nodes_with_profile_updates_by_branch_query = await GetNodesWithProfileUpdatesForBranchQuery.init(
@@ -131,7 +131,8 @@ async def _do_execute_for_branch(self, db: InfrahubDatabase, branch: Branch) ->
         console.print("done")
 
         right_now = Timestamp()
-        with Progress() as progress:
+        console.log("Applying profiles to nodes...")
+        with Progress(console=console) as progress:
             apply_task = progress.add_task("Applying profiles to nodes...", total=len(node_ids_to_update))
             applier = self._get_profile_applier(db=db, branch=branch)
             for node_id in node_ids_to_update:
@@ -141,5 +142,6 @@ async def _do_execute_for_branch(self, db: InfrahubDatabase, branch: Branch) ->
                     if updated_field_names:
                         await node.save(db=db, fields=updated_field_names, at=right_now)
                 progress.update(apply_task, advance=1)
+        console.log("Completed applying profiles to nodes.")
 
         return result

backend/infrahub/core/migrations/graph/m039_ipam_reconcile.py

@@ -9,7 +9,7 @@
 from infrahub.core.constants import SchemaPathType
 from infrahub.core.initialization import get_root_node
 from infrahub.core.migrations.schema.node_attribute_add import NodeAttributeAddMigration
-from infrahub.core.migrations.shared import MigrationRequiringRebase, MigrationResult
+from infrahub.core.migrations.shared import MigrationRequiringRebase, MigrationResult, get_migration_console
 from infrahub.core.path import SchemaPath
 from infrahub.core.query import Query, QueryType
 
@@ -97,7 +97,7 @@ async def execute(self, db: InfrahubDatabase) -> MigrationResult:
                 ]
             )
 
-        with Progress() as progress:
+        with Progress(console=get_migration_console()) as progress:
             update_task = progress.add_task("Adding HFID and display label to nodes", total=len(migrations))
 
             for migration in migrations:
@@ -144,7 +144,7 @@ async def execute_against_branch(self, db: InfrahubDatabase, branch: Branch) ->
                 ]
             )
 
-        with Progress() as progress:
+        with Progress(console=get_migration_console()) as progress:
             update_task = progress.add_task(
                 f"Adding HFID and display label to nodes on branch {branch.name}", total=len(migrations)
             )