mirror of
https://github.com/qdrant/qdrant.git
synced 2026-07-23 11:11:00 -05:00
242c7dfa0a
* ci: parallelize consensus tests with pytest-xdist
Enable pytest-xdist for consensus_tests to run tests across multiple
workers in parallel, significantly reducing CI wall time (~20min → ~5-7min).
Changes:
- Add `-n auto --dist=loadfile` to the consensus test pytest invocation
- Remove hardcoded port_seed from tests that don't need fixed ports for
restart/rejoin (test_order_by, test_consensus_compaction,
test_named_vector_crud, test_listener_node)
- Give test_cluster_rejoin its own PORT_SEED=15000 to avoid port
conflicts with auth tests (PORT_SEED=10000)
- Derive restart ports from killed PeerProcess objects instead of
hardcoded arithmetic where possible
- Add xdist_group("auth") marker to auth test files to ensure they
run on the same worker (they share PORT_SEED=10000)
Made-with: Cursor
* fix: remove remaining hardcoded port_seed=20000 causing parallel test conflicts
8 test files were using port_seed=20000 as a positional argument to
start_cluster(), which was missed in the initial change. When running
in parallel with pytest-xdist, multiple workers would try to bind to
the same port range (20000-20x02), causing port conflicts and cascading
test failures.
Also remove port_seed=23000 from test_snapshot_recovery_kill.py since
it doesn't need fixed ports for restart.
Made-with: Cursor
* fix: use saved port for restart in test_two_follower_nodes_down
The test was restarting killed peers on hardcoded ports (20200/20100)
that previously matched port_seed=20000. After switching to random
ports, the restart ports no longer match the original peer ports,
causing raft state URI mismatches and peer startup failures.
Save the p2p_port from the killed PeerProcess and reuse it for restart.
Made-with: Cursor
* Reuse p2p ports when restarting killed peers in consensus tests
When a peer is killed and restarted with random ports, it gets a new
consensus URI. The cluster needs a Raft operation to update this URI,
which under CPU contention from parallel test workers can exceed the
30-second timeout. Fix by capturing each peer's p2p_port before killing
and reusing it on restart, so the URI stays the same and no consensus
update is needed.
Made-with: Cursor
* A few improvements for parallel runs (#8731)
* fix: make auth tests' PORT_SEED per-worker to avoid port collisions
* ci: improve failure visibility for parallel consensus tests
* Three small changes to make hangs, interleaved output, and coverage runs behave predictably under pytest-xdist
* fix: two test bugs surfaced by parallel runs and revert drop PR_SET_PDEATHSIG helper
* fix: wait for count convergence in test_triple_replication
* fix: clean leaked peer processes at test start
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Cursor Agent <agent@cursor.com>
Co-authored-by: tellet-q <166374656+tellet-q@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
import multiprocessing
|
|
import pathlib
|
|
|
|
from .fixtures import upsert_random_points, create_collection
|
|
from .utils import *
|
|
|
|
N_PEERS = 3
|
|
N_SHARDS = 1
|
|
N_REPLICA = 3
|
|
|
|
|
|
def update_points_in_loop(peer_url, collection_name):
|
|
offset = 0
|
|
limit = 3
|
|
while True:
|
|
upsert_random_points(peer_url, limit, collection_name, offset=offset)
|
|
offset += limit
|
|
# time.sleep(0.01)
|
|
|
|
|
|
def run_update_points_in_background(peer_url, collection_name):
|
|
p = multiprocessing.Process(target=update_points_in_loop, args=(peer_url, collection_name))
|
|
p.start()
|
|
return p
|
|
|
|
|
|
# Test that a 3 nodes cluster can recover from two follower nodes down.
|
|
def test_two_follower_nodes_down(tmp_path: pathlib.Path):
|
|
assert_project_root()
|
|
|
|
peer_api_uris, peer_dirs, bootstrap_uri = start_cluster(tmp_path, N_PEERS)
|
|
|
|
create_collection(peer_api_uris[0], shard_number=N_SHARDS, replication_factor=N_REPLICA)
|
|
wait_collection_exists_and_active_on_all_peers(
|
|
collection_name="test_collection",
|
|
peer_api_uris=peer_api_uris
|
|
)
|
|
|
|
# upload points to the leader
|
|
upload_process = run_update_points_in_background(peer_api_uris[0], "test_collection")
|
|
|
|
time.sleep(0.3)
|
|
|
|
# Kill third peer
|
|
p = processes.pop()
|
|
restart_port_3 = p.p2p_port
|
|
p.kill()
|
|
peer_api_uris.pop()
|
|
|
|
# Kill second peer
|
|
p = processes.pop()
|
|
restart_port_2 = p.p2p_port
|
|
p.kill()
|
|
peer_api_uris.pop()
|
|
|
|
time.sleep(0.3)
|
|
|
|
# Stop pushing points to the leader
|
|
upload_process.kill()
|
|
|
|
# Restart third peer on same ports
|
|
new_url_3 = start_peer(peer_dirs[2], f"peer_{3}_restarted.log", bootstrap_uri, restart_port_3)
|
|
peer_api_uris.append(new_url_3)
|
|
|
|
# Restart second peer on same ports
|
|
new_url_2 = start_peer(peer_dirs[1], f"peer_{2}_restarted.log", bootstrap_uri, restart_port_2)
|
|
peer_api_uris.append(new_url_2)
|
|
|
|
# Wait for peers to be online
|
|
wait_for_peer_online(new_url_3)
|
|
time.sleep(1)
|
|
wait_for_peer_online(new_url_2)
|
|
|
|
timeout = 10
|
|
while True:
|
|
if timeout < 0:
|
|
raise Exception("Timeout waiting for all replicas to be active")
|
|
|
|
all_active = True
|
|
points_counts = set()
|
|
for peer_api_uri in peer_api_uris:
|
|
res = check_collection_cluster(peer_api_uri, "test_collection")
|
|
points_counts.add(res['points_count'])
|
|
if res['state'] != 'Active':
|
|
all_active = False
|
|
|
|
if all_active:
|
|
if len(points_counts) != 1:
|
|
with open("test_triple_replication.log", "w") as f:
|
|
for peer_api_uri in peer_api_uris:
|
|
collection_name = "test_collection"
|
|
res = requests.get(f"{peer_api_uri}/collections/{collection_name}/cluster", timeout=10)
|
|
f.write(f"{peer_api_uri} {res.json()['result']}\n")
|
|
for peer_api_uri in peer_api_uris:
|
|
res = requests.get(f"{peer_api_uri}/cluster", timeout=10)
|
|
f.write(f"{peer_api_uri} {res.json()['result']}\n")
|
|
|
|
for peer_api_uri in peer_api_uris:
|
|
res = requests.post(f"{peer_api_uri}/collections/test_collection/points/count", json={"exact": True})
|
|
print(res.json())
|
|
|
|
assert False, f"Points count is not equal on all peers: {points_counts}"
|
|
break
|
|
|
|
time.sleep(1)
|
|
timeout -= 1
|
|
|