Skip to content
This repository was archived by the owner on Mar 23, 2026. It is now read-only.

Improve annotation of the CloudWatch store #13511

Merged
giograno merged 4 commits into from
Dec 16, 2025
Merged

Improve annotation of the CloudWatch store #13511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c116cfd
type cloudwatch store
giograno Dec 11, 2025
893e783
datetime fix
giograno Dec 11, 2025
fecfb60
fix typo
giograno Dec 11, 2025
5f146ff
simplify history access
giograno Dec 11, 2025
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
12 changes: 10 additions & 2 deletions localstack-core/localstack/services/cloudwatch/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,12 @@
import datetime

from localstack.aws.api.cloudwatch import CompositeAlarm, DashboardBody, MetricAlarm, StateValue
from localstack.aws.api.cloudwatch import (
AlarmHistoryItem,
CompositeAlarm,
DashboardBody,
MetricAlarm,
StateValue,
)
from localstack.services.stores import (
AccountRegionBundle,
BaseStore,
Expand Down Expand Up @@ -72,6 +78,8 @@ class LocalStackDashboard:
dashboard_name: str
dashboard_arn: str
dashboard_body: DashboardBody
last_modified: datetime.datetime
size: int

def __init__(
self, account_id: str, region: str, dashboard_name: str, dashboard_body: DashboardBody
Expand Down Expand Up @@ -99,7 +107,7 @@ class CloudWatchStore(BaseStore):

# Contains all the Alarm Histories. Per documentation, an alarm history is retained even if the alarm is deleted,
# making it necessary to save this at store level
histories: list[dict] = LocalAttribute(default=list)
histories: list[dict[str, AlarmHistoryItem]] = LocalAttribute(default=list)

dashboards: dict[str, LocalStackDashboard] = LocalAttribute(default=dict)

Expand Down
33 changes: 14 additions & 19 deletions localstack-core/localstack/services/cloudwatch/provider_v2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from localstack.aws.api.cloudwatch import (
AccountId,
ActionPrefix,
AlarmHistoryItem,
AlarmName,
AlarmNamePrefix,
AlarmNames,
Expand Down Expand Up @@ -760,7 +761,7 @@ def _update_state(
self,
context: RequestContext,
alarm: LocalStackAlarm,
state_value: str,
state_value: StateValue,
state_reason: str,
state_reason_data: dict = None,
):
Expand All @@ -780,18 +781,17 @@ def _update_state(
"stateReasonData": state_reason_data,
},
}
store.histories.append(
{
"Timestamp": timestamp_millis(alarm.alarm["StateUpdatedTimestamp"]),
"HistoryItemType": HistoryItemType.StateUpdate,
"AlarmName": alarm.alarm["AlarmName"],
"HistoryData": json.dumps(history_data),
"HistorySummary": f"Alarm updated from {old_state} to {state_value}",
"AlarmType": "MetricAlarm"
if isinstance(alarm, LocalStackMetricAlarm)
else "CompositeAlarm",
}
alarm_history_item = AlarmHistoryItem(
Timestamp=alarm.alarm["StateUpdatedTimestamp"],
HistoryItemType=HistoryItemType.StateUpdate,
AlarmName=alarm.alarm["AlarmName"],
HistoryData=json.dumps(history_data),
HistorySummary=f"Alarm updated from {old_state} to {state_value}",
AlarmType="MetricAlarm"
if isinstance(alarm, LocalStackMetricAlarm)
else "CompositeAlarm",
)
store.histories.append(alarm_history_item)
alarm.alarm["StateValue"] = state_value
alarm.alarm["StateReason"] = state_reason
if state_reason_data:
Expand Down Expand Up @@ -837,15 +837,10 @@ def describe_alarm_history(
if alarm_name:
history = [h for h in history if h["AlarmName"] == alarm_name]

def _get_timestamp(input: dict):
if timestamp_string := input.get("Timestamp"):
return datetime.datetime.fromisoformat(timestamp_string)
return None

if start_date:
history = [h for h in history if (date := _get_timestamp(h)) and date >= start_date]
history = [h for h in history if (date := h.get("Timestamp")) and date >= start_date]
if end_date:
history = [h for h in history if (date := _get_timestamp(h)) and date <= end_date]
history = [h for h in history if (date := h.get("Timestamp")) and date <= end_date]
return DescribeAlarmHistoryOutput(AlarmHistoryItems=history)

def _evaluate_composite_alarms(self, context: RequestContext, triggering_alarm):
Expand Down
Loading