From 5a51fa3dd346458be9518e1a45c169c160a21a4b Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 19:39:51 +0800 Subject: [PATCH] feat(signoz): add isolated metadata restore controller --- agent99-signoz-metadata-executor.ps1 | 408 +++- config/signoz/metadata-export-policy.json | 67 + .../signoz-metadata-isolated-restore.py | 1721 +++++++++++++++++ .../backup/signoz-metadata-restore-drill.py | 303 ++- .../test_signoz_metadata_export_contract.py | 153 +- .../test_signoz_metadata_isolated_restore.py | 597 ++++++ .../ops/deploy-signoz-metadata-toolchain.sh | 32 +- ...oz_metadata_executor_runtime_entrypoint.py | 96 +- .../test_signoz_metadata_toolchain_deploy.py | 16 +- 9 files changed, 3311 insertions(+), 82 deletions(-) create mode 100755 scripts/backup/signoz-metadata-isolated-restore.py create mode 100644 scripts/backup/tests/test_signoz_metadata_isolated_restore.py diff --git a/agent99-signoz-metadata-executor.ps1 b/agent99-signoz-metadata-executor.ps1 index 3db682cca..d2e64b43b 100644 --- a/agent99-signoz-metadata-executor.ps1 +++ b/agent99-signoz-metadata-executor.ps1 @@ -1,5 +1,5 @@ param( - [ValidateSet("Check", "Apply", "Verify")] + [ValidateSet("Check", "Apply", "Verify", "RestoreCheck", "RestoreApply", "RestoreVerifyCleanup")] [string]$Mode = "Check", [Parameter(Mandatory = $true)] [string]$TraceId, @@ -22,30 +22,57 @@ $EvidenceDir = Join-Path $AgentRoot "evidence" $RuntimeManifestPath = Join-Path $AgentRoot "state\runtime-manifest.json" $EntrypointPath = [string]$MyInvocation.MyCommand.Path $SafeIdPattern = "^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$" -$BundleId = "d1318c412aa303c5d8a5795f22ee62dce16ca79354c5a15e95bfad41a5e35533" +$BundleId = "ed4e5fbfc5f732b64d6a78f6ab26b90118e1d42a4f38b40d583876445c32e3aa" $ToolchainRoot = "/backup/toolchains/signoz-metadata/$BundleId" $PolicyPath = "$ToolchainRoot/metadata-export-policy.json" $ExporterPath = "$ToolchainRoot/signoz-metadata-export.py" $VerifierPath = "$ToolchainRoot/verify-signoz-metadata-export.py" $RestoreDriverPath = "$ToolchainRoot/signoz-metadata-restore-drill.py" +$IsolatedRestorePath = "$ToolchainRoot/signoz-metadata-isolated-restore.py" +$IsolatedClusterConfigPath = "$ToolchainRoot/isolated-cluster.xml" $SharedContractPath = "$ToolchainRoot/signoz_metadata_contract.py" $CredentialFile = "/etc/awoooi/secrets/signoz-metadata-api-key" $BaseUrl = "http://127.0.0.1:8080" $OutputDir = "/var/backups/awoooi/signoz-metadata-export-$RunId" $RemoteReceipt = "/backup/deploy-receipts/signoz-metadata-toolchain/export-$RunId.json" +$RestoreRemoteReceipt = "/backup/deploy-receipts/signoz-metadata-toolchain/restore-$RunId.json" $SignozImage = "signoz/signoz:v0.113.0" $SignozImageId = "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b" $SignozContainer = "signoz" $OperationLock = "/tmp/awoooi-signoz-backup-operation.lock" -$TargetKillAfterSeconds = 15 +$DefaultTargetKillAfterSeconds = 15 +$RestoreCleanupKillAfterSeconds = 600 $TargetGuardWorstCaseSeconds = 95 $CheckVerifyTargetTimeoutSeconds = 90 $ApplyTargetTimeoutSeconds = 720 +$RestoreCheckTargetTimeoutSeconds = 180 +$RestoreApplyTargetTimeoutSeconds = 1800 +$RestoreCleanupTargetTimeoutSeconds = 300 $CheckVerifyControllerWaitSeconds = 300 $ApplyControllerWaitSeconds = 900 +$RestoreCheckControllerWaitSeconds = 420 +$RestoreApplyControllerWaitSeconds = 2550 +$RestoreCleanupControllerWaitSeconds = 1050 $ControllerTransportReserveSeconds = 30 -$TargetProgramTimeoutSeconds = if ($Mode -eq "Apply") { $ApplyTargetTimeoutSeconds } else { $CheckVerifyTargetTimeoutSeconds } -$ControllerWaitSeconds = if ($Mode -eq "Apply") { $ApplyControllerWaitSeconds } else { $CheckVerifyControllerWaitSeconds } +$TargetProgramTimeoutSeconds = switch ($Mode) { + "Apply" { $ApplyTargetTimeoutSeconds } + "RestoreCheck" { $RestoreCheckTargetTimeoutSeconds } + "RestoreApply" { $RestoreApplyTargetTimeoutSeconds } + "RestoreVerifyCleanup" { $RestoreCleanupTargetTimeoutSeconds } + default { $CheckVerifyTargetTimeoutSeconds } +} +$ControllerWaitSeconds = switch ($Mode) { + "Apply" { $ApplyControllerWaitSeconds } + "RestoreCheck" { $RestoreCheckControllerWaitSeconds } + "RestoreApply" { $RestoreApplyControllerWaitSeconds } + "RestoreVerifyCleanup" { $RestoreCleanupControllerWaitSeconds } + default { $CheckVerifyControllerWaitSeconds } +} +$TargetKillAfterSeconds = if ($Mode -in @("RestoreApply", "RestoreVerifyCleanup")) { + $RestoreCleanupKillAfterSeconds +} else { + $DefaultTargetKillAfterSeconds +} $TargetWorstCaseSeconds = $TargetGuardWorstCaseSeconds + $TargetProgramTimeoutSeconds + $TargetKillAfterSeconds $EvidenceRunId = if ($RunId -match $SafeIdPattern) { $RunId } else { "invalid-$PID" } $EvidencePath = Join-Path $EvidenceDir "agent99-signoz-metadata-$($Mode.ToLowerInvariant())-$EvidenceRunId.json" @@ -53,14 +80,17 @@ $EvidenceReservationPath = "$EvidencePath.reserve" $script:TransientTransportOutputUsed = $false $script:TransportCleanupSucceeded = $true $script:EvidenceReserved = $false -$script:ApplyDispatchAttempted = $false +$script:ExportCredentialDispatchAttempted = $false +$script:IsolatedApplyDispatchAttempted = $false $ToolchainFiles = [ordered]@{ - $PolicyPath = "9eac4e39515ef007763faacdb4c386e030aeb142872cec7a7a8d2a842befc2f0" + $PolicyPath = "1365c6406b522f2678693cebe99914255d44b4f219a78a4625ac2c2b5eec2557" $ExporterPath = "8cd5cabf245e3814da32e33920c431ec28ed7180edc0495a2d0b9379fb4025fb" $SharedContractPath = "d2b2281da9ec89afc44d5153711b2d61e972545ad107611b74a5e6dfb01defe8" $VerifierPath = "0bac6d8ef8063f3584fc744cf205de08fc50d1085b292ffbd061f6090c344c86" - $RestoreDriverPath = "5688f3212f970b3a0d0f93dfdb1c59c3147729bb5da09610572cd12da5dabd36" + $RestoreDriverPath = "eb766f1209af08c3b339756d58653165d55b3c23bc8dbd66438dfb46d8098156" + $IsolatedRestorePath = "a91fed22ae0c2682c8298eb075d6ede5c37608c83ac15f03c55e4bad52e93946" + $IsolatedClusterConfigPath = "e78a5cedd8b211c5cc30777d2c9e52cbc22f37b396eef78d0f39c7842fe7700b" } function Invoke-Agent99SignozFixedRemote { @@ -143,6 +173,23 @@ function New-Agent99SignozPathStateCommand { ) } +function Get-Agent99SignozRemotePathState { + param( + [Parameter(Mandatory = $true)][string]$Path, + [ValidateSet("directory", "file")][string]$ExpectedKind + ) + $remote = Invoke-Agent99SignozFixedRemote ( + New-Agent99SignozPathStateCommand $Path $ExpectedKind + ) + if ( + (Test-Agent99SignozTransportOutcome $remote) -and + [string]$remote.stdout -match "^agent99_path_state:(absent|directory|file|other|symlink)$" + ) { + return [string]$Matches[1] + } + return "unverified" +} + function Convert-Agent99JsonReceipt { param([object]$RemoteResult) if ( @@ -180,7 +227,31 @@ function New-Agent99SignozLockedRemoteCommand { "[ `"`$(/usr/bin/timeout --kill-after=5s 15s /usr/bin/docker inspect --format={{.Config.Image}} $SignozContainer)`" = `"$SignozImage`" ]; " + "[ `"`$(/usr/bin/timeout --kill-after=5s 15s /usr/bin/docker inspect --format={{.State.Running}} $SignozContainer)`" = `"true`" ]; " + "[ `"`$(/usr/bin/timeout --kill-after=5s 15s /usr/bin/curl -sS -o /dev/null -w `"%{http_code}`" $BaseUrl/api/v1/health)`" = `"200`" ]; " + - "set +e; active_output=`$(/usr/bin/timeout --kill-after=5s 10s /usr/bin/pgrep -af `"([s]ignoz-metadata-export[.]py|[s]ignoz-metadata-restore-drill[.]py|(^|/|[[:space:]])([b]ackup-signoz|[c]lickhouse-native-backup|[c]lickhouse-native-restore-drill|[r]un-signoz-backup-canary)[.]sh([[:space:]]|$))`"); active_rc=`$?; set -e; " + + "set +e; active_output=`$(/usr/bin/timeout --kill-after=5s 10s /usr/bin/pgrep -af `"([s]ignoz-metadata-export[.]py|[s]ignoz-metadata-restore-drill[.]py|[s]ignoz-metadata-isolated-restore[.]py|(^|/|[[:space:]])([b]ackup-signoz|[c]lickhouse-native-backup|[c]lickhouse-native-restore-drill|[r]un-signoz-backup-canary)[.]sh([[:space:]]|$))`"); active_rc=`$?; set -e; " + + "[ `"`$active_rc`" -eq 1 ] && [ -z `"`$active_output`" ]; " + + "exec /usr/bin/timeout --signal=TERM --kill-after=$($TargetKillAfterSeconds)s " + + "$($TargetTimeoutSeconds)s $FixedCommand" + ) + $encodedProgram = [Convert]::ToBase64String( + [Text.Encoding]::UTF8.GetBytes($bashProgram) + ) + return ( + "/usr/bin/printf %s $encodedProgram | /usr/bin/base64 -d | " + + "sudo -n /bin/bash" + ) +} + +function New-Agent99SignozCleanupLockedRemoteCommand { + param( + [Parameter(Mandatory = $true)][string]$FixedCommand, + [Parameter(Mandatory = $true)][int]$TargetTimeoutSeconds + ) + $bashProgram = ( + "set -euo pipefail; " + + "[ -f `"$OperationLock`" ] && [ ! -L `"$OperationLock`" ]; " + + "exec 8<`"$OperationLock`"; /usr/bin/flock -n 8; " + + "/usr/bin/timeout --kill-after=5s 15s /usr/bin/docker info >/dev/null; " + + "set +e; active_output=`$(/usr/bin/timeout --kill-after=5s 10s /usr/bin/pgrep -af `"([s]ignoz-metadata-export[.]py|[s]ignoz-metadata-restore-drill[.]py|[s]ignoz-metadata-isolated-restore[.]py|(^|/|[[:space:]])([b]ackup-signoz|[c]lickhouse-native-backup|[c]lickhouse-native-restore-drill|[r]un-signoz-backup-canary)[.]sh([[:space:]]|$))`"); active_rc=`$?; set -e; " + "[ `"`$active_rc`" -eq 1 ] && [ -z `"`$active_output`" ]; " + "exec /usr/bin/timeout --signal=TERM --kill-after=$($TargetKillAfterSeconds)s " + "$($TargetTimeoutSeconds)s $FixedCommand" @@ -256,6 +327,66 @@ function Test-Agent99SignozReceiptIdentity { ) } +function Get-Agent99SignozCleanupPreflight { + $program = ( + "set -euo pipefail; " + + "/usr/bin/echo agent99_hashes_begin; " + + "/usr/bin/sha256sum " + (($ToolchainFiles.Keys) -join " ") + "; " + + "/usr/bin/echo agent99_hashes_end; " + + "/usr/bin/printf agent99_docker:; " + + "/usr/bin/timeout --kill-after=5s 15s /usr/bin/docker info --format={{.ServerVersion}}" + ) + $encoded = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($program)) + $remote = Invoke-Agent99SignozFixedRemote ( + "/usr/bin/printf %s $encoded | /usr/bin/base64 -d | sudo -n /bin/bash" + ) + $lines = if (Test-Agent99SignozTransportOutcome $remote) { + @(([string]$remote.stdout -split "`r?`n") | Where-Object { [string]$_ -ne "" }) + } else { @() } + $shapeValid = [bool]( + $lines.Count -eq $ToolchainFiles.Count + 3 -and + [string]$lines[0] -eq "agent99_hashes_begin" -and + [string]$lines[$lines.Count - 2] -eq "agent99_hashes_end" -and + [string]$lines[$lines.Count - 1] -match "^agent99_docker:[A-Za-z0-9._+-]+$" + ) + $actualHashes = @{} + if ($shapeValid) { + foreach ($line in @($lines[1..($lines.Count - 3)])) { + if ($line -match "^([0-9a-f]{64})\s+(.+)$") { + $path = [string]$Matches[2] + if ($actualHashes.ContainsKey($path) -or -not $ToolchainFiles.Contains($path)) { + $shapeValid = $false + break + } + $actualHashes[$path] = [string]$Matches[1] + } else { + $shapeValid = $false + break + } + } + } + $toolchainExact = [bool]($shapeValid -and $actualHashes.Count -eq $ToolchainFiles.Count) + foreach ($entry in $ToolchainFiles.GetEnumerator()) { + if (-not $actualHashes.ContainsKey([string]$entry.Key) -or $actualHashes[[string]$entry.Key] -ne [string]$entry.Value) { + $toolchainExact = $false + } + } + return [pscustomobject]@{ + ready = [bool]($shapeValid -and $toolchainExact) + transportReadbackVerified = $shapeValid + toolchainBundleId = $BundleId + toolchainFileCount = $ToolchainFiles.Count + toolchainReadbackVerified = $shapeValid + toolchainExact = $toolchainExact + dockerDaemonReadbackVerified = $shapeValid + credentialValueRead = $false + credentialValueOutput = $false + productionHealthRequired = $false + exportArtifactRequired = $false + deterministicOwnershipEnforcedByController = $true + } +} + function Get-Agent99SignozPreflight { param( [ValidateSet("absent", "present")][string]$ExpectedArtifactState, @@ -379,6 +510,65 @@ function Get-Agent99SignozPreflight { $runtimeImageId -eq $SignozImageId -and $runtimeImageRef -eq $SignozImage ) + + # Typed restore-startup observation. It emits only fixed booleans and a + # mount count: no environment value, source mount, secret, or raw Env array. + $restoreStartupProgram = @' +set -euo pipefail +entrypoint="$(/usr/bin/docker inspect --format='{{json .Config.Entrypoint}}' signoz)" +entrypoint_exact=0 +[ "$entrypoint" = '["./signoz","server"]' ] && entrypoint_exact=1 +mounts="$(/usr/bin/docker inspect --format='{{range .Mounts}}{{println .Destination .Type .RW}}{{end}}' signoz)" +mount_count="$(/usr/bin/printf '%s\n' "$mounts" | /usr/bin/awk 'NF { count += 1 } END { print count + 0 }')" +mount_exact=0 +[ "$mount_count" -eq 1 ] && [ "$(/usr/bin/printf '%s\n' "$mounts" | /usr/bin/grep -Fxc '/var/lib/signoz volume true')" -eq 1 ] && mount_exact=1 +dsn_present=0 +dsn_class_exact=0 +sqlite_present=0 +sqlite_exact=0 +if /usr/bin/docker exec signoz /bin/sh -c 'test "${SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN+x}" = x' >/dev/null 2>&1; then dsn_present=1; fi +if /usr/bin/docker exec signoz /bin/sh -c 'case "${SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN:-}" in tcp://clickhouse:9000*) exit 0;; *) exit 1;; esac' >/dev/null 2>&1; then dsn_class_exact=1; fi +if /usr/bin/docker exec signoz /bin/sh -c 'test "${SIGNOZ_SQLSTORE_SQLITE_PATH+x}" = x' >/dev/null 2>&1; then sqlite_present=1; fi +if /usr/bin/docker exec signoz /bin/sh -c 'test "${SIGNOZ_SQLSTORE_SQLITE_PATH:-}" = /var/lib/signoz/signoz.db' >/dev/null 2>&1; then sqlite_exact=1; fi +/usr/bin/printf 'agent99_restore_startup:%s,%s,%s,%s,%s,%s,%s' "$entrypoint_exact" "$mount_count" "$mount_exact" "$dsn_present" "$dsn_class_exact" "$sqlite_present" "$sqlite_exact" +'@ + $restoreStartupEncoded = [Convert]::ToBase64String( + [Text.Encoding]::UTF8.GetBytes($restoreStartupProgram) + ) + $restoreStartup = Invoke-Agent99SignozFixedRemote ( + "/usr/bin/printf %s $restoreStartupEncoded | /usr/bin/base64 -d | sudo -n /bin/bash" + ) + $restoreStartupReadbackVerified = $false + $restoreStartupEntrypointExact = $false + $restoreStartupMountCount = $null + $restoreStartupMountExact = $false + $restoreStartupDsnPresent = $false + $restoreStartupDsnClassExact = $false + $restoreStartupSqlitePathPresent = $false + $restoreStartupSqlitePathExact = $false + if ( + (Test-Agent99SignozTransportOutcome $restoreStartup) -and + [string]$restoreStartup.stdout -match "^agent99_restore_startup:(0|1),([0-9]+),(0|1),(0|1),(0|1),(0|1),(0|1)$" + ) { + $restoreStartupReadbackVerified = $true + $restoreStartupEntrypointExact = [bool]([string]$Matches[1] -eq "1") + $restoreStartupMountCount = [int]$Matches[2] + $restoreStartupMountExact = [bool]([string]$Matches[3] -eq "1") + $restoreStartupDsnPresent = [bool]([string]$Matches[4] -eq "1") + $restoreStartupDsnClassExact = [bool]([string]$Matches[5] -eq "1") + $restoreStartupSqlitePathPresent = [bool]([string]$Matches[6] -eq "1") + $restoreStartupSqlitePathExact = [bool]([string]$Matches[7] -eq "1") + } + $restoreStartupContractReady = [bool]( + $restoreStartupReadbackVerified -and + $restoreStartupEntrypointExact -and + $restoreStartupMountCount -eq 1 -and + $restoreStartupMountExact -and + $restoreStartupDsnPresent -and + $restoreStartupDsnClassExact -and + $restoreStartupSqlitePathPresent -and + $restoreStartupSqlitePathExact + ) $health = Invoke-Agent99SignozFixedRemote ( "sudo -n /bin/bash -c 'set -euo pipefail; " + "/usr/bin/printf `"agent99_health:`"; " + @@ -401,7 +591,7 @@ function Get-Agent99SignozPreflight { $active = Invoke-Agent99SignozFixedRemote ( "sudo -n /bin/bash -c 'set -euo pipefail; set +e; " + - "active_output=`$(/usr/bin/pgrep -fc `"[s]ignoz-metadata-export.py.*--apply`" ); " + + "active_output=`$(/usr/bin/pgrep -fc `"([s]ignoz-metadata-export.py.*--apply|[s]ignoz-metadata-isolated-restore.py.*--apply|[s]ignoz-metadata-restore-drill.py.*--apply)`" ); " + "active_rc=`$?; set -e; " + "[ `"`$active_rc`" -eq 0 ] || [ `"`$active_rc`" -eq 1 ]; " + "/usr/bin/printf `"%s`" `"`$active_output`" | /usr/bin/grep -Eq `"^[0-9]+$`"; " + @@ -454,6 +644,7 @@ function Get-Agent99SignozPreflight { $toolchainReadbackVerified -and $imageReadbackVerified -and $runtimeReadbackVerified -and + $restoreStartupReadbackVerified -and $healthReadbackVerified -and $activeReadbackVerified -and $stateReadable @@ -465,6 +656,7 @@ function Get-Agent99SignozPreflight { $toolchainExact -and $imageExact -and $runtimeImageExact -and + $restoreStartupContractReady -and $healthOk -and $activeOperationsClear -and $stateReadable -and @@ -498,6 +690,20 @@ function Get-Agent99SignozPreflight { signozRuntimeRunning = $runtimeRunning signozRuntimeReadbackVerified = $runtimeReadbackVerified signozRuntimeImageExact = $runtimeImageExact + restoreStartupContractReadbackVerified = $restoreStartupReadbackVerified + restoreStartupContractReady = $restoreStartupContractReady + restoreStartupEntrypointExact = $restoreStartupEntrypointExact + restoreStartupMountDestination = if ($restoreStartupMountExact) { "/var/lib/signoz" } else { $null } + restoreStartupMountCount = $restoreStartupMountCount + restoreStartupMountExact = $restoreStartupMountExact + restoreStartupRequiredKeyPresence = [pscustomobject]@{ + SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN = $restoreStartupDsnPresent + SIGNOZ_SQLSTORE_SQLITE_PATH = $restoreStartupSqlitePathPresent + } + restoreStartupDsnClass = if ($restoreStartupDsnClassExact) { "tcp://clickhouse:9000*" } else { "mismatch_or_unverified" } + restoreStartupSqlitePathClass = if ($restoreStartupSqlitePathExact) { "/var/lib/signoz/signoz.db" } else { "mismatch_or_unverified" } + restoreStartupEnvironmentValuesOutput = $false + restoreStartupFullEnvironmentOutput = $false apiHealthReadbackVerified = $healthReadbackVerified apiHealthOk = $healthOk activeOperationCount = $activeOperationCount @@ -570,16 +776,27 @@ function Write-Agent99SignozTerminal { [object]$ArtifactWritePerformed ) + $isRestoreMode = [bool]($Mode -like "Restore*") + $restoreCompleted = [bool]( + $Terminal -eq "portable_metadata_restore_independently_verified_zero_residue" + ) + $zeroResidueVerified = [bool]( + $Terminal -in @( + "portable_metadata_restore_verified_zero_residue_pending_independent_verifier", + "portable_metadata_restore_independently_verified_zero_residue", + "restore_zero_residue_verified_no_completion" + ) + ) $controlledApplyPhases = [ordered]@{ sensorSource = if ($Preflight) { "observed" } else { "blocked" } normalizedAsset = if ($Preflight) { $CanonicalAsset } else { "blocked" } sourceTruthDiff = if ($Preflight -and $Preflight.toolchainExact) { "exact" } else { "blocked_or_drift" } - aiDecision = "fixed_allowlisted_metadata_export" - riskPolicy = "high_bounded_read_only_source_private_artifact_no_raw_sqlite" + aiDecision = if ($isRestoreMode) { "fixed_isolated_portable_metadata_restore" } else { "fixed_allowlisted_metadata_export" } + riskPolicy = if ($isRestoreMode) { "high_bounded_ephemeral_internal_only_zero_residue" } else { "high_bounded_read_only_source_private_artifact_no_raw_sqlite" } checkMode = if ($CheckReceipt) { [string]$CheckReceipt.terminal } elseif ($Preflight -and $Preflight.ready) { "preflight_pass" } else { "blocked" } boundedExecution = if ($ApplyReceipt) { [string]$ApplyReceipt.terminal } elseif ($Mode -eq "Apply") { "not_verified" } else { "not_run" } independentVerifier = if ($VerifierReceipt) { [string]$VerifierReceipt.terminal } else { "not_run" } - rollbackOrNoWrite = if ($Mode -in @("Check", "Verify")) { "no_write" } elseif ($Terminal -eq "export_verified_restore_pending") { "not_required_additive_immutable_artifact" } else { "artifact_preserved_no_delete" } + rollbackOrNoWrite = if ($Mode -in @("Check", "Verify", "RestoreCheck")) { "no_write" } elseif ($zeroResidueVerified) { "ephemeral_resources_cleaned_zero_residue" } elseif ($Mode -eq "RestoreVerifyCleanup") { "reconciliation_failed_or_incomplete" } elseif ($Terminal -eq "export_verified_restore_pending") { "not_required_additive_immutable_artifact" } else { "artifact_preserved_no_delete" } learningWriteback = "pending_telegram_km_rag_mcp_playbook_ack" } $payload = [pscustomobject]@{ @@ -587,7 +804,7 @@ function Write-Agent99SignozTerminal { canonicalAsset = $CanonicalAsset domain = "backup_observability" executor = "agent99_signoz_metadata_executor" - verifier = "signoz_metadata_export_independent_verifier" + verifier = if ($Mode -eq "RestoreVerifyCleanup") { "signoz_metadata_restore_cleanup_verifier" } elseif ($isRestoreMode) { "pending_distinct_restore_cleanup_verifier" } else { "signoz_metadata_export_independent_verifier" } executionHub = "host:192.168.0.99" dispatchPath = "windows99_agent99_to_host110_fixed_metadata_toolchain" dispatchOnly = $true @@ -599,6 +816,7 @@ function Write-Agent99SignozTerminal { targetHost = $TargetHost toolchainBundleId = $BundleId credentialReference = $CredentialFile + restoreReceipt = if ($isRestoreMode) { $RestoreRemoteReceipt } else { $null } targetOperationLock = $OperationLock targetOperationLockRequiredExistingNonSymlink = $true targetTimeoutKillAfterSeconds = $TargetKillAfterSeconds @@ -619,13 +837,14 @@ function Write-Agent99SignozTerminal { controlledApplyPhases = [pscustomobject]$controlledApplyPhases artifactWritePerformed = $ArtifactWritePerformed productionRuntimeMutationPerformed = $false + isolatedEphemeralRuntimeCreated = [bool]($Mode -eq "RestoreApply" -and $script:IsolatedApplyDispatchAttempted) rawSqliteRead = $false rawVolumeRead = $false secretValueReadByController = $false secretValueTransmitted = $false secretValuePersistedInEvidence = $false - credentialReferenceDispatchAttempted = $script:ApplyDispatchAttempted - credentialReferenceConsumedOnTarget = if (-not $script:ApplyDispatchAttempted) { + credentialReferenceDispatchAttempted = $script:ExportCredentialDispatchAttempted + credentialReferenceConsumedOnTarget = if (-not $script:ExportCredentialDispatchAttempted) { $false } elseif (Test-Agent99SignozReceiptIdentity $ApplyReceipt "terminal" @("partial_degraded")) { $true @@ -634,9 +853,13 @@ function Write-Agent99SignozTerminal { } arbitraryCommandAllowed = $false crossDomainFallbackAllowed = $false - restoreTerminal = "pending_isolated_same_version_target" - zeroResidueTerminal = "pending" - completionClaim = $false + restoreTerminal = if ($restoreCompleted) { "portable_metadata_restore_verified" } elseif ($isRestoreMode) { $Terminal } else { "pending_isolated_same_version_target" } + zeroResidueTerminal = if ($zeroResidueVerified) { "verified" } elseif ($isRestoreMode) { "not_verified" } else { "pending" } + completionScope = if ($restoreCompleted) { "portable_assets_only" } else { $null } + completionClaim = $restoreCompleted + fullSqliteCompletionClaim = $false + rolesPreferencesGlobalConfigRestoreClaim = $false + secretBearingAssetsRestoreClaim = $false telegramAcknowledged = $false kmRagMcpPlaybookAcknowledged = $false transientTransportOutputUsed = $script:TransientTransportOutputUsed @@ -721,6 +944,151 @@ if (-not $runtimeSourceBinding.ok) { Stop-Agent99SignozRun "blocked_with_safe_next_action" "runtime_source_binding_failed" $runtimeSourceBinding $null $null $null $false 65 } +if ($Mode -eq "RestoreVerifyCleanup") { + $preflight = Get-Agent99SignozCleanupPreflight + if (-not $preflight.transportReadbackVerified -or -not $preflight.ready) { + Stop-Agent99SignozRun "restore_cleanup_blocked_with_safe_next_action" "restore_cleanup_toolchain_or_docker_preflight_failed" $preflight $null $null $null $false 1 + } + $restoreReceiptStateBefore = Get-Agent99SignozRemotePathState $RestoreRemoteReceipt "file" + if ($restoreReceiptStateBefore -eq "unverified" -or $restoreReceiptStateBefore -in @("symlink", "other", "directory")) { + Stop-Agent99SignozRun "restore_cleanup_blocked_with_safe_next_action" "restore_receipt_state_unverified_or_unsafe" $preflight $null $null $null $false 1 + } + $restoreProgram = ( + "/usr/bin/python3 $IsolatedRestorePath --verify-cleanup " + + "--policy $PolicyPath --cluster-config $IsolatedClusterConfigPath " + + "--trace-id $TraceId --run-id $RunId --work-item-id $WorkItemId" + ) + if ($restoreReceiptStateBefore -eq "file") { + $restoreProgram += " --apply-receipt-file $RestoreRemoteReceipt" + } + $restoreCommand = New-Agent99SignozCleanupLockedRemoteCommand $restoreProgram $RestoreCleanupTargetTimeoutSeconds + $restoreRemote = Invoke-Agent99SignozFixedRemote $restoreCommand $RestoreCleanupControllerWaitSeconds + $restoreReceipt = Convert-Agent99JsonReceipt $restoreRemote + $transportAccepted = Test-Agent99SignozTransportOutcome $restoreRemote + $failureReceiptAccepted = Test-Agent99SignozReceiptIdentity $restoreReceipt "terminal" @( + "failed_zero_residue_verified_no_completion", + "failed_cleanup_reconciliation_required" + ) + if (-not $transportAccepted) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_cleanup_controller_transport_failed_or_timed_out" $preflight $null $null $restoreReceipt $false 1 + } + if ($failureReceiptAccepted) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_cleanup_controller_failed" $preflight $null $null $restoreReceipt $false 1 + } + $independentCompletion = Test-Agent99SignozReceiptIdentity $restoreReceipt "independent_restore_verifier" @( + "portable_metadata_restore_independently_verified_zero_residue" + ) + $cleanupOnlyPass = Test-Agent99SignozReceiptIdentity $restoreReceipt "independent_restore_verifier" @( + "pass_zero_residue_verified" + ) + if (-not $independentCompletion -and -not $cleanupOnlyPass) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_cleanup_terminal_invalid" $preflight $null $null $restoreReceipt $false 1 + } + if ($independentCompletion -and $restoreReceipt.completionClaim -ne $true) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_cleanup_completion_claim_invalid" $preflight $null $null $restoreReceipt $false 1 + } + if ($cleanupOnlyPass -and $restoreReceipt.completionClaim -ne $false) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_cleanup_only_claim_invalid" $preflight $null $null $restoreReceipt $false 1 + } + $restoreReceiptStateAfter = Get-Agent99SignozRemotePathState $RestoreRemoteReceipt "file" + if ($restoreReceiptStateAfter -ne $restoreReceiptStateBefore) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_cleanup_changed_durable_receipt_state" $preflight $null $null $restoreReceipt $false 1 + } + $postflight = Get-Agent99SignozCleanupPreflight + if (-not $postflight.transportReadbackVerified -or -not $postflight.ready) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_cleanup_postflight_failed" $postflight $null $null $restoreReceipt $false 1 + } + if ($independentCompletion) { + Stop-Agent99SignozRun "portable_metadata_restore_independently_verified_zero_residue" "" $postflight $null $null $restoreReceipt $false 0 + } + Stop-Agent99SignozRun "restore_zero_residue_verified_no_completion" "" $postflight $null $null $restoreReceipt $false 0 +} + +if ($Mode -in @("RestoreCheck", "RestoreApply")) { + $preflight = Get-Agent99SignozPreflight -ExpectedArtifactState "present" -RequireCredential $false + if (-not $preflight.transportReadbackVerified) { + Stop-Agent99SignozRun "restore_blocked_with_safe_next_action" "restore_preflight_transport_unverified" $preflight $null $null $null $false 1 + } + if (-not $preflight.ready) { + Stop-Agent99SignozRun "restore_blocked_with_safe_next_action" "restore_preflight_failed" $preflight $null $null $null $false 1 + } + + $restoreReceiptStateBefore = Get-Agent99SignozRemotePathState $RestoreRemoteReceipt "file" + if ($restoreReceiptStateBefore -eq "unverified" -or $restoreReceiptStateBefore -in @("symlink", "other", "directory")) { + Stop-Agent99SignozRun "restore_blocked_with_safe_next_action" "restore_receipt_state_unverified_or_unsafe" $preflight $null $null $null $false 1 + } + if ($Mode -in @("RestoreCheck", "RestoreApply") -and $restoreReceiptStateBefore -ne "absent") { + Stop-Agent99SignozRun "restore_blocked_with_safe_next_action" "restore_receipt_identity_collision" $preflight $null $null $null $false 66 + } + + $restoreModeArgument = switch ($Mode) { + "RestoreCheck" { "--check" } + "RestoreApply" { "--apply" } + default { "" } + } + if (-not $restoreModeArgument) { + Stop-Agent99SignozRun "restore_blocked_with_safe_next_action" "restore_mode_not_routable" $preflight $null $null $null $false 64 + } + $restoreProgram = ( + "/usr/bin/python3 $IsolatedRestorePath $restoreModeArgument " + + "--bundle-dir $OutputDir --policy $PolicyPath " + + "--cluster-config $IsolatedClusterConfigPath " + + "--trace-id $TraceId --run-id $RunId --work-item-id $WorkItemId" + ) + if ($Mode -eq "RestoreApply") { + $restoreProgram += " --receipt-file $RestoreRemoteReceipt" + $script:IsolatedApplyDispatchAttempted = $true + } + $restoreCommand = New-Agent99SignozLockedRemoteCommand $restoreProgram $TargetProgramTimeoutSeconds "present" + $restoreRemote = Invoke-Agent99SignozFixedRemote $restoreCommand $ControllerWaitSeconds + $restoreReceipt = Convert-Agent99JsonReceipt $restoreRemote + $transportAccepted = Test-Agent99SignozTransportOutcome $restoreRemote + $failureReceiptAccepted = Test-Agent99SignozReceiptIdentity $restoreReceipt "terminal" @( + "failed_zero_residue_verified_no_completion", + "failed_cleanup_reconciliation_required" + ) + if (-not $transportAccepted) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_controller_transport_failed_or_timed_out" $preflight $null $restoreReceipt $null $false 1 + } + if ($failureReceiptAccepted) { + $failureTerminal = if ([string]$restoreReceipt.terminal -eq "failed_zero_residue_verified_no_completion") { + "restore_failed_zero_residue_verified_no_completion" + } else { + "restore_failed_cleanup_reconciliation_required" + } + Stop-Agent99SignozRun $failureTerminal "isolated_restore_controller_failed" $preflight $null $restoreReceipt $null $false 1 + } + + $allowedRestoreTerminal = switch ($Mode) { + "RestoreCheck" { "check_pass_no_write" } + "RestoreApply" { "portable_metadata_restore_verified_zero_residue_pending_independent_verifier" } + } + if (-not (Test-Agent99SignozReceiptIdentity $restoreReceipt "terminal" @($allowedRestoreTerminal))) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_controller_terminal_invalid" $preflight $null $restoreReceipt $null $false 1 + } + + $restoreReceiptStateAfter = Get-Agent99SignozRemotePathState $RestoreRemoteReceipt "file" + if ($Mode -eq "RestoreApply" -and $restoreReceiptStateAfter -ne "file") { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_durable_receipt_missing" $preflight $null $restoreReceipt $null $false 1 + } + if ($Mode -eq "RestoreCheck" -and $restoreReceiptStateAfter -ne "absent") { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_check_wrote_receipt" $preflight $restoreReceipt $null $null $false 1 + } + $postflight = Get-Agent99SignozPreflight -ExpectedArtifactState "present" -RequireCredential $false + if (-not $postflight.transportReadbackVerified -or -not $postflight.ready) { + Stop-Agent99SignozRun "restore_failed_cleanup_reconciliation_required" "restore_postflight_runtime_or_artifact_drift" $postflight $null $restoreReceipt $null $false 1 + } + switch ($Mode) { + "RestoreCheck" { + Stop-Agent99SignozRun "restore_check_ready" "" $postflight $restoreReceipt $null $null $false 0 + } + "RestoreApply" { + Stop-Agent99SignozRun "portable_metadata_restore_verified_zero_residue_pending_independent_verifier" "" $postflight $null $restoreReceipt $null $false 0 + } + } + Stop-Agent99SignozRun "restore_blocked_with_safe_next_action" "restore_mode_fallthrough" $preflight $null $null $null $false 64 +} + if ($Mode -eq "Verify") { $preflight = Get-Agent99SignozPreflight -ExpectedArtifactState "present" -RequireCredential $false if (-not $preflight.transportReadbackVerified) { @@ -790,7 +1158,7 @@ $applyProgram = ( "--receipt-file $RemoteReceipt" ) $applyCommand = New-Agent99SignozLockedRemoteCommand $applyProgram $ApplyTargetTimeoutSeconds "absent" -$script:ApplyDispatchAttempted = $true +$script:ExportCredentialDispatchAttempted = $true $apply = Invoke-Agent99SignozFixedRemote $applyCommand $ApplyControllerWaitSeconds $applyReceipt = Convert-Agent99JsonReceipt $apply if (-not (Test-Agent99SignozTransportOutcome $apply) -or -not (Test-Agent99SignozReceiptIdentity $applyReceipt "terminal" @("partial_degraded"))) { diff --git a/config/signoz/metadata-export-policy.json b/config/signoz/metadata-export-policy.json index 937ca1c62..944bcf16c 100644 --- a/config/signoz/metadata-export-policy.json +++ b/config/signoz/metadata-export-policy.json @@ -270,7 +270,74 @@ "restore_isolation": { "target_url_scope": "loopback_only", "required_image_ref": "signoz/signoz:v0.113.0", + "required_image_id": "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b", "image_id_sha256_required": true, + "images": { + "signoz": { + "ref": "signoz/signoz:v0.113.0", + "id": "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b" + }, + "clickhouse": { + "ref": "clickhouse/clickhouse-server:25.5.6", + "id": "sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6" + }, + "zookeeper": { + "ref": "signoz/zookeeper:3.7.1", + "id": "sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0" + } + }, + "startup_contract": { + "reviewed": true, + "review_source": "agent99_typed_live_nonsecret_preflight_2026-07-22", + "server_entrypoint": [ + "./signoz", + "server" + ], + "server_environment": { + "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN": "tcp://clickhouse:9000", + "SIGNOZ_SQLSTORE_SQLITE_PATH": "/var/lib/signoz/signoz.db" + }, + "server_data_mount_destination": "/var/lib/signoz", + "clickhouse_network_alias": "clickhouse", + "zookeeper_network_alias": "restore-zookeeper", + "required_production_key_presence": [ + "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN", + "SIGNOZ_SQLSTORE_SQLITE_PATH" + ], + "production_dsn_class": "tcp://clickhouse:9000*", + "production_sqlite_path": "/var/lib/signoz/signoz.db", + "secret_value_output_allowed": false, + "full_environment_output_allowed": false + }, + "resource_prefix": "awoooi-signoz-md-restore-", + "canonical_id_prefix": "ephemeral-signoz-metadata-restore-", + "secondary_resource_label_key": "com.awoooi.restore.identity", + "private_workspace_parent": "/tmp", + "server_container_port": 8080, + "loopback_publish_host": "127.0.0.1", + "isolated_cluster_config": "isolated-cluster.xml", + "production_continuity": { + "signoz": { + "container": "signoz", + "image_ref": "signoz/signoz:v0.113.0", + "image_id": "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b" + }, + "clickhouse": { + "container": "signoz-clickhouse", + "image_ref": "clickhouse/clickhouse-server:25.5.6", + "image_id": "sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6" + }, + "zookeeper": { + "container": "signoz-zookeeper-1", + "image_ref": "signoz/zookeeper:3.7.1", + "image_id": "sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0" + }, + "otel_collector": { + "container": "signoz-otel-collector", + "image_ref": "signoz/signoz-otel-collector:v0.144.1", + "image_id": "sha256:47cbed488288df7bac6aab3665275bec6d6e24fe5c545c66914827a891e538d8" + } + }, "image_pull_allowed": false, "production_network_attachment_allowed": false, "production_volume_attachment_allowed": false, diff --git a/scripts/backup/signoz-metadata-isolated-restore.py b/scripts/backup/signoz-metadata-isolated-restore.py new file mode 100755 index 000000000..f9ad99a55 --- /dev/null +++ b/scripts/backup/signoz-metadata-isolated-restore.py @@ -0,0 +1,1721 @@ +#!/usr/bin/env python3 +"""Run a bounded SigNoz portable-metadata restore in an isolated Docker stack. + +The controller creates only deterministic, run-owned Docker resources. It +never pulls or builds an image, never joins a production network or volume, +and removes its three containers, four volumes, internal network and private +workspace on every terminal. Secrets are generated in memory or stored only +in mode-0400 files below the private workspace; they are never command-line +arguments, stdout fields or receipt fields. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import secrets +import shutil +import signal +import socket +import stat +import string +import subprocess # nosec B404 -- fixed executable and argv vectors only +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + +from signoz_metadata_contract import ( + DEFAULT_POLICY, + ContractError, + canonical_json_bytes, + load_policy, + read_json_object, + receipt, + require_no_symlink_components, + sha256_bytes, + validate_identity, + validate_new_private_destination, + write_new_private_atomic, +) + + +DOCKER = "/usr/bin/docker" +PYTHON = "/usr/bin/python3" +RESTORE_DRIVER = Path(__file__).with_name("signoz-metadata-restore-drill.py") +EXPORT_VERIFIER = Path(__file__).with_name("verify-signoz-metadata-export.py") +DEFAULT_CLUSTER_CONFIG = Path(__file__).with_name("isolated-cluster.xml") +PRIMARY_LABEL = "awoooi.signoz.metadata.restore.run_id" +IDENTITY_LABEL = "com.awoooi.restore.identity" +EXPECTED_IMAGES = { + "signoz": ( + "signoz/signoz:v0.113.0", + "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b", + ), + "clickhouse": ( + "clickhouse/clickhouse-server:25.5.6", + "sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6", + ), + "zookeeper": ( + "signoz/zookeeper:3.7.1", + "sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0", + ), +} +EXPECTED_PRODUCTION = { + "signoz": EXPECTED_IMAGES["signoz"], + "signoz-clickhouse": EXPECTED_IMAGES["clickhouse"], + "signoz-zookeeper-1": EXPECTED_IMAGES["zookeeper"], + "signoz-otel-collector": ( + "signoz/signoz-otel-collector:v0.144.1", + "sha256:47cbed488288df7bac6aab3665275bec6d6e24fe5c545c66914827a891e538d8", + ), +} +EXPECTED_ENTRYPOINT = ["./signoz", "server"] +EXPECTED_SERVER_ENV = { + "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN": "tcp://clickhouse:9000", + "SIGNOZ_SQLSTORE_SQLITE_PATH": "/var/lib/signoz/signoz.db", +} +EXPECTED_SERVER_MOUNT = "/var/lib/signoz" +PORT_RETRIES = 120 +PORT_RETRY_INTERVAL_SECONDS = 1.0 +CLICKHOUSE_RETRIES = 90 +CLICKHOUSE_RETRY_INTERVAL_SECONDS = 1.0 +COMMAND_TIMEOUT_SECONDS = 60 +CLEANUP_COMMAND_TIMEOUT_SECONDS = 3 +CLEANUP_RECONCILIATION_ATTEMPTS = 5 +CLEANUP_STABLE_PASSES = 3 +CLEANUP_RECONCILIATION_DELAY_SECONDS = 1.0 + + +class SignalAbort(ContractError): + """A handled termination signal that still requires owned cleanup.""" + + +@dataclass(frozen=True) +class ResourcePlan: + identity: str + short_id: str + prefix: str + canonical_id: str + network: str + zookeeper_container: str + clickhouse_container: str + server_container: str + zookeeper_data_volume: str + zookeeper_log_volume: str + clickhouse_data_volume: str + server_data_volume: str + workspace: Path + + @property + def containers(self) -> tuple[str, ...]: + return ( + self.server_container, + self.clickhouse_container, + self.zookeeper_container, + ) + + @property + def volumes(self) -> tuple[str, ...]: + return ( + self.zookeeper_data_volume, + self.zookeeper_log_volume, + self.clickhouse_data_volume, + self.server_data_volume, + ) + + +@dataclass(frozen=True) +class CommandResult: + returncode: int + stdout: str + + +Runner = Callable[[list[str], int], CommandResult] + + +def derive_identity(trace_id: str, run_id: str, work_item_id: str) -> str: + material = ( + trace_id.encode("utf-8") + + b"\0" + + run_id.encode("utf-8") + + b"\0" + + work_item_id.encode("utf-8") + ) + return hashlib.sha256(material).hexdigest() + + +def build_resource_plan( + trace_id: str, + run_id: str, + work_item_id: str, + *, + workspace_parent: Path = Path("/tmp"), +) -> ResourcePlan: + identity = derive_identity(trace_id, run_id, work_item_id) + short_id = identity[:16] + prefix = f"awoooi-signoz-md-restore-{short_id}" + return ResourcePlan( + identity=identity, + short_id=short_id, + prefix=prefix, + canonical_id=f"ephemeral-signoz-metadata-restore-{short_id}", + network=f"{prefix}-net", + zookeeper_container=f"{prefix}-zookeeper", + clickhouse_container=f"{prefix}-clickhouse", + server_container=f"{prefix}-server", + zookeeper_data_volume=f"{prefix}-zk-data", + zookeeper_log_volume=f"{prefix}-zk-log", + clickhouse_data_volume=f"{prefix}-ch-data", + server_data_volume=f"{prefix}-server-data", + workspace=workspace_parent / f"{prefix}-private", + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument("--check", action="store_true") + mode.add_argument("--apply", action="store_true") + mode.add_argument("--verify-cleanup", action="store_true") + parser.add_argument("--bundle-dir") + parser.add_argument("--policy", default=str(DEFAULT_POLICY)) + parser.add_argument("--cluster-config", default=str(DEFAULT_CLUSTER_CONFIG)) + parser.add_argument("--trace-id", required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--work-item-id", default="P0-OBS-002") + parser.add_argument("--receipt-file") + parser.add_argument("--apply-receipt-file") + args = parser.parse_args() + if args.apply and not args.receipt_file: + parser.error("receipt_file_required_for_restore_apply") + return args + + +def default_runner(argv: list[str], timeout: int) -> CommandResult: + try: + result = subprocess.run( # nosec B603 + argv, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + timeout=timeout, + check=False, + env={ + "PATH": "/usr/sbin:/usr/bin:/sbin:/bin", + "LANG": "C.UTF-8", + "LC_ALL": "C.UTF-8", + }, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ContractError("bounded_command_transport_failed") from exc + return CommandResult(result.returncode, result.stdout) + + +def run_required( + runner: Runner, + argv: list[str], + *, + label: str, + timeout: int = COMMAND_TIMEOUT_SECONDS, +) -> str: + result = runner(argv, timeout) + if result.returncode != 0: + raise ContractError(f"{label}_failed") + return result.stdout.strip() + + +def run_optional( + runner: Runner, + argv: list[str], + *, + timeout: int = COMMAND_TIMEOUT_SECONDS, +) -> CommandResult: + return runner(argv, timeout) + + +def validate_restore_policy(policy: dict[str, Any]) -> dict[str, Any]: + isolation = policy.get("restore_isolation") + if not isinstance(isolation, dict): + raise ContractError("restore_isolation_policy_missing") + if isolation.get("images") != { + key: {"ref": ref, "id": image_id} + for key, (ref, image_id) in EXPECTED_IMAGES.items() + }: + raise ContractError("restore_image_policy_not_exact") + if isolation.get("required_image_ref") != EXPECTED_IMAGES["signoz"][0]: + raise ContractError("restore_signoz_image_ref_not_exact") + if isolation.get("required_image_id") != EXPECTED_IMAGES["signoz"][1]: + raise ContractError("restore_signoz_image_id_not_exact") + if isolation.get("resource_prefix") != "awoooi-signoz-md-restore-": + raise ContractError("restore_resource_prefix_not_exact") + if isolation.get("canonical_id_prefix") != ( + "ephemeral-signoz-metadata-restore-" + ): + raise ContractError("restore_canonical_prefix_not_exact") + if isolation.get("resource_label_key") != PRIMARY_LABEL: + raise ContractError("restore_primary_label_not_exact") + if isolation.get("secondary_resource_label_key") != IDENTITY_LABEL: + raise ContractError("restore_identity_label_not_exact") + if isolation.get("private_workspace_parent") != "/tmp": + raise ContractError("restore_workspace_parent_not_exact") + if isolation.get("server_container_port") != 8080: + raise ContractError("restore_server_port_not_exact") + if isolation.get("loopback_publish_host") != "127.0.0.1": + raise ContractError("restore_publish_host_not_exact") + if any( + isolation.get(key) is not expected + for key, expected in ( + ("image_pull_allowed", False), + ("production_network_attachment_allowed", False), + ("production_volume_attachment_allowed", False), + ("ephemeral_volumes_only", True), + ("cleanup_controller_required", True), + ("zero_residue_verifier_required", True), + ) + ): + raise ContractError("restore_isolation_boolean_contract_invalid") + + startup = isolation.get("startup_contract") + if not isinstance(startup, dict) or startup.get("reviewed") is not True: + raise ContractError("restore_startup_contract_not_reviewed") + if startup.get("server_entrypoint") != EXPECTED_ENTRYPOINT: + raise ContractError("restore_server_entrypoint_not_exact") + if startup.get("server_environment") != EXPECTED_SERVER_ENV: + raise ContractError("restore_server_environment_not_exact") + if startup.get("server_data_mount_destination") != EXPECTED_SERVER_MOUNT: + raise ContractError("restore_server_mount_not_exact") + if startup.get("clickhouse_network_alias") != "clickhouse": + raise ContractError("restore_clickhouse_alias_not_exact") + if startup.get("zookeeper_network_alias") != "restore-zookeeper": + raise ContractError("restore_zookeeper_alias_not_exact") + key_presence = startup.get("required_production_key_presence") + if ( + not isinstance(key_presence, list) + or len(key_presence) != len(EXPECTED_SERVER_ENV) + or sorted(key_presence) != sorted(EXPECTED_SERVER_ENV) + ): + raise ContractError("restore_production_key_presence_not_exact") + if startup.get("production_dsn_class") != "tcp://clickhouse:9000*": + raise ContractError("restore_production_dsn_class_not_exact") + if startup.get("production_sqlite_path") != EXPECTED_SERVER_ENV[ + "SIGNOZ_SQLSTORE_SQLITE_PATH" + ]: + raise ContractError("restore_production_sqlite_path_not_exact") + if startup.get("secret_value_output_allowed") is not False: + raise ContractError("restore_secret_output_must_be_forbidden") + if startup.get("full_environment_output_allowed") is not False: + raise ContractError("restore_full_environment_output_must_be_forbidden") + + expected_continuity = { + key.replace("-", "_"): { + "container": key, + "image_ref": ref, + "image_id": image_id, + } + for key, (ref, image_id) in EXPECTED_PRODUCTION.items() + } + # The policy uses semantic keys for the production containers. + actual_continuity = isolation.get("production_continuity") + if not isinstance(actual_continuity, dict): + raise ContractError("restore_production_continuity_missing") + normalized_actual = { + str(value.get("container", "")).replace("-", "_"): value + for value in actual_continuity.values() + if isinstance(value, dict) + } + if normalized_actual != expected_continuity: + raise ContractError("restore_production_continuity_not_exact") + return isolation + + +def require_regular_cluster_config(path: Path) -> None: + try: + metadata = path.lstat() + except OSError as exc: + raise ContractError("isolated_cluster_config_unavailable") from exc + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + raise ContractError("isolated_cluster_config_not_regular") + content = path.read_text(encoding="utf-8") + for marker in ( + "restore-zookeeper", + "restore-clickhouse", + '', + ): + if marker not in content: + raise ContractError("isolated_cluster_config_contract_invalid") + + +def label_args(plan: ResourcePlan, run_id: str) -> list[str]: + return [ + "--label", + f"{PRIMARY_LABEL}={run_id}", + "--label", + f"{IDENTITY_LABEL}={plan.identity}", + ] + + +def inspect_image_id(runner: Runner, ref: str) -> str: + return run_required( + runner, + [DOCKER, "image", "inspect", "--format={{.Id}}", ref], + label="docker_image_inspect", + ) + + +def verify_exact_images(runner: Runner) -> dict[str, dict[str, str]]: + observed: dict[str, dict[str, str]] = {} + for role, (ref, expected_id) in EXPECTED_IMAGES.items(): + actual_id = inspect_image_id(runner, ref) + if actual_id != expected_id: + raise ContractError(f"restore_image_id_mismatch_{role}") + observed[role] = {"ref": ref, "id": actual_id} + return observed + + +def inspect_json( + runner: Runner, + kind: str, + name: str, + *, + timeout: int = COMMAND_TIMEOUT_SECONDS, +) -> dict[str, Any] | None: + result = run_optional(runner, [DOCKER, kind, "inspect", name], timeout=timeout) + if result.returncode != 0: + return None + try: + value = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise ContractError(f"docker_{kind}_inspect_invalid") from exc + if not isinstance(value, list) or len(value) != 1 or not isinstance(value[0], dict): + raise ContractError(f"docker_{kind}_inspect_invalid") + return value[0] + + +def production_snapshot( + runner: Runner, + *, + timeout: int = COMMAND_TIMEOUT_SECONDS, +) -> dict[str, dict[str, Any]]: + snapshot: dict[str, dict[str, Any]] = {} + for container, (expected_ref, expected_image_id) in EXPECTED_PRODUCTION.items(): + data = inspect_json(runner, "container", container, timeout=timeout) + if data is None: + raise ContractError(f"production_container_missing_{container}") + config = data.get("Config") + state = data.get("State") + if not isinstance(config, dict) or not isinstance(state, dict): + raise ContractError("production_container_inspect_shape_invalid") + row = { + "id": data.get("Id"), + "image_id": data.get("Image"), + "image_ref": config.get("Image"), + "running": state.get("Running"), + "started_at": state.get("StartedAt"), + "restart_count": data.get("RestartCount"), + } + if ( + not isinstance(row["id"], str) + or len(row["id"]) != 64 + or row["image_id"] != expected_image_id + or row["image_ref"] != expected_ref + or row["running"] is not True + or not isinstance(row["started_at"], str) + or not isinstance(row["restart_count"], int) + ): + raise ContractError(f"production_container_identity_invalid_{container}") + snapshot[container] = row + return snapshot + + +def image_inventory( + runner: Runner, + *, + timeout: int = COMMAND_TIMEOUT_SECONDS, +) -> list[str]: + output = run_required( + runner, + [ + DOCKER, + "image", + "ls", + "--no-trunc", + "--format={{.ID}}|{{.Repository}}:{{.Tag}}", + ], + label="docker_image_inventory", + timeout=timeout, + ) + rows = sorted(line for line in output.splitlines() if line) + if not rows: + raise ContractError("docker_image_inventory_empty") + return rows + + +def require_daemon_readable( + runner: Runner, + *, + timeout: int = COMMAND_TIMEOUT_SECONDS, +) -> None: + output = run_required( + runner, + [DOCKER, "info", "--format={{.ServerVersion}}"], + label="docker_daemon_inventory", + timeout=timeout, + ) + if not output or any(character.isspace() for character in output): + raise ContractError("docker_daemon_inventory_invalid") + + +def inspect_resource_labels( + runner: Runner, + kind: str, + name: str, + *, + timeout: int = COMMAND_TIMEOUT_SECONDS, +) -> dict[str, str] | None: + data = inspect_json(runner, kind, name, timeout=timeout) + if data is None: + return None + if kind == "container": + labels = data.get("Config", {}).get("Labels") + else: + labels = data.get("Labels") + if labels is None: + return {} + if not isinstance(labels, dict) or not all( + isinstance(key, str) and isinstance(value, str) + for key, value in labels.items() + ): + raise ContractError(f"docker_{kind}_labels_invalid") + return labels + + +def assert_absent_or_owned( + runner: Runner, + plan: ResourcePlan, + run_id: str, + kind: str, + name: str, +) -> bool: + labels = inspect_resource_labels( + runner, + kind, + name, + timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS, + ) + if labels is None: + return False + if labels.get(PRIMARY_LABEL) != run_id or labels.get(IDENTITY_LABEL) != plan.identity: + raise ContractError(f"restore_{kind}_name_collision_not_owned") + return True + + +def require_no_preexisting_resources( + runner: Runner, + plan: ResourcePlan, + run_id: str, +) -> None: + for name in plan.containers: + if assert_absent_or_owned(runner, plan, run_id, "container", name): + raise ContractError("restore_owned_container_residue_requires_reconcile") + for name in plan.volumes: + if assert_absent_or_owned(runner, plan, run_id, "volume", name): + raise ContractError("restore_owned_volume_residue_requires_reconcile") + if assert_absent_or_owned(runner, plan, run_id, "network", plan.network): + raise ContractError("restore_owned_network_residue_requires_reconcile") + if plan.workspace.exists() or plan.workspace.is_symlink(): + raise ContractError("restore_private_workspace_residue_requires_reconcile") + + +def verify_bundle(args: argparse.Namespace, runner: Runner) -> None: + if not args.bundle_dir: + raise ContractError("bundle_dir_required") + bundle = Path(args.bundle_dir) + output = run_required( + runner, + [ + PYTHON, + str(EXPORT_VERIFIER), + "--bundle-dir", + str(bundle), + "--policy", + str(args.policy), + "--expected-trace-id", + args.trace_id, + "--expected-run-id", + args.run_id, + "--expected-work-item-id", + args.work_item_id, + ], + label="independent_bundle_verifier", + ) + try: + value = json.loads(output) + except json.JSONDecodeError as exc: + raise ContractError("independent_bundle_verifier_receipt_invalid") from exc + if value.get("terminal") != "pass_export_verified_restore_pending": + raise ContractError("independent_bundle_verifier_terminal_invalid") + + +def create_workspace(plan: ResourcePlan) -> None: + plan.workspace.mkdir(mode=0o700) + metadata = plan.workspace.lstat() + if ( + not stat.S_ISDIR(metadata.st_mode) + or metadata.st_uid != os.geteuid() + or stat.S_IMODE(metadata.st_mode) != 0o700 + ): + raise ContractError("restore_workspace_owner_or_mode_invalid") + + +def write_private(path: Path, data: bytes) -> None: + write_new_private_atomic(path, data, label="restore_private_material") + path.chmod(0o400) + + +def write_state(path: Path, value: dict[str, Any]) -> None: + payload = canonical_json_bytes(value) + if path.exists(): + metadata = path.lstat() + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or metadata.st_uid != os.geteuid() + or stat.S_IMODE(metadata.st_mode) != 0o600 + ): + raise ContractError("restore_state_owner_or_mode_invalid") + temp = path.with_name(f".{path.name}.{os.getpid()}.tmp") + write_new_private_atomic(temp, payload, label="restore_state_update") + os.replace(temp, path) + path.chmod(0o600) + else: + write_new_private_atomic(path, payload, label="restore_state") + + +def create_resources( + runner: Runner, + plan: ResourcePlan, + run_id: str, + cluster_config: Path, +) -> None: + labels = label_args(plan, run_id) + run_required( + runner, + [DOCKER, "network", "create", "--internal", *labels, plan.network], + label="restore_network_create", + ) + for name in plan.volumes: + run_required( + runner, + [DOCKER, "volume", "create", *labels, name], + label="restore_volume_create", + ) + + zk_ref = EXPECTED_IMAGES["zookeeper"][0] + run_required( + runner, + [ + DOCKER, + "run", + "--detach", + "--pull=never", + "--restart=no", + "--name", + plan.zookeeper_container, + "--network", + plan.network, + "--network-alias", + "restore-zookeeper", + *labels, + "--env", + "ZOO_MY_ID=1", + "--env", + "ZOO_SERVERS=server.1=0.0.0.0:2888:3888;2181", + "--volume", + f"{plan.zookeeper_data_volume}:/data", + "--volume", + f"{plan.zookeeper_log_volume}:/datalog", + zk_ref, + ], + label="restore_zookeeper_start", + ) + + clickhouse_ref = EXPECTED_IMAGES["clickhouse"][0] + run_required( + runner, + [ + DOCKER, + "run", + "--detach", + "--pull=never", + "--restart=no", + "--name", + plan.clickhouse_container, + "--network", + plan.network, + "--network-alias", + "clickhouse", + "--network-alias", + "restore-clickhouse", + *labels, + "--env", + f"CLICKHOUSE_RESTORE_REPLICA=restore-{plan.short_id}", + "--env", + "CLICKHOUSE_SKIP_USER_SETUP=1", + "--volume", + f"{plan.clickhouse_data_volume}:/var/lib/clickhouse", + "--volume", + f"{cluster_config}:/etc/clickhouse-server/config.d/isolated-cluster.xml:ro", + clickhouse_ref, + ], + label="restore_clickhouse_start", + ) + wait_for_clickhouse(runner, plan) + + signoz_ref = EXPECTED_IMAGES["signoz"][0] + run_required( + runner, + [ + DOCKER, + "run", + "--detach", + "--pull=never", + "--restart=no", + "--name", + plan.server_container, + "--network", + plan.network, + *labels, + "--publish", + "127.0.0.1::8080", + "--env", + "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN=tcp://clickhouse:9000", + "--env", + "SIGNOZ_SQLSTORE_SQLITE_PATH=/var/lib/signoz/signoz.db", + "--volume", + f"{plan.server_data_volume}:/var/lib/signoz", + signoz_ref, + ], + label="restore_signoz_start", + ) + + +def wait_for_clickhouse(runner: Runner, plan: ResourcePlan) -> None: + for attempt in range(CLICKHOUSE_RETRIES): + result = run_optional( + runner, + [ + DOCKER, + "exec", + plan.clickhouse_container, + "clickhouse-client", + "--query", + "SELECT 1", + ], + timeout=5, + ) + if result.returncode == 0 and result.stdout.strip() == "1": + return + if attempt + 1 < CLICKHOUSE_RETRIES: + time.sleep(CLICKHOUSE_RETRY_INTERVAL_SECONDS) + raise ContractError("isolated_clickhouse_readiness_timeout") + + +def verify_runtime_container_images(runner: Runner, plan: ResourcePlan) -> None: + expected = { + plan.server_container: EXPECTED_IMAGES["signoz"], + plan.clickhouse_container: EXPECTED_IMAGES["clickhouse"], + plan.zookeeper_container: EXPECTED_IMAGES["zookeeper"], + } + for name, (expected_ref, expected_id) in expected.items(): + data = inspect_json(runner, "container", name) + if data is None: + raise ContractError("restore_runtime_container_missing") + config = data.get("Config", {}) + host_config = data.get("HostConfig", {}) + if ( + data.get("Image") != expected_id + or config.get("Image") != expected_ref + or host_config.get("RestartPolicy", {}).get("Name") not in {"", "no"} + ): + raise ContractError("restore_runtime_container_image_or_restart_drift") + + +def discover_loopback_port(runner: Runner, plan: ResourcePlan) -> int: + output = run_required( + runner, + [DOCKER, "port", plan.server_container, "8080/tcp"], + label="restore_listener_discovery", + ) + lines = [line.strip() for line in output.splitlines() if line.strip()] + if len(lines) != 1 or not lines[0].startswith("127.0.0.1:"): + raise ContractError("restore_listener_not_loopback_ephemeral") + port_text = lines[0].rsplit(":", 1)[-1] + if not port_text.isdigit(): + raise ContractError("restore_listener_port_invalid") + port = int(port_text) + if port <= 0 or port > 65535 or port == 8080: + raise ContractError("restore_listener_port_not_ephemeral") + return port + + +def request_json( + url: str, + *, + method: str = "GET", + payload: dict[str, Any] | None = None, + authorization: str | None = None, + timeout: int = 10, +) -> dict[str, Any]: + body = canonical_json_bytes(payload) if payload is not None else None + request = urllib.request.Request(url, data=body, method=method) + request.add_header("Accept", "application/json") + if body is not None: + request.add_header("Content-Type", "application/json") + if authorization is not None: + request.add_header("Authorization", authorization) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: # nosec B310 + if response.status < 200 or response.status >= 300: + raise ContractError("isolated_api_unexpected_status") + raw = response.read(1024 * 1024 + 1) + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise ContractError("isolated_api_request_failed") from exc + if len(raw) > 1024 * 1024: + raise ContractError("isolated_api_response_too_large") + try: + value = json.loads(raw) + except (UnicodeError, json.JSONDecodeError) as exc: + raise ContractError("isolated_api_response_invalid") from exc + if not isinstance(value, dict): + raise ContractError("isolated_api_response_not_object") + return value + + +def wait_for_health_and_version(base_url: str) -> None: + last_health_ok = False + for _ in range(PORT_RETRIES): + try: + health = request_json(f"{base_url}/api/v1/health", timeout=2) + last_health_ok = health.get("status") == "ok" + if last_health_ok: + break + except ContractError: + pass + time.sleep(PORT_RETRY_INTERVAL_SECONDS) + if not last_health_ok: + raise ContractError("isolated_signoz_health_timeout") + version = request_json(f"{base_url}/api/v1/version", timeout=5) + if version.get("version") != "v0.113.0": + raise ContractError("isolated_signoz_version_mismatch") + + +def random_password() -> str: + alphabet = string.ascii_letters + string.digits + "-_" + return "R!" + "".join(secrets.choice(alphabet) for _ in range(46)) + + +def bootstrap_isolated_session(base_url: str, workspace: Path) -> Path: + email = f"restore-{secrets.token_hex(12)}@example.invalid" + password = random_password() + register = request_json( + f"{base_url}/api/v1/register", + method="POST", + payload={ + "email": email, + "orgDisplayName": "AWOOOI isolated restore", + "password": password, + }, + ) + data = register.get("data") + register_body = data if isinstance(data, dict) else register + org_id = register_body.get("orgId") + if not isinstance(org_id, str) or not org_id: + raise ContractError("isolated_registration_org_id_missing") + login = request_json( + f"{base_url}/api/v2/sessions/email_password", + method="POST", + payload={"email": email, "password": password, "orgId": org_id}, + ) + login_data = login.get("data") + if not isinstance(login_data, dict): + login_data = login + access_token = ( + login_data.get("accessToken") if isinstance(login_data, dict) else None + ) + refresh_token = ( + login_data.get("refreshToken") if isinstance(login_data, dict) else None + ) + if ( + not isinstance(access_token, str) + or len(access_token) < 16 + or not isinstance(refresh_token, str) + or len(refresh_token) < 16 + ): + raise ContractError("isolated_session_token_shape_invalid") + token_path = workspace / "session-access-token" + write_private(token_path, access_token.encode("ascii") + b"\n") + return token_path + + +def isolation_receipt_value( + args: argparse.Namespace, + plan: ResourcePlan, + base_url: str, + images: dict[str, dict[str, str]], + production_before: dict[str, dict[str, Any]], + image_inventory_before: list[str], +) -> dict[str, Any]: + return { + "schema": "awoooi_signoz_metadata_restore_isolation_v1", + "trace_id": args.trace_id, + "run_id": args.run_id, + "work_item_id": args.work_item_id, + "target": { + "identity": plan.identity, + "prefix": plan.prefix, + "canonical_id": plan.canonical_id, + "base_url_sha256": sha256_bytes(base_url.encode("utf-8")), + "image_ref": EXPECTED_IMAGES["signoz"][0], + "image_id": EXPECTED_IMAGES["signoz"][1], + "images": images, + "image_present_locally": True, + "image_pull_performed": False, + "image_build_performed": False, + "production_network_attached": False, + "production_volume_attached": False, + "ephemeral_volumes_only": True, + "cleanup_controller_armed": True, + "zero_residue_verifier_armed": True, + "network_scope": "loopback_and_internal_only", + "resource_label_key": PRIMARY_LABEL, + "resource_label_value": args.run_id, + "identity_label_key": IDENTITY_LABEL, + "identity_label_value": plan.identity, + "resources": { + "network": plan.network, + "containers": list(plan.containers), + "volumes": list(plan.volumes), + "workspace": str(plan.workspace), + }, + "listener": {"host": "127.0.0.1", "port": int(base_url.rsplit(":", 1)[1])}, + "authentication_mode": "isolated_session_bearer", + }, + "production_before": production_before, + "image_inventory_before": image_inventory_before, + } + + +def invoke_restore_driver( + args: argparse.Namespace, + runner: Runner, + base_url: str, + isolation_receipt: Path, + token_file: Path, + mode: str, +) -> dict[str, Any]: + argv = [ + PYTHON, + str(RESTORE_DRIVER), + mode, + "--bundle-dir", + str(args.bundle_dir), + "--policy", + str(args.policy), + "--target-base-url", + base_url, + "--credential-file", + str(token_file), + "--auth-mode", + "isolated_session_bearer", + "--isolation-receipt", + str(isolation_receipt), + "--trace-id", + args.trace_id, + "--run-id", + args.run_id, + "--work-item-id", + args.work_item_id, + ] + output = run_required( + runner, + argv, + label=f"isolated_restore_driver_{mode.removeprefix('--')}", + timeout=900, + ) + try: + value = json.loads(output) + except json.JSONDecodeError as exc: + raise ContractError("isolated_restore_driver_receipt_invalid") from exc + expected = ( + "check_pass_no_write" + if mode == "--check" + else "partial_degraded_cleanup_pending" + ) + if value.get("terminal") != expected: + raise ContractError("isolated_restore_driver_terminal_invalid") + return value + + +def tcp_connectable(host: str, port: int) -> bool: + try: + with socket.create_connection((host, port), timeout=2): + return True + except OSError: + return False + + +def remove_owned_resource( + runner: Runner, + plan: ResourcePlan, + run_id: str, + kind: str, + name: str, +) -> bool: + labels = inspect_resource_labels(runner, kind, name) + if labels is None: + return False + if labels.get(PRIMARY_LABEL) != run_id or labels.get(IDENTITY_LABEL) != plan.identity: + raise ContractError(f"cleanup_{kind}_ownership_mismatch") + if kind == "container": + argv = [DOCKER, "rm", "--force", name] + elif kind == "volume": + argv = [DOCKER, "volume", "rm", name] + else: + argv = [DOCKER, "network", "rm", name] + run_required( + runner, + argv, + label=f"cleanup_{kind}_remove", + timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS, + ) + return True + + +def remove_private_workspace(plan: ResourcePlan) -> None: + if not plan.workspace.exists() and not plan.workspace.is_symlink(): + return + metadata = plan.workspace.lstat() + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISDIR(metadata.st_mode) + or metadata.st_uid != os.geteuid() + or stat.S_IMODE(metadata.st_mode) != 0o700 + ): + raise ContractError("cleanup_workspace_ownership_or_mode_mismatch") + shutil.rmtree(plan.workspace) + + +def label_inventory_count( + runner: Runner, + kind: str, + label: str, +) -> int: + if kind == "container": + argv = [DOCKER, "ps", "-a", "--quiet", "--filter", f"label={label}"] + else: + argv = [DOCKER, kind, "ls", "--quiet", "--filter", f"label={label}"] + output = run_required( + runner, + argv, + label=f"cleanup_{kind}_inventory", + timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS, + ) + return len([line for line in output.splitlines() if line.strip()]) + + +def capture_owned_listener_before_cleanup( + runner: Runner, + plan: ResourcePlan, + run_id: str, +) -> dict[str, Any] | None: + data = inspect_json( + runner, + "container", + plan.server_container, + timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS, + ) + if data is None: + return None + labels = data.get("Config", {}).get("Labels") + if ( + not isinstance(labels, dict) + or labels.get(PRIMARY_LABEL) != run_id + or labels.get(IDENTITY_LABEL) != plan.identity + ): + raise ContractError("cleanup_server_listener_ownership_mismatch") + ports = data.get("NetworkSettings", {}).get("Ports") + bindings = ports.get("8080/tcp") if isinstance(ports, dict) else None + if not isinstance(bindings, list) or len(bindings) != 1: + raise ContractError("cleanup_server_listener_binding_unknown") + binding = bindings[0] + if not isinstance(binding, dict): + raise ContractError("cleanup_server_listener_binding_invalid") + host = binding.get("HostIp") + port_text = binding.get("HostPort") + if host != "127.0.0.1" or not isinstance(port_text, str) or not port_text.isdigit(): + raise ContractError("cleanup_server_listener_not_loopback") + port = int(port_text) + if port <= 0 or port > 65535: + raise ContractError("cleanup_server_listener_port_invalid") + return {"host": host, "port": port} + + +def cleanup_and_verify( + runner: Runner, + plan: ResourcePlan, + run_id: str, + state: dict[str, Any] | None, +) -> list[str]: + baseline_errors: list[str] = [] + listener: dict[str, Any] | None = None + production_before: dict[str, Any] | None = None + image_inventory_before: list[str] | None = None + if state is not None: + state_listener = state.get("listener") + if isinstance(state_listener, dict): + listener = state_listener + state_production = state.get("production_before") + if isinstance(state_production, dict): + production_before = state_production + state_images = state.get("image_inventory_before") + if isinstance(state_images, list) and all( + isinstance(item, str) for item in state_images + ): + image_inventory_before = state_images + try: + if listener is None: + listener = capture_owned_listener_before_cleanup( + runner, plan, run_id + ) + if production_before is None: + production_before = production_snapshot( + runner, timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS + ) + if image_inventory_before is None: + image_inventory_before = image_inventory( + runner, timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS + ) + except ContractError as exc: + baseline_errors.append(str(exc)) + + stable_passes = 0 + last_errors: list[str] = [] + for attempt in range(CLEANUP_RECONCILIATION_ATTEMPTS): + attempt_errors = list(baseline_errors) + removed_during_attempt = False + + # Re-run removal on every pass. A Docker API request interrupted after + # daemon acceptance may materialize a labeled resource after the first + # inventory; repeated ownership-gated removal closes that race. + for name in plan.containers: + try: + removed_during_attempt = ( + remove_owned_resource(runner, plan, run_id, "container", name) + or removed_during_attempt + ) + except ContractError as exc: + attempt_errors.append(str(exc)) + for name in plan.volumes: + try: + removed_during_attempt = ( + remove_owned_resource(runner, plan, run_id, "volume", name) + or removed_during_attempt + ) + except ContractError as exc: + attempt_errors.append(str(exc)) + try: + removed_during_attempt = ( + remove_owned_resource(runner, plan, run_id, "network", plan.network) + or removed_during_attempt + ) + except ContractError as exc: + attempt_errors.append(str(exc)) + + try: + require_daemon_readable( + runner, timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS + ) + for name in plan.containers: + if inspect_json( + runner, + "container", + name, + timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS, + ) is not None: + attempt_errors.append("cleanup_exact_container_residue") + for name in plan.volumes: + if inspect_json( + runner, + "volume", + name, + timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS, + ) is not None: + attempt_errors.append("cleanup_exact_volume_residue") + if inspect_json( + runner, + "network", + plan.network, + timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS, + ) is not None: + attempt_errors.append("cleanup_exact_network_residue") + for label in ( + f"{PRIMARY_LABEL}={run_id}", + f"{IDENTITY_LABEL}={plan.identity}", + ): + for kind in ("container", "volume", "network"): + if label_inventory_count(runner, kind, label) != 0: + attempt_errors.append(f"cleanup_{kind}_label_residue") + except ContractError as exc: + attempt_errors.append(str(exc)) + + if listener is not None: + host = listener.get("host") + port = listener.get("port") + if isinstance(host, str) and isinstance(port, int): + if tcp_connectable(host, port): + attempt_errors.append("cleanup_tcp_listener_still_present") + else: + attempt_errors.append("cleanup_listener_state_invalid") + if production_before is None or image_inventory_before is None: + attempt_errors.append("cleanup_continuity_baseline_unavailable") + else: + try: + if production_snapshot( + runner, timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS + ) != production_before: + attempt_errors.append( + "cleanup_production_identity_continuity_failed" + ) + if image_inventory( + runner, timeout=CLEANUP_COMMAND_TIMEOUT_SECONDS + ) != image_inventory_before: + attempt_errors.append("cleanup_image_inventory_drift") + except ContractError as exc: + attempt_errors.append(str(exc)) + + try: + remove_private_workspace(plan) + except ContractError as exc: + attempt_errors.append(str(exc)) + if plan.workspace.exists() or plan.workspace.is_symlink(): + attempt_errors.append("cleanup_private_workspace_residue") + + if removed_during_attempt: + attempt_errors.append("cleanup_owned_resource_reappeared") + + last_errors = sorted(set(attempt_errors)) + if last_errors: + stable_passes = 0 + else: + stable_passes += 1 + if stable_passes >= CLEANUP_STABLE_PASSES: + return [] + if attempt + 1 < CLEANUP_RECONCILIATION_ATTEMPTS: + time.sleep(CLEANUP_RECONCILIATION_DELAY_SECONDS) + + return sorted(set(last_errors + ["cleanup_stable_zero_residue_not_observed"])) + + +def load_state_if_safe(plan: ResourcePlan) -> dict[str, Any] | None: + path = plan.workspace / "state.json" + if not path.exists(): + return None + value = read_json_object(path, label="restore_state") + if value.get("schema") != "awoooi_signoz_metadata_restore_state_v1": + raise ContractError("restore_state_schema_invalid") + if value.get("identity") != plan.identity: + raise ContractError("restore_state_identity_mismatch") + if not isinstance(value.get("production_before"), dict): + raise ContractError("restore_state_production_baseline_invalid") + images = value.get("image_inventory_before") + if not isinstance(images, list) or not all(isinstance(item, str) for item in images): + raise ContractError("restore_state_image_baseline_invalid") + listener = value.get("listener") + if listener is not None and ( + not isinstance(listener, dict) + or listener.get("host") != "127.0.0.1" + or not isinstance(listener.get("port"), int) + or listener["port"] <= 0 + or listener["port"] > 65535 + ): + raise ContractError("restore_state_listener_invalid") + return value + + +def run_check( + args: argparse.Namespace, + runner: Runner, + plan: ResourcePlan, + policy: dict[str, Any], + cluster_config: Path, +) -> dict[str, Any]: + validate_restore_policy(policy) + require_regular_cluster_config(cluster_config) + verify_bundle(args, runner) + require_daemon_readable(runner) + verify_exact_images(runner) + production_snapshot(runner) + image_inventory(runner) + require_no_preexisting_resources(runner, plan, args.run_id) + value = receipt( + trace_id=args.trace_id, + run_id=args.run_id, + work_item_id=args.work_item_id, + phase="terminal", + terminal="check_pass_no_write", + detail="exact_images_startup_contract_production_continuity_and_zero_collision_pass", + ) + value.update( + { + "identity": plan.identity, + "resourcePrefix": plan.prefix, + "resourceWritePerformed": False, + "secretValueRead": False, + "completionClaim": False, + } + ) + return value + + +def run_apply( + args: argparse.Namespace, + runner: Runner, + plan: ResourcePlan, + policy: dict[str, Any], + cluster_config: Path, +) -> dict[str, Any]: + run_check(args, runner, plan, policy, cluster_config) + create_workspace(plan) + state_path = plan.workspace / "state.json" + images = verify_exact_images(runner) + before = production_snapshot(runner) + inventory_before = image_inventory(runner) + state: dict[str, Any] = { + "schema": "awoooi_signoz_metadata_restore_state_v1", + "identity": plan.identity, + "trace_id": args.trace_id, + "run_id": args.run_id, + "work_item_id": args.work_item_id, + "production_before": before, + "image_inventory_before": inventory_before, + "listener": None, + } + write_state(state_path, state) + create_resources(runner, plan, args.run_id, cluster_config) + verify_runtime_container_images(runner, plan) + port = discover_loopback_port(runner, plan) + state["listener"] = {"host": "127.0.0.1", "port": port} + write_state(state_path, state) + base_url = f"http://127.0.0.1:{port}" + wait_for_health_and_version(base_url) + token_path = bootstrap_isolated_session(base_url, plan.workspace) + isolation_path = plan.workspace / "isolation-receipt.json" + isolation = isolation_receipt_value( + args, + plan, + base_url, + images, + before, + inventory_before, + ) + write_new_private_atomic( + isolation_path, + canonical_json_bytes(isolation), + label="isolation_receipt", + ) + invoke_restore_driver( + args, + runner, + base_url, + isolation_path, + token_path, + "--check", + ) + restore = invoke_restore_driver( + args, + runner, + base_url, + isolation_path, + token_path, + "--apply", + ) + value = receipt( + trace_id=args.trace_id, + run_id=args.run_id, + work_item_id=args.work_item_id, + phase="terminal", + terminal="portable_metadata_restore_verified_cleanup_pending", + detail="portable_assets_semantic_readback_pass_cleanup_pending", + ) + value["restoreSemanticDigest"] = restore.get("portable_semantic_sha256") + value["restoredAssetCounts"] = restore.get("restored_asset_counts") + value["completionClaim"] = False + return value + + +def executor_success( + args: argparse.Namespace, + plan: ResourcePlan, + restore_result: dict[str, Any], +) -> dict[str, Any]: + value = receipt( + trace_id=args.trace_id, + run_id=args.run_id, + work_item_id=args.work_item_id, + phase="terminal", + terminal=( + "portable_metadata_restore_verified_zero_residue_" + "pending_independent_verifier" + ), + detail=( + "portable_assets_semantic_parity_and_executor_cleanup_verified_" + "pending_independent_verifier" + ), + ) + value.update( + { + "identity": plan.identity, + "resourcePrefix": plan.prefix, + "completionScope": "portable_assets_only", + "portableAssetIds": [ + "dashboards", + "alert_rules", + "explorer_views", + ], + "completionClaim": False, + "executorZeroResidueVerified": True, + "independentZeroResidueVerified": False, + "fullSqliteCompletionClaim": False, + "rolesPreferencesGlobalConfigRestoreClaim": False, + "secretBearingAssetsRestoreClaim": False, + "productionMutationPerformed": False, + "imagePullPerformed": False, + "imageBuildPerformed": False, + "secretValueInArguments": False, + "secretValueInReceipt": False, + "restoreSemanticDigest": restore_result.get("restoreSemanticDigest"), + "restoredAssetCounts": restore_result.get("restoredAssetCounts"), + "zeroResidue": { + "exactContainers": 0, + "exactVolumes": 0, + "exactNetworks": 0, + "labelInventory": 0, + "tcpListener": 0, + "privateWorkspace": 0, + }, + } + ) + return value + + +def validate_apply_receipt( + path: Path, + args: argparse.Namespace, + plan: ResourcePlan, +) -> dict[str, Any]: + if not path.is_absolute(): + raise ContractError("apply_receipt_path_must_be_absolute") + require_no_symlink_components(path.parent, label="apply_receipt_parent") + try: + metadata = path.lstat() + except OSError as exc: + raise ContractError("apply_receipt_unavailable") from exc + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + raise ContractError("apply_receipt_not_regular") + if metadata.st_uid != os.geteuid(): + raise ContractError("apply_receipt_owner_mismatch") + if stat.S_IMODE(metadata.st_mode) not in {0o400, 0o600}: + raise ContractError("apply_receipt_mode_invalid") + if metadata.st_size <= 0 or metadata.st_size > 65536: + raise ContractError("apply_receipt_size_invalid") + value = read_json_object(path, label="apply_receipt") + expected = { + "schema": "awoooi_signoz_metadata_receipt_v1", + "trace_id": args.trace_id, + "run_id": args.run_id, + "work_item_id": args.work_item_id, + "phase": "terminal", + "terminal": ( + "portable_metadata_restore_verified_zero_residue_" + "pending_independent_verifier" + ), + "identity": plan.identity, + "resourcePrefix": plan.prefix, + "completionClaim": False, + "executorZeroResidueVerified": True, + "independentZeroResidueVerified": False, + "completionScope": "portable_assets_only", + "fullSqliteCompletionClaim": False, + "rolesPreferencesGlobalConfigRestoreClaim": False, + "secretBearingAssetsRestoreClaim": False, + "productionMutationPerformed": False, + "imagePullPerformed": False, + "imageBuildPerformed": False, + "secretValueInArguments": False, + "secretValueInReceipt": False, + } + if any(value.get(key) != expected_value for key, expected_value in expected.items()): + raise ContractError("apply_receipt_identity_or_terminal_invalid") + digest = value.get("restoreSemanticDigest") + counts = value.get("restoredAssetCounts") + zero_residue = value.get("zeroResidue") + if ( + not isinstance(digest, str) + or len(digest) != 64 + or any(character not in "0123456789abcdef" for character in digest) + or not isinstance(counts, dict) + or sorted(counts) != ["alert_rules", "dashboards", "explorer_views"] + or not all(isinstance(count, int) and count >= 0 for count in counts.values()) + or zero_residue + != { + "exactContainers": 0, + "exactVolumes": 0, + "exactNetworks": 0, + "labelInventory": 0, + "tcpListener": 0, + "privateWorkspace": 0, + } + ): + raise ContractError("apply_receipt_semantic_evidence_invalid") + return value + + +def independent_success( + args: argparse.Namespace, + plan: ResourcePlan, + apply_receipt: dict[str, Any], +) -> dict[str, Any]: + value = receipt( + trace_id=args.trace_id, + run_id=args.run_id, + work_item_id=args.work_item_id, + phase="independent_restore_verifier", + terminal="portable_metadata_restore_independently_verified_zero_residue", + detail="immutable_apply_receipt_and_stable_zero_residue_readback_pass", + ) + value.update( + { + "verifierIdentity": "signoz_metadata_restore_cleanup_verifier", + "applyReceiptValidated": True, + "applyReceiptSha256": sha256_bytes( + canonical_json_bytes(apply_receipt) + ), + "identity": plan.identity, + "resourcePrefix": plan.prefix, + "completionScope": "portable_assets_only", + "completionClaim": True, + "executorZeroResidueVerified": True, + "independentZeroResidueVerified": True, + "fullSqliteCompletionClaim": False, + "rolesPreferencesGlobalConfigRestoreClaim": False, + "secretBearingAssetsRestoreClaim": False, + "restoreSemanticDigest": apply_receipt["restoreSemanticDigest"], + "restoredAssetCounts": apply_receipt["restoredAssetCounts"], + "zeroResidue": { + "stablePasses": CLEANUP_STABLE_PASSES, + "exactContainers": 0, + "exactVolumes": 0, + "exactNetworks": 0, + "labelInventory": 0, + "tcpListener": 0, + "privateWorkspace": 0, + }, + } + ) + return value + + +def write_terminal(path: Path, value: dict[str, Any]) -> None: + write_new_private_atomic(path, canonical_json_bytes(value), label="restore_receipt") + + +def emit(value: dict[str, Any]) -> None: + print(json.dumps(value, sort_keys=True, separators=(",", ":"))) + + +def main() -> int: + args = parse_args() + runner = default_runner + plan: ResourcePlan | None = None + cleanup_errors: list[str] = [] + restore_result: dict[str, Any] | None = None + caught_error: str | None = None + old_handlers: dict[int, Any] = {} + cleanup_reconciliation_completed = False + signals_held_until_terminal = False + + def handle_signal(signum: int, _frame: Any) -> None: + # The first handled signal transfers control to finally. Ignore later + # handled signals so the owned cleanup cannot be interrupted mid-way. + for handled in (signal.SIGHUP, signal.SIGINT, signal.SIGTERM): + signal.signal(handled, signal.SIG_IGN) + raise SignalAbort(f"signal_{signum}_cleanup_required") + + def restore_signal_handlers() -> None: + for signum, handler in old_handlers.items(): + signal.signal(signum, handler) + + def hold_cleanup_signals() -> None: + nonlocal signals_held_until_terminal + for handled in (signal.SIGHUP, signal.SIGINT, signal.SIGTERM): + signal.signal(handled, signal.SIG_IGN) + signals_held_until_terminal = True + + try: + for value, label in ( + (args.trace_id, "trace_id"), + (args.run_id, "run_id"), + (args.work_item_id, "work_item_id"), + ): + validate_identity(value, label=label) + if args.work_item_id != "P0-OBS-002": + raise ContractError("work_item_not_allowlisted") + if args.receipt_file: + validate_new_private_destination( + Path(args.receipt_file), label="restore_receipt" + ) + plan = build_resource_plan(args.trace_id, args.run_id, args.work_item_id) + policy = load_policy(Path(args.policy)) + cluster_config = Path(args.cluster_config) + for signum in (signal.SIGHUP, signal.SIGINT, signal.SIGTERM): + old_handlers[signum] = signal.getsignal(signum) + signal.signal(signum, handle_signal) + + if args.check: + result = run_check(args, runner, plan, policy, cluster_config) + if args.receipt_file: + write_terminal(Path(args.receipt_file), result) + emit(result) + return 0 + + if args.verify_cleanup: + try: + state = load_state_if_safe(plan) + except (ContractError, OSError, UnicodeError) as exc: + state = None + cleanup_errors.append(str(exc)) + reconciliation_errors = cleanup_and_verify( + runner, plan, args.run_id, state + ) + cleanup_errors.extend(reconciliation_errors) + cleanup_reconciliation_completed = True + if cleanup_errors: + raise ContractError("cleanup_reconciliation_incomplete") + if args.apply_receipt_file: + apply_receipt = validate_apply_receipt( + Path(args.apply_receipt_file), args, plan + ) + result = independent_success(args, plan, apply_receipt) + else: + result = receipt( + trace_id=args.trace_id, + run_id=args.run_id, + work_item_id=args.work_item_id, + phase="independent_restore_verifier", + terminal="pass_zero_residue_verified", + detail="deterministic_reconciliation_zero_residue_pass", + ) + result["completionClaim"] = False + result["productionContinuityScope"] = ( + "reconciliation_window_only" + ) + result["restoreCompletionClaim"] = False + if args.receipt_file: + write_terminal(Path(args.receipt_file), result) + emit(result) + return 0 + + restore_result = run_apply(args, runner, plan, policy, cluster_config) + except (ContractError, OSError, UnicodeError) as exc: + caught_error = str(exc) + finally: + cleanup_retry_required = bool( + plan is not None + and ( + args.apply + or (args.verify_cleanup and not cleanup_reconciliation_completed) + ) + ) + if cleanup_retry_required and plan is not None: + # The target timeout sends TERM before its kill-after window. Once + # reconciliation starts here, keep all handled signals ignored + # through the durable terminal write so cleanup cannot be cut in + # half or separated from its receipt. + hold_cleanup_signals() + try: + state = load_state_if_safe(plan) + if state is None and args.apply: + cleanup_errors.append( + "restore_state_missing_for_apply_continuity" + ) + except (ContractError, OSError, UnicodeError) as exc: + state = None + cleanup_errors.append(str(exc)) + try: + reconciliation_errors = cleanup_and_verify( + runner, plan, args.run_id, state + ) + cleanup_errors.extend(reconciliation_errors) + cleanup_reconciliation_completed = True + except (ContractError, OSError, UnicodeError) as exc: + cleanup_errors.append(str(exc)) + if plan is not None and args.verify_cleanup and caught_error is not None: + # Cleanup may have completed just before a signal or verifier + # receipt validation failure. Preserve the failure receipt window + # even when no second reconciliation pass is required. + hold_cleanup_signals() + if not args.apply and not signals_held_until_terminal: + restore_signal_handlers() + + if plan is None: + return 1 + if caught_error is None and not cleanup_errors and restore_result is not None: + result = executor_success(args, plan, restore_result) + exit_code = 0 + else: + error_code = caught_error or "cleanup_reconciliation_incomplete" + result = receipt( + trace_id=args.trace_id, + run_id=args.run_id, + work_item_id=args.work_item_id, + phase="terminal", + terminal=( + "failed_zero_residue_verified_no_completion" + if not cleanup_errors + else "failed_cleanup_reconciliation_required" + ), + detail=f"isolated_restore_failed_{error_code}", + ) + result.update( + { + "completionClaim": False, + "fullSqliteCompletionClaim": False, + "secretValueInArguments": False, + "secretValueInReceipt": False, + "zeroResidueVerified": not cleanup_errors, + "cleanupErrorCodes": cleanup_errors, + } + ) + exit_code = 1 + try: + if args.receipt_file: + write_terminal(Path(args.receipt_file), result) + emit(result) + return exit_code + except (ContractError, OSError): + return 74 + finally: + # Restore signal behavior only after the durable terminal and stdout + # receipt are complete. A second TERM cannot interrupt cleanup or the + # write-once receipt persistence window. + restore_signal_handlers() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/backup/signoz-metadata-restore-drill.py b/scripts/backup/signoz-metadata-restore-drill.py index a0e07ebcd..b89648d43 100755 --- a/scripts/backup/signoz-metadata-restore-drill.py +++ b/scripts/backup/signoz-metadata-restore-drill.py @@ -4,21 +4,23 @@ from __future__ import annotations import argparse +import hashlib import json import os +import socket import stat # Child commands use fixed argv without a shell. import subprocess # nosec B404 import sys import urllib.error +import urllib.parse import urllib.request from pathlib import Path from typing import Any from signoz_metadata_contract import ( DEFAULT_POLICY, - SHA256, ApiClient, ContractError, canonical_json_bytes, @@ -49,6 +51,11 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--policy", default=str(DEFAULT_POLICY)) parser.add_argument("--target-base-url", required=True) parser.add_argument("--credential-file") + parser.add_argument( + "--auth-mode", + choices=("signoz_api_key", "isolated_session_bearer"), + default="signoz_api_key", + ) parser.add_argument("--isolation-receipt", required=True) parser.add_argument("--trace-id", required=True) parser.add_argument("--run-id", required=True) @@ -57,6 +64,57 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() +EXPECTED_IMAGES = { + "signoz": { + "ref": "signoz/signoz:v0.113.0", + "id": "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b", + }, + "clickhouse": { + "ref": "clickhouse/clickhouse-server:25.5.6", + "id": "sha256:8248e5926d7304400e44ecdf0c7181fbde28e6a9b1d647780eca839f34c00cc6", + }, + "zookeeper": { + "ref": "signoz/zookeeper:3.7.1", + "id": "sha256:3ab0e8f032ab58f14ac1e929ac40305a4a464cf320bdfe4151d62d6922729ba0", + }, +} +PRIMARY_LABEL = "awoooi.signoz.metadata.restore.run_id" +IDENTITY_LABEL = "com.awoooi.restore.identity" + + +def restore_identity(trace_id: str, run_id: str, work_item_id: str) -> str: + return hashlib.sha256( + trace_id.encode("utf-8") + + b"\0" + + run_id.encode("utf-8") + + b"\0" + + work_item_id.encode("utf-8") + ).hexdigest() + + +def expected_resources(identity: str) -> dict[str, Any]: + short_id = identity[:16] + prefix = f"awoooi-signoz-md-restore-{short_id}" + return { + "identity": identity, + "prefix": prefix, + "canonical_id": f"ephemeral-signoz-metadata-restore-{short_id}", + "network": f"{prefix}-net", + "containers": [ + f"{prefix}-server", + f"{prefix}-clickhouse", + f"{prefix}-zookeeper", + ], + "volumes": [ + f"{prefix}-zk-data", + f"{prefix}-zk-log", + f"{prefix}-ch-data", + f"{prefix}-server-data", + ], + "workspace": f"/tmp/{prefix}-private", + } + + def run_independent_bundle_verifier(args: argparse.Namespace) -> None: command = [ sys.executable, @@ -122,34 +180,60 @@ def validate_isolation_receipt( if not isinstance(target, dict): raise ContractError("isolation_target_missing") isolation_policy = policy["restore_isolation"] + identity = restore_identity(args.trace_id, args.run_id, args.work_item_id) + resources = expected_resources(identity) target_url = validate_base_url(args.target_base_url, loopback_only=True) if target.get("base_url_sha256") != sha256_bytes(target_url.encode("utf-8")): raise ContractError("isolation_target_url_hash_mismatch") - canonical_id = target.get("canonical_id") - if not isinstance(canonical_id, str) or not canonical_id.startswith( - "ephemeral-signoz-metadata-restore-" - ): + if target.get("identity") != identity: + raise ContractError("isolation_target_identity_mismatch") + if target.get("prefix") != resources["prefix"]: + raise ContractError("isolation_target_prefix_mismatch") + if target.get("canonical_id") != resources["canonical_id"]: raise ContractError("isolation_target_canonical_id_invalid") if target.get("image_ref") != isolation_policy["required_image_ref"]: raise ContractError("isolation_target_image_ref_mismatch") - image_id = target.get("image_id") + if target.get("image_id") != isolation_policy["required_image_id"]: + raise ContractError("isolation_target_image_id_mismatch") + if target.get("images") != EXPECTED_IMAGES: + raise ContractError("isolation_target_image_set_mismatch") + if target.get("resources") != { + "network": resources["network"], + "containers": resources["containers"], + "volumes": resources["volumes"], + "workspace": resources["workspace"], + }: + raise ContractError("isolation_target_resource_names_mismatch") + listener = target.get("listener") if ( - not isinstance(image_id, str) - or not image_id.startswith("sha256:") - or not SHA256.fullmatch(image_id.removeprefix("sha256:")) + not isinstance(listener, dict) + or listener.get("host") != "127.0.0.1" + or not isinstance(listener.get("port"), int) + or listener["port"] <= 0 + or listener["port"] > 65535 ): - raise ContractError("isolation_target_image_id_invalid") + raise ContractError("isolation_target_listener_invalid") + expected_url = f"http://127.0.0.1:{listener['port']}" + if target_url != expected_url: + raise ContractError("isolation_target_listener_url_mismatch") + if args.auth_mode != "isolated_session_bearer": + raise ContractError("restore_auth_mode_must_be_isolated_session_bearer") + if target.get("authentication_mode") != args.auth_mode: + raise ContractError("isolation_target_auth_mode_mismatch") required_values = { "image_present_locally": True, "image_pull_performed": False, + "image_build_performed": False, "production_network_attached": False, "production_volume_attached": False, "ephemeral_volumes_only": True, "cleanup_controller_armed": True, "zero_residue_verifier_armed": True, "network_scope": "loopback_and_internal_only", - "resource_label_key": isolation_policy["resource_label_key"], + "resource_label_key": PRIMARY_LABEL, "resource_label_value": args.run_id, + "identity_label_key": IDENTITY_LABEL, + "identity_label_value": identity, } for key, expected in required_values.items(): if target.get(key) != expected: @@ -201,9 +285,14 @@ def make_client( return None if token is None: raise ContractError("credential_value_unavailable") + if args.auth_mode == "isolated_session_bearer": + header_name = "Authorization" + token = f"Bearer {token}" + else: + header_name = policy["authentication"]["header_name"] return ApiClient( base_url=validate_base_url(args.target_base_url, loopback_only=True), - header_name=policy["authentication"]["header_name"], + header_name=header_name, token=token, timeout_seconds=policy["limits"]["request_timeout_seconds"], max_response_bytes=policy["limits"]["max_response_bytes"], @@ -278,6 +367,30 @@ def run_apply( expected = {asset["id"]: items for asset, items in assets} if canonical_json_bytes(after) != canonical_json_bytes(expected): raise ContractError("restore_semantic_readback_mismatch") + restored_asset_counts: dict[str, int] = {} + semantic_sha256: dict[str, str] = {} + for asset, items in assets: + asset_id = asset["id"] + restored_items = after[asset_id] + expected_digest = sha256_bytes(canonical_json_bytes(items)) + restored_digest = sha256_bytes(canonical_json_bytes(restored_items)) + if len(restored_items) != len(items): + raise ContractError(f"restore_semantic_count_mismatch_{asset_id}") + if restored_digest != expected_digest: + raise ContractError(f"restore_semantic_digest_mismatch_{asset_id}") + restored_asset_counts[asset_id] = len(restored_items) + semantic_sha256[asset_id] = restored_digest + portable_semantic_sha256 = sha256_bytes( + canonical_json_bytes( + { + asset_id: { + "count": restored_asset_counts[asset_id], + "sha256": semantic_sha256[asset_id], + } + for asset_id in sorted(restored_asset_counts) + } + ) + ) terminal = receipt( trace_id=args.trace_id, run_id=args.run_id, @@ -289,6 +402,14 @@ def run_apply( "cleanup_and_zero_residue_verifier_pending" ), ) + terminal["restored_asset_counts"] = restored_asset_counts + terminal["restored_asset_semantic_sha256"] = semantic_sha256 + terminal["portable_semantic_sha256"] = portable_semantic_sha256 + terminal["completion_scope"] = "portable_assets_only" + terminal["completion_claim"] = False + terminal["full_sqlite_completion_claim"] = False + terminal["roles_preferences_global_config_restore_claim"] = False + terminal["secret_bearing_assets_restore_claim"] = False terminal["phase_receipts"] = [ receipt( trace_id=args.trace_id, @@ -397,41 +518,153 @@ def docker_label_count(resource: str, label: str) -> int: return len([line for line in result.stdout.splitlines() if line.strip()]) -def require_target_unreachable(base_url: str) -> None: - base_url = validate_base_url(base_url, loopback_only=True) - request = urllib.request.Request(base_url, method="GET") +def docker_inspect(resource: str, name: str) -> dict[str, Any] | None: + command = ["docker", resource, "inspect", name] try: - # base_url is validated as loopback HTTP(S) immediately above. - with urllib.request.urlopen( # nosec B310 - request, timeout=2 - ): - pass - except urllib.error.HTTPError as exc: - raise ContractError("cleanup_target_listener_still_present") from exc - except (urllib.error.URLError, TimeoutError, OSError): + result = subprocess.run( # nosec B603 + command, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + timeout=15, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ContractError(f"cleanup_{resource}_inventory_failed") from exc + if result.returncode != 0: + return None + try: + value = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise ContractError(f"cleanup_{resource}_inventory_invalid") from exc + if not isinstance(value, list) or len(value) != 1 or not isinstance(value[0], dict): + raise ContractError(f"cleanup_{resource}_inventory_invalid") + return value[0] + + +def require_target_tcp_unreachable(base_url: str) -> None: + base_url = validate_base_url(base_url, loopback_only=True) + parsed = urllib.parse.urlsplit(base_url) + if parsed.hostname != "127.0.0.1" or parsed.port is None: + raise ContractError("cleanup_target_listener_invalid") + try: + with socket.create_connection((parsed.hostname, parsed.port), timeout=2): + raise ContractError("cleanup_target_tcp_listener_still_present") + except ContractError: + raise + except OSError: return - raise ContractError("cleanup_target_listener_still_present") + + +def cleanup_production_snapshot() -> dict[str, dict[str, Any]]: + expected = { + "signoz": EXPECTED_IMAGES["signoz"], + "signoz-clickhouse": EXPECTED_IMAGES["clickhouse"], + "signoz-zookeeper-1": EXPECTED_IMAGES["zookeeper"], + "signoz-otel-collector": { + "ref": "signoz/signoz-otel-collector:v0.144.1", + "id": "sha256:47cbed488288df7bac6aab3665275bec6d6e24fe5c545c66914827a891e538d8", + }, + } + snapshot: dict[str, dict[str, Any]] = {} + for name, image in expected.items(): + data = docker_inspect("container", name) + if data is None: + raise ContractError("cleanup_production_container_missing") + config = data.get("Config") + state = data.get("State") + if not isinstance(config, dict) or not isinstance(state, dict): + raise ContractError("cleanup_production_container_shape_invalid") + row = { + "id": data.get("Id"), + "image_id": data.get("Image"), + "image_ref": config.get("Image"), + "running": state.get("Running"), + "started_at": state.get("StartedAt"), + "restart_count": data.get("RestartCount"), + } + if row["image_id"] != image["id"] or row["image_ref"] != image["ref"]: + raise ContractError("cleanup_production_image_identity_invalid") + snapshot[name] = row + return snapshot + + +def cleanup_image_inventory() -> list[str]: + try: + result = subprocess.run( # nosec B603 + [ + "docker", + "image", + "ls", + "--no-trunc", + "--format={{.ID}}|{{.Repository}}:{{.Tag}}", + ], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + timeout=15, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ContractError("cleanup_image_inventory_failed") from exc + if result.returncode != 0: + raise ContractError("cleanup_image_inventory_failed") + rows = sorted(line for line in result.stdout.splitlines() if line.strip()) + if not rows: + raise ContractError("cleanup_image_inventory_empty") + return rows def run_cleanup_verifier( args: argparse.Namespace, policy: dict[str, Any], + isolation: dict[str, Any], ) -> dict[str, Any]: - label = f"{policy['restore_isolation']['resource_label_key']}={args.run_id}" - counts = { - resource: docker_label_count(resource, label) - for resource in ("container", "volume", "network") - } - if any(counts.values()): - raise ContractError("cleanup_resource_residue_present") - require_target_unreachable(args.target_base_url) + identity = restore_identity(args.trace_id, args.run_id, args.work_item_id) + resources = expected_resources(identity) + for name in resources["containers"]: + if docker_inspect("container", name) is not None: + raise ContractError("cleanup_exact_container_residue_present") + for name in resources["volumes"]: + if docker_inspect("volume", name) is not None: + raise ContractError("cleanup_exact_volume_residue_present") + if docker_inspect("network", resources["network"]) is not None: + raise ContractError("cleanup_exact_network_residue_present") + labels = ( + f"{PRIMARY_LABEL}={args.run_id}", + f"{IDENTITY_LABEL}={identity}", + ) + for label in labels: + counts = { + resource: docker_label_count(resource, label) + for resource in ("container", "volume", "network") + } + if any(counts.values()): + raise ContractError("cleanup_resource_label_residue_present") + require_target_tcp_unreachable(args.target_base_url) + workspace = Path(resources["workspace"]) + if workspace.exists() or workspace.is_symlink(): + raise ContractError("cleanup_private_workspace_residue_present") + production_before = isolation.get("production_before") + if not isinstance(production_before, dict): + raise ContractError("cleanup_production_before_missing") + if cleanup_production_snapshot() != production_before: + raise ContractError("cleanup_production_identity_continuity_failed") + image_inventory_before = isolation.get("image_inventory_before") + if not isinstance(image_inventory_before, list): + raise ContractError("cleanup_image_inventory_before_missing") + if cleanup_image_inventory() != image_inventory_before: + raise ContractError("cleanup_image_inventory_drift") return receipt( trace_id=args.trace_id, run_id=args.run_id, work_item_id=args.work_item_id, phase="terminal", terminal="pass_zero_residue_verified", - detail="container_0_volume_0_network_0_listener_0", + detail=( + "exact_container_0_volume_0_network_0_label_inventory_0_" + "tcp_listener_0_workspace_0_production_identity_unchanged" + ), ) @@ -469,14 +702,14 @@ def main() -> int: ) policy = load_policy(Path(args.policy)) run_independent_bundle_verifier(args) - validate_isolation_receipt(args, policy) + isolation = validate_isolation_receipt(args, policy) assets = load_portable_assets(Path(args.bundle_dir), policy) if args.check: result = run_check(args, policy, assets) elif args.apply: result = run_apply(args, policy, assets) else: - result = run_cleanup_verifier(args, policy) + result = run_cleanup_verifier(args, policy, isolation) if args.receipt_file: write_receipt(Path(args.receipt_file), result) emit_result(result) diff --git a/scripts/backup/tests/test_signoz_metadata_export_contract.py b/scripts/backup/tests/test_signoz_metadata_export_contract.py index 60f1ac2d6..ae9f5f974 100644 --- a/scripts/backup/tests/test_signoz_metadata_export_contract.py +++ b/scripts/backup/tests/test_signoz_metadata_export_contract.py @@ -60,7 +60,10 @@ class MetadataHandler(BaseHTTPRequestHandler): return def _authorized(self) -> bool: - return self.headers.get("SIGNOZ-API-KEY") == TOKEN + return ( + self.headers.get("SIGNOZ-API-KEY") == TOKEN + or self.headers.get("Authorization") == f"Bearer {TOKEN}" + ) def _send_json(self, status: int, value: Any) -> None: payload = json.dumps(value, separators=(",", ":")).encode("utf-8") @@ -219,20 +222,46 @@ def create_bundle(tmp_path: Path) -> tuple[Path, Path]: def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path: policy = json.loads(POLICY.read_text(encoding="utf-8")) + identity = hashlib.sha256( + TRACE_ID.encode("utf-8") + + b"\0" + + RUN_ID.encode("utf-8") + + b"\0" + + WORK_ITEM_ID.encode("utf-8") + ).hexdigest() + short_id = identity[:16] + prefix = f"awoooi-signoz-md-restore-{short_id}" + port_text = target_base_url.rsplit(":", 1)[-1] + port = int(port_text) if port_text.isdigit() else 443 + production_before = { + entry["container"]: { + "id": hashlib.sha256(entry["container"].encode()).hexdigest(), + "image_id": entry["image_id"], + "image_ref": entry["image_ref"], + "running": True, + "started_at": "2026-07-22T00:00:00Z", + "restart_count": 0, + } + for entry in policy["restore_isolation"]["production_continuity"].values() + } value = { "schema": "awoooi_signoz_metadata_restore_isolation_v1", "trace_id": TRACE_ID, "run_id": RUN_ID, "work_item_id": WORK_ITEM_ID, "target": { - "canonical_id": f"ephemeral-signoz-metadata-restore-{RUN_ID}", + "identity": identity, + "prefix": prefix, + "canonical_id": f"ephemeral-signoz-metadata-restore-{short_id}", "base_url_sha256": hashlib.sha256( target_base_url.encode("utf-8") ).hexdigest(), "image_ref": "signoz/signoz:v0.113.0", - "image_id": f"sha256:{'a' * 64}", + "image_id": policy["restore_isolation"]["required_image_id"], + "images": policy["restore_isolation"]["images"], "image_present_locally": True, "image_pull_performed": False, + "image_build_performed": False, "production_network_attached": False, "production_volume_attached": False, "ephemeral_volumes_only": True, @@ -241,7 +270,28 @@ def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path: "network_scope": "loopback_and_internal_only", "resource_label_key": policy["restore_isolation"]["resource_label_key"], "resource_label_value": RUN_ID, + "identity_label_key": "com.awoooi.restore.identity", + "identity_label_value": identity, + "resources": { + "network": f"{prefix}-net", + "containers": [ + f"{prefix}-server", + f"{prefix}-clickhouse", + f"{prefix}-zookeeper", + ], + "volumes": [ + f"{prefix}-zk-data", + f"{prefix}-zk-log", + f"{prefix}-ch-data", + f"{prefix}-server-data", + ], + "workspace": f"/tmp/{prefix}-private", + }, + "listener": {"host": "127.0.0.1", "port": port}, + "authentication_mode": "isolated_session_bearer", }, + "production_before": production_before, + "image_inventory_before": ["sha256:test|test:fixture"], } path = tmp_path / "isolation.json" path.write_text(json.dumps(value), encoding="utf-8") @@ -266,6 +316,8 @@ def restore_command( str(bundle), "--target-base-url", target_base_url, + "--auth-mode", + "isolated_session_bearer", "--isolation-receipt", str(isolation), "--trace-id", @@ -349,6 +401,23 @@ def test_policy_rejects_unhashable_empty_collection_sentinel_as_contract_error( assert "Traceback" not in result.stderr +def test_policy_pins_reviewed_isolated_restore_runtime() -> None: + policy = json.loads(POLICY.read_text(encoding="utf-8")) + isolation = policy["restore_isolation"] + assert isolation["required_image_id"] == ( + "sha256:c17991d10966aedcb0d4d83805a0baf5310f1553d90611629036fcfaee8d969b" + ) + assert isolation["startup_contract"]["reviewed"] is True + assert isolation["startup_contract"]["server_entrypoint"] == [ + "./signoz", + "server", + ] + assert isolation["startup_contract"]["server_environment"] == { + "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN": "tcp://clickhouse:9000", + "SIGNOZ_SQLSTORE_SQLITE_PATH": "/var/lib/signoz/signoz.db", + } + + def test_check_mode_writes_nothing(tmp_path: Path) -> None: output = tmp_path / "must-not-exist" result = subprocess.run( @@ -654,7 +723,19 @@ def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None: assert json.loads(apply_result.stdout)["terminal"] == ( "partial_degraded_cleanup_pending" ) - phase_receipts = json.loads(apply_result.stdout)["phase_receipts"] + apply_receipt = json.loads(apply_result.stdout) + assert apply_receipt["restored_asset_counts"] == { + "dashboards": 1, + "alert_rules": 1, + "explorer_views_logs": 1, + "explorer_views_meter": 0, + "explorer_views_metrics": 0, + "explorer_views_traces": 0, + } + assert len(apply_receipt["portable_semantic_sha256"]) == 64 + assert apply_receipt["completion_scope"] == "portable_assets_only" + assert apply_receipt["full_sqlite_completion_claim"] is False + phase_receipts = apply_receipt["phase_receipts"] assert [item["phase"] for item in phase_receipts] == [ "sensor_source", "normalized_asset_identity", @@ -693,12 +774,74 @@ def test_restore_rejects_non_loopback_target_before_writes(tmp_path: Path) -> No assert "restore_target_must_be_loopback" in result.stderr +def test_restore_rejects_non_exact_isolation_image_before_writes( + tmp_path: Path, +) -> None: + bundle, credential = create_bundle(tmp_path) + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + target = f"http://127.0.0.1:{port}" + isolation = isolation_receipt(tmp_path, target) + value = json.loads(isolation.read_text(encoding="utf-8")) + value["target"]["image_id"] = f"sha256:{'a' * 64}" + isolation.write_text(json.dumps(value), encoding="utf-8") + isolation.chmod(0o600) + + result = subprocess.run( + restore_command( + mode="--apply", + bundle=bundle, + target_base_url=target, + credential=credential, + isolation=isolation, + ), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=20, + check=False, + ) + + assert result.returncode == 1 + assert "isolation_target_image_id_mismatch" in result.stderr + + def test_cleanup_verifier_requires_zero_labeled_residue(tmp_path: Path) -> None: bundle, _ = create_bundle(tmp_path) fake_bin = tmp_path / "bin" fake_bin.mkdir() docker = fake_bin / "docker" - docker.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + policy = json.loads(POLICY.read_text(encoding="utf-8")) + production = { + entry["container"]: { + "Id": hashlib.sha256(entry["container"].encode()).hexdigest(), + "Image": entry["image_id"], + "Config": {"Image": entry["image_ref"]}, + "State": { + "Running": True, + "StartedAt": "2026-07-22T00:00:00Z", + }, + "RestartCount": 0, + } + for entry in policy["restore_isolation"]["production_continuity"].values() + } + docker.write_text( + "#!/usr/bin/env python3\n" + "import json, sys\n" + f"production = {production!r}\n" + "args = sys.argv[1:]\n" + "if args[:2] == ['container', 'inspect'] and args[2] in production:\n" + " print(json.dumps([production[args[2]]]))\n" + " raise SystemExit(0)\n" + "if args[:3] == ['image', 'ls', '--no-trunc']:\n" + " print('sha256:test|test:fixture')\n" + " raise SystemExit(0)\n" + "if len(args) >= 2 and args[1] == 'inspect':\n" + " raise SystemExit(1)\n" + "raise SystemExit(0)\n", + encoding="utf-8", + ) docker.chmod(0o755) with socket.socket() as sock: diff --git a/scripts/backup/tests/test_signoz_metadata_isolated_restore.py b/scripts/backup/tests/test_signoz_metadata_isolated_restore.py new file mode 100644 index 000000000..f5664f198 --- /dev/null +++ b/scripts/backup/tests/test_signoz_metadata_isolated_restore.py @@ -0,0 +1,597 @@ +from __future__ import annotations + +import importlib.util +import json +import stat +import subprocess +import sys +from types import SimpleNamespace +from pathlib import Path +from typing import Any + +import pytest + + +ROOT = Path(__file__).resolve().parents[3] +CONTROLLER = ROOT / "scripts" / "backup" / "signoz-metadata-isolated-restore.py" +POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json" +TRACE_ID = "trace-P0-OBS-002-isolated" +RUN_ID = "run-P0-OBS-002-isolated" +WORK_ITEM_ID = "P0-OBS-002" + + +def load_controller() -> Any: + backup_dir = str(ROOT / "scripts" / "backup") + if backup_dir not in sys.path: + sys.path.insert(0, backup_dir) + spec = importlib.util.spec_from_file_location("signoz_isolated_restore", CONTROLLER) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +@pytest.fixture() +def controller() -> Any: + return load_controller() + + +class RecordingRunner: + def __init__(self, controller: Any) -> None: + self.controller = controller + self.commands: list[list[str]] = [] + self.resources: dict[tuple[str, str], dict[str, Any]] = {} + self.production = { + name: { + "Id": (name.encode().hex() + "0" * 64)[:64], + "Image": image_id, + "Config": {"Image": ref}, + "State": { + "Running": True, + "StartedAt": "2026-07-22T00:00:00Z", + }, + "RestartCount": 0, + } + for name, (ref, image_id) in controller.EXPECTED_PRODUCTION.items() + } + self.image_rows = ["sha256:test|test:fixture"] + + def __call__(self, argv: list[str], _timeout: int) -> Any: + self.commands.append(argv[:]) + c = self.controller + if argv[1:3] == ["image", "inspect"]: + ref = argv[-1] + for expected_ref, image_id in c.EXPECTED_IMAGES.values(): + if ref == expected_ref: + return c.CommandResult(0, f"{image_id}\n") + return c.CommandResult(1, "") + if argv[1:3] == ["image", "ls"]: + return c.CommandResult(0, "\n".join(self.image_rows) + "\n") + if argv[1:2] == ["info"]: + return c.CommandResult(0, "28.0.0\n") + if len(argv) >= 4 and argv[2] == "inspect": + kind, name = argv[1], argv[3] + if kind == "container" and name in self.production: + return c.CommandResult(0, json.dumps([self.production[name]])) + value = self.resources.get((kind, name)) + if value is None: + return c.CommandResult(1, "") + return c.CommandResult(0, json.dumps([value])) + if argv[1:3] == ["network", "create"]: + name = argv[-1] + labels = labels_from_command(argv) + self.resources[("network", name)] = {"Name": name, "Labels": labels} + return c.CommandResult(0, f"{name}\n") + if argv[1:3] == ["volume", "create"]: + name = argv[-1] + labels = labels_from_command(argv) + self.resources[("volume", name)] = {"Name": name, "Labels": labels} + return c.CommandResult(0, f"{name}\n") + if argv[1:2] == ["run"]: + name = argv[argv.index("--name") + 1] + ref = argv[-1] + expected_id = next( + image_id + for expected_ref, image_id in c.EXPECTED_IMAGES.values() + if expected_ref == ref + ) + self.resources[("container", name)] = { + "Id": (name.encode().hex() + "0" * 64)[:64], + "Image": expected_id, + "Config": {"Image": ref, "Labels": labels_from_command(argv)}, + "HostConfig": {"RestartPolicy": {"Name": "no"}}, + "State": {"Running": True}, + "NetworkSettings": { + "Ports": { + "8080/tcp": [ + {"HostIp": "127.0.0.1", "HostPort": "49173"} + ] + } + if name.endswith("-server") + else {} + }, + "RestartCount": 0, + } + return c.CommandResult(0, f"{name}\n") + if argv[1:2] == ["port"]: + return c.CommandResult(0, "127.0.0.1:49173\n") + if argv[1:2] == ["exec"] and argv[-2:] == ["--query", "SELECT 1"]: + return c.CommandResult(0, "1\n") + if argv[1:3] == ["rm", "--force"]: + self.resources.pop(("container", argv[-1]), None) + return c.CommandResult(0, "") + if argv[1:3] == ["volume", "rm"]: + self.resources.pop(("volume", argv[-1]), None) + return c.CommandResult(0, "") + if argv[1:3] == ["network", "rm"]: + self.resources.pop(("network", argv[-1]), None) + return c.CommandResult(0, "") + if argv[1:2] == ["ps"] or ( + len(argv) >= 3 and argv[1] in {"volume", "network"} and argv[2] == "ls" + ): + label = argv[-1].removeprefix("label=") + key, expected = label.split("=", 1) + matches = [] + for (kind, name), value in self.resources.items(): + if argv[1] == "ps" and kind != "container": + continue + if argv[1] in {"volume", "network"} and kind != argv[1]: + continue + labels = ( + value.get("Config", {}).get("Labels", {}) + if kind == "container" + else value.get("Labels", {}) + ) + if labels.get(key) == expected: + matches.append(name) + return c.CommandResult(0, "\n".join(matches)) + return c.CommandResult(1, "") + + +def labels_from_command(argv: list[str]) -> dict[str, str]: + labels: dict[str, str] = {} + for index, value in enumerate(argv): + if value == "--label": + key, label_value = argv[index + 1].split("=", 1) + labels[key] = label_value + return labels + + +def test_identity_and_exact_resource_names(controller: Any) -> None: + plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID) + + assert plan.identity == controller.derive_identity( + TRACE_ID, RUN_ID, WORK_ITEM_ID + ) + assert len(plan.identity) == 64 + assert plan.prefix == f"awoooi-signoz-md-restore-{plan.identity[:16]}" + assert plan.canonical_id == ( + f"ephemeral-signoz-metadata-restore-{plan.identity[:16]}" + ) + assert plan.containers == ( + f"{plan.prefix}-server", + f"{plan.prefix}-clickhouse", + f"{plan.prefix}-zookeeper", + ) + assert len(plan.volumes) == 4 + assert plan.workspace == Path(f"/tmp/{plan.prefix}-private") + + +def test_policy_requires_reviewed_exact_startup_and_images(controller: Any) -> None: + policy = json.loads(POLICY.read_text(encoding="utf-8")) + assert controller.validate_restore_policy(policy)["startup_contract"][ + "reviewed" + ] is True + + policy["restore_isolation"]["startup_contract"]["reviewed"] = False + with pytest.raises( + controller.ContractError, match="restore_startup_contract_not_reviewed" + ): + controller.validate_restore_policy(policy) + + policy = json.loads(POLICY.read_text(encoding="utf-8")) + policy["restore_isolation"]["images"]["clickhouse"]["id"] = ( + f"sha256:{'a' * 64}" + ) + with pytest.raises( + controller.ContractError, match="restore_image_policy_not_exact" + ): + controller.validate_restore_policy(policy) + + +def test_resource_creation_is_internal_no_pull_no_restart_and_loopback( + controller: Any, +) -> None: + runner = RecordingRunner(controller) + plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID) + cluster = ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" + + controller.create_resources(runner, plan, RUN_ID, cluster) + + commands = runner.commands + network = next(command for command in commands if command[1:3] == ["network", "create"]) + assert "--internal" in network + run_commands = [command for command in commands if command[1:2] == ["run"]] + assert len(run_commands) == 3 + for command in run_commands: + assert "--pull=never" in command + assert "--restart=no" in command + assert f"{controller.PRIMARY_LABEL}={RUN_ID}" in command + assert f"{controller.IDENTITY_LABEL}={plan.identity}" in command + assert not any("signoz-clickhouse:" in value for value in command) + server = next(command for command in run_commands if plan.server_container in command) + assert "127.0.0.1::8080" in server + assert f"{plan.server_data_volume}:/var/lib/signoz" in server + assert "SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN=tcp://clickhouse:9000" in server + clickhouse_run_index = commands.index( + next(command for command in run_commands if plan.clickhouse_container in command) + ) + readiness_index = next( + index + for index, command in enumerate(commands) + if command[1:2] == ["exec"] + and command[-2:] == ["--query", "SELECT 1"] + ) + server_run_index = commands.index(server) + assert clickhouse_run_index < readiness_index < server_run_index + + +def test_exact_image_drift_fails_before_resource_creation(controller: Any) -> None: + runner = RecordingRunner(controller) + + def drift(argv: list[str], timeout: int) -> Any: + result = runner(argv, timeout) + if argv[-1] == controller.EXPECTED_IMAGES["clickhouse"][0]: + return controller.CommandResult(0, f"sha256:{'b' * 64}\n") + return result + + with pytest.raises( + controller.ContractError, match="restore_image_id_mismatch_clickhouse" + ): + controller.verify_exact_images(drift) + assert not runner.resources + + +def test_cleanup_refuses_same_name_with_wrong_ownership(controller: Any) -> None: + runner = RecordingRunner(controller) + plan = controller.build_resource_plan(TRACE_ID, RUN_ID, WORK_ITEM_ID) + runner.resources[("container", plan.server_container)] = { + "Config": { + "Labels": { + controller.PRIMARY_LABEL: "some-other-run", + controller.IDENTITY_LABEL: plan.identity, + } + } + } + + with pytest.raises( + controller.ContractError, match="cleanup_container_ownership_mismatch" + ): + controller.remove_owned_resource( + runner, + plan, + RUN_ID, + "container", + plan.server_container, + ) + assert ("container", plan.server_container) in runner.resources + assert not any(command[1:3] == ["rm", "--force"] for command in runner.commands) + + +def test_cleanup_removes_exact_resources_and_proves_continuity( + controller: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runner = RecordingRunner(controller) + plan = controller.build_resource_plan( + TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path + ) + cluster = ROOT / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" + controller.create_workspace(plan) + controller.create_resources(runner, plan, RUN_ID, cluster) + state = { + "production_before": controller.production_snapshot(runner), + "image_inventory_before": controller.image_inventory(runner), + "listener": {"host": "127.0.0.1", "port": 49173}, + } + monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) + monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None) + + errors = controller.cleanup_and_verify(runner, plan, RUN_ID, state) + + assert errors == [] + assert not runner.resources + assert not plan.workspace.exists() + + +@pytest.mark.parametrize("server_created", [False, True]) +def test_cleanup_reconciles_without_listener_state_after_interruption( + controller: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + server_created: bool, +) -> None: + runner = RecordingRunner(controller) + plan = controller.build_resource_plan( + TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path + ) + controller.create_workspace(plan) + if server_created: + cluster = ( + ROOT + / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" + ) + controller.create_resources(runner, plan, RUN_ID, cluster) + state = { + "production_before": controller.production_snapshot(runner), + "image_inventory_before": controller.image_inventory(runner), + "listener": None, + } + monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) + monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None) + + errors = controller.cleanup_and_verify(runner, plan, RUN_ID, state) + + assert errors == [] + assert not runner.resources + assert not plan.workspace.exists() + + +def test_cleanup_already_clean_without_state_uses_fresh_continuity( + controller: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runner = RecordingRunner(controller) + plan = controller.build_resource_plan( + TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path + ) + monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) + monkeypatch.setattr(controller.time, "sleep", lambda _seconds: None) + + errors = controller.cleanup_and_verify(runner, plan, RUN_ID, None) + + assert errors == [] + assert not plan.workspace.exists() + + +def test_cleanup_reconciles_resource_materialized_after_first_inventory( + controller: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runner = RecordingRunner(controller) + plan = controller.build_resource_plan( + TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path + ) + sleeps = 0 + + def materialize_after_first_pass(_seconds: float) -> None: + nonlocal sleeps + sleeps += 1 + if sleeps == 1: + runner.resources[("volume", plan.server_data_volume)] = { + "Name": plan.server_data_volume, + "Labels": { + controller.PRIMARY_LABEL: RUN_ID, + controller.IDENTITY_LABEL: plan.identity, + }, + } + + monkeypatch.setattr(controller, "tcp_connectable", lambda _host, _port: False) + monkeypatch.setattr(controller.time, "sleep", materialize_after_first_pass) + + errors = controller.cleanup_and_verify(runner, plan, RUN_ID, None) + + assert errors == [] + assert not runner.resources + removals = [ + command + for command in runner.commands + if command[1:3] == ["volume", "rm"] + and command[-1] == plan.server_data_volume + ] + assert len(removals) == 1 + + +def test_apply_cli_requires_durable_receipt_before_runtime_dispatch() -> None: + result = subprocess.run( + [ + sys.executable, + str(CONTROLLER), + "--apply", + "--trace-id", + TRACE_ID, + "--run-id", + RUN_ID, + "--work-item-id", + WORK_ITEM_ID, + ], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + check=False, + ) + + assert result.returncode == 2 + assert "receipt_file_required_for_restore_apply" in result.stderr + assert result.stdout == "" + + +def test_distinct_verifier_requires_immutable_apply_receipt( + controller: Any, + tmp_path: Path, +) -> None: + args = SimpleNamespace( + trace_id=TRACE_ID, + run_id=RUN_ID, + work_item_id=WORK_ITEM_ID, + ) + plan = controller.build_resource_plan( + TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path + ) + executor_receipt = controller.executor_success( + args, + plan, + { + "restoreSemanticDigest": "a" * 64, + "restoredAssetCounts": { + "dashboards": 1, + "alert_rules": 2, + "explorer_views": 3, + }, + }, + ) + path = tmp_path / "apply.json" + controller.write_terminal(path, executor_receipt) + + validated = controller.validate_apply_receipt(path, args, plan) + verified = controller.independent_success(args, plan, validated) + + assert executor_receipt["completionClaim"] is False + assert executor_receipt["executorZeroResidueVerified"] is True + assert verified["phase"] == "independent_restore_verifier" + assert verified["completionClaim"] is True + assert verified["independentZeroResidueVerified"] is True + assert verified["restoreSemanticDigest"] == "a" * 64 + assert verified["applyReceiptValidated"] is True + assert len(verified["applyReceiptSha256"]) == 64 + + +def test_interrupted_verify_cleanup_reconciles_again_before_zero_residue_receipt( + controller: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + plan = controller.build_resource_plan( + TRACE_ID, RUN_ID, WORK_ITEM_ID, workspace_parent=tmp_path + ) + calls = 0 + emitted: list[dict[str, Any]] = [] + + def interrupt_once( + _runner: Any, + _plan: Any, + _run_id: str, + _state: dict[str, Any] | None, + ) -> list[str]: + nonlocal calls + calls += 1 + if calls == 1: + raise controller.SignalAbort("signal_15_cleanup_required") + return [] + + monkeypatch.setattr( + sys, + "argv", + [ + str(CONTROLLER), + "--verify-cleanup", + "--trace-id", + TRACE_ID, + "--run-id", + RUN_ID, + "--work-item-id", + WORK_ITEM_ID, + ], + ) + monkeypatch.setattr(controller, "build_resource_plan", lambda *_args: plan) + monkeypatch.setattr(controller, "load_policy", lambda _path: {}) + monkeypatch.setattr(controller, "load_state_if_safe", lambda _plan: None) + monkeypatch.setattr(controller, "cleanup_and_verify", interrupt_once) + monkeypatch.setattr(controller, "emit", emitted.append) + + exit_code = controller.main() + + assert exit_code == 1 + assert calls == 2 + assert emitted[-1]["terminal"] == "failed_zero_residue_verified_no_completion" + assert emitted[-1]["zeroResidueVerified"] is True + assert emitted[-1]["cleanupErrorCodes"] == [] + assert "signal_15_cleanup_required" in emitted[-1]["detail"] + + +def test_bootstrap_secrets_stay_in_mode_0400_files( + controller: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + calls: list[tuple[str, dict[str, Any] | None]] = [] + + def fake_request( + url: str, + *, + method: str = "GET", + payload: dict[str, Any] | None = None, + authorization: str | None = None, + timeout: int = 10, + ) -> dict[str, Any]: + assert authorization is None + calls.append((url, payload)) + if url.endswith("/api/v1/register"): + return {"data": {"orgId": "isolated-org"}, "status": "success"} + return { + "data": { + "accessToken": "a" * 64, + "refreshToken": "r" * 64, + }, + "status": "success", + } + + monkeypatch.setattr(controller, "request_json", fake_request) + + token = controller.bootstrap_isolated_session("http://127.0.0.1:49173", tmp_path) + + assert token == tmp_path / "session-access-token" + assert stat.S_IMODE(token.stat().st_mode) == 0o400 + assert sorted(path.name for path in tmp_path.iterdir()) == [ + "session-access-token" + ] + assert calls[0][0].endswith("/api/v1/register") + assert calls[1][0].endswith("/api/v2/sessions/email_password") + assert capsys.readouterr().out == "" + + +def test_bootstrap_accepts_live_root_response_envelope( + controller: Any, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def fake_request( + url: str, + **_kwargs: Any, + ) -> dict[str, Any]: + if url.endswith("/api/v1/register"): + return {"orgId": "isolated-org"} + return {"accessToken": "a" * 64, "refreshToken": "r" * 64} + + monkeypatch.setattr(controller, "request_json", fake_request) + + token = controller.bootstrap_isolated_session( + "http://127.0.0.1:49173", tmp_path + ) + + assert token.read_text(encoding="ascii").strip() == "a" * 64 + assert stat.S_IMODE(token.stat().st_mode) == 0o400 + + +def test_controller_source_has_signal_cleanup_and_no_secret_argv() -> None: + source = CONTROLLER.read_text(encoding="utf-8") + assert "signal.SIGHUP" in source + assert "signal.SIGINT" in source + assert "signal.SIGTERM" in source + assert "finally:" in source + assert "cleanup_and_verify(" in source + assert '"--pull=never"' in source + assert '"--restart=no"' in source + assert '"--auth-mode"' in source + assert "access_token" not in source[source.index("argv = [") : source.index("output = run_required", source.index("argv = ["))] + assert "password" not in source[source.index("argv = [") : source.index("output = run_required", source.index("argv = ["))] + durable_write = source.rindex("write_terminal(Path(args.receipt_file), result)") + final_handler_restore = source.rindex("restore_signal_handlers()") + assert durable_write < final_handler_restore + assert "if not args.apply and not signals_held_until_terminal:" in source diff --git a/scripts/ops/deploy-signoz-metadata-toolchain.sh b/scripts/ops/deploy-signoz-metadata-toolchain.sh index 3c438f62b..c26ae2b7f 100755 --- a/scripts/ops/deploy-signoz-metadata-toolchain.sh +++ b/scripts/ops/deploy-signoz-metadata-toolchain.sh @@ -26,6 +26,8 @@ LABELS=( signoz-metadata-export.py verify-signoz-metadata-export.py signoz-metadata-restore-drill.py + signoz-metadata-isolated-restore.py + isolated-cluster.xml ) LOCAL_SOURCES=( "${ROOT_DIR}/config/signoz/metadata-export-policy.json" @@ -33,8 +35,10 @@ LOCAL_SOURCES=( "${ROOT_DIR}/scripts/backup/signoz-metadata-export.py" "${ROOT_DIR}/scripts/backup/verify-signoz-metadata-export.py" "${ROOT_DIR}/scripts/backup/signoz-metadata-restore-drill.py" + "${ROOT_DIR}/scripts/backup/signoz-metadata-isolated-restore.py" + "${ROOT_DIR}/ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml" ) -MODES=(0644 0644 0755 0755 0755) +MODES=(0644 0644 0755 0755 0755 0755 0644) SSH_OPTIONS=( -o BatchMode=yes -o IdentitiesOnly=yes @@ -72,7 +76,7 @@ Optional fixed transport paths: SIGNOZ_METADATA_SSH_IDENTITY_FILE SIGNOZ_METADATA_KNOWN_HOSTS_FILE ---check validates the five local sources and performs a no-write host110 diff. +--check validates the seven local sources and performs a no-write host110 diff. --apply creates one immutable exact-hash directory. It does not create or change an active pointer and does not run an authenticated metadata export. USAGE @@ -153,7 +157,8 @@ import sys for value in sys.argv[1:]: path = Path(value) - compile(path.read_text(encoding="utf-8"), str(path), "exec") + if path.suffix == ".py": + compile(path.read_text(encoding="utf-8"), str(path), "exec") PY PYTHONPATH="${ROOT_DIR}/scripts/backup" "${LOCAL_PYTHON_BIN}" - \ "${LOCAL_SOURCES[0]}" <<'PY' @@ -182,7 +187,7 @@ BUNDLE_ID="$(printf '%s\n' "${SOURCE_HASHES[@]}" | sha256sum | awk '{print $1}') TARGET_DIR="${REMOTE_ROOT}/${BUNDLE_ID}" STAGE_DIR="${STAGE_ROOT}/awoooi-signoz-metadata-toolchain.${RUN_ID}" -emit sensor_source pass "local_exact_sources_5" +emit sensor_source pass "local_exact_sources_7" emit normalized_asset_identity pass "bundle_${BUNDLE_ID}" emit ai_decision pass "candidate_additive_immutable_inactive_source_deploy" emit risk_policy_decision pass "medium_no_active_pointer_no_secret_no_raw_sqlite" @@ -209,11 +214,13 @@ LABELS=( signoz-metadata-export.py verify-signoz-metadata-export.py signoz-metadata-restore-drill.py + signoz-metadata-isolated-restore.py + isolated-cluster.xml ) -MODES=(0644 0644 0755 0755 0755) +MODES=(0644 0644 0755 0755 0755 0755 0644) TARGET_DIR="${REMOTE_ROOT}/${BUNDLE_ID}" -[ "${#EXPECTED_HASHES[@]}" -eq 5 ] +[ "${#EXPECTED_HASHES[@]}" -eq 7 ] [ -d /backup ] && [ ! -L /backup ] [ ! -e "${REMOTE_ROOT}/current" ] && [ ! -L "${REMOTE_ROOT}/current" ] for command_name in curl docker pgrep python3 sha256sum stat timeout; do @@ -230,7 +237,7 @@ api_status="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ [ "${api_status}" = "200" ] active_operations="$( - (pgrep -af '(^|/|[[:space:]])(backup-signoz|clickhouse-native-backup|clickhouse-native-restore-drill|run-signoz-backup-canary)[.]sh([[:space:]]|$)' || true) \ + (pgrep -af '([s]ignoz-metadata-isolated-restore[.]py|(^|/|[[:space:]])(backup-signoz|clickhouse-native-backup|clickhouse-native-restore-drill|run-signoz-backup-canary)[.]sh([[:space:]]|$))' || true) \ | wc -l | tr -d ' ' )" [ "${active_operations}" = "0" ] @@ -369,8 +376,10 @@ LABELS=( signoz-metadata-export.py verify-signoz-metadata-export.py signoz-metadata-restore-drill.py + signoz-metadata-isolated-restore.py + isolated-cluster.xml ) -MODES=(0644 0644 0755 0755 0755) +MODES=(0644 0644 0755 0755 0755 0755 0644) TARGET_DIR="${REMOTE_ROOT}/${BUNDLE_ID}" CANDIDATE_DIR="${REMOTE_ROOT}/.${BUNDLE_ID}.candidate.${RUN_ID}" RECEIPT_DIR="${RECEIPT_ROOT}/${RUN_ID}" @@ -408,7 +417,7 @@ finalize() { } trap finalize EXIT HUP INT TERM -[ "${#EXPECTED_HASHES[@]}" -eq 5 ] +[ "${#EXPECTED_HASHES[@]}" -eq 7 ] [ -d "${STAGE_DIR}" ] && [ ! -L "${STAGE_DIR}" ] [ ! -e "${TARGET_DIR}" ] && [ ! -L "${TARGET_DIR}" ] [ ! -e "${CANDIDATE_DIR}" ] && [ ! -L "${CANDIDATE_DIR}" ] @@ -431,7 +440,7 @@ OPERATION_LOCK=/tmp/awoooi-signoz-backup-operation.lock exec 8<"${OPERATION_LOCK}" flock -n 8 active_operations="$( - (pgrep -af '(^|/|[[:space:]])(backup-signoz|clickhouse-native-backup|clickhouse-native-restore-drill|run-signoz-backup-canary)[.]sh([[:space:]]|$)' || true) \ + (pgrep -af '([s]ignoz-metadata-isolated-restore[.]py|(^|/|[[:space:]])(backup-signoz|clickhouse-native-backup|clickhouse-native-restore-drill|run-signoz-backup-canary)[.]sh([[:space:]]|$))' || true) \ | wc -l | tr -d ' ' )" [ "${active_operations}" = "0" ] @@ -443,7 +452,7 @@ api_before="$(timeout --kill-after="${KILL_AFTER_SECONDS}" \ "${REMOTE_TIMEOUT_SECONDS}" curl -sS -o /dev/null -w '%{http_code}' \ http://127.0.0.1:8080/api/v1/health)" [ "${api_before}" = "200" ] -emit_remote sensor_source pass "staged_sources_5_runtime_api_200" +emit_remote sensor_source pass "staged_sources_7_runtime_api_200" for index in "${!LABELS[@]}"; do staged="${STAGE_DIR}/${LABELS[index]}" @@ -463,6 +472,7 @@ for name in ( "signoz-metadata-export.py", "verify-signoz-metadata-export.py", "signoz-metadata-restore-drill.py", + "signoz-metadata-isolated-restore.py", ): path = root / name compile(path.read_text(encoding="utf-8"), str(path), "exec") diff --git a/scripts/ops/tests/test_agent99_signoz_metadata_executor_runtime_entrypoint.py b/scripts/ops/tests/test_agent99_signoz_metadata_executor_runtime_entrypoint.py index a79f2ee8e..1169397bc 100644 --- a/scripts/ops/tests/test_agent99_signoz_metadata_executor_runtime_entrypoint.py +++ b/scripts/ops/tests/test_agent99_signoz_metadata_executor_runtime_entrypoint.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import re from pathlib import Path @@ -17,13 +18,16 @@ def _quoted_array(source: str, start: str, end: str) -> tuple[str, ...]: def test_entrypoint_is_fixed_to_windows99_host110_and_p0_obs_002() -> None: source = ENTRYPOINT.read_text(encoding="utf-8") - assert '[ValidateSet("Check", "Apply", "Verify")]' in source + assert ( + '[ValidateSet("Check", "Apply", "Verify", "RestoreCheck", ' + '"RestoreApply", "RestoreVerifyCleanup")]' in source + ) assert '$CanonicalAsset = "signoz-110-control-plane-metadata"' in source assert '$TargetHost = "192.168.0.110"' in source assert '$RemoteUser = "wooo"' in source assert '$WorkItemId -ne "P0-OBS-002"' in source assert '$SafeIdPattern = "^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"' in source - assert "._+" not in source + assert '$SafeIdPattern = "^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"' in source assert '$EvidenceRunId = if ($RunId -match $SafeIdPattern)' in source assert "Get-Agent99RuntimeSourceBinding" in source assert "$result.fileCount -eq 20" in source @@ -31,9 +35,12 @@ def test_entrypoint_is_fixed_to_windows99_host110_and_p0_obs_002() -> None: assert "$result.entrypointHashMatched" in source assert '$CredentialFile = "/etc/awoooi/secrets/signoz-metadata-api-key"' in source assert ( - '$BundleId = "d1318c412aa303c5d8a5795f22ee62dce16ca79354c5a15e95bfad41a5e35533"' + '$BundleId = "ed4e5fbfc5f732b64d6a78f6ab26b90118e1d42a4f38b40d583876445c32e3aa"' in source ) + assert re.search(r'\$BundleId = "[0-9a-f]{64}"', source) + assert '$IsolatedRestorePath = "$ToolchainRoot/signoz-metadata-isolated-restore.py"' in source + assert '$IsolatedClusterConfigPath = "$ToolchainRoot/isolated-cluster.xml"' in source assert '$BaseUrl = "http://127.0.0.1:8080"' in source assert ( '$OutputDir = "/var/backups/awoooi/signoz-metadata-export-$RunId"' @@ -48,6 +55,34 @@ def test_entrypoint_is_fixed_to_windows99_host110_and_p0_obs_002() -> None: ) +def test_entrypoint_pins_current_seven_file_toolchain_bundle() -> None: + source = ENTRYPOINT.read_text(encoding="utf-8") + rows = ( + ("PolicyPath", ROOT / "config/signoz/metadata-export-policy.json"), + ("SharedContractPath", ROOT / "scripts/backup/signoz_metadata_contract.py"), + ("ExporterPath", ROOT / "scripts/backup/signoz-metadata-export.py"), + ("VerifierPath", ROOT / "scripts/backup/verify-signoz-metadata-export.py"), + ("RestoreDriverPath", ROOT / "scripts/backup/signoz-metadata-restore-drill.py"), + ("IsolatedRestorePath", ROOT / "scripts/backup/signoz-metadata-isolated-restore.py"), + ( + "IsolatedClusterConfigPath", + ROOT + / "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml", + ), + ) + hashes = [hashlib.sha256(path.read_bytes()).hexdigest() for _, path in rows] + bundle_id = hashlib.sha256( + "".join(f"{value}\n" for value in hashes).encode("ascii") + ).hexdigest() + + assert f'$BundleId = "{bundle_id}"' in source + for variable, expected_hash in zip( + (variable for variable, _path in rows), hashes, strict=True + ): + assert f'${variable} = "{expected_hash}"' in source + assert "PENDING_ISOLATED_RESTORE_SHA256" not in source + + def test_entrypoint_uses_fixed_public_key_transport_and_no_arbitrary_shell() -> None: source = ENTRYPOINT.read_text(encoding="utf-8") lock_helper = source[ @@ -184,6 +219,16 @@ def test_entrypoint_typed_readbacks_are_exact_and_preserve_unknown_state() -> No assert "outputPresent = if ($outputStateReadbackVerified)" in preflight assert "receiptPresent = if ($receiptStateReadbackVerified)" in preflight assert "credentialReferencePresent = if ($credentialReadbackVerified)" in preflight + assert "agent99_restore_startup:" in preflight + assert "restoreStartupContractReadbackVerified" in preflight + assert "restoreStartupContractReady" in preflight + assert "restoreStartupEnvironmentValuesOutput = $false" in preflight + assert "restoreStartupFullEnvironmentOutput = $false" in preflight + assert "{{json .Config.Env}}" not in preflight + assert "{{range .Config.Env}}" not in preflight + assert "docker exec signoz /bin/sh -c" in preflight + assert "tcp://clickhouse:9000*" in preflight + assert "/var/lib/signoz/signoz.db" in preflight credential_unverified = main.index("credential_reference_readback_unverified") credential_absent = main.index("credential_reference_absent") @@ -205,6 +250,9 @@ def test_entrypoint_typed_readbacks_are_exact_and_preserve_unknown_state() -> No transport_reserve = 30 assert guard + 90 + kill_after + transport_reserve < 300 assert guard + 720 + kill_after + transport_reserve < 900 + restore_cleanup_kill_after = 600 + assert guard + 1800 + restore_cleanup_kill_after + transport_reserve < 2550 + assert guard + 300 + restore_cleanup_kill_after + transport_reserve < 1050 def test_entrypoint_never_reads_or_persists_credential_value_in_controller() -> None: @@ -217,7 +265,9 @@ def test_entrypoint_never_reads_or_persists_credential_value_in_controller() -> assert 'secretValueTransmitted = $false' in source assert 'secretValuePersistedInEvidence = $false' in source assert "credentialReferenceDispatchAttempted" in source - assert "$script:ApplyDispatchAttempted = $true" in source + assert "$script:ExportCredentialDispatchAttempted = $true" in source + assert "$script:IsolatedApplyDispatchAttempted = $true" in source + assert "credentialReferenceDispatchAttempted = $script:ExportCredentialDispatchAttempted" in source assert "credentialReferenceConsumedOnTarget = if" in source assert "elseif (Test-Agent99SignozReceiptIdentity $ApplyReceipt" in source assert 'rawSqliteRead = $false' in source @@ -248,8 +298,8 @@ def test_entrypoint_requires_check_apply_independent_verify_and_partial_terminal assert '--receipt-file $RemoteReceipt' in source assert 'pass_export_verified_restore_pending' in source assert 'export_verified_restore_pending' in source - assert 'restoreTerminal = "pending_isolated_same_version_target"' in source - assert 'zeroResidueTerminal = "pending"' in source + assert 'else { "pending_isolated_same_version_target" }' in source + assert 'else { "pending" }' in source assert 'completionClaim = $false' in source assert "controlledApplyPhases = [pscustomobject]$controlledApplyPhases" in source assert 'credential_reference_absent' in source @@ -258,6 +308,40 @@ def test_entrypoint_requires_check_apply_independent_verify_and_partial_terminal assert 'post_verifier_runtime_or_artifact_drift' in source +def test_restore_modes_are_bounded_distinct_and_cleanup_remains_available() -> None: + source = ENTRYPOINT.read_text(encoding="utf-8") + main_start = source.index("foreach ($identity") + cleanup_start = source.index( + 'if ($Mode -eq "RestoreVerifyCleanup")', main_start + ) + restore_apply_start = source.index( + 'if ($Mode -in @("RestoreCheck", "RestoreApply"))', cleanup_start + ) + export_verify_start = source.index('if ($Mode -eq "Verify")', restore_apply_start) + cleanup_helper = source[ + source.index("function New-Agent99SignozCleanupLockedRemoteCommand") : + source.index("function Get-Agent99RuntimeSourceBinding") + ] + cleanup_branch = source[cleanup_start:restore_apply_start] + restore_apply_branch = source[restore_apply_start:export_verify_start] + + assert "Get-Agent99SignozCleanupPreflight" in cleanup_branch + assert "New-Agent99SignozCleanupLockedRemoteCommand" in cleanup_branch + assert "Get-Agent99SignozPreflight" not in cleanup_branch + assert "$OutputDir" not in cleanup_branch + assert "$BaseUrl/api/v1/health" not in cleanup_helper + assert "ExpectedArtifactState" not in cleanup_helper + assert "/usr/bin/docker info" in cleanup_helper + assert "--apply-receipt-file $RestoreRemoteReceipt" in cleanup_branch + assert '"independent_restore_verifier"' in cleanup_branch + assert '"portable_metadata_restore_independently_verified_zero_residue"' in cleanup_branch + assert '"portable_metadata_restore_verified_zero_residue_pending_independent_verifier"' in restore_apply_branch + assert cleanup_start < restore_apply_start < export_verify_start + assert "$RestoreCleanupKillAfterSeconds = 600" in source + assert "$RestoreApplyControllerWaitSeconds = 2550" in source + assert "$RestoreCleanupControllerWaitSeconds = 1050" in source + + def test_entrypoint_reserves_durable_evidence_before_any_remote_dispatch() -> None: source = ENTRYPOINT.read_text(encoding="utf-8") main = source[source.index("foreach ($identity") :] diff --git a/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py b/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py index d49dc59bc..3c8782c41 100644 --- a/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py +++ b/scripts/ops/tests/test_signoz_metadata_toolchain_deploy.py @@ -143,14 +143,14 @@ def test_check_mode_fails_closed_on_remote_drift(tmp_path: Path) -> None: assert rows[-1]["terminal"] == "failed_no_write" -def test_apply_uses_five_transfers_and_exact_postcheck(tmp_path: Path) -> None: +def test_apply_uses_seven_transfers_and_exact_postcheck(tmp_path: Path) -> None: result = run_deployer(tmp_path, mode="apply", apply=True) assert result.returncode == 0, result.stderr rows = receipt_rows(result.stdout) assert rows[-1]["terminal"] == "pass_source_deployed_inactive" assert rows[-1]["detail"] == "runtime_export_not_executed" events = (tmp_path / "events.log").read_text(encoding="utf-8").splitlines() - assert len([line for line in events if line.startswith("scp ")]) == 5 + assert len([line for line in events if line.startswith("scp ")]) == 7 assert len([line for line in events if line.startswith("ssh ")]) == 4 @@ -170,6 +170,10 @@ def test_deployer_has_no_active_pointer_or_destructive_fallback() -> None: assert "curl -fsSL" not in source assert "\nrm " not in source assert "ln -s" not in source + assert "signoz-metadata-isolated-restore.py" in source + assert "isolated-cluster.xml" in source + assert "local_exact_sources_7" in source + assert "[ \"${#EXPECTED_HASHES[@]}\" -eq 7 ]" in source def test_transport_propagates_strict_fixed_paths_with_spaces(tmp_path: Path) -> None: @@ -247,7 +251,7 @@ def test_remote_check_uses_privileged_read_only_visibility() -> None: assert "REMOTE_CHECK contains only stat/hash/process/API" in remote_check -def test_exact_five_file_supply_chain_and_modes_are_fixed() -> None: +def test_exact_seven_file_supply_chain_and_modes_are_fixed() -> None: source = DEPLOYER.read_text(encoding="utf-8") for path in ( "config/signoz/metadata-export-policy.json", @@ -255,7 +259,9 @@ def test_exact_five_file_supply_chain_and_modes_are_fixed() -> None: "scripts/backup/signoz-metadata-export.py", "scripts/backup/verify-signoz-metadata-export.py", "scripts/backup/signoz-metadata-restore-drill.py", + "scripts/backup/signoz-metadata-isolated-restore.py", + "ops/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml", ): assert path in source - assert "MODES=(0644 0644 0755 0755 0755)" in source - assert '[ "${#EXPECTED_HASHES[@]}" -eq 5 ]' in source + assert "MODES=(0644 0644 0755 0755 0755 0755 0644)" in source + assert '[ "${#EXPECTED_HASHES[@]}" -eq 7 ]' in source