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

fix(scan_operations): add retry policy to cql query #9600

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(scan_operations): add retry policy to cql query #9600

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
29 changes: 21 additions & 8 deletions sdcm/scan_operation_thread.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,13 @@
from abc import abstractmethod
from string import Template
from typing import Optional, Type, NamedTuple, TYPE_CHECKING
from contextlib import contextmanager

from pytz import utc
from cassandra import ConsistencyLevel, OperationTimedOut, ReadTimeout
from cassandra.cluster import ResponseFuture, ResultSet # pylint: disable=no-name-in-module
from cassandra.query import SimpleStatement # pylint: disable=no-name-in-module
from cassandra.policies import ExponentialBackoffRetryPolicy

from sdcm.remote import LocalCmdRunner
from sdcm.sct_events import Severity
Expand Down Expand Up @@ -106,6 +108,10 @@ def __init__(self, generator: random.Random, thread_params: ThreadParams, thread
self.db_node = self._get_random_node()
self.current_operation_stat = None
self.log.info("FullscanOperationBase init finished")
self._exp_backoff_retry_policy_params = {
"max_num_retries": 15.0, "min_interval": 1.0, "max_interval": 1800.0
}
self._request_default_timeout = 1800

def _get_random_node(self) -> BaseNode:
return self.generator.choice(self.fullscan_params.db_cluster.data_nodes)
Expand Down Expand Up @@ -141,11 +147,7 @@ def run_scan_event(self, cmd: str,
cmd=cmd
)

with self.fullscan_params.db_cluster.cql_connection_patient(
node=self.db_node,
connect_timeout=300,
user=self.fullscan_params.user,
password=self.fullscan_params.user_password) as session:
with self.cql_connection(connect_timeout=300) as session:
try:
scan_op_event.message = ''
start_time = time.time()
Expand Down Expand Up @@ -191,6 +193,18 @@ def fetch_result_pages(self, result, read_pages):
if read_pages > 0:
pages += 1

@contextmanager
def cql_connection(self, **kwargs):
node = kwargs.get("node", self.db_node)
Copy link
Contributor

Choose a reason for hiding this comment

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

you need to pop out the node from kwargs, if you want to specify it later
or use setdefault

the unittest demonstrate it's currently broken

with self.fullscan_params.db_cluster.cql_connection_patient(
node=node,
user=self.fullscan_params.user,
password=self.fullscan_params.user_password, **kwargs) as session:
session.cluster.default_retry_policy = ExponentialBackoffRetryPolicy(
**self._exp_backoff_retry_policy_params)
session.default_timeout = self._request_default_timeout
yield session


class FullScanOperation(FullscanOperationBase):
def __init__(self, generator, **kwargs):
Expand Down Expand Up @@ -239,7 +253,7 @@ def __init__(self, generator, **kwargs):
def get_table_clustering_order(self) -> str:
node = self._get_random_node()
try:
with self.fullscan_params.db_cluster.cql_connection_patient(node=node, connect_timeout=300) as session:
with self.cql_connection(node=node, connect_timeout=300) as session:
# Using CL ONE. No need for a quorum since querying a constant fixed attribute of a table.
session.default_consistency_level = ConsistencyLevel.ONE
return get_table_clustering_order(ks_cf=self.fullscan_params.ks_cf,
Expand All @@ -264,8 +278,7 @@ def randomly_form_cql_statement(self) -> Optional[tuple[str, str]]: # pylint: d
"""
db_node = self._get_random_node()

with self.fullscan_params.db_cluster.cql_connection_patient(
node=db_node, connect_timeout=300) as session:
with self.cql_connection(node=db_node, connect_timeout=300) as session:
ck_random_min_value = self.generator.randint(a=1, b=self.fullscan_params.rows_count)
ck_random_max_value = self.generator.randint(a=ck_random_min_value, b=self.fullscan_params.rows_count)
self.ck_filter = ck_filter = self.generator.choice(list(self.reversed_query_filter_ck_by.keys()))
Expand Down
Loading