|
7 | 7 | # the Business Source License, use of this software will be governed |
8 | 8 | # by the Apache License, Version 2.0 |
9 | 9 |
|
| 10 | +import time |
10 | 11 | from typing import Callable |
11 | 12 |
|
12 | 13 | from connectrpc.errors import ConnectError, ConnectErrorCode |
13 | 14 |
|
14 | 15 | import google.protobuf.duration_pb2 as duration_pb2 |
15 | 16 | import google.protobuf.field_mask_pb2 as field_mask_pb2 |
16 | 17 |
|
| 18 | +from ducktape.mark import parametrize |
17 | 19 | from ducktape.tests.test import TestContext |
18 | 20 | from ducktape.utils.util import wait_until |
19 | 21 |
|
@@ -548,3 +550,153 @@ def test_permission_grant_activates_sync(self): |
548 | 550 | backoff_sec=1, |
549 | 551 | err_msg="role did not mirror after the permission grant", |
550 | 552 | ) |
| 553 | + |
| 554 | + |
| 555 | +class ShadowLinkRoleSyncScaleTest(RoleSyncTestBase): |
| 556 | + """Scale characterization for the roles migrator. Each parametrized point |
| 557 | + runs the full create -> update -> delete lifecycle (exercising all three |
| 558 | + migrator apply loops at volume) while pushing one cost axis: role count, |
| 559 | + total membership, or a single role's member set. |
| 560 | + """ |
| 561 | + |
| 562 | + PREFIX = "synced-" |
| 563 | + |
| 564 | + def __init__(self, test_context: TestContext): |
| 565 | + super().__init__( |
| 566 | + test_context=test_context, |
| 567 | + num_brokers=3, |
| 568 | + secondary_cluster_args=SecondaryClusterArgs(), |
| 569 | + ) |
| 570 | + |
| 571 | + def setUp(self): |
| 572 | + super().setUp() |
| 573 | + self._src = AdminV2RoleWrapper(AdminV2(self.source_cluster_service)) |
| 574 | + |
| 575 | + def _retry_transient( |
| 576 | + self, fn: Callable[..., object], *args: object, **kwargs: object |
| 577 | + ) -> None: |
| 578 | + """Run a source mutation, retrying the transient controller-backpressure |
| 579 | + errors -- raft not_leader/timeout, surfaced as a ConnectError 'internal' |
| 580 | + -- that thousands of rapid role mutations can provoke. Other error codes |
| 581 | + and the final failure propagate.""" |
| 582 | + last: ConnectError | None = None |
| 583 | + for attempt in range(6): |
| 584 | + try: |
| 585 | + fn(*args, **kwargs) |
| 586 | + return |
| 587 | + except ConnectError as e: |
| 588 | + if e.code != ConnectErrorCode.INTERNAL: |
| 589 | + raise |
| 590 | + last = e |
| 591 | + self.logger.debug(f"retrying transient source mutation: {e}") |
| 592 | + time.sleep(0.5 * (attempt + 1)) |
| 593 | + assert last is not None |
| 594 | + raise last |
| 595 | + |
| 596 | + def _seed_source_roles(self, num_roles: int, members_per_role: int) -> set[str]: |
| 597 | + """Create num_roles in-scope roles on the source, each with |
| 598 | + members_per_role distinct user members; return the role names. Members |
| 599 | + are free-form principal names that need no backing user accounts -- the |
| 600 | + migrator mirrors them as data.""" |
| 601 | + names: set[str] = set() |
| 602 | + for i in range(num_roles): |
| 603 | + name = f"{self.PREFIX}role-{i}" |
| 604 | + members = [_user(f"u-{i}-{m}") for m in range(members_per_role)] |
| 605 | + self._retry_transient(self._src.create_role, role=name, members=members) |
| 606 | + names.add(name) |
| 607 | + return names |
| 608 | + |
| 609 | + def _synced_role_names(self) -> set[str]: |
| 610 | + return {n for n in self._dst.list_role_names() if n.startswith(self.PREFIX)} |
| 611 | + |
| 612 | + @cluster(num_nodes=6) |
| 613 | + @parametrize(num_roles=5000, members_per_role=1) |
| 614 | + @parametrize(num_roles=50, members_per_role=100) |
| 615 | + @parametrize(num_roles=5, members_per_role=1000) |
| 616 | + def test_role_sync_at_scale(self, num_roles: int, members_per_role: int) -> None: |
| 617 | + # create + update + delete each write ~1 controller record per role, on |
| 618 | + # both the source (seed/modify/delete) and the target (mirror), so each |
| 619 | + # controller log grows by ~3*num_roles over the lifecycle. Raise the |
| 620 | + # guard above its 1000 default to admit that while still catching runaway |
| 621 | + # writes. |
| 622 | + max_controller_records = 1000 + num_roles * 3 |
| 623 | + self.redpanda.set_expected_controller_records(max_controller_records) |
| 624 | + self.source_cluster_service.set_expected_controller_records( |
| 625 | + max_controller_records |
| 626 | + ) |
| 627 | + |
| 628 | + first = f"{self.PREFIX}role-0" |
| 629 | + last = f"{self.PREFIX}role-{num_roles - 1}" |
| 630 | + mirror_timeout_sec = 240 |
| 631 | + |
| 632 | + # CREATE: seed the source, then mirror it to the destination. |
| 633 | + seed_start = time.time() |
| 634 | + expected = self._seed_source_roles(num_roles, members_per_role) |
| 635 | + self.logger.info( |
| 636 | + f"seeded {num_roles} source roles x {members_per_role} members " |
| 637 | + f"in {time.time() - seed_start:.1f}s" |
| 638 | + ) |
| 639 | + create_start = time.time() |
| 640 | + self._create_link_with_role_sync() |
| 641 | + wait_until( |
| 642 | + lambda: self._synced_role_names() >= expected, |
| 643 | + timeout_sec=mirror_timeout_sec, |
| 644 | + backoff_sec=2, |
| 645 | + err_msg=f"{num_roles} roles did not fully mirror within " |
| 646 | + f"{mirror_timeout_sec}s", |
| 647 | + ) |
| 648 | + self.logger.info( |
| 649 | + f"role sync created {num_roles} roles in {time.time() - create_start:.1f}s" |
| 650 | + ) |
| 651 | + # Membership integrity at the range ends (a full per-role check would add |
| 652 | + # num_roles RPCs); the first and last seeded roles bound the range. |
| 653 | + for i in sorted({0, num_roles - 1}): |
| 654 | + name = f"{self.PREFIX}role-{i}" |
| 655 | + want = {f"u-{i}-{m}" for m in range(members_per_role)} |
| 656 | + assert self._dst_role_members(name) == want, ( |
| 657 | + f"membership mismatch for {name} after create" |
| 658 | + ) |
| 659 | + |
| 660 | + # UPDATE: change every role's membership -- drop one member from each -- |
| 661 | + # so all num_roles roles flow through the update apply loop. |
| 662 | + update_start = time.time() |
| 663 | + for i in range(num_roles): |
| 664 | + self._retry_transient( |
| 665 | + self._src.remove_role_members, |
| 666 | + role=f"{self.PREFIX}role-{i}", |
| 667 | + members=[_user(f"u-{i}-0")], |
| 668 | + ) |
| 669 | + wait_until( |
| 670 | + lambda: ( |
| 671 | + "u-0-0" not in self._dst_role_members(first) |
| 672 | + and f"u-{num_roles - 1}-0" not in self._dst_role_members(last) |
| 673 | + ), |
| 674 | + timeout_sec=mirror_timeout_sec, |
| 675 | + backoff_sec=2, |
| 676 | + err_msg=f"membership update did not mirror for {num_roles} roles", |
| 677 | + ) |
| 678 | + self.logger.info( |
| 679 | + f"role sync updated {num_roles} roles in {time.time() - update_start:.1f}s" |
| 680 | + ) |
| 681 | + |
| 682 | + # DELETE: drop every source role and confirm the destination is pruned. |
| 683 | + delete_start = time.time() |
| 684 | + for name in expected: |
| 685 | + self._retry_transient(self._src.delete_role, name, delete_acls=False) |
| 686 | + wait_until( |
| 687 | + lambda: len(self._synced_role_names()) == 0, |
| 688 | + timeout_sec=mirror_timeout_sec, |
| 689 | + backoff_sec=2, |
| 690 | + err_msg=f"{num_roles} roles were not pruned from the destination", |
| 691 | + ) |
| 692 | + self.logger.info( |
| 693 | + f"role sync deleted {num_roles} roles in {time.time() - delete_start:.1f}s" |
| 694 | + ) |
| 695 | + |
| 696 | + # The full lifecycle ran fault-free -> the task settles ACTIVE. |
| 697 | + wait_until( |
| 698 | + lambda: self._roles_task_state() == shadow_link_pb2.TASK_STATE_ACTIVE, |
| 699 | + timeout_sec=30, |
| 700 | + backoff_sec=1, |
| 701 | + err_msg="roles task did not settle ACTIVE after the lifecycle", |
| 702 | + ) |
0 commit comments