fix(cloudwatch): publish SNS actions on alarm state transitions (#5918)

(cherry picked from commit f5811c3a0735be634b6122503237330826e0ba4a)

docs/services/cloudwatch.md

@@ -331,6 +331,14 @@ aws logs put-retention-policy \
 Metric streams are stored as definitions with their `running` or `stopped` state. Floci never
 delivers metrics to the Firehose delivery stream a metric stream names.
 
+### SNS alarm actions
+
+State transitions publish to same-account standard SNS topics in `AlarmActions`, `OKActions`, or
+`InsufficientDataActions`, including transitions from `SetAlarmState`. Unchanged states and
+`ActionsEnabled=false` suppress notifications. Scaling action retries remain independent.
+Failures are logged per target without blocking other targets. Notifications include a
+single-metric `Trigger`; durable retries, FIFO topics, and metric-math payloads are not covered.
+
 ### Metric-math alarm definitions
 
 `PutMetricAlarm` stores the `Metrics` query definitions and `DescribeAlarms` returns them over

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifier.java

@@ -0,0 +1,112 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.core.common.AwsArnUtils;
+import io.github.hectorvent.floci.core.common.AwsPartition;
+import io.github.hectorvent.floci.core.common.AwsPartitions;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.services.sns.SnsService;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.jboss.logging.Logger;
+
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+/** Publishes SNS actions once per state transition, independently of scaling action retries. */
+@ApplicationScoped
+public class AlarmSnsNotifier {
+
+    private static final Logger LOG = Logger.getLogger(AlarmSnsNotifier.class);
+    private static final DateTimeFormatter TIMESTAMP_FORMAT = DateTimeFormatter
+            .ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ", Locale.ROOT).withZone(ZoneOffset.UTC);
+    private final SnsService snsService;
+    private final ObjectMapper objectMapper;
+
+    public AlarmSnsNotifier(SnsService snsService, ObjectMapper objectMapper) {
+        this.snsService = snsService;
+        this.objectMapper = objectMapper;
+    }
+
+    public void notifyTransition(MetricAlarm alarm, String oldState) {
+        if (!alarm.isActionsEnabled() || alarm.getStateValue().equals(oldState)) {
+            return;
+        }
+        List<String> actions = switch (alarm.getStateValue()) {
+            case "ALARM" -> alarm.getAlarmActions();
+            case "OK" -> alarm.getOkActions();
+            case "INSUFFICIENT_DATA" -> alarm.getInsufficientDataActions();
+            default -> List.of();
+        };
+        if (actions == null) {
+            return;
+        }
+        for (String action : actions) {
+            if (!AwsArnUtils.isArnFor(action, "sns")) {
+                continue;
+            }
+            try {
+                AwsArnUtils.Arn topic = AwsArnUtils.parse(action);
+                String subject = alarm.getStateValue() + ": \"" + alarm.getAlarmName() + "\"";
+                if (subject.length() > 100) {
+                    subject = subject.substring(0, 100);
+                }
+                snsService.publish(action, null, message(alarm, oldState), subject, topic.region());
+            } catch (Exception e) {
+                LOG.warnv(e, "Unable to publish alarm {0} state {1} to SNS action {2}",
+                        alarm.getAlarmName(), alarm.getStateValue(), action);
+            }
+        }
+    }
+
+    private String message(MetricAlarm alarm, String oldState) throws JsonProcessingException {
+        AwsArnUtils.Arn arn = AwsArnUtils.parse(alarm.getAlarmArn());
+        Map<String, Object> payload = new LinkedHashMap<>();
+        payload.put("AlarmName", alarm.getAlarmName());
+        payload.put("AlarmDescription", alarm.getAlarmDescription());
+        payload.put("AWSAccountId", arn.accountId());
+        payload.put("AlarmConfigurationUpdatedTimestamp", timestamp(alarm.getAlarmConfigurationUpdatedTimestamp()));
+        payload.put("NewStateValue", alarm.getStateValue());
+        payload.put("NewStateReason", alarm.getStateReason());
+        payload.put("StateChangeTime", timestamp(alarm.getStateUpdatedTimestamp()));
+        String regionName = AwsPartitions.forRegion(arn.region()).stream()
+                .flatMap(partition -> partition.regions().stream())
+                .filter(region -> region.id().equals(arn.region()))
+                .map(AwsPartition.Region::description).findFirst().orElse(arn.region());
+        payload.put("Region", regionName);
+        payload.put("AlarmArn", alarm.getAlarmArn());
+        payload.put("OldStateValue", oldState);
+        payload.put("OKActions", alarm.getOkActions());
+        payload.put("AlarmActions", alarm.getAlarmActions());
+        payload.put("InsufficientDataActions", alarm.getInsufficientDataActions());
+        Map<String, Object> trigger = new LinkedHashMap<>();
+        trigger.put("MetricName", alarm.getMetricName());
+        trigger.put("Namespace", alarm.getNamespace());
+        trigger.put("StatisticType", "Statistic");
+        trigger.put("Statistic", alarm.getStatistic() != null
+                ? alarm.getStatistic().toUpperCase(Locale.ROOT) : null);
+        trigger.put("Unit", alarm.getUnit());
+        trigger.put("Dimensions", alarm.getDimensions());
+        trigger.put("Period", alarm.getPeriod());
+        trigger.put("EvaluationPeriods", alarm.getEvaluationPeriods());
+        if (alarm.getDatapointsToAlarm() != null) {
+            trigger.put("DatapointsToAlarm", alarm.getDatapointsToAlarm());
+        }
+        trigger.put("ComparisonOperator", alarm.getComparisonOperator());
+        trigger.put("Threshold", alarm.getThreshold());
+        trigger.put("TreatMissingData", alarm.getTreatMissingData());
+        trigger.put("EvaluateLowSampleCountPercentile", alarm.getEvaluateLowSampleCountPercentile() != null
+                ? alarm.getEvaluateLowSampleCountPercentile() : "");
+        payload.put("Trigger", trigger);
+        return objectMapper.writeValueAsString(payload);
+    }
+
+    private String timestamp(long epochSeconds) {
+        return TIMESTAMP_FORMAT.format(Instant.ofEpochSecond(epochSeconds));
+    }
+}

src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsService.java

@@ -30,22 +30,36 @@ public class CloudWatchMetricsService {
     private final StorageBackend<String, MetricDatum> metricStore;
     private final StorageBackend<String, MetricAlarm> alarmStore;
     private final RegionResolver regionResolver;
+    private final AlarmSnsNotifier alarmSnsNotifier;
+
+    public CloudWatchMetricsService(StorageFactory storageFactory, RegionResolver regionResolver) {
+        this(storageFactory, regionResolver, null);
+    }
 
     @Inject
-    public CloudWatchMetricsService(StorageFactory storageFactory, RegionResolver regionResolver) {
+    public CloudWatchMetricsService(StorageFactory storageFactory, RegionResolver regionResolver,
+                                    AlarmSnsNotifier alarmSnsNotifier) {
         this.metricStore = storageFactory.create("cloudwatchmetrics", "cwmetrics.json",
                 new TypeReference<Map<String, MetricDatum>>() {});
         this.alarmStore = storageFactory.create("cloudwatchmetrics", "cwalarms.json",
                 new TypeReference<Map<String, MetricAlarm>>() {});
         this.regionResolver = regionResolver;
+        this.alarmSnsNotifier = alarmSnsNotifier;
     }
 
     CloudWatchMetricsService(StorageBackend<String, MetricDatum> metricStore,
                              StorageBackend<String, MetricAlarm> alarmStore,
                              RegionResolver regionResolver) {
+        this(metricStore, alarmStore, regionResolver, null);
+    }
+
+    CloudWatchMetricsService(StorageBackend<String, MetricDatum> metricStore,
+                             StorageBackend<String, MetricAlarm> alarmStore,
+                             RegionResolver regionResolver, AlarmSnsNotifier alarmSnsNotifier) {
         this.metricStore = metricStore;
         this.alarmStore = alarmStore;
         this.regionResolver = regionResolver;
+        this.alarmSnsNotifier = alarmSnsNotifier;
     }
 
     public void putMetricData(String namespace, List<MetricDatum> datums, String region) {
@@ -323,6 +337,7 @@ public class CloudWatchMetricsService {
         MetricAlarm alarm = alarmStore.get(key)
                 .orElseThrow(() -> new AwsException("ResourceNotFound", "Alarm not found: " + alarmName, 404));
 
+        String oldState = alarm.getStateValue();
         alarm.setStateValue(stateValue);
         alarm.setStateReason(stateReason);
         alarm.setStateReasonData(stateReasonData);
@@ -330,6 +345,9 @@ public class CloudWatchMetricsService {
 
         alarmStore.put(key, alarm);
         LOG.infov("SetAlarmState: {0} -> {1}", alarmName, stateValue);
+        if (alarmSnsNotifier != null) {
+            alarmSnsNotifier.notifyTransition(alarm, oldState);
+        }
     }
 
     /**

src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifierTest.java

@@ -0,0 +1,127 @@
+package io.github.hectorvent.floci.services.cloudwatch.metrics;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.github.hectorvent.floci.core.common.RegionResolver;
+import io.github.hectorvent.floci.core.storage.InMemoryStorage;
+import io.github.hectorvent.floci.services.cloudwatch.metrics.model.MetricAlarm;
+import io.github.hectorvent.floci.services.sns.SnsService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.time.Instant;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+class AlarmSnsNotifierTest {
+    private static final String REGION = "us-east-1";
+    private static final String TOPIC = "arn:aws:sns:us-east-1:000000000000:alarms";
+    private final ObjectMapper mapper = new ObjectMapper();
+    private SnsService sns;
+    private CloudWatchMetricsService service;
+    private MetricAlarm alarm;
+
+    @BeforeEach
+    void setUp() {
+        sns = mock(SnsService.class);
+        service = new CloudWatchMetricsService(new InMemoryStorage<>(), new InMemoryStorage<>(),
+                new RegionResolver(REGION, "000000000000"), new AlarmSnsNotifier(sns, mapper));
+        alarm = new MetricAlarm();
+        alarm.setAlarmName("test-alarm");
+        alarm.setActionsEnabled(true);
+        alarm.setAlarmActions(List.of(TOPIC));
+        alarm.setOkActions(List.of(TOPIC));
+        alarm.setInsufficientDataActions(List.of(TOPIC));
+        alarm.setEvaluationPeriods(3);
+        alarm.setStatistic("Average");
+        alarm.setMetricName("Errors");
+        alarm.setNamespace("Local/PoC");
+        service.putMetricAlarm(alarm, REGION);
+    }
+
+    @Test
+    void publishesAwsPayloadForEachTransitionButNotRepeatedState() throws Exception {
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        service.setAlarmState("test-alarm", "ALARM", "still breached", null, REGION);
+        service.setAlarmState("test-alarm", "OK", "recovered", null, REGION);
+        service.setAlarmState("test-alarm", "INSUFFICIENT_DATA", "missing", null, REGION);
+        ArgumentCaptor<String> messages = ArgumentCaptor.forClass(String.class);
+        verify(sns, times(3)).publish(eq(TOPIC), isNull(), messages.capture(), anyString(), eq(REGION));
+        JsonNode first = mapper.readTree(messages.getAllValues().getFirst());
+        assertEquals("test-alarm", first.path("AlarmName").asText());
+        assertEquals("INSUFFICIENT_DATA", first.path("OldStateValue").asText());
+        assertEquals("ALARM", first.path("NewStateValue").asText());
+        assertEquals("000000000000", first.path("AWSAccountId").asText());
+        assertEquals(alarm.getAlarmArn(), first.path("AlarmArn").asText());
+        assertEquals("Errors", first.path("Trigger").path("MetricName").asText());
+        assertFalse(first.path("Trigger").has("DatapointsToAlarm"));
+        assertEquals("AVERAGE", first.path("Trigger").path("Statistic").asText());
+        assertEquals("", first.path("Trigger").path("EvaluateLowSampleCountPercentile").asText());
+        assertTrue(first.path("Trigger").path("EvaluateLowSampleCountPercentile").isTextual());
+        assertNull(alarm.getDatapointsToAlarm());
+        assertEquals("OK", mapper.readTree(messages.getAllValues().get(2)).path("OldStateValue").asText());
+    }
+
+    @Test
+    void formatsBothTimestampsWithMillisAndUtcOffset() throws Exception {
+        alarm.setAlarmConfigurationUpdatedTimestamp(1716280510L);
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        ArgumentCaptor<String> message = ArgumentCaptor.forClass(String.class);
+        verify(sns).publish(eq(TOPIC), isNull(), message.capture(), anyString(), eq(REGION));
+        JsonNode payload = mapper.readTree(message.getValue());
+        assertEquals("2024-05-21T08:35:10.000+0000",
+                payload.path("AlarmConfigurationUpdatedTimestamp").asText());
+        String stateChangeTime = payload.path("StateChangeTime").asText();
+        assertTrue(stateChangeTime
+                .matches("[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}\\.[0-9]{3}\\+0000"));
+        assertEquals(alarm.getStateUpdatedTimestamp(),
+                Instant.parse(stateChangeTime.replace("+0000", "Z")).getEpochSecond());
+    }
+
+    @Test
+    void preservesExplicitDatapointsToAlarmInPayload() throws Exception {
+        alarm.setDatapointsToAlarm(2);
+        alarm.setStatistic("Maximum");
+        alarm.setEvaluateLowSampleCountPercentile("ignore");
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        ArgumentCaptor<String> message = ArgumentCaptor.forClass(String.class);
+        verify(sns).publish(eq(TOPIC), isNull(), message.capture(), anyString(), eq(REGION));
+        assertEquals(2, mapper.readTree(message.getValue()).path("Trigger").path("DatapointsToAlarm").intValue());
+        JsonNode trigger = mapper.readTree(message.getValue()).path("Trigger");
+        assertEquals("MAXIMUM", trigger.path("Statistic").asText());
+        assertEquals("ignore", trigger.path("EvaluateLowSampleCountPercentile").asText());
+        assertEquals(2, alarm.getDatapointsToAlarm());
+    }
+
+    @Test
+    void disabledActionsDoNotPublish() {
+        alarm.setActionsEnabled(false);
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        verifyNoInteractions(sns);
+        assertEquals("ALARM", alarm.getStateValue());
+    }
+
+    @Test
+    void failedTargetDoesNotPreventOtherTargetsOrStateChange() {
+        String missing = "arn:aws:sns:us-east-1:000000000000:missing";
+        alarm.setAlarmActions(List.of(missing, "arn:aws:autoscaling:us-east-1:000000000000:scalingPolicy:x", TOPIC));
+        doThrow(new IllegalStateException("missing topic"))
+                .when(sns).publish(eq(missing), isNull(), anyString(), anyString(), eq(REGION));
+        assertDoesNotThrow(() -> service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION));
+        verify(sns).publish(eq(TOPIC), isNull(), anyString(), anyString(), eq(REGION));
+        assertEquals("ALARM", alarm.getStateValue());
+    }
+
+    @Test
+    void usesTopicRegionAndPreservesChinaPartition() {
+        String topic = "arn:aws-cn:sns:cn-north-1:000000000000:alarms";
+        alarm.setAlarmArn("arn:aws-cn:cloudwatch:cn-north-1:000000000000:alarm:test-alarm");
+        alarm.setAlarmActions(List.of(topic));
+        service.setAlarmState("test-alarm", "ALARM", "breached", null, REGION);
+        verify(sns).publish(eq(topic), isNull(), contains("arn:aws-cn:cloudwatch:"), anyString(), eq("cn-north-1"));
+    }
+}