clickhouse-migrate-test-support 0.2.0.0 → 0.2.1.0
raw patch · 9 files changed
+296/−94 lines, 9 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- CHANGELOG.md +12/−0
- clickhouse-migrate-test-support.cabal +1/−1
- test/RepairBoundaries.hs +7/−1
- test/RepairIntegration.hs +115/−1
- test/RunnerIntegration.hs +18/−0
- test/SurvivingWork.hs +7/−1
- test/data/clickhouse-matrix.json +43/−1
- test/scripts/native_fixture.py +63/−63
- test/scripts/replicated_fixture.py +30/−26
CHANGELOG.md view
@@ -1,5 +1,17 @@ # Changelog for clickhouse-migrate-test-support +## 0.2.1.0 — 2026-11-14++### Other Changes++- Use Apple Container with exact digest/platform inspection, owned bind mounts+ and inspected replica addresses.+- Cover denied/own-user process observations, callback failure/cancellation,+ exact interrupted receipts and active-work refusal under explicit external+ authority.+- Qualify bounded read-only inspection and the direct Linux arm64 26.8.28.14+ LTS image.+ ## 0.2.0.0 — 2026-11-08 ### New Features
clickhouse-migrate-test-support.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.12 name: clickhouse-migrate-test-support-version: 0.2.0.0+version: 0.2.1.0 synopsis: Isolated server fixtures for ClickHouse migration tests description: Isolated server fixtures for ClickHouse migration tests.
test/RepairBoundaries.hs view
@@ -11,6 +11,7 @@ import ClickHouseMigrate.TestSupport.Server import Control.Monad (forM_, when) import Data.Aeson (Value)+import Data.IORef (modifyIORef', newIORef, readIORef) import Data.Text qualified as T import Data.Text.IO qualified as TIO import RunnerIntegration (assert, field, fixture, fromJson, planFor, provider, right)@@ -82,7 +83,10 @@ newPosition <- maybe (fail "Missing new position") pure currentPosition (newAttempt, newSteps) <- case current of [R.MigrationSummary _ _ (Just value) values] -> pure (value, values); _ -> fail "Missing new active attempt" assert "Authorization did not change attempt" (newAttempt /= attempt)- freshEvidence <- right (Repair.recoveryEvidence audit quiet receipt)+ -- Exercise the explicit external policy across the real lost-authorization+ -- receipts, with all original position/step/receipt checks still in force.+ processObservations <- newIORef []+ freshEvidence <- right (Repair.externalDrainRecoveryEvidence audit quiet receipt (\value -> modifyIORef' processObservations (<> [value]))) fresh <- case action of "accept" -> do step <- selected newSteps@@ -104,6 +108,8 @@ pure () else pure () _ <- I.verify I.Observe bound (I.verifyOptions True) plan >>= right+ observations <- readIORef processObservations+ assert "Receipt recovery lost its explicit diagnostic distinction" (not (null observations) && all (== Repair.NoVisibleRecordedQueries) observations) assert "Recovery replayed completed steps or executed acceptance SQL" . (== 2) =<< count proxy putStrLn ("repair authorization " ++ T.unpack action ++ ": lost response=" ++ show lost ++ "; ordinary restart blocked; fresh request succeeds without replay") -- Acceptance and finalization have their own dependent append acknowledgements.
test/RepairIntegration.hs view
@@ -1,6 +1,7 @@ module RepairIntegration (repairEarlyCases) where import ClickHouseMigrate+import ClickHouseMigrate.CLI qualified as CLI import ClickHouseMigrate.Event qualified as Event import ClickHouseMigrate.Inspection qualified as Inspect import ClickHouseMigrate.Ledger qualified as Ledger@@ -9,7 +10,8 @@ import ClickHouseMigrate.Runner qualified as Run import ClickHouseMigrate.TestSupport.FaultProxy import ClickHouseMigrate.TestSupport.Server-import Control.Monad (forM_, when)+import Control.Exception (AsyncException (ThreadKilled), SomeAsyncException, SomeException, fromException, throwIO, try)+import Control.Monad (forM_, void, when) import Data.Aeson (Value) import Data.IORef (modifyIORef', newIORef, readIORef) import Data.List.NonEmpty (NonEmpty (..))@@ -61,6 +63,8 @@ repairEarlyCases :: Server -> IO () repairEarlyCases server = do+ externalDrainCase server+ externalOperationCases server oversizedRecovery server access <- provider audit <- right (Ledger.auditContext "runner fixture" "explicit first use" ["owned server; parent serializes whole job"])@@ -197,3 +201,113 @@ _ <- Inspect.verify Inspect.Observe bound (Inspect.verifyOptions True) plan >>= right assert "accept/finalize submitted application SQL" . (== 0) =<< applicationCount proxy putStrLn "recovery size: oversized acceptance preserves (3,3) events, position and attempt; combined audits reject authorization; replacement against original position accepts and finalizes without SQL"++-- First real embedding producer/consumer: separate app-scoped credentials may+-- not inspect processes, while the parent owns every earlier completed request.+externalDrainCase :: Server -> IO ()+externalDrainCase server = do+ let label = "repair_external_drain"+ username = "repair_external_user"+ ledger <- newLedgerId+ key <- right (targetKey ("runner-tests/" <> label))+ access <- provider+ audit <- right (Ledger.auditContext "owned external-drain fixture" "all predecessor calls joined under parent exclusion" ["owned fixture: no runtime, schedule or predecessor remains"])+ quiet <- right (Ledger.attestQuiescence key audit)+ plan <- planWithTail label+ owner <- fixture server label (serverEndpoint server) FirstUse+ let stop (Event.RunEvent phase _ _ _ step _ _) = when (phase == Event.BeforeSql && step == Just "insert") (fail "controlled before-SQL stop")+ initial <- Run.up (Run.mutationContext owner access stop Event.redactedDiagnostics) (Run.upOptions audit (Just (Ledger.authorizeInitialization key ledger audit))) plan+ report <- case initial of Left (R.RunFailure (R.CallbackFailed _) acknowledged _) -> pure acknowledged; _ -> fail "External fixture did not retain an uncertain intent"+ (position, attempt, steps) <- checkpoint report+ step <- selected steps+ config <- scopedRepairFixture server label username ledger False+ let context = Run.mutationContext config access Event.quietEvents Event.redactedDiagnostics+ strict <- right (Repair.recoveryEvidence audit quiet []) >>= right . Repair.retryStepRequest position attempt step+ rejected <- Repair.repair context plan strict+ assert "Strict repair did not refuse denied process metadata" (case rejected of Left (R.RunFailure (R.LedgerRequestFailed _ _ _) _ _) -> True; _ -> False)+ -- The observer is part of pre-authorization policy and cannot fail open.+ throwing <- right (Repair.externalDrainRecoveryEvidence audit quiet [] (const (fail "SQL_SECRET_MARKER"))) >>= right . Repair.retryStepRequest position attempt step+ failed <- Repair.repair context plan throwing+ assert "External observer failure did not stop authorization" (case failed of Left (R.RunFailure (R.CallbackFailed (R.EventCallback "RepairProcessObservation")) _ _) -> not ("SQL_SECRET_MARKER" `Text.isInfixOf` Text.pack (show failed)); _ -> False)+ cancelling <- right (Repair.externalDrainRecoveryEvidence audit quiet [] (const (throwIO ThreadKilled))) >>= right . Repair.retryStepRequest position attempt step+ cancelled <- try @SomeException (Repair.repair context plan cancelling)+ assert+ "Observer cancellation became an ordinary repair result"+ ( case cancelled of+ Left exception -> case fromException exception :: Maybe SomeAsyncException of+ Just _ -> True+ _ -> False+ Right _ -> False+ )+ observations <- newIORef []+ let sink observation = modifyIORef' observations (<> [observation])+ environment = CLI.commandEnvironment (Just config) (Just access) Event.quietEvents Event.redactedDiagnostics CLI.AdministrativeCommands+ input = CLI.AuditInput "owned embedding" "independent predecessor drain retained" ["owned fixture: all calls joined; no other submitter exists"]+ command = CLI.RepairCommand CLI.RetryStep (CLI.RepairInput key position attempt (Just step) input True [])+ result <- CLI.handleCommandWithExternalDrain sink environment plan command+ retried <- case result of CLI.CommandResult _ (Right (CLI.RunReport value)) -> pure value; _ -> fail (Text.unpack (Encoding.decodeUtf8 (CLI.renderJson result)))+ assert "Denied observation was erased into an empty result" . (== [Repair.ProcessObservationUnavailable]) =<< readIORef observations+ assert "External retry did not execute the exact selected step" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)+ staleEvidence <- right (Repair.externalDrainRecoveryEvidence audit quiet [] sink)+ staleRequest <- right (Repair.retryStepRequest position attempt step staleEvidence)+ stale <- Repair.repair context plan staleRequest+ assert "External authority bypassed a stale position" (case stale of Left (R.RunFailure (R.StalePosition _ _) _ _) -> True; _ -> False)+ assert "Stale repair reached optional diagnostics" . (== 1) . length =<< readIORef observations+ (after, nextAttempt, _) <- checkpoint retried+ resumed <- right (Repair.resumeRequest after nextAttempt staleEvidence) >>= Repair.repair context plan >>= right+ assert "External resume failed the exact pending suffix" (case resumed of R.RunReport _ _ _ _ _ [R.MigrationSummary _ R.Applied _ _, R.MigrationSummary _ R.Pending _ _] _ _ -> True; _ -> False)+ assert "External resume replayed the selected insert" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)+ putStrLn "external repair: actual denied metadata; strict refusal; callback failure/cancellation; CLI retry and SDK resume; stale-position refusal"++scopedRepairFixture :: Server -> Text.Text -> Text.Text -> LedgerId -> Bool -> IO TargetConfig+scopedRepairFixture server label username ledger ownProcesses = do+ key <- right (targetKey ("runner-tests/" <> label))+ void (queryServer server ("CREATE USER " <> username <> " IDENTIFIED WITH no_password"))+ void (queryServer server ("GRANT SELECT,INSERT,ALTER ON default." <> label <> " TO " <> username))+ void (queryServer server ("GRANT SELECT,INSERT ON " <> label <> "_history.* TO " <> username))+ forM_ ["databases", "tables", "columns", "storage_policies", "disks"] $ \table ->+ void (queryServer server ("GRANT SELECT ON system." <> table <> " TO " <> username))+ rows <- queryServer server "SELECT toString(serverUUID()) AS id" >>= fromJson :: IO [Value]+ identity <- case rows of [row] -> field "id" row; _ -> fail "Owned server identity missing"+ config <- targetConfig key <$> right (directEndpoint (serverEndpoint server)) <*> right (serverId identity) <*> right (ledgerNamespace (label <> "_history")) <*> pure (ExistingLedger ledger) <*> right (basicCredentials username "") <*> pure defaultHttpLimits+ when ownProcesses (void (queryServer server ("GRANT SELECT ON system.processes TO " <> username)))+ pure config++externalOperationCases :: Server -> IO ()+externalOperationCases server = forM_ [(visible, action) | visible <- [False, True], action <- ["retry", "accept", "finalize"], visible || action /= "retry"] $ \(visible, action) -> do+ let suffix = if visible then "own" else "denied"+ label = "repair_external_" <> action <> "_" <> suffix+ username = label <> "_user"+ ledger <- newLedgerId+ key <- right (targetKey ("runner-tests/" <> label))+ access <- provider+ audit <- right (Ledger.auditContext "scoped recovery fixture" "all previous native requests joined; parent retains target exclusion" ["owned serialized fixture: no unjoined client or schedule"])+ quiet <- right (Ledger.attestQuiescence key audit)+ plan <- planWithTail label+ owner <- fixture server label (serverEndpoint server) FirstUse+ let stop (Event.RunEvent phase _ _ _ step _ _) = when ((action == "finalize" && phase == Event.BeforeMigrationComplete) || (action /= "finalize" && phase == Event.BeforeSql && step == Just "insert")) (fail "controlled recovery boundary")+ initial <- Run.up (Run.mutationContext owner access stop Event.redactedDiagnostics) (Run.upOptions audit (Just (Ledger.authorizeInitialization key ledger audit))) plan+ report <- case initial of Left (R.RunFailure (R.CallbackFailed _) value _) -> pure value; _ -> fail "Scoped recovery interruption missing"+ (position, attempt, steps) <- checkpoint report+ when (action == "accept") (void (queryServer server ("INSERT INTO default." <> label <> " VALUES(7,'SQL_SECRET_MARKER')")))+ config <- scopedRepairFixture server label username ledger visible+ observations <- newIORef []+ evidence <- right (Repair.externalDrainRecoveryEvidence audit quiet [] (\observation -> modifyIORef' observations (<> [observation])))+ request <-+ if action == "finalize"+ then right (Repair.finalizeRequest position attempt evidence)+ else do+ step <- selected steps+ right (if action == "accept" then Repair.acceptStepRequest position attempt step evidence else Repair.retryStepRequest position attempt step evidence)+ let context = Run.mutationContext config access Event.quietEvents Event.redactedDiagnostics+ expectedObservation = if visible then Repair.NoVisibleRecordedQueries else Repair.ProcessObservationUnavailable+ repaired <- Repair.repair context plan request >>= right+ assert "Scoped observation lost its visibility distinction" . (== [expectedObservation]) =<< readIORef observations+ assert "Acceptance/finalization replayed or omitted actual effects" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)+ when (action /= "finalize") $ do+ assert "Selected recovery ran the remaining suffix" . (== 0) =<< scalar server ("SELECT toUInt32(count()) AS value FROM system.columns WHERE database='default' AND table='" <> label <> "' AND name='later'")+ (after, next, _) <- checkpoint repaired+ _ <- right (Repair.resumeRequest after next evidence) >>= Repair.repair context plan >>= right+ assert "Scoped resume did not execute the pending suffix" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM system.columns WHERE database='default' AND table='" <> label <> "' AND name='later'")+ assert "Scoped recovery started another migration" . (== 1) =<< scalar server ("SELECT toUInt32(count()) AS value FROM default." <> label)+ putStrLn ("external repair: " ++ Text.unpack action ++ "; process visibility=" ++ Text.unpack suffix ++ "; exact step and pending suffix preserved")
test/RunnerIntegration.hs view
@@ -134,6 +134,24 @@ kinds <- mapM (field "stage") received :: IO [Maybe Text.Text] assert "Completed migration replayed" (length (filter (== Just "application-sql") kinds) == 2) _ <- Inspect.verify Inspect.Observe bound (Inspect.verifyOptions True) plan >>= right+ -- A constrained reader rejects the legacy unlimited-result requests.+ -- Explicit bounded observation succeeds; overflow cannot return a prefix.+ let reader = "bounded_history_reader"+ namespace = label <> "_history"+ _ <- queryServer server ("CREATE USER " <> reader <> " IDENTIFIED BY 'bounded-test' SETTINGS readonly=2 CONST, max_result_rows=10000 MIN 1 MAX 10000, max_result_bytes=1048576 MIN 1 MAX 1048576")+ _ <- queryServer server ("GRANT SELECT ON " <> namespace <> ".* TO " <> reader)+ _ <- queryServer server ("GRANT SELECT ON system.disks TO " <> reader)+ _ <- queryServer server ("GRANT SELECT ON system.storage_policies TO " <> reader)+ sid <- case configuredTopology bound of DirectServer value -> pure value; _ -> fail "expected direct fixture"+ readerConfig <- targetConfig (configuredTargetKey bound) <$> right (directEndpoint (serverEndpoint server)) <*> pure sid <*> right (ledgerNamespace namespace) <*> pure (ExistingLedger persisted) <*> right (basicCredentials reader "bounded-test") <*> pure defaultHttpLimits+ legacy <- Inspect.verify Inspect.Observe readerConfig (Inspect.verifyOptions True) plan+ assert "unbounded request unexpectedly admitted by constrained role" (case legacy of Left _ -> True; _ -> False)+ bounded <- maybe (fail "invalid bounds") pure (Inspect.boundedObservation 10000 1048576)+ _ <- Inspect.verify bounded readerConfig (Inspect.verifyOptions True) plan >>= right+ tiny <- maybe (fail "invalid tiny bound") pure (Inspect.boundedObservation 1 1048576)+ overflow <- Inspect.verify tiny readerConfig (Inspect.verifyOptions True) plan+ assert "bounded ledger overflow returned false success" (case overflow of Left _ -> True; _ -> False)+ putStrLn "bounded readonly inspection: constrained role accepted; overflowing history refuses without prefix success" changed <- planFor label 8 blocked <- Run.up (Run.mutationContext bound access Event.quietEvents Event.redactedDiagnostics) (Run.upOptions audit Nothing) changed assert "Modified applied SQL accepted" (case blocked of Left (R.RunFailure (R.IncompatibleHistory _) _ _) -> True; _ -> False)
test/SurvivingWork.hs view
@@ -15,7 +15,7 @@ import Control.Exception (SomeAsyncException, finally, fromException) import Control.Monad (forM_, void) import Data.Aeson (Value)-import Data.IORef (atomicModifyIORef', newIORef)+import Data.IORef (atomicModifyIORef', modifyIORef', newIORef, readIORef) import Data.List.NonEmpty (NonEmpty (..)) import Data.Set qualified as Set import Data.Text qualified as T@@ -122,6 +122,12 @@ unsafe <- right (Repair.acceptStepRequest expected attempt selected falseEvidence) rejected <- Repair.repair (Run.mutationContext bound broken E.quietEvents E.redactedDiagnostics) plan unsafe assert "Recovery accepted a positively active predecessor" (case rejected of Left (R.RunFailure (R.PriorQueriesActive queries) _ _) -> query `elem` queries; _ -> False)+ observations <- newIORef []+ externalEvidence <- right (Repair.externalDrainRecoveryEvidence audit falseQuiet [] (\value -> modifyIORef' observations (<> [value])))+ externalRequest <- right (Repair.acceptStepRequest expected attempt selected externalEvidence)+ externallyRejected <- Repair.repair (Run.mutationContext bound broken E.quietEvents E.redactedDiagnostics) plan externalRequest+ assert "External policy accepted a positively active predecessor" (case externallyRejected of Left (R.RunFailure (R.PriorQueriesActive queries) _ _) -> query `elem` queries; _ -> False)+ assert "Active work was not retained as a positive diagnostic" . any (\case Repair.RecordedQueriesActive queries -> query `elem` queries; _ -> False) =<< readIORef observations stopClient assert "Server mutation did not survive client shutdown" =<< activeMutation blocked <- hostLocalExclusiveAccess root 1000000 Nothing >>= right
test/data/clickhouse-matrix.json view
@@ -1,6 +1,6 @@ { "matrixVersion": 1,- "checkedAt": "2026-11-07",+ "checkedAt": "2026-11-14", "policy": "Only exact qualified patches and declared topology groups; no prereleases or managed-cloud certification", "entries": [ {@@ -153,6 +153,48 @@ ], "evidence": "docs/release-evidence/2026-11-07-replicated.md" }+ },+ {+ "id": "lts-linux-arm64",+ "version": "26.8.28.14",+ "tag": "v26.8.28.14-lts",+ "commit": "3694fe07040369cef78c937311cbaac5f0555484",+ "platform": "aarch64-linux",+ "architecture": "arm64",+ "image": "docker.io/clickhouse/clickhouse-server@sha256:597add6eeadf0b3c562b58c9a63d9712b9a0cde4e49b68e64a697dbf2b363b25",+ "indexDigest": "sha256:63e02621435cde1146ab0fde66057091ee75b312d12917c0109a6a0c99bc7ce1",+ "status": "native-qualified",+ "groups": [+ "transport",+ "completion",+ "ledger",+ "exclusive",+ "runner",+ "repair",+ "boundaries",+ "survivors",+ "adoption",+ "consumer",+ "cli",+ "example",+ "remediation"+ ],+ "scope": "Owned direct Atomic Linux arm64 server; no replicated, managed-cloud or Linux-client qualification",+ "qualification": {+ "date": "2026-11-14",+ "cohort": "f4bc4e6d32decfaaf1a19e1033a22f3e28fc167c",+ "compiler": "ghc-9.12.4",+ "clientHost": "macOS arm64",+ "requiredGroups": 13,+ "artifactVariants": [+ "working-tree Cabal binaries",+ "seven extracted sibling source distributions",+ "Nix installed suite, driver and example"+ ],+ "evidence": "docs/release-evidence/2026-11-14-lts-linux-arm64.md",+ "scope": "Direct Atomic local synchronous effects on exact Linux arm64 OCI artifact; no replicated, managed-cloud or Linux-client certification"+ },+ "tagObject": "cb8b2f96f32feb4203ab9edbf4f03b274bbb6952" } ], "unqualifiedReleases": [
test/scripts/native_fixture.py view
@@ -58,6 +58,18 @@ return binary +def verify_apple_image(entry):+ """Require the exact local matrix digest and Linux architecture; never a tag."""+ image = json.loads(subprocess.check_output(+ ['container', 'image', 'inspect', entry['image']], text=True))[0]+ digest = entry['image'].split('@', 1)[1]+ if not any(v.get('digest') == digest and+ v.get('platform', {}).get('architecture') == entry['architecture'] and+ v.get('platform', {}).get('os') == 'linux'+ for v in image.get('variants', [])):+ raise RuntimeError('Apple Container candidate digest/platform mismatch; pre-pull the exact matrix digest')++ class NativeServer: def __init__(self, entry, deployment=None): self.entry = entry@@ -65,18 +77,10 @@ self.port = port() self.tcp_port = port() self.container = None- self.docker = None+ self.runtime = None if 'image' in entry:- context = os.environ.get('CLICKHOUSE_MIGRATE_DOCKER_CONTEXT')- if not context:- raise RuntimeError('OCI candidate requires an explicit isolated Docker context')- self.docker = ['docker', '--context', context]- endpoint = json.loads(subprocess.check_output(self.docker + ['context', 'inspect', context], text=True))[0]['Endpoints']['docker']['Host']- if not endpoint.startswith('unix://'):- raise RuntimeError('Owned OCI fixture requires a local Unix Docker socket')- image = json.loads(subprocess.check_output(self.docker + ['image', 'inspect', entry['image']], text=True))[0]- if image['Architecture'] != entry['architecture'] or image['Os'] != 'linux':- raise RuntimeError('OCI candidate platform mismatch; pre-pull the exact matrix digest')+ self.runtime = ['container']+ verify_apple_image(entry) self.binary = None else: self.binary = binary_for(entry)@@ -92,13 +96,13 @@ <profile>default</profile><quota>default</quota><access_management>1</access_management></default></users> <quotas><default><interval><duration>3600</duration><queries>0</queries><errors>0</errors> <result_rows>0</result_rows><read_rows>0</read_rows><execution_time>0</execution_time></interval></default></quotas></clickhouse>''')- root = '/fixture' if self.docker else escape(str(self.root))- if self.docker:+ root = '/fixture' if self.runtime else escape(str(self.root))+ if self.runtime: users.write_text(users.read_text().replace('<ip>127.0.0.1</ip>', '<ip>::/0</ip>'))- users_path = '/fixture/users.xml' if self.docker else str(users)- listener = '0.0.0.0' if self.docker else '127.0.0.1'- http_port = 8123 if self.docker else self.port- tcp_port = 9000 if self.docker else self.tcp_port+ users_path = '/fixture/users.xml' if self.runtime else str(users)+ listener = '0.0.0.0' if self.runtime else '127.0.0.1'+ http_port = 8123 if self.runtime else self.port+ tcp_port = 9000 if self.runtime else self.tcp_port (self.root / 'config.xml').write_text(f'''<clickhouse> <logger><level>warning</level><log>{root}/server.log</log><errorlog>{root}/error.log</errorlog></logger> <path>{root}/data/</path><tmp_path>{root}/tmp/</tmp_path><user_files_path>{root}/user_files/</user_files_path>@@ -118,47 +122,48 @@ config = self.root / 'config.xml' config.write_text(config.read_text().replace('</clickhouse>', extra + '</clickhouse>')) try:- if self.docker:+ if self.runtime: name = deployment['name'] if deployment else 'clickhouse-migrate-' + uuid.uuid4().hex- subprocess.run(self.docker + ['create', '--name', name, '--label', 'clickhouse-migrate.fixture=owned',+ self.container = name+ subprocess.run(self.runtime + ['run', '--detach', '--name', name,+ '--label', 'clickhouse-migrate.fixture=owned', *(['--network', deployment['network']] if deployment else []), '--user', str(os.getuid()), '--memory', '2g', '--cpus', '2',- '-p', '127.0.0.1::8123', '-p', '127.0.0.1::9000',+ '--platform', 'linux/' + entry['architecture'],+ '--volume', str(self.root) + ':/fixture',+ '-p', f'127.0.0.1:{self.port}:8123',+ '-p', f'127.0.0.1:{self.tcp_port}:9000', '--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'],- check=True, capture_output=True, timeout=20)- self.container = name- subprocess.run(self.docker + ['cp', str(self.root) + '/.', name + ':/fixture'], check=True, capture_output=True, timeout=20)- # Docker allocates ports on its own host, not the client OS.- # Keep the namespace alive while restarting only ClickHouse so- # these exact bindings remain reserved for the fixture lifetime.- subprocess.run(self.docker + ['start', name], check=True, capture_output=True, timeout=20)- self.port = int(subprocess.check_output(self.docker + ['port', name, '8123/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])- self.tcp_port = int(subprocess.check_output(self.docker + ['port', name, '9000/tcp'], text=True, timeout=5).strip().rsplit(':', 1)[1])- self.start()- deadline = time.monotonic() + 30- while time.monotonic() < deadline:- if self.process.poll() is not None:- raise RuntimeError(f'Owned server exited before readiness: {self.process.returncode}')- try:- with urllib.request.urlopen(self.endpoint + '/ping', timeout=.2) as response:- if response.read() == b'Ok.\n':- break- except OSError:- pass- time.sleep(.05)- else:- raise RuntimeError('Owned server readiness deadline expired')- identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0]- version = identity['version']- if identity['server_id'] != self.retained_uuid():- raise RuntimeError('Endpoint is not the owned server data directory')- if version != entry['version']:- raise RuntimeError('Candidate server version mismatch')+ check=True, capture_output=True, timeout=60)+ if not (deployment and deployment.get('defer_start')):+ self.ready() except BaseException: print(f"Owned server failure evidence: {self.root}", file=sys.stderr, flush=True) self.close(False) raise + def ready(self):+ self.start()+ deadline = time.monotonic() + 30+ while time.monotonic() < deadline:+ if self.process.poll() is not None:+ raise RuntimeError(f'Owned server exited before readiness: {self.process.returncode}')+ try:+ with urllib.request.urlopen(self.endpoint + '/ping', timeout=.2) as response:+ if response.read() == b'Ok.\n':+ break+ except OSError:+ pass+ time.sleep(.05)+ else:+ raise RuntimeError('Owned server readiness deadline expired')+ identity = self.query('SELECT version() AS version, toString(serverUUID()) AS server_id')[0]+ version = identity['version']+ if identity['server_id'] != self.retained_uuid():+ raise RuntimeError('Endpoint is not the owned server data directory')+ if version != self.entry['version']:+ raise RuntimeError('Candidate server version mismatch')+ @property def endpoint(self): return f'http://127.0.0.1:{self.port}'@@ -169,25 +174,25 @@ return [json.loads(line) for line in response.read().splitlines() if line.strip()] def retained_uuid(self):- if self.docker:- return subprocess.check_output(self.docker + ['exec', self.container, 'cat', '/fixture/data/uuid'], text=True, timeout=5).strip()+ if self.runtime:+ return subprocess.check_output(self.runtime + ['exec', self.container, 'cat', '/fixture/data/uuid'], text=True, timeout=5).strip() return (self.root / 'data/uuid').read_text().strip() def start(self):- command = self.docker + ['exec', self.container, '/bin/sh', '-c', 'echo $$ > /fixture/owned-server.pid; exec clickhouse server --config-file=/fixture/config.xml'] if self.docker else [str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')]+ command = self.runtime + ['exec', self.container, '/bin/sh', '-c', 'echo $$ > /fixture/owned-server.pid; exec clickhouse server --config-file=/fixture/config.xml'] if self.runtime else [str(self.binary), 'server', '--config-file', str(self.root / 'config.xml')] self.process = subprocess.Popen(command, stdout=self.log, stderr=subprocess.STDOUT, start_new_session=True) def stop(self, sig): if self.process is not None and self.process.poll() is None:- if self.docker:- subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -s "$1" "$(cat /fixture/owned-server.pid)"', 'sh', signal.Signals(sig).name.removeprefix('SIG')], check=True, capture_output=True, timeout=10)+ if self.runtime:+ subprocess.run(self.runtime + ['exec', self.container, '/bin/sh', '-c', 'kill -s "$1" "$(cat /fixture/owned-server.pid)"', 'sh', signal.Signals(sig).name.removeprefix('SIG')], check=True, capture_output=True, timeout=10) else: os.killpg(self.process.pid, sig) try: self.process.wait(timeout=10) except subprocess.TimeoutExpired:- if self.docker:- subprocess.run(self.docker + ['exec', self.container, '/bin/sh', '-c', 'kill -KILL "$(cat /fixture/owned-server.pid)"'], check=True, capture_output=True, timeout=10)+ if self.runtime:+ subprocess.run(self.runtime + ['exec', self.container, '/bin/sh', '-c', 'kill -KILL "$(cat /fixture/owned-server.pid)"'], check=True, capture_output=True, timeout=10) else: os.killpg(self.process.pid, signal.SIGKILL) self.process.wait(timeout=5)@@ -216,13 +221,8 @@ self.stop(signal.SIGTERM) finally: if self.container:- # Copy only this fixture's private data/logs before removing its- # recorded container. No ambient container or host volume is used.- if not success:- evidence = self.root / 'container-evidence'- evidence.mkdir()- subprocess.run(self.docker + ['cp', self.container + ':/fixture/.', str(evidence)], capture_output=True, timeout=30, check=True)- subprocess.run(self.docker + ['rm', '--force', self.container], capture_output=True, timeout=15, check=True)+ # The private bind mount retains failure evidence on the host.+ subprocess.run(self.runtime + ['delete', '--force', self.container], capture_output=True, timeout=30, check=True) self.log.close() if success: shutil.rmtree(self.root)
test/scripts/replicated_fixture.py view
@@ -1,5 +1,5 @@ #!/usr/bin/env python3-"""Owned Keeper plus two replicas; no shared server or Docker network."""+"""Owned Keeper plus two replicas; no shared server or container network.""" import base64 import http.client import json@@ -17,7 +17,7 @@ import urllib.parse import uuid -from native_fixture import NativeServer, PASSWORD+from native_fixture import NativeServer, PASSWORD, port, verify_apple_image class RoundRobin:@@ -70,23 +70,15 @@ raise RuntimeError('Replicated fixture requires an OCI candidate') self.entry = entry self.root = Path(tempfile.mkdtemp(prefix='clickhouse-migrate-replicated-', dir='/tmp'))- context = os.environ.get('CLICKHOUSE_MIGRATE_DOCKER_CONTEXT')- if not context:- raise RuntimeError('Replicated fixture requires an explicit isolated Docker context')- self.docker = ['docker', '--context', context]- endpoint = json.loads(subprocess.check_output(self.docker + ['context', 'inspect', context], text=True))[0]['Endpoints']['docker']['Host']- if not endpoint.startswith('unix://'):- raise RuntimeError('Owned replicated fixture requires a local Unix Docker socket')- image = json.loads(subprocess.check_output(self.docker + ['image', 'inspect', entry['image']], text=True))[0]- if image['Architecture'] != entry['architecture'] or image['Os'] != 'linux':- raise RuntimeError('Replicated OCI candidate platform mismatch; pre-pull the exact matrix digest')+ self.runtime = ['container']+ verify_apple_image(entry) self.network = 'clickhouse-migrate-' + uuid.uuid4().hex self.keeper = None self.replicas = [] self.proxy = None self.closed = False try:- subprocess.run(self.docker + ['network', 'create', '--label', 'clickhouse-migrate.fixture=owned', self.network], check=True, capture_output=True)+ subprocess.run(self.runtime + ['network', 'create', '--label', 'clickhouse-migrate.fixture=owned', self.network], check=True, capture_output=True) keeper_name = self.network + '-keeper' self.keeper = keeper_name keeper_config = self.root / 'keeper.xml'@@ -96,12 +88,15 @@ <coordination_settings><operation_timeout_ms>10000</operation_timeout_ms><session_timeout_ms>10000</session_timeout_ms></coordination_settings> <raft_configuration><server><id>1</id><hostname>localhost</hostname><port>9234</port></server></raft_configuration> </keeper_server></clickhouse>''')- subprocess.run(self.docker + ['create', '--name', keeper_name, '--network', self.network, '--label', 'clickhouse-migrate.fixture=owned', '--user', str(os.getuid()), '-p', '127.0.0.1::9181', '--memory', '512m', '--entrypoint', '/bin/sh', entry['image'], '-c', 'exec sleep 2147483647'], check=True, capture_output=True)- subprocess.run(self.docker + ['cp', str(self.root) + '/.', keeper_name + ':/fixture'], check=True, capture_output=True)- subprocess.run(self.docker + ['start', keeper_name], check=True, capture_output=True)+ keeper_port = port()+ subprocess.run(self.runtime + ['run', '--detach', '--name', keeper_name,+ '--network', self.network, '--label', 'clickhouse-migrate.fixture=owned',+ '--user', str(os.getuid()), '-p', f'127.0.0.1:{keeper_port}:9181',+ '--memory', '512m', '--platform', 'linux/' + entry['architecture'],+ '--volume', str(self.root) + ':/fixture', '--entrypoint', '/bin/sh',+ entry['image'], '-c', 'exec sleep 2147483647'], check=True, capture_output=True, timeout=60) self.keeper_log = (self.root / 'keeper.log').open('w')- self.keeper_process = subprocess.Popen(self.docker + ['exec', keeper_name, 'clickhouse', 'keeper', '--config-file=/fixture/keeper.xml'], stdout=self.keeper_log, stderr=subprocess.STDOUT)- keeper_port = int(subprocess.check_output(self.docker + ['port', keeper_name, '9181/tcp'], text=True).strip().rsplit(':', 1)[1])+ self.keeper_process = subprocess.Popen(self.runtime + ['exec', keeper_name, 'clickhouse', 'keeper', '--config-file=/fixture/keeper.xml'], stdout=self.keeper_log, stderr=subprocess.STDOUT) deadline = time.monotonic() + 20 while time.monotonic() < deadline: if self.keeper_process.poll() is not None:@@ -117,11 +112,20 @@ else: raise RuntimeError('Keeper readiness deadline expired') names = [self.network + '-r1', self.network + '-r2']- cluster = '<remote_servers><fixture><shard>' + ''.join('<replica><host>' + name + '</host><port>9000</port><user>default</user><password>' + PASSWORD + '</password></replica>' for name in names) + '</shard></fixture></remote_servers>'- for i, name in enumerate(names):- extra = cluster + '<zookeeper><node><host>' + keeper_name + '</host><port>9181</port></node></zookeeper><macros><cluster>fixture</cluster><shard>1</shard><replica>r' + str(i + 1) + '</replica></macros><interserver_http_port>9009</interserver_http_port><interserver_http_host>' + name + '</interserver_http_host><distributed_ddl><path>/clickhouse/task_queue/ddl</path></distributed_ddl>'- self.replicas.append(NativeServer(entry, dict(name=name, network=self.network, config=extra)))- (self.root / 'replicas.json').write_text(json.dumps([dict(replica='r' + str(n + 1), root=str(replica.root), endpoint=replica.endpoint) for n, replica in enumerate(self.replicas)], indent=2) + '\n')+ for name in names:+ self.replicas.append(NativeServer(entry, dict(name=name, network=self.network, config='', defer_start=True)))+ def address(name):+ inspected = json.loads(subprocess.check_output(['container', 'inspect', name], text=True))[0]+ return inspected['status']['networks'][0]['ipv4Address'].split('/')[0]+ addresses = [address(name) for name in names]+ keeper_address = address(keeper_name)+ cluster = '<remote_servers><fixture><shard>' + ''.join('<replica><host>' + ip + '</host><port>9000</port><user>default</user><password>' + PASSWORD + '</password></replica>' for ip in addresses) + '</shard></fixture></remote_servers>'+ for i, replica in enumerate(self.replicas):+ extra = cluster + '<zookeeper><node><host>' + keeper_address + '</host><port>9181</port></node></zookeeper><macros><cluster>fixture</cluster><shard>1</shard><replica>r' + str(i + 1) + '</replica></macros><interserver_http_port>9009</interserver_http_port><interserver_http_host>' + addresses[i] + '</interserver_http_host><distributed_ddl><path>/clickhouse/task_queue/ddl</path></distributed_ddl>'+ config = replica.root / 'config.xml'+ config.write_text(config.read_text().replace('</clickhouse>', extra + '</clickhouse>'))+ replica.ready()+ (self.root / 'replicas.json').write_text(json.dumps([dict(replica='r' + str(n + 1), root=str(r.root), endpoint=r.endpoint) for n, r in enumerate(self.replicas)], indent=2) + '\n') # Administrative bootstrap exactly mirrors consumer infrastructure. for database in ['fixture_app', 'fixture_ledger']: result = self.request(0, "CREATE DATABASE " + database + " ON CLUSTER fixture ENGINE=Replicated('/clickhouse/databases/{uuid}', '{shard}', '{replica}')", user='default')@@ -182,13 +186,13 @@ for replica in reversed(self.replicas): replica.close(success) if self.keeper:- logs = subprocess.run(self.docker + ['logs', self.keeper], capture_output=True, text=True)+ logs = subprocess.run(self.runtime + ['logs', self.keeper], capture_output=True, text=True) (self.root / 'keeper-container.log').write_text(logs.stdout + logs.stderr)- subprocess.run(self.docker + ['rm', '-f', self.keeper], capture_output=True, timeout=15)+ subprocess.run(self.runtime + ['delete', '--force', self.keeper], capture_output=True, timeout=15) if hasattr(self, 'keeper_process'): self.keeper_process.wait(timeout=10) self.keeper_log.close()- subprocess.run(self.docker + ['network', 'rm', self.network], capture_output=True, timeout=15)+ subprocess.run(self.runtime + ['network', 'delete', self.network], capture_output=True, timeout=15) if success: shutil.rmtree(self.root) else: