feat(resource_monitor): manage triggers, several notify users, and non-ACCOUNTADMIN owners

snowcap/blueprint.py

@@ -2592,10 +2592,14 @@ def execution_strategy_for_change(
         return default_role, False
 
     elif change.urn.resource_type == ResourceType.RESOURCE_MONITOR:
-        # For some reason Snowflake chose to not have a priv type for resource monitors.
-        # Only ACCOUNTADMIN can create them.
+        # Snowflake has no privilege for creating resource monitors: only ACCOUNTADMIN can, and
+        # it can hand one to another role. Once it has, only that owning role can alter or drop
+        # it -- ACCOUNTADMIN is refused -- so those changes run as the owner.
+        if isinstance(change, (UpdateResource, DropResource)) and change_owner:
+            return change_owner, False
         if "ACCOUNTADMIN" in available_roles:
-            return ResourceName("ACCOUNTADMIN"), False
+            transfer_ownership = isinstance(change, CreateResource) and change_owner != "ACCOUNTADMIN"
+            return ResourceName("ACCOUNTADMIN"), transfer_ownership
         raise MissingPrivilegeException(
             "ACCOUNTADMIN role is required to manage resource monitors.\n"
             "  Grant ACCOUNTADMIN to your user or use a different connection."

snowcap/data_provider.py

@@ -3667,14 +3667,27 @@ def fetch_resource_monitor(session: SnowflakeConnection, fqn: FQN):
     if len(resource_monitors) > 1:
         raise Exception(f"Found multiple resource monitors matching {fqn}")
     data = resource_monitors[0]
+    # SHOW reports each action's thresholds as a comma-separated list of percentages
+    # ("50%,75%"), and the notify users as one comma-separated string.
+    triggers = [
+        {"threshold": int(percent.rstrip("%")), "action": action}
+        for column, action in (
+            ("notify_at", "NOTIFY"),
+            ("suspend_at", "SUSPEND"),
+            ("suspend_immediately_at", "SUSPEND_IMMEDIATE"),
+        )
+        if data[column]
+        for percent in data[column].split(",")
+    ]
     return {
         "name": _quote_snowflake_identifier(data["name"]),
         "owner": _get_owner_identifier(data),
         "credit_quota": int(float(data["credit_quota"])) if data["credit_quota"] else None,
         "frequency": data["frequency"],
         "start_timestamp": _convert_to_gmt(data["start_time"], "%Y-%m-%d %H:%M"),
         "end_timestamp": _convert_to_gmt(data["end_time"], "%Y-%m-%d %H:%M"),
-        "notify_users": data["notify_users"] or None,
+        "notify_users": [user.strip() for user in data["notify_users"].split(",")] if data["notify_users"] else None,
+        "triggers": triggers or None,
     }
 
 

snowcap/lifecycle.py

@@ -444,6 +444,22 @@ def update_masking_policy(urn: URN, data: dict, props: Props) -> Union[str, list
         return update__default(urn, {attr: new_value}, props)
 
 
+def update_resource_monitor(urn: URN, data: dict, props: Props) -> Union[str, list[str]]:
+    # TRIGGERS is a clause of its own: it follows SET rather than appearing inside it, and it
+    # replaces every trigger on the monitor, so the delta always carries the full set.
+    if "triggers" not in data:
+        return update__default(urn, data, props)
+    set_data = {attr: value for attr, value in data.items() if attr != "triggers"}
+    return tidy_sql(
+        "ALTER",
+        urn.resource_type,
+        urn.fqn,
+        "SET" if set_data else "",
+        props.render(set_data),
+        props["triggers"].render(data["triggers"]),
+    )
+
+
 def update_mcp_server(urn: URN, data: dict, props: Props) -> str:
     # Snowflake has no ALTER MCP SERVER command. A rename is impossible regardless
     # of what else changed, so it takes precedence over a specification change.

snowcap/props.py

@@ -486,6 +486,37 @@ def render(self, value):
         return value
 
 
+class TriggersProp(Prop):
+    """
+    TRIGGERS ON <threshold> PERCENT DO { SUSPEND | SUSPEND_IMMEDIATE | NOTIFY } [ ... ]
+
+    Parses to and renders from a list of {"threshold": int, "action": <action>} dicts.
+    """
+
+    def __init__(self, label, action_enum):
+        self.action_enum = action_enum
+        action = pp.MatchFirst([Keyword(val.value) for val in action_enum])
+        trigger = pp.Group(
+            Keyword("ON").suppress()
+            + pp.Word(pp.nums)
+            + Keyword("PERCENT").suppress()
+            + Keyword("DO").suppress()
+            + action
+        )
+        super().__init__(label, value_expr=pp.OneOrMore(trigger), eq=False)
+
+    def typecheck(self, prop_value):
+        return [{"threshold": int(threshold), "action": self.action_enum(action)} for threshold, action in prop_value]
+
+    def render(self, values):
+        if values is None:
+            return ""
+        return tidy_sql(
+            self.label.upper(),
+            *[f"ON {trigger['threshold']} PERCENT DO {trigger['action']}" for trigger in values],
+        )
+
+
 class QueryProp(Prop):
     def __init__(self, label):
         value_expr = pp.Word(pp.printables + " \n")

snowcap/resources/resource_monitor.py

@@ -7,6 +7,7 @@
     Props,
     StringListProp,
     StringProp,
+    TriggersProp,
 )
 from ..resource_name import ResourceName
 from ..scope import AccountScope
@@ -22,6 +23,12 @@ class ResourceMonitorFrequency(ParseableEnum):
     NEVER = "NEVER"
 
 
+class ResourceMonitorAction(ParseableEnum):
+    NOTIFY = "NOTIFY"
+    SUSPEND = "SUSPEND"
+    SUSPEND_IMMEDIATE = "SUSPEND_IMMEDIATE"
+
+
 @dataclass(unsafe_hash=True)
 class _ResourceMonitor(ResourceSpec):
     name: ResourceName
@@ -31,15 +38,33 @@ class _ResourceMonitor(ResourceSpec):
     start_timestamp: str = None
     end_timestamp: str = None
     notify_users: list[str] = None
+    triggers: list[dict] = None
 
     def __post_init__(self):
         super().__post_init__()
         if self.credit_quota is not None and not isinstance(self.credit_quota, int):
             raise ValueError("credit_quota must be an integer or None")
         if self.start_timestamp and self.frequency is None:
             self.frequency = ResourceMonitorFrequency.MONTHLY
-        if self.owner.name != "ACCOUNTADMIN":
-            raise ValueError("ResourceMonitors can only be created by ACCOUNTADMIN")
+        # Snowflake reports both lists in its own order, and neither order carries meaning,
+        # so both sides of a diff are sorted the same way.
+        if self.notify_users:
+            self.notify_users = sorted(self.notify_users)
+        if self.triggers is not None:
+            if not self.triggers:
+                raise ValueError(
+                    "triggers cannot be empty: Snowflake has no statement that removes every trigger "
+                    "from a resource monitor. Omit triggers to leave them unmanaged."
+                )
+            if any(not isinstance(trigger["threshold"], int) for trigger in self.triggers):
+                raise ValueError("trigger thresholds must be whole percentages, for example 75")
+            self.triggers = sorted(
+                (
+                    {"threshold": trigger["threshold"], "action": ResourceMonitorAction(trigger["action"])}
+                    for trigger in self.triggers
+                ),
+                key=lambda trigger: (trigger["threshold"], trigger["action"].value),
+            )
 
 
 class ResourceMonitor(NamedResource, Resource):
@@ -57,6 +82,12 @@ class ResourceMonitor(NamedResource, Resource):
         start_timestamp (string): The start time for the monitoring period. Defaults to None.
         end_timestamp (string): The end time for the monitoring period. Defaults to None.
         notify_users (list): A list of users to notify when thresholds are reached. Defaults to None.
+        triggers (list): The actions to take at a percentage of the credit quota. Each trigger has a
+            `threshold` (an integer percentage, which can exceed 100) and an `action` (NOTIFY,
+            SUSPEND or SUSPEND_IMMEDIATE). Snowflake replaces every trigger whenever one changes.
+            Defaults to None, which leaves the triggers unmanaged.
+        owner (string or Role): The role that owns the monitor. Snowcap creates the monitor as
+            ACCOUNTADMIN and transfers it to this role. Defaults to "ACCOUNTADMIN".
 
     Python:
 
@@ -67,7 +98,12 @@ class ResourceMonitor(NamedResource, Resource):
             frequency="DAILY",
             start_timestamp="2049-01-01 00:00",
             end_timestamp="2049-12-31 23:59",
-            notify_users=["user1", "user2"]
+            notify_users=["user1", "user2"],
+            triggers=[
+                {"threshold": 75, "action": "NOTIFY"},
+                {"threshold": 100, "action": "SUSPEND"},
+                {"threshold": 110, "action": "SUSPEND_IMMEDIATE"},
+            ],
         )
         ```
 
@@ -83,6 +119,13 @@ class ResourceMonitor(NamedResource, Resource):
             notify_users:
               - user1
               - user2
+            triggers:
+              - threshold: 75
+                action: NOTIFY
+              - threshold: 100
+                action: SUSPEND
+              - threshold: 110
+                action: SUSPEND_IMMEDIATE
         ```
     """
 
@@ -94,6 +137,7 @@ class ResourceMonitor(NamedResource, Resource):
         start_timestamp=StringProp("start_timestamp", alt_tokens=["IMMEDIATELY"]),
         end_timestamp=StringProp("end_timestamp"),
         notify_users=StringListProp("notify_users", parens=True),
+        triggers=TriggersProp("triggers", ResourceMonitorAction),
     )
     scope = AccountScope()
     spec = _ResourceMonitor
@@ -106,6 +150,7 @@ def __init__(
         start_timestamp: str = None,
         end_timestamp: str = None,
         notify_users: list[str] = None,
+        triggers: list[dict] = None,
         owner: str = "ACCOUNTADMIN",
         **kwargs,
     ):
@@ -117,6 +162,7 @@ def __init__(
             start_timestamp=start_timestamp,
             end_timestamp=end_timestamp,
             notify_users=notify_users,
+            triggers=triggers,
             owner=owner,
         )
         # TODO: rely on notify_users

tests/fixtures/json/resource_monitor.json

@@ -4,6 +4,23 @@
     "frequency": "DAILY",
     "start_timestamp": "IMMEDIATELY",
     "end_timestamp": "2049-12-31 23:59",
-    "owner": "ACCOUNTADMIN",
-    "notify_users": null
+    "owner": "SYSADMIN",
+    "notify_users": [
+        "JACK",
+        "JILL"
+    ],
+    "triggers": [
+        {
+            "threshold": 75,
+            "action": "NOTIFY"
+        },
+        {
+            "threshold": 100,
+            "action": "SUSPEND"
+        },
+        {
+            "threshold": 110,
+            "action": "SUSPEND_IMMEDIATE"
+        }
+    ]
 }

tests/fixtures/sql/resource_monitor.sql

@@ -4,6 +4,10 @@ CREATE OR REPLACE RESOURCE MONITOR my_mon_1
         FREQUENCY = DAILY
         START_TIMESTAMP = '2020-01-01 00:00:00'
         NOTIFY_USERS = ( teej, jack, jill )
+    TRIGGERS
+        ON 75 PERCENT DO NOTIFY
+        ON 100 PERCENT DO SUSPEND
+        ON 110 PERCENT DO SUSPEND_IMMEDIATE
 ;
 
 

tests/integration/data_provider/test_fetch_resource_simple.py

@@ -235,6 +235,12 @@ def resource_fixtures() -> list:
             name="TEST_FETCH_RESOURCE_MONITOR",
             credit_quota=1000,
             start_timestamp="2049-01-01 00:00",
+            triggers=[
+                {"threshold": 50, "action": "NOTIFY"},
+                {"threshold": 75, "action": "NOTIFY"},
+                {"threshold": 100, "action": "SUSPEND"},
+                {"threshold": 110, "action": "SUSPEND_IMMEDIATE"},
+            ],
         ),
         res.Role(
             name="TEST_FETCH_ROLE",

tests/test_blueprint.py

@@ -2905,3 +2905,96 @@ def test_alert_body_under_ignore_changes_leaves_only_state(session_ctx):
     updates = [c for c in diff(remote_state, manifest) if isinstance(c, UpdateResource)]
     assert len(updates) == 1
     assert updates[0].delta == {"state": "SUSPENDED"}
+
+
+class TestResourceMonitorPlanning:
+    """A resource monitor round-trips through SHOW RESOURCE MONITORS with its triggers, the
+    users it notifies, and the role that owns it, and changes to any of them apply."""
+
+    OWNER = "MONITOR_ADMIN"
+
+    def _show_row(self, **overrides) -> dict:
+        # The shape Snowflake returns for a warehouse monitor with three triggers and two
+        # notify users: thresholds come back as percent strings, users as one string.
+        row = {
+            "name": "WH_MONITOR",
+            "owner": self.OWNER,
+            "owner_role_type": "ROLE",
+            "credit_quota": "1000.00",
+            "frequency": "MONTHLY",
+            "start_time": None,
+            "end_time": None,
+            "notify_at": "75%",
+            "suspend_at": "100%",
+            "suspend_immediately_at": "110%",
+            "notify_users": "BOB, ALICE",
+            "level": "WAREHOUSE",
+        }
+        row.update(overrides)
+        return row
+
+    def _monitor(self) -> res.ResourceMonitor:
+        return res.ResourceMonitor(
+            name="WH_MONITOR",
+            owner=self.OWNER,
+            credit_quota=1000,
+            notify_users=["ALICE", "BOB"],
+            triggers=[
+                {"threshold": 110, "action": "SUSPEND_IMMEDIATE"},
+                {"threshold": 75, "action": "NOTIFY"},
+                {"threshold": 100, "action": "SUSPEND"},
+            ],
+        )
+
+    def _plan(self, session_ctx, remote_state, show_row):
+        owner = res.Role(name=self.OWNER)
+        monitor = self._monitor()
+        manifest = Blueprint(resources=[owner, monitor]).generate_manifest(session_ctx)
+        remote_state = remote_state.copy()
+        remote_state[URN.from_resource(account_locator="ABCD123", resource=owner)] = owner.to_dict()
+        if show_row is not None:
+            with patch("snowcap.data_provider.execute", return_value=[show_row]):
+                fetched = data_provider.fetch_resource_monitor(MagicMock(), monitor.fqn)
+            urn = URN.from_resource(account_locator="ABCD123", resource=monitor)
+            remote_state[urn] = res.ResourceMonitor.spec(**fetched).to_dict(AccountEdition.ENTERPRISE)
+        return diff(remote_state, manifest)
+
+    def _sql(self, session_ctx, plan) -> list[str]:
+        session_ctx = {**session_ctx, "available_roles": session_ctx["available_roles"] + [self.OWNER]}
+        return flatten_sql_commands(compile_plan_to_sql(session_ctx, plan))
+
+    def test_a_monitor_that_matches_snowflake_plans_no_change(self, session_ctx, remote_state):
+        assert self._plan(session_ctx, remote_state, self._show_row()) == []
+
+    @pytest.mark.parametrize(
+        "show_overrides, alter",
+        [
+            (
+                {"suspend_at": "90%"},
+                "ALTER RESOURCE MONITOR WH_MONITOR"
+                " TRIGGERS ON 75 PERCENT DO NOTIFY ON 100 PERCENT DO SUSPEND ON 110 PERCENT DO SUSPEND_IMMEDIATE",
+            ),
+            (
+                {"credit_quota": "500.00", "notify_at": None},
+                "ALTER RESOURCE MONITOR WH_MONITOR SET CREDIT_QUOTA = 1000"
+                " TRIGGERS ON 75 PERCENT DO NOTIFY ON 100 PERCENT DO SUSPEND ON 110 PERCENT DO SUSPEND_IMMEDIATE",
+            ),
+        ],
+    )
+    def test_a_changed_trigger_replaces_the_whole_set_as_the_owner(
+        self, session_ctx, remote_state, show_overrides, alter
+    ):
+        """Snowflake's TRIGGERS clause replaces every trigger and cannot follow SET, and only
+        the owning role can alter a monitor, even when that role is not ACCOUNTADMIN."""
+        sql = self._sql(session_ctx, self._plan(session_ctx, remote_state, self._show_row(**show_overrides)))
+        assert sql[-2:] == [f"USE ROLE {self.OWNER}", alter]
+
+    def test_a_new_monitor_is_created_by_accountadmin_and_handed_to_its_owner(self, session_ctx, remote_state):
+        sql = self._sql(session_ctx, self._plan(session_ctx, remote_state, show_row=None))
+        assert sql[-3:] == [
+            "USE ROLE ACCOUNTADMIN",
+            "CREATE RESOURCE MONITOR WH_MONITOR CREDIT_QUOTA = 1000"
+            " NOTIFY_USERS = ($$ALICE$$, $$BOB$$)"
+            " TRIGGERS ON 75 PERCENT DO NOTIFY ON 100 PERCENT DO SUSPEND ON 110 PERCENT DO SUSPEND_IMMEDIATE",
+            f"GRANT OWNERSHIP ON RESOURCE MONITOR WH_MONITOR TO ROLE {self.OWNER} COPY CURRENT GRANTS",
+        ]

tests/test_resource_types.py

@@ -1067,6 +1067,21 @@ def test_resource_monitor_all_properties(self):
         )
         assert rm._data.credit_quota == 1000
 
+    @pytest.mark.parametrize(
+        "triggers",
+        [
+            # Snowflake has no statement that removes every trigger from a monitor, so an
+            # empty list could never be applied to one that has triggers.
+            [],
+            # Snowflake reads a threshold as a whole percentage.
+            [{"threshold": 75.5, "action": "NOTIFY"}],
+            [{"threshold": "75", "action": "NOTIFY"}],
+        ],
+    )
+    def test_resource_monitor_rejects_triggers_snowflake_cannot_hold(self, triggers):
+        with pytest.raises(ValueError, match="trigger"):
+            res.ResourceMonitor(name="test_rm", triggers=triggers)
+
 
 class TestTag:
     """Tests for Tag resource."""