SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2024-10-18 11:52:43 +00:00
Code Issues Projects Releases Wiki Activity

feat: add _get_time_range

Browse source
This commit is contained in:
dettlaff 2024-09-20 01:38:50 +04:00 committed by Inex Code
parent cb471f5a8e
commit 77eaa181ca
1 changed files with 21 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
selfprivacy_api/utils/monitoring.py
View file

@ -111,6 +111,20 @@ class MonitoringQueries:
error=f"Prometheus request failed! Error: {str(error)}" error=f"Prometheus request failed! Error: {str(error)}"
) )
@staticmethod
def _get_time_range(
start: Optional[datetime] = None,
end: Optional[datetime] = None,
) -> Tuple[datetime, datetime]:
"""Get the start and end time for queries."""
if start is None:
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
return start, end
@staticmethod @staticmethod
def _prometheus_value_to_monitoring_value(x: Tuple[int, str]): def _prometheus_value_to_monitoring_value(x: Tuple[int, str]):
return MonitoringValue(timestamp=datetime.fromtimestamp(x[0]), value=x[1]) return MonitoringValue(timestamp=datetime.fromtimestamp(x[0]), value=x[1])
@ -193,11 +207,7 @@ class MonitoringQueries:
step (int): Interval in seconds for querying disk usage data. step (int): Interval in seconds for querying disk usage data.
""" """
if start is None: start, end = MonitoringQueries._get_time_range(start, end)
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
start_timestamp = int(start.timestamp()) start_timestamp = int(start.timestamp())
end_timestamp = int(end.timestamp()) end_timestamp = int(end.timestamp())
@ -237,11 +247,7 @@ class MonitoringQueries:
step (int): Interval in seconds for querying memory usage data. step (int): Interval in seconds for querying memory usage data.
""" """
if start is None: start, end = MonitoringQueries._get_time_range(start, end)
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
start_timestamp = int(start.timestamp()) start_timestamp = int(start.timestamp())
end_timestamp = int(end.timestamp()) end_timestamp = int(end.timestamp())
@ -281,11 +287,7 @@ class MonitoringQueries:
step (int): Interval in seconds for querying swap memory usage data. step (int): Interval in seconds for querying swap memory usage data.
""" """
if start is None: start, end = MonitoringQueries._get_time_range(start, end)
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
start_timestamp = int(start.timestamp()) start_timestamp = int(start.timestamp())
end_timestamp = int(end.timestamp()) end_timestamp = int(end.timestamp())
@ -325,11 +327,7 @@ class MonitoringQueries:
Defaults to current time if not provided. Defaults to current time if not provided.
""" """
if start is None: start, end = MonitoringQueries._get_time_range(start, end)
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
offset, duration = MonitoringQueries._calculate_offset_and_duration(start, end) offset, duration = MonitoringQueries._calculate_offset_and_duration(start, end)
@ -364,11 +362,7 @@ class MonitoringQueries:
Defaults to current time if not provided. Defaults to current time if not provided.
""" """
if start is None: start, end = MonitoringQueries._get_time_range(start, end)
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
offset, duration = MonitoringQueries._calculate_offset_and_duration(start, end) offset, duration = MonitoringQueries._calculate_offset_and_duration(start, end)
@ -405,11 +399,7 @@ class MonitoringQueries:
step (int): Interval in seconds for querying disk usage data. step (int): Interval in seconds for querying disk usage data.
""" """
if start is None: start, end = MonitoringQueries._get_time_range(start, end)
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
start_timestamp = int(start.timestamp()) start_timestamp = int(start.timestamp())
end_timestamp = int(end.timestamp()) end_timestamp = int(end.timestamp())
@ -446,11 +436,7 @@ class MonitoringQueries:
step (int): Interval in seconds for querying network data. step (int): Interval in seconds for querying network data.
""" """
if start is None: start, end = MonitoringQueries._get_time_range(start, end)
start = datetime.now() - timedelta(minutes=20)
if end is None:
end = datetime.now()
start_timestamp = int(start.timestamp()) start_timestamp = int(start.timestamp())
end_timestamp = int(end.timestamp()) end_timestamp = int(end.timestamp())