Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: fix RestoreStreamer to prevent buckets skipping #2830 #3119

Merged
merged 2 commits into from
Jun 4, 2024
Merged

fix: fix RestoreStreamer to prevent buckets skipping #2830 #3119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0642a47
fix: fix RestoreStreamer to prevent bucket skipping #2830
BorysTheDev Jun 3, 2024
ef0e5b7
refactor: address comments
BorysTheDev Jun 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/server/journal/streamer.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,8 @@ void RestoreStreamer::Start(io::Sink* dest, bool send_lsn) {

bool written = false;
cursor = pt->Traverse(cursor, [&](PrimeTable::bucket_iterator it) {
db_slice_->FlushChangeToEarlierCallbacks(0 /*db_id always 0 for cluster*/,
DbSlice::Iterator::FromPrime(it), snapshot_version_);
if (WriteBucket(it)) {
written = true;
}
Expand Down
21 changes: 4 additions & 17 deletions tests/dragonfly/cluster_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1177,11 +1177,6 @@ async def test_cluster_fuzzymigration(
seeder = df_seeder_factory.create(keys=keys, port=nodes[0].instance.port, cluster_mode=True)
await seeder.run(target_deviation=0.1)

fill_task = asyncio.create_task(seeder.run())
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do you remove this?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't see any point in having 2 calls for Seeder

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I didn't track the changes to this part of code, but there used to be a reason, likely now removed or just moved... Almost sure 😆

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok now I see that the seeder is not running while we are doing migration, right?
why so?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we didn't do it previously too, during migration, we run counters.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we don't control now when migration is finished so seeder can't be stopped before we migration is finished


# some time fo seeder
await asyncio.sleep(0.5)

# Counter that pushes values to a list
async def list_counter(key, client: aioredis.RedisCluster):
for i in itertools.count(start=1):
Expand All @@ -1197,9 +1192,6 @@ async def list_counter(key, client: aioredis.RedisCluster):
for key, conn in zip(counter_keys, counter_connections)
]

seeder.stop()
await fill_task

# Generate capture, capture ignores counter keys
capture = await seeder.capture()

Expand Down Expand Up @@ -1237,14 +1229,15 @@ async def list_counter(key, client: aioredis.RedisCluster):
keeping = node.slots[num_outgoing:]
node.next_slots.extend(keeping)

logging.debug("start migrations")
await push_config(json.dumps(generate_config(nodes)), [node.admin_client for node in nodes])

iterations = 0
while True:
is_all_finished = True
for node in nodes:
states = await node.admin_client.execute_command("DFLYCLUSTER", "SLOT-MIGRATION-STATUS")
print(states)
logging.debug(states)
is_all_finished = is_all_finished and (
all("FINISHED" in s for s in states) or states == "NO_STATE"
)
Expand All @@ -1257,23 +1250,17 @@ async def list_counter(key, client: aioredis.RedisCluster):

await asyncio.sleep(0.1)

# Stop counters
for counter in counters:
counter.cancel()

# clean migrations
for node in nodes:
node.migrations = []
node.slots = node.next_slots

# TODO this config should be pushed with new slots
# Push new config
logging.debug("remove finished migrations")
await push_config(json.dumps(generate_config(nodes)), [node.admin_client for node in nodes])

# Transfer nodes
for node in nodes:
node.slots = node.next_slots
node.new_slots = []

# Check counter consistency
cluster_client = aioredis.RedisCluster(host="localhost", port=nodes[0].instance.port)
for key in counter_keys:
Expand Down
Loading