dprevot/check-patroni
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Drop superfluous type annotation of 'self'

Browse source 
See https://mypy.readthedocs.io/en/stable/cheat_sheet_py3.html#classes
> For instance methods, omit type for "self"
This commit is contained in:
Denis Laxalde 2023-10-02 13:44:49 +02:00 committed by Denis Laxalde
parent de8b3daa7a
commit 95f21a133d
3 changed files with 43 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
check_patroni/cluster.py
View file

@ -14,7 +14,7 @@ def replace_chars(text: str) -> str:
class ClusterNodeCount(PatroniResource):
def probe(self: "ClusterNodeCount") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("cluster")
role_counters: Counter[str] = Counter()
roles = []
@ -48,7 +48,7 @@ class ClusterNodeCount(PatroniResource):
class ClusterHasLeader(PatroniResource):
def probe(self: "ClusterHasLeader") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("cluster")
is_leader_found = False
@ -69,24 +69,20 @@ class ClusterHasLeader(PatroniResource):
class ClusterHasLeaderSummary(nagiosplugin.Summary):
def ok(self: "ClusterHasLeaderSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return "The cluster has a running leader."
@handle_unknown
def problem(self: "ClusterHasLeaderSummary", results: nagiosplugin.Result) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
return "The cluster has no running leader."
class ClusterHasReplica(PatroniResource):
def __init__(
self: "ClusterHasReplica",
connection_info: ConnectionInfo,
max_lag: Union[int, None],
):
def __init__(self, connection_info: ConnectionInfo, max_lag: Union[int, None]):
super().__init__(connection_info)
self.max_lag = max_lag
def probe(self: "ClusterHasReplica") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("cluster")
replicas = []
@ -140,7 +136,7 @@ class ClusterHasReplica(PatroniResource):
class ClusterConfigHasChanged(PatroniResource):
def __init__(
self: "ClusterConfigHasChanged",
self,
connection_info: ConnectionInfo,
config_hash: str, # Always contains the old hash
state_file: str, # Only used to update the hash in the state_file (when needed)
@ -151,7 +147,7 @@ class ClusterConfigHasChanged(PatroniResource):
self.config_hash = config_hash
self.save = save
def probe(self: "ClusterConfigHasChanged") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("config")
new_hash = hashlib.md5(json.dumps(item_dict).encode()).hexdigest()
@ -183,23 +179,21 @@ class ClusterConfigHasChanged(PatroniResource):
class ClusterConfigHasChangedSummary(nagiosplugin.Summary):
def __init__(self: "ClusterConfigHasChangedSummary", config_hash: str) -> None:
def __init__(self, config_hash: str) -> None:
self.old_config_hash = config_hash
# Note: It would be helpful to display the old / new hash here. Unfortunately, it's not a metric.
# So we only have the old / expected one.
def ok(self: "ClusterConfigHasChangedSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return f"The hash of patroni's dynamic configuration has not changed ({self.old_config_hash})."
@handle_unknown
def problem(
self: "ClusterConfigHasChangedSummary", results: nagiosplugin.Result
) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
return f"The hash of patroni's dynamic configuration has changed. The old hash was {self.old_config_hash}."
class ClusterIsInMaintenance(PatroniResource):
def probe(self: "ClusterIsInMaintenance") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("cluster")
# The actual check
@ -212,7 +206,7 @@ class ClusterIsInMaintenance(PatroniResource):
class ClusterHasScheduledAction(PatroniResource):
def probe(self: "ClusterIsInMaintenance") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("cluster")
scheduled_switchover = 0

74
check_patroni/node.py
View file

@ -7,7 +7,7 @@ from .types import APIError, ConnectionInfo, PatroniResource, handle_unknown
class NodeIsPrimary(PatroniResource):
def probe(self: "NodeIsPrimary") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
try:
self.rest_api("primary")
except APIError:
@ -16,24 +16,22 @@ class NodeIsPrimary(PatroniResource):
class NodeIsPrimarySummary(nagiosplugin.Summary):
def ok(self: "NodeIsPrimarySummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return "This node is the primary with the leader lock."
@handle_unknown
def problem(self: "NodeIsPrimarySummary", results: nagiosplugin.Result) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
return "This node is not the primary with the leader lock."
class NodeIsLeader(PatroniResource):
def __init__(
self: "NodeIsLeader",
connection_info: ConnectionInfo,
check_is_standby_leader: bool,
self, connection_info: ConnectionInfo, check_is_standby_leader: bool
) -> None:
super().__init__(connection_info)
self.check_is_standby_leader = check_is_standby_leader
def probe(self: "NodeIsLeader") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
apiname = "leader"
if self.check_is_standby_leader:
apiname = "standby-leader"
@ -46,26 +44,23 @@ class NodeIsLeader(PatroniResource):
class NodeIsLeaderSummary(nagiosplugin.Summary):
def __init__(
self: "NodeIsLeaderSummary",
check_is_standby_leader: bool,
) -> None:
def __init__(self, check_is_standby_leader: bool) -> None:
if check_is_standby_leader:
self.leader_kind = "standby leader"
else:
self.leader_kind = "leader"
def ok(self: "NodeIsLeaderSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return f"This node is a {self.leader_kind} node."
@handle_unknown
def problem(self: "NodeIsLeaderSummary", results: nagiosplugin.Result) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
return f"This node is not a {self.leader_kind} node."
class NodeIsReplica(PatroniResource):
def __init__(
self: "NodeIsReplica",
self,
connection_info: ConnectionInfo,
max_lag: str,
check_is_sync: bool,
@ -76,7 +71,7 @@ class NodeIsReplica(PatroniResource):
self.check_is_sync = check_is_sync
self.check_is_async = check_is_async
def probe(self: "NodeIsReplica") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
try:
if self.check_is_sync:
api_name = "synchronous"
@ -95,12 +90,7 @@ class NodeIsReplica(PatroniResource):
class NodeIsReplicaSummary(nagiosplugin.Summary):
def __init__(
self: "NodeIsReplicaSummary",
lag: str,
check_is_sync: bool,
check_is_async: bool,
) -> None:
def __init__(self, lag: str, check_is_sync: bool, check_is_async: bool) -> None:
self.lag = lag
if check_is_sync:
self.replica_kind = "synchronous replica"
@ -109,7 +99,7 @@ class NodeIsReplicaSummary(nagiosplugin.Summary):
else:
self.replica_kind = "replica"
def ok(self: "NodeIsReplicaSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
if self.lag is None:
return (
f"This node is a running {self.replica_kind} with no noloadbalance tag."
@ -117,14 +107,14 @@ class NodeIsReplicaSummary(nagiosplugin.Summary):
return f"This node is a running {self.replica_kind} with no noloadbalance tag and the lag is under {self.lag}."
@handle_unknown
def problem(self: "NodeIsReplicaSummary", results: nagiosplugin.Result) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
if self.lag is None:
return f"This node is not a running {self.replica_kind} with no noloadbalance tag."
return f"This node is not a running {self.replica_kind} with no noloadbalance tag and a lag under {self.lag}."
class NodeIsPendingRestart(PatroniResource):
def probe(self: "NodeIsPendingRestart") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("patroni")
is_pending_restart = item_dict.get("pending_restart", False)
@ -137,19 +127,17 @@ class NodeIsPendingRestart(PatroniResource):
class NodeIsPendingRestartSummary(nagiosplugin.Summary):
def ok(self: "NodeIsPendingRestartSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return "This node doesn't have the pending restart flag."
@handle_unknown
def problem(
self: "NodeIsPendingRestartSummary", results: nagiosplugin.Result
) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
return "This node has the pending restart flag."
class NodeTLHasChanged(PatroniResource):
def __init__(
self: "NodeTLHasChanged",
self,
connection_info: ConnectionInfo,
timeline: str, # Always contains the old timeline
state_file: str, # Only used to update the timeline in the state_file (when needed)
@ -160,7 +148,7 @@ class NodeTLHasChanged(PatroniResource):
self.timeline = timeline
self.save = save
def probe(self: "NodeTLHasChanged") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("patroni")
new_tl = item_dict["timeline"]
@ -193,27 +181,23 @@ class NodeTLHasChanged(PatroniResource):
class NodeTLHasChangedSummary(nagiosplugin.Summary):
def __init__(self: "NodeTLHasChangedSummary", timeline: str) -> None:
def __init__(self, timeline: str) -> None:
self.timeline = timeline
def ok(self: "NodeTLHasChangedSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return f"The timeline is still {self.timeline}."
@handle_unknown
def problem(self: "NodeTLHasChangedSummary", results: nagiosplugin.Result) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
return f"The expected timeline was {self.timeline} got {results['timeline'].metric}."
class NodePatroniVersion(PatroniResource):
def __init__(
self: "NodePatroniVersion",
connection_info: ConnectionInfo,
patroni_version: str,
) -> None:
def __init__(self, connection_info: ConnectionInfo, patroni_version: str) -> None:
super().__init__(connection_info)
self.patroni_version = patroni_version
def probe(self: "NodePatroniVersion") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
item_dict = self.rest_api("patroni")
version = item_dict["patroni"]["version"]
@ -232,21 +216,21 @@ class NodePatroniVersion(PatroniResource):
class NodePatroniVersionSummary(nagiosplugin.Summary):
def __init__(self: "NodePatroniVersionSummary", patroni_version: str) -> None:
def __init__(self, patroni_version: str) -> None:
self.patroni_version = patroni_version
def ok(self: "NodePatroniVersionSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return f"Patroni's version is {self.patroni_version}."
@handle_unknown
def problem(self: "NodePatroniVersionSummary", results: nagiosplugin.Result) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
# FIXME find a way to make the following work, check is perf data can be strings
# return f"The expected patroni version was {self.patroni_version} got {results['patroni_version'].metric}."
return f"Patroni's version is not {self.patroni_version}."
class NodeIsAlive(PatroniResource):
def probe(self: "NodeIsAlive") -> Iterable[nagiosplugin.Metric]:
def probe(self) -> Iterable[nagiosplugin.Metric]:
try:
self.rest_api("liveness")
except APIError:
@ -255,9 +239,9 @@ class NodeIsAlive(PatroniResource):
class NodeIsAliveSummary(nagiosplugin.Summary):
def ok(self: "NodeIsAliveSummary", results: nagiosplugin.Result) -> str:
def ok(self, results: nagiosplugin.Result) -> str:
return "This node is alive (patroni is running)."
@handle_unknown
def problem(self: "NodeIsAliveSummary", results: nagiosplugin.Result) -> str:
def problem(self, results: nagiosplugin.Result) -> str:
return "This node is not alive (patroni is not running)."

2
check_patroni/types.py
View file

@ -32,7 +32,7 @@ class Parameters:
class PatroniResource(nagiosplugin.Resource):
conn_info: ConnectionInfo
def rest_api(self: "PatroniResource", service: str) -> Any:
def rest_api(self, service: str) -> Any:
"""Try to connect to all the provided endpoints for the requested service"""
for endpoint in self.conn_info.endpoints:
cert: Optional[Union[Tuple[str, str], str]] = None