Setting the file. One moment.
Expiry · 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/expiry.py
scripts/ expiry.py
Python · 125 lines · 6 KB
06:00 UTC on the 1st of the month. Before 06:00 UTC on the 1st the previous-month grace
15 is still open, so previous-month records are still submittable and are NOT expired.
16
17 Case (b) INCLUDES previous-month records still INSIDE the 24h window: once the grace has
18 closed the service rejects them, so the submitter (which reads the now-23h..now-1h window)
19 would otherwise re-attempt them every run until they cross 24h. The expiry sweep therefore
20 also covers the in-window range (1..23) for previous-month buckets when the grace is closed,
21 and its past-window sweep starts at hours_back == 24 so it is CONTIGUOUS with the submitter's
22 window (no bucket falls between the two). Current-month in-window buckets are still
23 submittable and are never expired.
24
25 Expiring = REMOVE meteringPending, SET meteringStatus=SubmissionExpired, emit the
26 UsageSubmissionExpired metric (distinct from the raw table's UsageAggregationExpired age-out,
27 so an operator can tell "aggregated but never submitted" from "aged out before aggregation").
28 The record itself is retained (audit).
29 """
30
31 from datetime import datetime, timedelta, timezone
32 from typing import Any, Dict
33
34 from handlers import metering_core as core
35
36 logger = core.logger
37
38 # How far back to sweep for stale pending buckets, and the empty-streak early stop — a
39 # sparse, self-terminating lookback (a backlog of non-empty buckets extends it).
40 EXPIRY_MAX_HOURS_BACK = 24 * 400
41 EXPIRY_EMPTY_STREAK_STOP = 6
42
43
44 def handler (event, context):
45 if core.aggregated_usage_table is None :
46 raise RuntimeError ( "AGGREGATED_USAGE_TABLE not configured" )
47
48 now = datetime.now(timezone.utc)
49 base = now.replace( minute = 0 , second = 0 , microsecond = 0 )
50 grace_open = core.previous_month_grace_open(now) # 1st of month, before 06:00 UTC
51
52 aged = 0
53
54 # (b) Month-end reconciliation: once the grace has CLOSED (on/after 06:00 UTC on the
55 # 1st), previous-month records INSIDE the 24h window are no longer submittable — the
56 # service rejects them — yet the submitter's now-23h..now-1h window would keep
57 # re-attempting them every run (failed submits + alarm noise) until they naturally
58 # cross 24h. So when the grace is closed, expire previous-month buckets in the in-window
59 # range 1..23 too. Current-month in-window buckets are still submittable → never touched.
60 if not grace_open:
61 for hours_back in range ( 1 , 24 ):
62 hour_bucket = (base - timedelta( hours = hours_back)).strftime(core. HOUR_FMT )
63 if core.is_previous_month_bucket(hour_bucket, now):
64 aged += _expire_bucket(hour_bucket)
65
66 # Past-window sweep: everything strictly older than 24h is unbillable regardless of
67 # month. Start at hours_back == 24 so this range is CONTIGUOUS with the submitter's
68 # now-1h..now-23h window (no bucket falls between the two). A previous-month bucket while
69 # the grace is still OPEN is skipped (still submittable). Sparse, self-terminating walk.
70 hours_back = 24
71 empty_streak = 0
72 while hours_back <= EXPIRY_MAX_HOURS_BACK and empty_streak < EXPIRY_EMPTY_STREAK_STOP :
73 hour_bucket = (base - timedelta( hours = hours_back)).strftime(core. HOUR_FMT )
74 hours_back += 1
75
76 is_prev_month = core.is_previous_month_bucket(hour_bucket, now)
77 # A previous-month bucket is only expirable once the grace has CLOSED (>=06:00 UTC
78 # on the 1st). While the grace is open, previous-month records are still submittable
79 # → skip. Current-month buckets past 24h are always expirable.
80 if is_prev_month and grace_open:
81 continue
82
83 found_any = _expire_bucket(hour_bucket)
84 aged += found_any
85 empty_streak = 0 if found_any else empty_streak + 1
86
87 if aged:
88 logger.error( f "Expired { aged } aggregated record(s) past the submittable window" )
89 else :
90 logger.info( "No aggregated records to expire" )
91 return { "expired" : aged}
92
93
94 def _expire_bucket (hour_bucket) -> int :
95 expired = 0
96 kwargs: Dict[ str , Any] = {
97 "IndexName" : "metering_pending" ,
98 "KeyConditionExpression" : "meteringPending = :hb" ,
99 "ExpressionAttributeValues" : { ":hb" : hour_bucket},
100 }
101 while True :
102 response = core.aggregated_usage_table.query( ** kwargs)
103 for item in response.get( "Items" , []):
104 core.emf_metric( "UsageSubmissionExpired" , 1 )
105 _expr, _vals, _ = core.with_audit_timestamps(
106 "REMOVE meteringPending SET meteringStatus = :st" ,
107 { ":st" : "SubmissionExpired" },
108 )
109 core.aggregated_usage_table.update_item(
110 Key = {
111 "licenseArn" : item[ "licenseArn" ],
112 "account_dimension_hour" : item[ "account_dimension_hour" ],
113 },
114 UpdateExpression = _expr,
115 ExpressionAttributeValues = _vals,
116 )
117 logger.warning(
118 "AGGREGATED EXPIRY: unbillable pending aggregated record "
119 f "license= { core.mask(item[ 'licenseArn' ]) } hour= { hour_bucket } "
120 )
121 expired += 1
122 if "LastEvaluatedKey" not in response:
123 break
124 kwargs[ "ExclusiveStartKey" ] = response[ "LastEvaluatedKey" ]
125 return expired