mirror of
https://github.com/Sea-Haven-Industries/procurement-ingest.git
synced 2026-09-30 10:43:14 +00:00
35dc32390c
Some checks failed
Deploy / deploy (push) Has been cancelled
The existing workorder-email-processor-errors alarm fires on any errored async invocation, but a message only reaches the DLQ after Lambda exhausts its async retries and gives up — a genuinely dropped work-order email that the errors alarm alone does not distinguish. Add an ALARM-only CloudWatch alarm (workorder-email-processor-dlq-messages) on the EmailProcessorDlq ApproximateNumberOfMessagesVisible metric (Statistic MAXIMUM, period 5m, evaluationPeriods 1, threshold > 0, treatMissingData NOT_BREACHING). Routes to the same shared site-alerts SNS topic via SnsAction, mirroring the errors-alarm construct style. Refs INFRA-41 / audit H-8.
351 lines
15 KiB
Python
351 lines
15 KiB
Python
"""CDK stack for the work order email ingestion pipeline."""
|
|
|
|
import aws_cdk as cdk
|
|
from aws_cdk import (
|
|
Duration,
|
|
RemovalPolicy,
|
|
Stack,
|
|
aws_cloudwatch as cloudwatch,
|
|
aws_cloudwatch_actions as cw_actions,
|
|
aws_dynamodb as dynamodb,
|
|
aws_kms as kms,
|
|
aws_lambda as lambda_,
|
|
aws_logs as logs,
|
|
aws_s3 as s3,
|
|
aws_s3_notifications as s3n,
|
|
aws_ses as ses,
|
|
aws_ses_actions as ses_actions,
|
|
aws_secretsmanager as secretsmanager,
|
|
aws_sns as sns,
|
|
aws_sqs as sqs,
|
|
aws_ssm as ssm,
|
|
)
|
|
from constructs import Construct
|
|
|
|
# Operations these tables actually issue (PutItem/UpdateItem/DeleteItem writes,
|
|
# GetItem/Query/BatchGetItem reads). DynamoDB emits ThrottledRequests/SystemErrors
|
|
# keyed by TableName + Operation only, so the CDK *_for_operations helpers (which
|
|
# render a SUM MathExpression across these per-operation metrics) are the correct,
|
|
# non-deprecated way to roll a table up to a single alarmable series.
|
|
_DDB_ALARM_OPERATIONS = [
|
|
dynamodb.Operation.GET_ITEM,
|
|
dynamodb.Operation.BATCH_GET_ITEM,
|
|
dynamodb.Operation.QUERY,
|
|
dynamodb.Operation.SCAN,
|
|
dynamodb.Operation.PUT_ITEM,
|
|
dynamodb.Operation.UPDATE_ITEM,
|
|
dynamodb.Operation.DELETE_ITEM,
|
|
dynamodb.Operation.BATCH_WRITE_ITEM,
|
|
]
|
|
|
|
|
|
def _add_ddb_alarms(scope, id_prefix, table, alarm_name_prefix, alarm_topic):
|
|
"""Add throttle + system-error alarms for a DynamoDB table.
|
|
|
|
Both fire on any non-zero datapoint in a 5-min window. ALARM-only SnsAction
|
|
to site-alerts (no OK action); TreatMissingData NOT_BREACHING.
|
|
"""
|
|
table.metric_throttled_requests_for_operations(
|
|
operations=_DDB_ALARM_OPERATIONS,
|
|
period=Duration.minutes(5),
|
|
statistic="Sum",
|
|
).create_alarm(
|
|
scope,
|
|
f"{id_prefix}ThrottlesAlarm",
|
|
alarm_name=f"{alarm_name_prefix}-throttles",
|
|
alarm_description=f"{alarm_name_prefix} DynamoDB throttled requests",
|
|
threshold=0,
|
|
evaluation_periods=1,
|
|
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
|
|
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
|
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
|
|
|
table.metric_system_errors_for_operations(
|
|
operations=_DDB_ALARM_OPERATIONS,
|
|
period=Duration.minutes(5),
|
|
statistic="Sum",
|
|
).create_alarm(
|
|
scope,
|
|
f"{id_prefix}SystemErrorsAlarm",
|
|
alarm_name=f"{alarm_name_prefix}-system-errors",
|
|
alarm_description=f"{alarm_name_prefix} DynamoDB server-side (5xx) errors",
|
|
threshold=0,
|
|
evaluation_periods=1,
|
|
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
|
|
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
|
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
|
|
|
|
|
class WorkorderIngestStack(Stack):
|
|
def __init__(self, scope: Construct, construct_id: str, **kwargs):
|
|
super().__init__(scope, construct_id, **kwargs)
|
|
|
|
# --- Shared alarm SNS topic (site-alerts) ---
|
|
# Imported once near the top so every alarm in this stack reuses the same
|
|
# Topic construct instance (avoids duplicate logical IDs). ALARM-only
|
|
# SnsAction; no OK action, per the CloudWatch-alarm preference. The
|
|
# topic's CMK (alias/seahaven-alarm-topics) lives on the topic itself.
|
|
alarm_topic = sns.Topic.from_topic_arn(
|
|
self,
|
|
"SiteAlertsTopic",
|
|
f"arn:aws:sns:{self.region}:{self.account}:site-alerts",
|
|
)
|
|
|
|
# --- S3 bucket for raw emails ---
|
|
email_bucket = s3.Bucket(
|
|
self,
|
|
"EmailBucket",
|
|
bucket_name=f"workorder-ingest-emails-{self.account}",
|
|
block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
|
|
removal_policy=RemovalPolicy.RETAIN,
|
|
lifecycle_rules=[
|
|
s3.LifecycleRule(expiration=Duration.days(90)),
|
|
],
|
|
)
|
|
|
|
# --- DynamoDB tables ---
|
|
work_orders_table = dynamodb.Table(
|
|
self,
|
|
"WorkOrdersTable",
|
|
table_name="WorkOrders",
|
|
partition_key=dynamodb.Attribute(
|
|
name="work_order_id",
|
|
type=dynamodb.AttributeType.STRING,
|
|
),
|
|
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
|
|
removal_policy=RemovalPolicy.RETAIN,
|
|
)
|
|
# site-code-index and status-index GSIs removed 2026-06-03 (audit M-20):
|
|
# 0 reads in 30d against ~50k WCU each of write amplification. Re-add if
|
|
# a site-code or status query path ships.
|
|
|
|
comments_table = dynamodb.Table(
|
|
self,
|
|
"CommentsTable",
|
|
table_name="WorkOrderComments",
|
|
partition_key=dynamodb.Attribute(
|
|
name="work_order_id",
|
|
type=dynamodb.AttributeType.STRING,
|
|
),
|
|
sort_key=dynamodb.Attribute(
|
|
name="comment_id",
|
|
type=dynamodb.AttributeType.STRING,
|
|
),
|
|
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
|
|
removal_policy=RemovalPolicy.RETAIN,
|
|
)
|
|
|
|
# --- Secrets Manager for Anthropic API key ---
|
|
anthropic_secret = secretsmanager.Secret(
|
|
self,
|
|
"AnthropicApiKey",
|
|
secret_name="workorder-ingest/anthropic-api-key",
|
|
description="Anthropic API key for work order email parsing",
|
|
removal_policy=RemovalPolicy.RETAIN,
|
|
)
|
|
|
|
# --- DLQ for failed async invocations (INFRA-41 / audit H-8) ---
|
|
# SES → S3 → Lambda is async; without an OnFailure destination a failed
|
|
# parse (bad email, transient error) is silently dropped after Lambda's
|
|
# retries. CDK generates the queue name to avoid colliding with the
|
|
# interim CLI-created workorder-email-processor-dlq (removed post-deploy).
|
|
email_processor_dlq = sqs.Queue(
|
|
self,
|
|
"EmailProcessorDlq",
|
|
retention_period=Duration.days(14),
|
|
enforce_ssl=True,
|
|
)
|
|
|
|
# --- Lambda function ---
|
|
email_processor = lambda_.Function(
|
|
self,
|
|
"EmailProcessor",
|
|
function_name="workorder-email-processor",
|
|
runtime=lambda_.Runtime.PYTHON_3_12,
|
|
architecture=lambda_.Architecture.ARM_64,
|
|
handler="handler.handler",
|
|
code=lambda_.Code.from_asset(
|
|
"../lambdas/wo/email_processor",
|
|
bundling=cdk.BundlingOptions(
|
|
image=lambda_.Runtime.PYTHON_3_12.bundling_image,
|
|
command=[
|
|
"bash",
|
|
"-c",
|
|
"pip install --platform manylinux2014_aarch64 --only-binary=:all: "
|
|
"-r requirements.txt -t /asset-output && "
|
|
"cp -r . /asset-output/",
|
|
],
|
|
),
|
|
),
|
|
timeout=Duration.seconds(60),
|
|
memory_size=256,
|
|
log_retention=logs.RetentionDays.TWO_MONTHS,
|
|
dead_letter_queue=email_processor_dlq,
|
|
environment={
|
|
"WORK_ORDERS_TABLE": work_orders_table.table_name,
|
|
"COMMENTS_TABLE": comments_table.table_name,
|
|
"ANTHROPIC_API_KEY_SECRET_ARN": anthropic_secret.secret_arn,
|
|
},
|
|
)
|
|
|
|
# Grant permissions
|
|
email_bucket.grant_read(email_processor)
|
|
work_orders_table.grant_read_write_data(email_processor)
|
|
comments_table.grant_read_write_data(email_processor)
|
|
anthropic_secret.grant_read(email_processor)
|
|
|
|
# Pre-emptive KMS grant on the shared DynamoDB CMK (alias/seahaven-dynamodb,
|
|
# /seahaven/dynamodb/cmk-arn). The WorkOrders/WorkOrderComments tables are
|
|
# NOT yet SSE-KMS encrypted, so this grant is currently unused; it is added
|
|
# ahead of the CMK migration so the processor role does not hit AccessDenied
|
|
# the moment those tables are migrated. The actual table migration + kebab
|
|
# rename is tracked in INFRA-6 (and the set-aside wo_stack CMK WIP); fold the
|
|
# web-ui read grant in there.
|
|
dynamodb_cmk = kms.Key.from_key_arn(
|
|
self,
|
|
"DynamoDbCmk",
|
|
ssm.StringParameter.value_for_string_parameter(
|
|
self, "/seahaven/dynamodb/cmk-arn"
|
|
),
|
|
)
|
|
dynamodb_cmk.grant_encrypt_decrypt(email_processor)
|
|
|
|
# --- Errors alarm (INFRA-41 / audit H-8) ---
|
|
# ALARM-only (no OK action, per the CloudWatch-alarm preference) to the
|
|
# shared site-alerts topic. Any errored invocation in a 5-min window pages.
|
|
email_processor.metric_errors(
|
|
period=Duration.minutes(5),
|
|
statistic="Sum",
|
|
).create_alarm(
|
|
self,
|
|
"EmailProcessorErrorsAlarm",
|
|
alarm_name="workorder-email-processor-errors",
|
|
alarm_description="workorder-email-processor async invocation errors",
|
|
threshold=0,
|
|
evaluation_periods=1,
|
|
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
|
|
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
|
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
|
|
|
# --- Throttles alarm: workorder-email-processor ---
|
|
# Any throttled invocation (concurrency cap hit) in a 5-min window pages.
|
|
# ALARM-only to site-alerts; no OK action; NOT_BREACHING when no data.
|
|
email_processor.metric_throttles(
|
|
period=Duration.minutes(5),
|
|
statistic="Sum",
|
|
).create_alarm(
|
|
self,
|
|
"EmailProcessorThrottlesAlarm",
|
|
alarm_name="workorder-email-processor-throttles",
|
|
alarm_description="workorder-email-processor invocation throttles",
|
|
threshold=0,
|
|
evaluation_periods=1,
|
|
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
|
|
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
|
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
|
|
|
# --- Duration alarm: workorder-email-processor (orphan adoption) ---
|
|
# Adopts the orphaned CLI alarm Lambda-Duration-workorder-email-processor
|
|
# under the repo's <fn>-duration naming (NEW logical name → no deploy
|
|
# collision; delete the orphan post-deploy). p95 /
|
|
# 45000 ms (75% of the 60s timeout) / eval 3 of which 2 datapoints —
|
|
# tighter than the orphan's Maximum>=48000 / 1-of-1.
|
|
email_processor.metric_duration(
|
|
period=Duration.minutes(5),
|
|
statistic="p95",
|
|
).create_alarm(
|
|
self,
|
|
"EmailProcessorDurationAlarm",
|
|
alarm_name="workorder-email-processor-duration",
|
|
alarm_description="workorder-email-processor p95 duration approaching the 60s timeout",
|
|
threshold=45000,
|
|
evaluation_periods=3,
|
|
datapoints_to_alarm=2,
|
|
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_OR_EQUAL_TO_THRESHOLD,
|
|
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
|
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
|
|
|
# --- DLQ messages-present alarm (INFRA-41 / audit H-8) ---
|
|
# The errors alarm above fires on any errored invocation, but a message
|
|
# only lands in the DLQ after Lambda exhausts its async retries and gives
|
|
# up — i.e. a genuinely dropped email. ALARM-only (no OK action) to the
|
|
# same shared site-alerts topic. MAXIMUM over a 5-min window so a single
|
|
# visible message pages even if it is later consumed/redriven.
|
|
email_processor_dlq.metric_approximate_number_of_messages_visible(
|
|
period=Duration.minutes(5),
|
|
statistic="Maximum",
|
|
).create_alarm(
|
|
self,
|
|
"EmailProcessorDlqMessagesAlarm",
|
|
alarm_name="workorder-email-processor-dlq-messages",
|
|
alarm_description="workorder-email-processor DLQ has visible messages (dropped emails)",
|
|
threshold=0,
|
|
evaluation_periods=1,
|
|
comparison_operator=cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD,
|
|
treat_missing_data=cloudwatch.TreatMissingData.NOT_BREACHING,
|
|
).add_alarm_action(cw_actions.SnsAction(alarm_topic))
|
|
|
|
# S3 event notification -> Lambda
|
|
email_bucket.add_event_notification(
|
|
s3.EventType.OBJECT_CREATED,
|
|
s3n.LambdaDestination(email_processor),
|
|
s3.NotificationKeyFilter(prefix="inbound/"),
|
|
)
|
|
|
|
# --- SES Receipt Rule ---
|
|
rule_set = ses.ReceiptRuleSet.from_receipt_rule_set_name(
|
|
self,
|
|
"ExistingRuleSet",
|
|
"INBOUND_MAIL",
|
|
)
|
|
|
|
rule_set.add_rule(
|
|
"WorkorderEmailRule",
|
|
recipients=["apm@int.seahaven.com"],
|
|
actions=[
|
|
ses_actions.S3(
|
|
bucket=email_bucket,
|
|
object_key_prefix="inbound/",
|
|
),
|
|
],
|
|
)
|
|
|
|
# --- Web UI Lambda ---
|
|
web_ui = lambda_.Function(
|
|
self,
|
|
"WebUI",
|
|
function_name="workorder-web-ui",
|
|
runtime=lambda_.Runtime.PYTHON_3_12,
|
|
architecture=lambda_.Architecture.ARM_64,
|
|
handler="handler.handler",
|
|
code=lambda_.Code.from_asset("../lambdas/wo/web_ui"),
|
|
timeout=Duration.seconds(15),
|
|
memory_size=128,
|
|
log_retention=logs.RetentionDays.TWO_MONTHS,
|
|
environment={
|
|
"WORK_ORDERS_TABLE": work_orders_table.table_name,
|
|
"COMMENTS_TABLE": comments_table.table_name,
|
|
},
|
|
)
|
|
|
|
work_orders_table.grant_read_data(web_ui)
|
|
comments_table.grant_read_data(web_ui)
|
|
|
|
# --- DynamoDB throttle + system-error alarms ---
|
|
# ThrottledRequests / SystemErrors emit at TableName + Operation only
|
|
# (verified against live CloudWatch: no TableName-only rollup exists, and
|
|
# metric_throttled_requests is deprecated/invalid in aws-cdk-lib 2.259.0).
|
|
# Each table currently has zero throttle/error datapoints, so the series
|
|
# only materialise on first occurrence — NOT_BREACHING keeps them OK until
|
|
# then.
|
|
_add_ddb_alarms(
|
|
self, "WorkOrdersTable", work_orders_table, "WorkOrders", alarm_topic
|
|
)
|
|
_add_ddb_alarms(
|
|
self, "WorkOrderComments", comments_table, "WorkOrderComments", alarm_topic
|
|
)
|
|
|
|
# Public Function URL removed 2026-06-08 (INFRA-74 / audit C-5): the
|
|
# unauthenticated FunctionUrlAuthType.NONE URL was deleted out-of-band
|
|
# via CLI. Removing the construct (and its auto-generated Principal:*
|
|
# invoke permission) reconciles IaC with the live state.
|