Setting the file. One moment.
Cleanup · AWS Marketplace Metering · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page Query Patterns
70
Creating Amazon Aurora Db Cluster With Instances
93
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
scripts/cleanup.py
scripts/ cleanup.py
Python · 103 lines · 5 KB
14
true writer persisted.
15
16 This makes a successfully-aggregated raw row observably distinct from a never-processed row
17 (previously only RejectedClientSide and AggregationExpired rows carried a status). The raw-row
18 meteringStatus domain is therefore {Aggregated, RejectedClientSide, AggregationExpired}; the
19 SUBMISSION outcome (meteringRecordId / Success / CustomerNotSubscribed / …) is NOT written
20 here — it lives only on the aggregated_usage record (the submitter owns it).
21
22 Why per-key UpdateItem and not BatchWriteItem
23 ---------------------------------------------
24 DynamoDB BatchWriteItem only supports full-item Put/Delete — it CANNOT partially update a
25 single attribute. Rewriting the whole item via a PutRequest would clobber any concurrent
26 seller write to that row. So we apply a surgical
27 ``UpdateItem ... REMOVE meteringPending SET meteringStatus, meteringStatusReason``
28 (idempotent: re-running overwrites the same values).
29
30 Avoiding a Lambda timeout without hot-spotting a partition
31 ----------------------------------------------------------
32 To get batch-like speed we issue the per-key UpdateItems concurrently with a BOUNDED
33 thread pool (CLEANUP_MAX_WORKERS, default 16) rather than serially. Every row of a group
34 shares the same ``licenseArn`` partition key, so a burst hits ONE DynamoDB partition
35 (~1000 WCU/sec limit). Bounded concurrency + the <=100-keys-per-message chunk keep
36 instantaneous WCU under that limit; a throttling error is retried with exponential
37 backoff. A key that still fails after retries is NOT swallowed — the whole message is
38 failed (raising) so SQS redelivers it and it eventually lands on the DLQ; cleanup is never
39 silently dropped (the raw row simply keeps meteringPending and is re-discovered, where the
40 aggregator's conditional put no-ops and re-enqueues cleanup).
41
42 Cleanup runs only AFTER the aggregator durably wrote the aggregated_usage record, so
43 finalizing the raw rows here never loses un-aggregated usage.
44 """
45
46 import json
47
48 from handlers import metering_core as core
49
50 logger = core.logger
51
52
53 def handler (event, context):
54 failures = []
55 for record in event.get( "Records" , []):
56 try :
57 _process_message(json.loads(record[ "body" ]))
58 except Exception as e:
59 logger.error( f "Cleanup failed for a message: { type (e). __name__ } : { e } " )
60 failures.append({ "itemIdentifier" : record[ "messageId" ]})
61 return { "batchItemFailures" : failures}
62
63
64 def _process_message (msg):
65 license_arn = msg[ "licenseArn" ]
66 row_keys = [k for k in msg.get( "rowKeys" , []) if k]
67 if not row_keys:
68 return
69
70 status = msg.get( "meteringStatus" ) # present only for a FRESH-write cleanup (rule a)
71 if status == "Aggregated" :
72 # Rule (d): the aggregator freshly wrote the record, so this message carries the
73 # authoritative sum + count. Clear meteringPending AND set the positive terminal
74 # bookkeeping status/reason. UNCONDITIONAL — performed even if meteringPending was
75 # already cleared by another process (a duplicate visit) — so the
76 # authoritative status/reason always lands. Idempotent per-key UpdateItem.
77 total_quantity = msg.get( "totalQuantity" , 0 )
78 record_count = msg.get( "recordCount" , len (row_keys))
79 reason = f "Aggregated total quantity { total_quantity } from { record_count } raw usage records"
80 core.update_rows_bounded(
81 license_arn,
82 row_keys,
83 "REMOVE meteringPending SET meteringStatus = :st, meteringStatusReason = :rsn" ,
84 { ":st" : "Aggregated" , ":rsn" : reason},
85 )
86 logger.info(
87 f "Finalized { len (row_keys) } raw row(s) as Aggregated ( { reason } ) "
88 f "for license= { core.mask(license_arn) } "
89 )
90 else :
91 # Rule (e): a status-less cleanup message (duplicate/concurrent visit).
92 # Only clear meteringPending so the row leaves the sparse GSI; do NOT stamp a status
93 # or a sum that might differ from what the true writer persisted.
94 core.update_rows_bounded(
95 license_arn,
96 row_keys,
97 "REMOVE meteringPending" ,
98 {},
99 )
100 logger.info(
101 f "Cleared meteringPending on { len (row_keys) } raw row(s) (no status) "
102 f "for license= { core.mask(license_arn) } "
103 )