Skip to content

Commit

Permalink
quota tool has been fixed (ydb-platform#13046)
Browse files Browse the repository at this point in the history
  • Loading branch information
dorooleg authored Dec 26, 2024
1 parent b9e3013 commit aa12ad5
Showing 1 changed file with 17 additions and 10 deletions.
27 changes: 17 additions & 10 deletions ydb/tests/tools/fq_runner/kikimr_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,13 @@
logging.getLogger("library.python.retry").setLevel("ERROR")


def plain_or_under_sanitizer_wrapper(plain, sanitized):
try:
return plain_or_under_sanitizer(plain, sanitized)
except Exception:
return plain


class BaseTenant(abc.ABC):
def __init__(
self,
Expand Down Expand Up @@ -215,14 +222,14 @@ def get_worker_count(self, node_index):
{"subsystem": "worker_manager", "sensor": "ActiveWorkers"})
return result if result is not None else 0

def wait_worker_count(self, node_index, activity, expected_count, timeout=plain_or_under_sanitizer(30, 150)):
def wait_worker_count(self, node_index, activity, expected_count, timeout=plain_or_under_sanitizer_wrapper(30, 150)):
deadline = time.time() + timeout
while True:
count = self.get_actor_count(node_index, activity)
if count >= expected_count:
break
assert time.time() < deadline, "Wait actor count failed"
time.sleep(plain_or_under_sanitizer(0.5, 2))
time.sleep(plain_or_under_sanitizer_wrapper(0.5, 2))
pass

def get_mkql_limit(self, node_index):
Expand Down Expand Up @@ -267,7 +274,7 @@ def ensure_is_alive(self):
self.wait_bootstrap(n)
assert self.get_actor_count(n, "GRPC_PROXY") > 0, "Node {} died".format(n)

def wait_bootstrap(self, node_index=None, wait_time=plain_or_under_sanitizer(90, 400)):
def wait_bootstrap(self, node_index=None, wait_time=plain_or_under_sanitizer_wrapper(90, 400)):
if node_index is None:
for n in self.kikimr_cluster.nodes:
self.wait_bootstrap(n, wait_time)
Expand All @@ -280,13 +287,13 @@ def wait_bootstrap(self, node_index=None, wait_time=plain_or_under_sanitizer(90,
if self.get_actor_count(node_index, "GRPC_PROXY") == 0:
continue
except Exception:
time.sleep(plain_or_under_sanitizer(0.3, 2))
time.sleep(plain_or_under_sanitizer_wrapper(0.3, 2))
continue
break
self.bootstraped_nodes.add(node_index)
logging.debug("Node {} has been bootstrapped".format(node_index))

def wait_discovery(self, node_index=None, wait_time=plain_or_under_sanitizer(30, 150)):
def wait_discovery(self, node_index=None, wait_time=plain_or_under_sanitizer_wrapper(30, 150)):
if node_index is None:
for n in self.kikimr_cluster.nodes:
self.wait_discovery(n, wait_time)
Expand All @@ -301,12 +308,12 @@ def wait_discovery(self, node_index=None, wait_time=plain_or_under_sanitizer(30,
if peer_count is None or peer_count < self.node_count:
continue
except Exception:
time.sleep(plain_or_under_sanitizer(0.3, 2))
time.sleep(plain_or_under_sanitizer_wrapper(0.3, 2))
continue
break
logging.debug("Node {} discovery finished".format(node_index))

def wait_workers(self, worker_count, wait_time=plain_or_under_sanitizer(30, 150)):
def wait_workers(self, worker_count, wait_time=plain_or_under_sanitizer_wrapper(30, 150)):
ca_count = worker_count * 2 # we count 2x CAs
deadline = time.time() + wait_time
while True:
Expand Down Expand Up @@ -357,17 +364,17 @@ def get_completed_checkpoints(self, query_id, expect_counters_exist=False):
expect_counters_exist=expect_counters_exist)

def wait_completed_checkpoints(self, query_id, checkpoints_count,
timeout=plain_or_under_sanitizer(30, 150),
timeout=plain_or_under_sanitizer_wrapper(30, 150),
expect_counters_exist=False):
deadline = time.time() + timeout
while True:
completed = self.get_completed_checkpoints(query_id, expect_counters_exist=expect_counters_exist)
if completed >= checkpoints_count:
break
assert time.time() < deadline, "Wait zero checkpoint failed, actual completed: " + str(completed)
time.sleep(plain_or_under_sanitizer(0.5, 2))
time.sleep(plain_or_under_sanitizer_wrapper(0.5, 2))

def wait_zero_checkpoint(self, query_id, timeout=plain_or_under_sanitizer(30, 150),
def wait_zero_checkpoint(self, query_id, timeout=plain_or_under_sanitizer_wrapper(30, 150),
expect_counters_exist=False):
self.wait_completed_checkpoints(query_id, 1, timeout, expect_counters_exist)

Expand Down

0 comments on commit aa12ad5

Please sign in to comment.