diff --git a/docs/services/cloudwatch.md b/docs/services/cloudwatch.md index 121e9c5..34b16ef 100644 --- a/docs/services/cloudwatch.md +++ b/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 diff --git a/src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifier.java b/src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifier.java new file mode 100644 index 0000000..a3b9f35 --- /dev/null +++ b/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 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 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 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)); + } +} diff --git a/src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsService.java b/src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsService.java index 6ea32a1..d7c35e2 100644 --- a/src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsService.java +++ b/src/main/java/io/github/hectorvent/floci/services/cloudwatch/metrics/CloudWatchMetricsService.java @@ -30,22 +30,36 @@ public class CloudWatchMetricsService { private final StorageBackend metricStore; private final StorageBackend 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>() {}); this.alarmStore = storageFactory.create("cloudwatchmetrics", "cwalarms.json", new TypeReference>() {}); this.regionResolver = regionResolver; + this.alarmSnsNotifier = alarmSnsNotifier; } CloudWatchMetricsService(StorageBackend metricStore, StorageBackend alarmStore, RegionResolver regionResolver) { + this(metricStore, alarmStore, regionResolver, null); + } + + CloudWatchMetricsService(StorageBackend metricStore, + StorageBackend alarmStore, + RegionResolver regionResolver, AlarmSnsNotifier alarmSnsNotifier) { this.metricStore = metricStore; this.alarmStore = alarmStore; this.regionResolver = regionResolver; + this.alarmSnsNotifier = alarmSnsNotifier; } public void putMetricData(String namespace, List 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); + } } /** diff --git a/src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifierTest.java b/src/test/java/io/github/hectorvent/floci/services/cloudwatch/metrics/AlarmSnsNotifierTest.java new file mode 100644 index 0000000..8a638b6 --- /dev/null +++ b/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 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 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 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")); + } +}