amadan
awsbilling: read CUR deliveries and classify charged S3 transfer
Object-store transfer is billed only when AWS charged Team Tito for it, and the Cost and Usage Report is AWS's own statement of that. This adds a pure reader for CUR 2.0 (and legacy) CSV, gzip or plain, with headers normalised across both spellings; a classifier that counts only charged S3 transfer-out lines (internet, regional, inter-region; never CloudFront, inbound, requests, storage, or zero-cost lines); and a delivery reader that takes only the newest manifest for a period, deduplicates by line identity, refuses conflicting duplicates, pins every file by digest, and reports Final only when every S3 line carries an invoice id. Amounts are exact big.Rat arithmetic, rounded down.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
9 files changed,
+1556
−0
internal/awsbilling/classify.go+179 −0internal/awsbilling/classify_test.go+163 −0internal/awsbilling/cur.go+381 −0internal/awsbilling/cur_test.go+153 −0internal/awsbilling/delivery.go+347 −0internal/awsbilling/delivery_test.go+261 −0internal/awsbilling/memsource.go+58 −0internal/awsbilling/testdata/cur2-2026-09.csv+11 −0internal/awsbilling/testdata/legacy-2026-09.csv+3 −0
diff --git a/internal/awsbilling/classify.go b/internal/awsbilling/classify.go| new file mode 100644 |
| index 000000000..7d3cb5b46 |
| --- /dev/null |
| +++ b/internal/awsbilling/classify.go |
| @@ -0,0 +1,179 @@ |
| +package awsbilling |
| + |
| +import ( |
| + "math/big" |
| + "regexp" |
| + "strings" |
| + "time" |
| +) |
| + |
| +// Classification reasons. Every line the classifier looks at gets exactly |
| +// one, so an uncounted byte is always accounted for by name. |
| +const ( |
| + // ReasonCharged: an S3 transfer-out line AWS charged for. The only |
| + // reason that counts bytes. |
| + ReasonCharged = "charged" |
| + |
| + // Not transfer at all. Never counted, never a hold. |
| + ReasonNotS3 = "not_s3" |
| + ReasonNotUsage = "not_usage_line" |
| + ReasonNotTransfer = "not_transfer" // storage, requests, retrieval, ... |
| + |
| + // Transfer AWS does not charge the bucket owner for, or charged at |
| + // zero. Never counted, never a hold. |
| + ReasonInbound = "transfer_in" |
| + ReasonCloudFront = "cloudfront_origin" |
| + ReasonFree = "zero_cost" // free tier, first-100-GB allowance, ... |
| + ReasonSubByte = "below_one_byte" // charged, but under one byte after rounding down |
| + |
| + // Charged lines the classifier cannot attribute to a bucket-hour or |
| + // cannot classify. Never counted, and IsHold is true: the month's |
| + // store-transfer meter must not be marked complete over them. |
| + ReasonUnknownTransfer = "unknown_transfer_type" |
| + ReasonNoResource = "no_bucket_resource_id" |
| + ReasonNotHourly = "not_an_hourly_line" |
| + ReasonBadAmount = "non_positive_amount" |
| +) |
| + |
| +// IsHold reports whether a reason describes a CHARGED line the pipeline |
| +// could not count. Such a line means AWS billed transfer that no tenant can |
| +// be shown to owe, so completeness for the month must be withheld until a |
| +// person has looked, rather than quietly billing everyone else and |
| +// absorbing it. |
| +func IsHold(reason string) bool { |
| + switch reason { |
| + case ReasonUnknownTransfer, ReasonNoResource, ReasonNotHourly, ReasonBadAmount: |
| + return true |
| + } |
| + return false |
| +} |
| + |
| +// Usage-type grammar. A region-prefixed type looks like "EUW1-...", while |
| +// us-east-1 historically carries no prefix at all, and eu-west-1 sometimes |
| +// carries the legacy "EU-" prefix, so the prefix is optional and loose. |
| +var ( |
| + regionPrefix = `(?:[A-Z]{2,4}[0-9]?-)?` |
| + // Internet egress and in-region cross-AZ/regional transfer. |
| + reTransferOut = regexp.MustCompile(`^` + regionPrefix + `DataTransfer-(?:Out|Regional)-Bytes$`) |
| + // Inter-region: "<FROM>-<TO>-AWS-Out-Bytes". |
| + reInterRegionOut = regexp.MustCompile(`^[A-Z]{2,4}[0-9]?-[A-Z]{2,4}[0-9]?-AWS-Out-Bytes$`) |
| + reCloudFront = regexp.MustCompile(`^` + regionPrefix + `CloudFront-(?:Out|In)-Bytes$`) |
| +) |
| + |
| +// classifyUsageType sorts an S3 usage type into transfer-out (true, ""), |
| +// or a non-counting reason. |
| +func classifyUsageType(ut string) (out bool, reason string) { |
| + switch { |
| + case reTransferOut.MatchString(ut), reInterRegionOut.MatchString(ut): |
| + return true, "" |
| + case reCloudFront.MatchString(ut): |
| + return false, ReasonCloudFront |
| + case strings.HasSuffix(ut, "-In-Bytes"): |
| + return false, ReasonInbound |
| + case strings.Contains(ut, "Bytes") && (strings.Contains(ut, "Out") || strings.Contains(ut, "DataTransfer")): |
| + // Looks like transfer and matches nothing we know. Not counted; |
| + // the caller turns a CHARGED one into a hold. |
| + return false, ReasonUnknownTransfer |
| + } |
| + return false, ReasonNotTransfer |
| +} |
| + |
| +var gigabyte = big.NewRat(1_000_000_000, 1) |
| + |
| +// ChargedS3Transfer decides whether a CUR line is object-store transfer that |
| +// AWS charged for, and if so, which bucket and how many bytes. |
| +// |
| +// A line counts only when all of these hold: |
| +// |
| +// - the service is AmazonS3; |
| +// - the line type is Usage or DiscountedUsage; |
| +// - the usage type is S3 transfer OUT: internet ("DataTransfer-Out-Bytes"), |
| +// regional ("DataTransfer-Regional-Bytes") or inter-region |
| +// ("<FROM>-<TO>-AWS-Out-Bytes"), never "CloudFront-Out-Bytes", never |
| +// an "-In-Bytes" type, never storage or requests; |
| +// - UnblendedCost is greater than zero; |
| +// - the resource id names a bucket; |
| +// - the line covers exactly one whole UTC hour. |
| +// |
| +// Bytes are UsageAmount (GB) x 1e9, rounded down. When the file carries an |
| +// unblended rate, the counted amount is also capped at cost / rate, so an |
| +// hour that straddled the end of a free allowance counts only the part AWS |
| +// priced. That cap can only lower a count, never raise one. |
| +// |
| +// A zero-cost line is uncharged with a reason and never counted: that is |
| +// the free tier and the CloudFront origin leg, which AWS does not bill. |
| +func ChargedS3Transfer(l Line) (bucket string, bytes int64, charged bool, reason string) { |
| + if l.ServiceCode != ServiceS3 { |
| + return "", 0, false, ReasonNotS3 |
| + } |
| + if !l.IsUsage() { |
| + return "", 0, false, ReasonNotUsage |
| + } |
| + out, why := classifyUsageType(l.UsageType) |
| + if !out && why != ReasonUnknownTransfer { |
| + return "", 0, false, why |
| + } |
| + if l.Cost == nil || l.Cost.Sign() <= 0 { |
| + return "", 0, false, ReasonFree |
| + } |
| + if !out { |
| + return "", 0, false, ReasonUnknownTransfer |
| + } |
| + bucket, ok := bucketFromResource(l.ResourceID) |
| + if !ok { |
| + return "", 0, false, ReasonNoResource |
| + } |
| + if !isWholeHour(l.UsageStart, l.UsageEnd) { |
| + return bucket, 0, false, ReasonNotHourly |
| + } |
| + if l.Amount == nil || l.Amount.Sign() <= 0 { |
| + return bucket, 0, false, ReasonBadAmount |
| + } |
| + gb := new(big.Rat).Set(l.Amount) |
| + if l.Rate != nil && l.Rate.Sign() > 0 { |
| + priced := new(big.Rat).Quo(l.Cost, l.Rate) |
| + if priced.Cmp(gb) < 0 { |
| + gb = priced |
| + } |
| + } |
| + n := floorBytes(gb) |
| + if n <= 0 { |
| + return bucket, 0, false, ReasonSubByte |
| + } |
| + return bucket, n, true, ReasonCharged |
| +} |
| + |
| +// TransferBytes is floor(UsageAmount x 1e9) for a line, whatever its |
| +// classification, for diagnostics on uncharged transfer. |
| +func TransferBytes(l Line) int64 { |
| + if l.Amount == nil || l.Amount.Sign() <= 0 { |
| + return 0 |
| + } |
| + return floorBytes(l.Amount) |
| +} |
| + |
| +func floorBytes(gb *big.Rat) int64 { |
| + x := new(big.Rat).Mul(gb, gigabyte) |
| + q := new(big.Int).Quo(x.Num(), x.Denom()) // both positive: truncation is floor |
| + if !q.IsInt64() { |
| + return 0 |
| + } |
| + return q.Int64() |
| +} |
| + |
| +// bucketFromResource extracts a bucket name from an S3 resource id. CUR |
| +// writes the bare bucket name; an ARN form is accepted too. Anything with a |
| +// path or an unexpected shape is not a bucket. |
| +func bucketFromResource(id string) (string, bool) { |
| + id = strings.TrimPrefix(id, "arn:aws:s3:::") |
| + if id == "" || len(id) > 63 || strings.ContainsAny(id, "/: ") { |
| + return "", false |
| + } |
| + return id, true |
| +} |
| + |
| +// isWholeHour reports whether [start, end) is exactly one UTC hour on the |
| +// hour. |
| +func isWholeHour(start, end time.Time) bool { |
| + return !start.IsZero() && start.Equal(start.Truncate(time.Hour)) && end.Sub(start) == time.Hour |
| +} |
diff --git a/internal/awsbilling/classify_test.go b/internal/awsbilling/classify_test.go| new file mode 100644 |
| index 000000000..4aa888121 |
| --- /dev/null |
| +++ b/internal/awsbilling/classify_test.go |
| @@ -0,0 +1,163 @@ |
| +package awsbilling |
| + |
| +import ( |
| + "math/big" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func rat(t *testing.T, s string) *big.Rat { |
| + t.Helper() |
| + x, err := ParseDecimal(s) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return x |
| +} |
| + |
| +func s3Line(t *testing.T, usageType, amount, cost string) Line { |
| + start := time.Date(2026, 9, 3, 10, 0, 0, 0, time.UTC) |
| + return Line{ |
| + ServiceCode: ServiceS3, |
| + LineItemType: LineTypeUsage, |
| + UsageType: usageType, |
| + ResourceID: "carlos-store-acc1-shop", |
| + UsageStart: start, |
| + UsageEnd: start.Add(time.Hour), |
| + PeriodStart: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC), |
| + Amount: rat(t, amount), |
| + Cost: rat(t, cost), |
| + } |
| +} |
| + |
| +func TestChargedS3TransferTable(t *testing.T) { |
| + type want struct { |
| + charged bool |
| + bytes int64 |
| + reason string |
| + } |
| + cases := []struct { |
| + name string |
| + line func() Line |
| + want want |
| + }{ |
| + {"internet out", func() Line { return s3Line(t, "EUW1-DataTransfer-Out-Bytes", "2.5", "0.225") }, want{true, 2_500_000_000, ReasonCharged}}, |
| + {"regional", func() Line { return s3Line(t, "EUW1-DataTransfer-Regional-Bytes", "1", "0.01") }, want{true, 1_000_000_000, ReasonCharged}}, |
| + {"inter-region", func() Line { return s3Line(t, "EUW1-USE1-AWS-Out-Bytes", "0.5", "0.01") }, want{true, 500_000_000, ReasonCharged}}, |
| + {"us-east-1 has no prefix", func() Line { return s3Line(t, "DataTransfer-Out-Bytes", "1", "0.09") }, want{true, 1_000_000_000, ReasonCharged}}, |
| + {"legacy EU prefix", func() Line { return s3Line(t, "EU-DataTransfer-Out-Bytes", "1", "0.09") }, want{true, 1_000_000_000, ReasonCharged}}, |
| + {"discounted usage counts", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "0.05") |
| + l.LineItemType = LineTypeDiscountedUsage |
| + return l |
| + }, want{true, 1_000_000_000, ReasonCharged}}, |
| + {"rounded down, exactly", func() Line { return s3Line(t, "EUW1-DataTransfer-Out-Bytes", "0.1234567899", "0.01") }, want{true, 123_456_789, ReasonCharged}}, |
| + {"exponent amount", func() Line { return s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1.5E-1", "0.0135") }, want{true, 150_000_000, ReasonCharged}}, |
| + {"free tier is zero cost", func() Line { return s3Line(t, "EUW1-DataTransfer-Out-Bytes", "40", "0") }, want{false, 0, ReasonFree}}, |
| + {"negative cost is not a charge", func() Line { return s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "-0.09") }, want{false, 0, ReasonFree}}, |
| + {"cloudfront origin", func() Line { return s3Line(t, "EUW1-CloudFront-Out-Bytes", "7", "0.02") }, want{false, 0, ReasonCloudFront}}, |
| + {"cloudfront in", func() Line { return s3Line(t, "EUW1-CloudFront-In-Bytes", "7", "0") }, want{false, 0, ReasonCloudFront}}, |
| + {"transfer in", func() Line { return s3Line(t, "EUW1-DataTransfer-In-Bytes", "9", "0.01") }, want{false, 0, ReasonInbound}}, |
| + {"inter-region in", func() Line { return s3Line(t, "EUW1-USE1-AWS-In-Bytes", "9", "0.01") }, want{false, 0, ReasonInbound}}, |
| + {"requests", func() Line { return s3Line(t, "EUW1-Requests-Tier2", "1000", "0.0004") }, want{false, 0, ReasonNotTransfer}}, |
| + {"storage", func() Line { return s3Line(t, "EUW1-TimedStorage-ByteHrs", "0.1", "0.0023") }, want{false, 0, ReasonNotTransfer}}, |
| + {"not s3", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "3", "0.27") |
| + l.ServiceCode = "AmazonEC2" |
| + return l |
| + }, want{false, 0, ReasonNotS3}}, |
| + {"credit line", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "0", "-0.0135") |
| + l.LineItemType = "Credit" |
| + return l |
| + }, want{false, 0, ReasonNotUsage}}, |
| + {"tax line", func() Line { |
| + l := s3Line(t, "", "0", "0.05") |
| + l.LineItemType = "Tax" |
| + return l |
| + }, want{false, 0, ReasonNotUsage}}, |
| + {"charged unknown transfer holds", func() Line { return s3Line(t, "EUW1-S3-Accel-Out-Bytes", "1", "0.04") }, want{false, 0, ReasonUnknownTransfer}}, |
| + {"free unknown transfer does not hold", func() Line { return s3Line(t, "EUW1-S3-Accel-Out-Bytes", "1", "0") }, want{false, 0, ReasonFree}}, |
| + {"charged with no bucket holds", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "0.09") |
| + l.ResourceID = "" |
| + return l |
| + }, want{false, 0, ReasonNoResource}}, |
| + {"charged with a path, not a bucket, holds", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "0.09") |
| + l.ResourceID = "bucket/key" |
| + return l |
| + }, want{false, 0, ReasonNoResource}}, |
| + {"charged daily line holds", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "0.09") |
| + l.UsageEnd = l.UsageStart.Add(24 * time.Hour) |
| + return l |
| + }, want{false, 0, ReasonNotHourly}}, |
| + {"charged off-the-hour line holds", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "0.09") |
| + l.UsageStart = l.UsageStart.Add(30 * time.Minute) |
| + l.UsageEnd = l.UsageStart.Add(time.Hour) |
| + return l |
| + }, want{false, 0, ReasonNotHourly}}, |
| + {"charged zero amount holds", func() Line { return s3Line(t, "EUW1-DataTransfer-Out-Bytes", "0", "0.09") }, want{false, 0, ReasonBadAmount}}, |
| + {"sub-byte charge counts nothing", func() Line { return s3Line(t, "EUW1-DataTransfer-Out-Bytes", "0.0000000001", "0.0000001") }, want{false, 0, ReasonSubByte}}, |
| + {"rate caps a partly free hour", func() Line { |
| + // 10 GB used, but only 4 GB priced at 0.09: the hour crossed |
| + // the end of the free allowance. |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "10", "0.36") |
| + l.Rate = rat(t, "0.09") |
| + return l |
| + }, want{true, 4_000_000_000, ReasonCharged}}, |
| + {"rate never raises a count", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "0.36") |
| + l.Rate = rat(t, "0.09") |
| + return l |
| + }, want{true, 1_000_000_000, ReasonCharged}}, |
| + {"arn resource id", func() Line { |
| + l := s3Line(t, "EUW1-DataTransfer-Out-Bytes", "1", "0.09") |
| + l.ResourceID = "arn:aws:s3:::carlos-store-acc1-shop" |
| + return l |
| + }, want{true, 1_000_000_000, ReasonCharged}}, |
| + } |
| + for _, c := range cases { |
| + bucket, n, charged, reason := ChargedS3Transfer(c.line()) |
| + if charged != c.want.charged || n != c.want.bytes || reason != c.want.reason { |
| + t.Errorf("%s: got charged=%v bytes=%d reason=%q, want %+v", c.name, charged, n, reason, c.want) |
| + } |
| + if charged && bucket != "carlos-store-acc1-shop" { |
| + t.Errorf("%s: bucket = %q", c.name, bucket) |
| + } |
| + if charged && IsHold(reason) { |
| + t.Errorf("%s: a counted line cannot be a hold", c.name) |
| + } |
| + } |
| +} |
| + |
| +func TestIsHold(t *testing.T) { |
| + for _, r := range []string{ReasonUnknownTransfer, ReasonNoResource, ReasonNotHourly, ReasonBadAmount} { |
| + if !IsHold(r) { |
| + t.Errorf("%s must hold", r) |
| + } |
| + } |
| + for _, r := range []string{ReasonCharged, ReasonNotS3, ReasonNotUsage, ReasonNotTransfer, ReasonInbound, ReasonCloudFront, ReasonFree, ReasonSubByte} { |
| + if IsHold(r) { |
| + t.Errorf("%s must not hold", r) |
| + } |
| + } |
| +} |
| + |
| +// Summing many sub-cent lines must stay exact: 1000 lines of 0.001 GB is |
| +// exactly 1 GB, which float accumulation famously misses. |
| +func TestChargedBytesSumExactly(t *testing.T) { |
| + var total int64 |
| + for i := 0; i < 1000; i++ { |
| + _, n, charged, _ := ChargedS3Transfer(s3Line(t, "EUW1-DataTransfer-Out-Bytes", "0.001", "0.00009")) |
| + if !charged { |
| + t.Fatal("line must count") |
| + } |
| + total += n |
| + } |
| + if total != 1_000_000_000 { |
| + t.Fatalf("total = %d, want exactly 1e9", total) |
| + } |
| +} |
diff --git a/internal/awsbilling/cur.go b/internal/awsbilling/cur.go| new file mode 100644 |
| index 000000000..cdd4922e5 |
| --- /dev/null |
| +++ b/internal/awsbilling/cur.go |
| @@ -0,0 +1,381 @@ |
| +// Package awsbilling reads the AWS Cost and Usage Report (CUR 2.0, delivered |
| +// by AWS Data Exports as CSV into a bucket the platform owns) and answers |
| +// one question from it: how many bytes of object-store transfer did AWS |
| +// actually charge for, per bucket, per hour. |
| +// |
| +// It exists because of a commercial ruling (Paul, 2026-09-30, superseding |
| +// the pricing spec's 2026-09-10 "best effort" section): object-storage |
| +// transfer is billed to a tenant ONLY when AWS charged Team Tito for it. |
| +// S3 server access logs record delivered bytes but cannot tell a charged |
| +// byte from a free one (free tier, same-region EC2, CloudFront origin |
| +// fetches), and they may omit records permanently. The CUR is AWS's own |
| +// statement of what it charges, so it is the only billing-grade source. |
| +// |
| +// Everything here is pure: no AWS SDK, no network. The Source seam is the |
| +// only way bytes come in, and tests drive it with fixture CSVs. |
| +// |
| +// Arithmetic is exact. Amounts and costs are parsed as math/big.Rat and |
| +// never pass through a float, and bytes are GB x 1e9 (AWS's decimal GB) |
| +// rounded down per line, which errs in the customer's favour. |
| +package awsbilling |
| + |
| +import ( |
| + "bufio" |
| + "compress/gzip" |
| + "encoding/csv" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "math/big" |
| + "strings" |
| + "time" |
| + "unicode" |
| +) |
| + |
| +// Normalised column names. CUR 2.0 spells them in snake_case already; the |
| +// legacy CUR spells them "lineItem/UsageType". NormaliseHeader maps both to |
| +// these. |
| +const ( |
| + ColResourceID = "line_item_resource_id" |
| + ColUsageType = "line_item_usage_type" |
| + ColUsageAmount = "line_item_usage_amount" |
| + ColUnblendedCost = "line_item_unblended_cost" |
| + ColUnblendedRate = "line_item_unblended_rate" // optional |
| + ColLineItemType = "line_item_line_item_type" |
| + ColDescription = "line_item_line_item_description" // optional |
| + ColUsageStart = "line_item_usage_start_date" |
| + ColUsageEnd = "line_item_usage_end_date" |
| + ColInvoiceID = "bill_invoice_id" |
| + ColLineItemID = "identity_line_item_id" |
| + ColTimeInterval = "identity_time_interval" // optional |
| + ColServiceCode = "product_servicecode" |
| + ColProductCode = "line_item_product_code" // accepted when servicecode is absent |
| + ColPeriodStart = "bill_billing_period_start_date" |
| + ColPeriodEnd = "bill_billing_period_end_date" // optional |
| + ColUsageAccountID = "line_item_usage_account_id" |
| + ColPayerAccountID = "bill_payer_account_id" // optional |
| +) |
| + |
| +// requiredColumns must each appear exactly once in a file's header. The |
| +// service column is checked separately: product_servicecode is preferred, |
| +// and line_item_product_code (which both CUR generations carry) is accepted |
| +// in its place, because AWS has moved product columns between generations. |
| +var requiredColumns = []string{ |
| + ColResourceID, ColUsageType, ColUsageAmount, ColUnblendedCost, |
| + ColLineItemType, ColUsageStart, ColUsageEnd, ColInvoiceID, |
| + ColLineItemID, ColPeriodStart, ColUsageAccountID, |
| +} |
| + |
| +var optionalColumns = []string{ |
| + ColUnblendedRate, ColDescription, ColTimeInterval, ColServiceCode, |
| + ColProductCode, ColPeriodEnd, ColPayerAccountID, |
| +} |
| + |
| +// Line types the classifier counts. Everything else (Tax, Credit, Refund, |
| +// Fee, RIFee, SavingsPlan*, BundledDiscount, ...) is an adjustment to money, |
| +// not a measurement of transfer, and is never counted. |
| +const ( |
| + LineTypeUsage = "Usage" |
| + LineTypeDiscountedUsage = "DiscountedUsage" |
| +) |
| + |
| +// ServiceS3 is the CUR service code for Amazon S3. |
| +const ServiceS3 = "AmazonS3" |
| + |
| +// Line is one CUR row, parsed. Amount and Cost are never nil (an empty cell |
| +// on a non-usage line reads as zero); Rate is nil when the file carries no |
| +// rate column or the cell is empty. |
| +type Line struct { |
| + File string // the object key the line came from, for diagnostics |
| + Row int // 1-based data row within the file |
| + |
| + ServiceCode string |
| + LineItemType string |
| + UsageType string |
| + ResourceID string |
| + UsageAccountID string |
| + PayerAccountID string |
| + InvoiceID string |
| + LineItemID string |
| + TimeInterval string |
| + Description string |
| + |
| + UsageStart, UsageEnd time.Time |
| + PeriodStart time.Time |
| + PeriodEnd time.Time // zero when the column is absent |
| + |
| + Amount *big.Rat // GB for transfer usage types |
| + Cost *big.Rat |
| + Rate *big.Rat |
| +} |
| + |
| +// IsUsage reports whether the line measures usage rather than adjusting |
| +// money. |
| +func (l Line) IsUsage() bool { |
| + return l.LineItemType == LineTypeUsage || l.LineItemType == LineTypeDiscountedUsage |
| +} |
| + |
| +// ErrMalformed marks a file the reader refuses. A refused file refuses its |
| +// whole delivery: a partly read report is not evidence of anything. |
| +var ErrMalformed = errors.New("awsbilling: malformed report") |
| + |
| +func malformed(format string, args ...any) error { |
| + return fmt.Errorf("%w: %s", ErrMalformed, fmt.Sprintf(format, args...)) |
| +} |
| + |
| +// NormaliseHeader maps a CUR column header to its snake_case form: |
| +// "lineItem/UsageType" -> "line_item_usage_type", "bill/InvoiceId" -> |
| +// "bill_invoice_id", "product/servicecode" -> "product_servicecode". A |
| +// header already in snake_case (CUR 2.0) comes back lower-cased and |
| +// otherwise unchanged. A UTF-8 byte-order mark is stripped. |
| +func NormaliseHeader(h string) string { |
| + h = strings.TrimSpace(strings.TrimPrefix(h, "\ufeff")) |
| + parts := strings.Split(h, "/") |
| + for i, p := range parts { |
| + parts[i] = camelToSnake(p) |
| + } |
| + return strings.Join(parts, "_") |
| +} |
| + |
| +func camelToSnake(s string) string { |
| + rs := []rune(s) |
| + var b strings.Builder |
| + for i, r := range rs { |
| + if unicode.IsUpper(r) && i > 0 { |
| + prev := rs[i-1] |
| + nextLower := i+1 < len(rs) && unicode.IsLower(rs[i+1]) |
| + if unicode.IsLower(prev) || unicode.IsDigit(prev) || (unicode.IsUpper(prev) && nextLower) { |
| + b.WriteByte('_') |
| + } |
| + } |
| + b.WriteRune(unicode.ToLower(r)) |
| + } |
| + return b.String() |
| +} |
| + |
| +// ParseDecimal parses a CUR numeric cell exactly. It accepts plain and |
| +// exponent decimal notation ("0.0123", "1.5E-7") and refuses anything else, |
| +// including the "a/b" fraction form big.Rat would otherwise accept, NaN and |
| +// infinities. |
| +func ParseDecimal(s string) (*big.Rat, error) { |
| + s = strings.TrimSpace(s) |
| + if s == "" { |
| + return nil, fmt.Errorf("empty number") |
| + } |
| + for _, r := range s { |
| + switch { |
| + case r >= '0' && r <= '9', r == '.', r == 'e', r == 'E', r == '+', r == '-': |
| + default: |
| + return nil, fmt.Errorf("%q is not a decimal number", s) |
| + } |
| + } |
| + x, ok := new(big.Rat).SetString(s) |
| + if !ok { |
| + return nil, fmt.Errorf("%q is not a decimal number", s) |
| + } |
| + return x, nil |
| +} |
| + |
| +var timeLayouts = []string{ |
| + time.RFC3339Nano, |
| + "2006-01-02T15:04:05Z", |
| + "2006-01-02T15:04:05.000Z", |
| + "2006-01-02 15:04:05", |
| + "2006-01-02 15:04:05.000", |
| + "2006-01-02T15:04:05", |
| + "20060102T150405.000Z", |
| + "20060102T150405Z", |
| +} |
| + |
| +// ParseTime parses a CUR timestamp. CUR timestamps are UTC; a layout with |
| +// no zone is read as UTC, and an explicit offset is converted to UTC. |
| +func ParseTime(s string) (time.Time, error) { |
| + s = strings.TrimSpace(s) |
| + for _, layout := range timeLayouts { |
| + if t, err := time.Parse(layout, s); err == nil { |
| + return t.UTC(), nil |
| + } |
| + } |
| + return time.Time{}, fmt.Errorf("%q is not a CUR timestamp", s) |
| +} |
| + |
| +// OpenReport returns a reader over a report file's CSV text, transparently |
| +// decompressing gzip (detected by its magic bytes, not by the key name). |
| +func OpenReport(r io.Reader) (io.Reader, error) { |
| + br := bufio.NewReader(r) |
| + magic, err := br.Peek(2) |
| + if err == nil && magic[0] == 0x1f && magic[1] == 0x8b { |
| + zr, err := gzip.NewReader(br) |
| + if err != nil { |
| + return nil, malformed("gzip: %v", err) |
| + } |
| + return zr, nil |
| + } |
| + return br, nil |
| +} |
| + |
| +// ReadLines parses one report file and calls fn for every data row, in |
| +// order. It is header-driven and tolerant of columns it does not use, and |
| +// strict on the ones it does: |
| +// |
| +// - every required column must be present exactly once (a used column |
| +// appearing twice after normalisation is ambiguous and refused); |
| +// - every amount and cost must parse exactly; an empty cell is accepted |
| +// only on a line that is not Usage or DiscountedUsage; |
| +// - the billing period start must parse on every line; |
| +// - a Usage or DiscountedUsage line must sit wholly inside its billing |
| +// period, with a start before its end. |
| +// |
| +// The window check is scoped to usage lines on purpose: a Refund or Credit |
| +// may legitimately carry dates from an earlier period, and refusing the |
| +// file over one would hold the month forever for a line that is never |
| +// counted anyway. |
| +// |
| +// Any violation refuses the file. A caller must treat an error as "this |
| +// file, and so this delivery, proves nothing", whatever fn had already |
| +// been given. |
| +func ReadLines(name string, r io.Reader, fn func(Line) error) error { |
| + text, err := OpenReport(r) |
| + if err != nil { |
| + return err |
| + } |
| + cr := csv.NewReader(text) |
| + cr.ReuseRecord = true |
| + header, err := cr.Read() |
| + if err != nil { |
| + if errors.Is(err, io.EOF) { |
| + return malformed("%s: empty file", name) |
| + } |
| + return malformed("%s: header: %v", name, err) |
| + } |
| + idx := map[string]int{} |
| + dup := map[string]bool{} |
| + for i, h := range header { |
| + n := NormaliseHeader(h) |
| + if _, seen := idx[n]; seen { |
| + dup[n] = true |
| + } |
| + idx[n] = i |
| + } |
| + for _, c := range append(append([]string{}, requiredColumns...), optionalColumns...) { |
| + if dup[c] { |
| + return malformed("%s: column %s appears more than once", name, c) |
| + } |
| + } |
| + for _, c := range requiredColumns { |
| + if _, ok := idx[c]; !ok { |
| + return malformed("%s: missing required column %s", name, c) |
| + } |
| + } |
| + _, haveService := idx[ColServiceCode] |
| + _, haveProduct := idx[ColProductCode] |
| + if !haveService && !haveProduct { |
| + return malformed("%s: missing required column %s (or %s)", name, ColServiceCode, ColProductCode) |
| + } |
| + get := func(rec []string, col string) string { |
| + i, ok := idx[col] |
| + if !ok || i >= len(rec) { |
| + return "" |
| + } |
| + return strings.TrimSpace(rec[i]) |
| + } |
| + row := 0 |
| + for { |
| + rec, err := cr.Read() |
| + if errors.Is(err, io.EOF) { |
| + return nil |
| + } |
| + if err != nil { |
| + return malformed("%s: row %d: %v", name, row+1, err) |
| + } |
| + row++ |
| + l := Line{ |
| + File: name, |
| + Row: row, |
| + LineItemType: get(rec, ColLineItemType), |
| + UsageType: get(rec, ColUsageType), |
| + ResourceID: get(rec, ColResourceID), |
| + UsageAccountID: get(rec, ColUsageAccountID), |
| + PayerAccountID: get(rec, ColPayerAccountID), |
| + InvoiceID: get(rec, ColInvoiceID), |
| + LineItemID: get(rec, ColLineItemID), |
| + TimeInterval: get(rec, ColTimeInterval), |
| + Description: get(rec, ColDescription), |
| + } |
| + l.ServiceCode = get(rec, ColServiceCode) |
| + if l.ServiceCode == "" { |
| + l.ServiceCode = get(rec, ColProductCode) |
| + } |
| + usage := l.IsUsage() |
| + if l.Amount, err = parseCell(get(rec, ColUsageAmount), usage); err != nil { |
| + return malformed("%s: row %d: usage amount: %v", name, row, err) |
| + } |
| + if l.Cost, err = parseCell(get(rec, ColUnblendedCost), usage); err != nil { |
| + return malformed("%s: row %d: unblended cost: %v", name, row, err) |
| + } |
| + if raw := get(rec, ColUnblendedRate); raw != "" { |
| + if l.Rate, err = ParseDecimal(raw); err != nil { |
| + return malformed("%s: row %d: unblended rate: %v", name, row, err) |
| + } |
| + } |
| + if l.PeriodStart, err = ParseTime(get(rec, ColPeriodStart)); err != nil { |
| + return malformed("%s: row %d: billing period start: %v", name, row, err) |
| + } |
| + if raw := get(rec, ColPeriodEnd); raw != "" { |
| + if l.PeriodEnd, err = ParseTime(raw); err != nil { |
| + return malformed("%s: row %d: billing period end: %v", name, row, err) |
| + } |
| + } |
| + if err := parseWindow(&l, get(rec, ColUsageStart), get(rec, ColUsageEnd), usage); err != nil { |
| + return malformed("%s: row %d: %v", name, row, err) |
| + } |
| + if err := fn(l); err != nil { |
| + return err |
| + } |
| + } |
| +} |
| + |
| +// parseCell parses a numeric cell. Empty is zero, but only where strict is |
| +// false: a usage line with no amount or cost is not a measurement. |
| +func parseCell(raw string, strict bool) (*big.Rat, error) { |
| + if raw == "" { |
| + if strict { |
| + return nil, fmt.Errorf("empty on a usage line") |
| + } |
| + return new(big.Rat), nil |
| + } |
| + return ParseDecimal(raw) |
| +} |
| + |
| +// parseWindow parses the usage window and, for usage lines, requires it to |
| +// lie inside the billing period. The period's end is the explicit column |
| +// when present, else the start plus one calendar month (AWS bills calendar |
| +// months). |
| +func parseWindow(l *Line, rawStart, rawEnd string, usage bool) error { |
| + var err error |
| + if rawStart == "" && rawEnd == "" && !usage { |
| + return nil |
| + } |
| + if l.UsageStart, err = ParseTime(rawStart); err != nil { |
| + return fmt.Errorf("usage start: %v", err) |
| + } |
| + if l.UsageEnd, err = ParseTime(rawEnd); err != nil { |
| + return fmt.Errorf("usage end: %v", err) |
| + } |
| + if !usage { |
| + return nil |
| + } |
| + if !l.UsageStart.Before(l.UsageEnd) { |
| + return fmt.Errorf("usage window %s..%s is empty or reversed", l.UsageStart.Format(time.RFC3339), l.UsageEnd.Format(time.RFC3339)) |
| + } |
| + periodEnd := l.PeriodEnd |
| + if periodEnd.IsZero() { |
| + periodEnd = l.PeriodStart.AddDate(0, 1, 0) |
| + } |
| + if l.UsageStart.Before(l.PeriodStart) || l.UsageEnd.After(periodEnd) { |
| + return fmt.Errorf("usage window %s..%s lies outside its billing period %s..%s", |
| + l.UsageStart.Format(time.RFC3339), l.UsageEnd.Format(time.RFC3339), |
| + l.PeriodStart.Format(time.RFC3339), periodEnd.Format(time.RFC3339)) |
| + } |
| + return nil |
| +} |
diff --git a/internal/awsbilling/cur_test.go b/internal/awsbilling/cur_test.go| new file mode 100644 |
| index 000000000..65311c796 |
| --- /dev/null |
| +++ b/internal/awsbilling/cur_test.go |
| @@ -0,0 +1,153 @@ |
| +package awsbilling |
| + |
| +import ( |
| + "bytes" |
| + "compress/gzip" |
| + "errors" |
| + "math/big" |
| + "os" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func readFixture(t *testing.T, name string) []byte { |
| + t.Helper() |
| + b, err := os.ReadFile("testdata/" + name) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return b |
| +} |
| + |
| +func collect(t *testing.T, name string, body []byte) ([]Line, error) { |
| + t.Helper() |
| + var out []Line |
| + err := ReadLines(name, bytes.NewReader(body), func(l Line) error { |
| + out = append(out, l) |
| + return nil |
| + }) |
| + return out, err |
| +} |
| + |
| +func TestNormaliseHeader(t *testing.T) { |
| + cases := map[string]string{ |
| + "lineItem/UsageType": ColUsageType, |
| + "lineItem/ResourceId": ColResourceID, |
| + "lineItem/UsageAmount": ColUsageAmount, |
| + "lineItem/UnblendedCost": ColUnblendedCost, |
| + "lineItem/LineItemType": ColLineItemType, |
| + "lineItem/UsageStartDate": ColUsageStart, |
| + "lineItem/UsageEndDate": ColUsageEnd, |
| + "lineItem/UsageAccountId": ColUsageAccountID, |
| + "bill/InvoiceId": ColInvoiceID, |
| + "bill/BillingPeriodStartDate": ColPeriodStart, |
| + "bill/PayerAccountId": ColPayerAccountID, |
| + "identity/LineItemId": ColLineItemID, |
| + "identity/TimeInterval": ColTimeInterval, |
| + "product/servicecode": ColServiceCode, |
| + "lineItem/ProductCode": ColProductCode, |
| + "\ufeffidentity_line_item_id": ColLineItemID, |
| + "line_item_usage_type": ColUsageType, |
| + " bill_invoice_id ": ColInvoiceID, |
| + } |
| + for in, want := range cases { |
| + if got := NormaliseHeader(in); got != want { |
| + t.Errorf("NormaliseHeader(%q) = %q, want %q", in, got, want) |
| + } |
| + } |
| +} |
| + |
| +// Both spellings of the same report must read identically, and an extra |
| +// column neither generation promises (resource tags) must not matter. |
| +func TestReadLinesBothGenerations(t *testing.T) { |
| + cur2, err := collect(t, "cur2", readFixture(t, "cur2-2026-09.csv")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(cur2) != 10 { |
| + t.Fatalf("cur2 rows = %d, want 10", len(cur2)) |
| + } |
| + l := cur2[0] |
| + if l.ServiceCode != ServiceS3 || l.UsageType != "EUW1-DataTransfer-Out-Bytes" || l.ResourceID != "carlos-store-acc1-shop" || |
| + l.InvoiceID != "EUINIE26-0001" || l.LineItemID != "id-internet" || l.UsageAccountID != "271376211898" { |
| + t.Fatalf("first line misread: %+v", l) |
| + } |
| + if l.Amount.Cmp(big.NewRat(5, 2)) != 0 || l.Cost.Cmp(big.NewRat(225, 1000)) != 0 { |
| + t.Fatalf("amount/cost = %s/%s", l.Amount.RatString(), l.Cost.RatString()) |
| + } |
| + if !l.UsageStart.Equal(time.Date(2026, 9, 3, 10, 0, 0, 0, time.UTC)) { |
| + t.Fatalf("start = %s", l.UsageStart) |
| + } |
| + tax := cur2[9] |
| + if tax.LineItemType != "Tax" || tax.Amount.Sign() != 0 { |
| + t.Fatalf("tax line: %+v", tax) |
| + } |
| + |
| + legacy, err := collect(t, "legacy", readFixture(t, "legacy-2026-09.csv")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(legacy) != 2 { |
| + t.Fatalf("legacy rows = %d", len(legacy)) |
| + } |
| + if legacy[0].Amount.Cmp(big.NewRat(15, 100)) != 0 { |
| + t.Fatalf("exponent amount = %s, want 3/20", legacy[0].Amount.RatString()) |
| + } |
| + if legacy[0].InvoiceID != "" { |
| + t.Fatal("an unissued invoice id must read as empty") |
| + } |
| +} |
| + |
| +func TestReadLinesGzip(t *testing.T) { |
| + var buf bytes.Buffer |
| + zw := gzip.NewWriter(&buf) |
| + zw.Write(readFixture(t, "cur2-2026-09.csv")) |
| + zw.Close() |
| + lines, err := collect(t, "cur2.csv.gz", buf.Bytes()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(lines) != 10 { |
| + t.Fatalf("gzip rows = %d, want 10", len(lines)) |
| + } |
| +} |
| + |
| +func TestReadLinesRefusals(t *testing.T) { |
| + good := string(readFixture(t, "cur2-2026-09.csv")) |
| + header, rows, _ := strings.Cut(good, "\n") |
| + first, _, _ := strings.Cut(rows, "\n") |
| + cases := map[string]string{ |
| + "missing required column": strings.Replace(header, "bill_invoice_id", "bill_something_else", 1) + "\n" + first + "\n", |
| + "missing service column": strings.Replace(strings.Replace(header, "product_servicecode", "product_x", 1), |
| + "line_item_product_code", "line_item_y", 1) + "\n" + first + "\n", |
| + "duplicate used column": strings.Replace(header, "resource_tags_user_app", "line_item_usage_type", 1) + "\n" + first + "\n", |
| + "unparsable amount": header + "\n" + strings.Replace(first, ",2.5,", ",2.5GB,", 1) + "\n", |
| + "fraction amount": header + "\n" + strings.Replace(first, ",2.5,", ",5/2,", 1) + "\n", |
| + "empty usage amount": header + "\n" + strings.Replace(first, ",2.5,", ",,", 1) + "\n", |
| + "unparsable cost": header + "\n" + strings.Replace(first, ",0.225,", ",NaN,", 1) + "\n", |
| + "hour before period": header + "\n" + strings.Replace(first, "Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z", "Usage,2026-08-31T23:00:00Z,2026-09-01T00:00:00Z", 1) + "\n", |
| + "hour after period": header + "\n" + strings.Replace(first, "Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z", "Usage,2026-10-01T00:00:00Z,2026-10-01T01:00:00Z", 1) + "\n", |
| + "reversed window": header + "\n" + strings.Replace(first, "Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z", "Usage,2026-09-03T11:00:00Z,2026-09-03T10:00:00Z", 1) + "\n", |
| + "unparsable period start": header + "\n" + strings.Replace(first, "EUINIE26-0001,2026-09-01T00:00:00Z", "EUINIE26-0001,September", 1) + "\n", |
| + "short row": header + "\n" + "id-x,a,b\n", |
| + "empty file": "", |
| + "unparsable usage start": header + "\n" + strings.Replace(first, "Usage,2026-09-03T10:00:00Z", "Usage,yesterday", 1) + "\n", |
| + "unparsable rate on a line": header + "\n" + strings.Replace(first, ",0.09,0.225,", ",cheap,0.225,", 1) + "\n", |
| + } |
| + for name, body := range cases { |
| + if _, err := collect(t, name, []byte(body)); !errors.Is(err, ErrMalformed) { |
| + t.Errorf("%s: err = %v, want ErrMalformed", name, err) |
| + } |
| + } |
| +} |
| + |
| +func TestParseTimeLayouts(t *testing.T) { |
| + want := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) |
| + for _, s := range []string{"2026-09-01T00:00:00Z", "2026-09-01T00:00:00.000Z", "2026-09-01 00:00:00", "20260901T000000.000Z", "2026-09-01T01:00:00+01:00"} { |
| + got, err := ParseTime(s) |
| + if err != nil || !got.Equal(want) || got.Location() != time.UTC { |
| + t.Errorf("ParseTime(%q) = %v, %v", s, got, err) |
| + } |
| + } |
| +} |
diff --git a/internal/awsbilling/delivery.go b/internal/awsbilling/delivery.go| new file mode 100644 |
| index 000000000..a041ce0e3 |
| --- /dev/null |
| +++ b/internal/awsbilling/delivery.go |
| @@ -0,0 +1,347 @@ |
| +package awsbilling |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "sort" |
| + "strings" |
| + "time" |
| +) |
| + |
| +// Source is the read-only seam over the report bucket. release.Store |
| +// satisfies it; tests use an in-memory map. Get returns the body and its |
| +// ETag. |
| +type Source interface { |
| + List(ctx context.Context, prefix string) ([]string, error) |
| + Get(ctx context.Context, key string) ([]byte, string, error) |
| +} |
| + |
| +// ErrNoDelivery means no manifest exists yet for the period: AWS has not |
| +// delivered it, or the export is not configured. It is a hold, never a zero. |
| +var ErrNoDelivery = errors.New("awsbilling: no report delivery for the period") |
| + |
| +// ErrAmbiguousDelivery means several manifests exist for the period and |
| +// their order cannot be established from their keys. Picking one would be a |
| +// guess, and summing them would count restatements twice. |
| +var ErrAmbiguousDelivery = errors.New("awsbilling: cannot tell which delivery is newest") |
| + |
| +// ErrDeliveryChanged means the manifest changed while its files were being |
| +// read, so the files may belong to two different deliveries. |
| +var ErrDeliveryChanged = errors.New("awsbilling: delivery changed during the read") |
| + |
| +// ErrConflictingDuplicate means one line identity appeared twice in a |
| +// delivery with different values. A delivery that disagrees with itself |
| +// proves nothing. |
| +var ErrConflictingDuplicate = errors.New("awsbilling: a line id repeats with different values") |
| + |
| +// FileEvidence pins one data file as it was read. |
| +type FileEvidence struct { |
| + Key string `json:"key"` |
| + ETag string `json:"etag,omitempty"` |
| + SHA256 string `json:"sha256"` |
| + Rows int `json:"rows"` |
| +} |
| + |
| +// Delivery is the newest CUR delivery for one billing period, as read. |
| +// |
| +// Only the newest delivery is authoritative: AWS redelivers a period many |
| +// times as it restates it, and an earlier delivery is superseded, never |
| +// summed. Final means every S3 line in it carries a bill/InvoiceId, which |
| +// AWS fills in only once it has issued the invoice for the month; before |
| +// that the numbers are estimates and may still move. |
| +type Delivery struct { |
| + Period time.Time `json:"period"` // first instant of the month, UTC |
| + AssemblyID string `json:"assembly_id"` |
| + ManifestKey string `json:"manifest_key"` |
| + ManifestETag string `json:"manifest_etag,omitempty"` |
| + ManifestSHA256 string `json:"manifest_sha256"` |
| + Files []FileEvidence `json:"files"` |
| + Final bool `json:"final"` |
| + // NotFinal names why Final is false: "no_s3_lines" or |
| + // "invoice_not_issued". Empty when Final. |
| + NotFinal string `json:"not_final,omitempty"` |
| + // S3Lines counts distinct S3 lines after deduplication; Duplicates |
| + // counts identical repeats that were dropped. |
| + S3Lines int `json:"s3_lines"` |
| + Duplicates int `json:"duplicates,omitempty"` |
| +} |
| + |
| +// Manifest is the part of a CUR manifest the reader needs. CUR 2.0 (Data |
| +// Exports) spells the delivery id executionId and the files dataFiles, as |
| +// s3:// URIs; the legacy CUR spells them assemblyId and reportKeys, as bare |
| +// keys. Both are accepted. |
| +type Manifest struct { |
| + ExecutionID string `json:"executionId"` |
| + AssemblyID string `json:"assemblyId"` |
| + DataFiles []string `json:"dataFiles"` |
| + ReportKeys []string `json:"reportKeys"` |
| + BillingPeriod struct { |
| + Start string `json:"start"` |
| + End string `json:"end"` |
| + } `json:"billingPeriod"` |
| +} |
| + |
| +// PeriodLabel is the CUR 2.0 partition label for a month: "2026-09". |
| +func PeriodLabel(period time.Time) string { return period.UTC().Format("2006-01") } |
| + |
| +// MetadataPrefix is where a Data Export writes a period's manifests, under |
| +// the export's own prefix (which should end in "/"). |
| +func MetadataPrefix(prefix string, period time.Time) string { |
| + return prefix + "metadata/BILLING_PERIOD=" + PeriodLabel(period) + "/" |
| +} |
| + |
| +// DataPrefix is where a Data Export writes a period's data files. |
| +func DataPrefix(prefix string, period time.Time) string { |
| + return prefix + "data/BILLING_PERIOD=" + PeriodLabel(period) + "/" |
| +} |
| + |
| +// deliveryStampLen is the length of the "20060102T150405Z" stamp a |
| +// versioned ("create new export file") delivery directory starts with. |
| +const deliveryStampLen = len("20060102T150405Z") |
| + |
| +// newestManifest picks the authoritative manifest among a period's |
| +// metadata keys. |
| +// |
| +// One manifest (the overwrite layout) is the delivery. Several (the |
| +// versioned layout, one directory per delivery) are ordered by the UTC |
| +// stamp their directory name starts with. A key whose directory carries no |
| +// parseable stamp, or two deliveries with the same stamp, make the order |
| +// unknowable, and the answer is ErrAmbiguousDelivery rather than a guess. |
| +func newestManifest(metaPrefix string, keys []string) (string, error) { |
| + var manifests []string |
| + for _, k := range keys { |
| + if strings.HasSuffix(k, "Manifest.json") && strings.HasPrefix(k, metaPrefix) { |
| + manifests = append(manifests, k) |
| + } |
| + } |
| + switch len(manifests) { |
| + case 0: |
| + return "", ErrNoDelivery |
| + case 1: |
| + return manifests[0], nil |
| + } |
| + type stamped struct { |
| + key string |
| + at time.Time |
| + } |
| + var all []stamped |
| + seen := map[time.Time]bool{} |
| + for _, k := range manifests { |
| + rest := strings.TrimPrefix(k, metaPrefix) |
| + dir, _, ok := strings.Cut(rest, "/") |
| + if !ok || len(dir) < deliveryStampLen { |
| + return "", fmt.Errorf("%w: %s has no delivery stamp", ErrAmbiguousDelivery, k) |
| + } |
| + at, err := time.Parse("20060102T150405Z", dir[:deliveryStampLen]) |
| + if err != nil { |
| + return "", fmt.Errorf("%w: %s has no delivery stamp", ErrAmbiguousDelivery, k) |
| + } |
| + if seen[at] { |
| + return "", fmt.Errorf("%w: two deliveries stamped %s", ErrAmbiguousDelivery, dir[:deliveryStampLen]) |
| + } |
| + seen[at] = true |
| + all = append(all, stamped{k, at}) |
| + } |
| + sort.Slice(all, func(i, j int) bool { return all[i].at.Before(all[j].at) }) |
| + return all[len(all)-1].key, nil |
| +} |
| + |
| +func digest(b []byte) string { |
| + sum := sha256.Sum256(b) |
| + return hex.EncodeToString(sum[:]) |
| +} |
| + |
| +// dataFileKey turns a manifest's file reference into a key in the report |
| +// bucket, and refuses one outside the period's data prefix: a manifest must |
| +// not be able to steer the reader at arbitrary objects. |
| +func dataFileKey(ref, dataPrefix string) (string, error) { |
| + key := ref |
| + if strings.HasPrefix(ref, "s3://") { |
| + rest := strings.TrimPrefix(ref, "s3://") |
| + _, k, ok := strings.Cut(rest, "/") |
| + if !ok { |
| + return "", fmt.Errorf("%w: manifest file %q has no key", ErrMalformed, ref) |
| + } |
| + key = k |
| + } |
| + if !strings.HasPrefix(key, dataPrefix) || strings.Contains(key, "..") { |
| + return "", fmt.Errorf("%w: manifest file %q lies outside %s", ErrMalformed, ref, dataPrefix) |
| + } |
| + return key, nil |
| +} |
| + |
| +// lineIdentity is the deduplication key. AWS documents identity/LineItemId |
| +// as unique within a partition, not across a whole delivery, so the usage |
| +// window rides along: the same id for two different hours is two lines, |
| +// the same id for the same hour is one line delivered twice. |
| +func lineIdentity(l Line) string { |
| + ti := l.TimeInterval |
| + if ti == "" { |
| + ti = l.UsageStart.Format(time.RFC3339) + "/" + l.UsageEnd.Format(time.RFC3339) |
| + } |
| + return l.LineItemID + "|" + ti |
| +} |
| + |
| +// lineFingerprint covers every value the pipeline reads from a line, so two |
| +// rows with one identity are either identical (a duplicate, dropped) or in |
| +// conflict (the delivery is refused). |
| +func lineFingerprint(l Line) string { |
| + rate := "" |
| + if l.Rate != nil { |
| + rate = l.Rate.RatString() |
| + } |
| + return digest([]byte(strings.Join([]string{ |
| + l.ServiceCode, l.LineItemType, l.UsageType, l.ResourceID, |
| + l.UsageAccountID, l.InvoiceID, |
| + l.UsageStart.Format(time.RFC3339Nano), l.UsageEnd.Format(time.RFC3339Nano), |
| + l.PeriodStart.Format(time.RFC3339Nano), |
| + l.Amount.RatString(), l.Cost.RatString(), rate, |
| + }, "\x00"))) |
| +} |
| + |
| +// ReadDelivery reads the newest delivery for period (the first instant of a |
| +// UTC month) under prefix, and calls fn once for every distinct AmazonS3 |
| +// line in it. Lines of other services are read and checked, but not passed |
| +// on. |
| +// |
| +// The returned Delivery pins the manifest and every file by digest, so a |
| +// billing record can name exactly what it was computed from. |
| +// |
| +// Refusals, each of which proves nothing and must hold the month: |
| +// - no manifest (ErrNoDelivery), or no way to order several |
| +// (ErrAmbiguousDelivery); |
| +// - a manifest whose billing period is not exactly the month, with no |
| +// delivery id, or with no files; |
| +// - a file the line reader refuses (ErrMalformed), or a line whose |
| +// billing period is not this one; |
| +// - an S3 line with no line item id; |
| +// - one identity carrying two different lines (ErrConflictingDuplicate); |
| +// - a manifest that changed while its files were read |
| +// (ErrDeliveryChanged). In the overwrite layout AWS replaces files in |
| +// place, so the manifest is read again after the files and must be |
| +// byte-identical, or the files may straddle two deliveries. |
| +// |
| +// fn may be called before a refusal is discovered. A caller must discard |
| +// everything it accumulated when ReadDelivery returns an error. |
| +func ReadDelivery(ctx context.Context, src Source, prefix string, period time.Time, fn func(Line) error) (*Delivery, error) { |
| + period = period.UTC() |
| + if !period.Equal(time.Date(period.Year(), period.Month(), 1, 0, 0, 0, 0, time.UTC)) { |
| + return nil, fmt.Errorf("awsbilling: period %s is not the start of a month", period.Format(time.RFC3339)) |
| + } |
| + periodEnd := period.AddDate(0, 1, 0) |
| + metaPrefix := MetadataPrefix(prefix, period) |
| + keys, err := src.List(ctx, metaPrefix) |
| + if err != nil { |
| + return nil, fmt.Errorf("awsbilling: list %s: %w", metaPrefix, err) |
| + } |
| + manifestKey, err := newestManifest(metaPrefix, keys) |
| + if err != nil { |
| + return nil, err |
| + } |
| + rawManifest, manifestETag, err := src.Get(ctx, manifestKey) |
| + if err != nil { |
| + return nil, fmt.Errorf("awsbilling: read %s: %w", manifestKey, err) |
| + } |
| + var m Manifest |
| + if err := json.Unmarshal(rawManifest, &m); err != nil { |
| + return nil, fmt.Errorf("%w: manifest %s: %v", ErrMalformed, manifestKey, err) |
| + } |
| + id := m.ExecutionID |
| + if id == "" { |
| + id = m.AssemblyID |
| + } |
| + if id == "" { |
| + return nil, fmt.Errorf("%w: manifest %s names no delivery id", ErrMalformed, manifestKey) |
| + } |
| + ps, perr := ParseTime(m.BillingPeriod.Start) |
| + pe, eerr := ParseTime(m.BillingPeriod.End) |
| + if perr != nil || eerr != nil || !ps.Equal(period) || !pe.Equal(periodEnd) { |
| + return nil, fmt.Errorf("%w: manifest %s covers %q..%q, not %s", ErrMalformed, manifestKey, |
| + m.BillingPeriod.Start, m.BillingPeriod.End, PeriodLabel(period)) |
| + } |
| + refs := m.DataFiles |
| + if len(refs) == 0 { |
| + refs = m.ReportKeys |
| + } |
| + if len(refs) == 0 { |
| + return nil, fmt.Errorf("%w: manifest %s lists no data files", ErrMalformed, manifestKey) |
| + } |
| + dataPrefix := DataPrefix(prefix, period) |
| + d := &Delivery{ |
| + Period: period, |
| + AssemblyID: id, |
| + ManifestKey: manifestKey, |
| + ManifestETag: manifestETag, |
| + ManifestSHA256: digest(rawManifest), |
| + } |
| + seenFile := map[string]bool{} |
| + seenLine := map[string]string{} |
| + invoiced := true |
| + for _, ref := range refs { |
| + key, err := dataFileKey(ref, dataPrefix) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if seenFile[key] { |
| + return nil, fmt.Errorf("%w: manifest %s lists %s twice", ErrMalformed, manifestKey, key) |
| + } |
| + seenFile[key] = true |
| + body, etag, err := src.Get(ctx, key) |
| + if err != nil { |
| + return nil, fmt.Errorf("awsbilling: read %s: %w", key, err) |
| + } |
| + fe := FileEvidence{Key: key, ETag: etag, SHA256: digest(body)} |
| + err = ReadLines(key, bytes.NewReader(body), func(l Line) error { |
| + fe.Rows++ |
| + if !l.PeriodStart.Equal(period) { |
| + return fmt.Errorf("%w: %s: row %d: billing period %s is not %s", ErrMalformed, key, l.Row, |
| + l.PeriodStart.Format(time.RFC3339), PeriodLabel(period)) |
| + } |
| + if l.ServiceCode != ServiceS3 { |
| + return nil |
| + } |
| + if l.LineItemID == "" { |
| + return fmt.Errorf("%w: %s: row %d: S3 line has no line item id", ErrMalformed, key, l.Row) |
| + } |
| + ident, fp := lineIdentity(l), lineFingerprint(l) |
| + if prev, dup := seenLine[ident]; dup { |
| + if prev != fp { |
| + return fmt.Errorf("%w: %s (%s row %d)", ErrConflictingDuplicate, l.LineItemID, key, l.Row) |
| + } |
| + d.Duplicates++ |
| + return nil |
| + } |
| + seenLine[ident] = fp |
| + d.S3Lines++ |
| + if l.InvoiceID == "" { |
| + invoiced = false |
| + } |
| + return fn(l) |
| + }) |
| + if err != nil { |
| + return nil, err |
| + } |
| + d.Files = append(d.Files, fe) |
| + } |
| + again, _, err := src.Get(ctx, manifestKey) |
| + if err != nil { |
| + return nil, fmt.Errorf("awsbilling: re-read %s: %w", manifestKey, err) |
| + } |
| + if digest(again) != d.ManifestSHA256 { |
| + return nil, fmt.Errorf("%w: %s", ErrDeliveryChanged, manifestKey) |
| + } |
| + switch { |
| + case d.S3Lines == 0: |
| + d.NotFinal = "no_s3_lines" |
| + case !invoiced: |
| + d.NotFinal = "invoice_not_issued" |
| + default: |
| + d.Final = true |
| + } |
| + return d, nil |
| +} |
diff --git a/internal/awsbilling/delivery_test.go b/internal/awsbilling/delivery_test.go| new file mode 100644 |
| index 000000000..8405e8415 |
| --- /dev/null |
| +++ b/internal/awsbilling/delivery_test.go |
| @@ -0,0 +1,261 @@ |
| +package awsbilling |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +const testPrefix = "cur/carloku-transfer/" |
| + |
| +var sept = time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) |
| + |
| +// putDelivery writes a CUR 2.0 manifest and its data files. dir is "" for |
| +// the overwrite layout, or a "<stamp>-<id>" directory for the versioned one. |
| +func putDelivery(t *testing.T, src *MemSource, dir, id string, files map[string]string) string { |
| + t.Helper() |
| + sub := "" |
| + if dir != "" { |
| + sub = dir + "/" |
| + } |
| + var refs []string |
| + for name, body := range files { |
| + key := DataPrefix(testPrefix, sept) + sub + name |
| + src.Put(key, []byte(body)) |
| + refs = append(refs, "s3://carloku-billing-cur-271376211898-eu-west-1/"+key) |
| + } |
| + m := map[string]any{ |
| + "executionId": id, |
| + "billingPeriod": map[string]string{"start": "2026-09-01T00:00:00.000Z", "end": "2026-10-01T00:00:00.000Z"}, |
| + "dataFiles": refs, |
| + } |
| + raw, _ := json.Marshal(m) |
| + key := MetadataPrefix(testPrefix, sept) + sub + "carloku-transfer-Manifest.json" |
| + src.Put(key, raw) |
| + return key |
| +} |
| + |
| +func readAll(t *testing.T, src Source) (*Delivery, []Line, error) { |
| + t.Helper() |
| + var lines []Line |
| + d, err := ReadDelivery(context.Background(), src, testPrefix, sept, func(l Line) error { |
| + lines = append(lines, l) |
| + return nil |
| + }) |
| + return d, lines, err |
| +} |
| + |
| +func TestReadDeliveryFinalAndPinned(t *testing.T) { |
| + src := NewMemSource() |
| + body := string(readFixture(t, "cur2-2026-09.csv")) |
| + putDelivery(t, src, "", "exec-1", map[string]string{"part-00001.csv": body}) |
| + d, lines, err := readAll(t, src) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !d.Final || d.NotFinal != "" { |
| + t.Fatalf("every S3 line carries an invoice id: Final=%v NotFinal=%q", d.Final, d.NotFinal) |
| + } |
| + if d.AssemblyID != "exec-1" || len(d.Files) != 1 || len(d.Files[0].SHA256) != 64 || d.Files[0].Rows != 10 || len(d.ManifestSHA256) != 64 { |
| + t.Fatalf("evidence not pinned: %+v", d) |
| + } |
| + // Nine of the ten rows are S3; the EC2 row is read and checked but not |
| + // passed on. |
| + if len(lines) != 9 || d.S3Lines != 9 { |
| + t.Fatalf("S3 lines = %d (%d), want 9", len(lines), d.S3Lines) |
| + } |
| + for _, l := range lines { |
| + if l.ServiceCode != ServiceS3 { |
| + t.Fatalf("non-S3 line passed on: %+v", l) |
| + } |
| + } |
| +} |
| + |
| +func TestReadDeliveryNotFinalWithoutInvoice(t *testing.T) { |
| + src := NewMemSource() |
| + putDelivery(t, src, "", "exec-legacy", map[string]string{"part-00001.csv": string(readFixture(t, "legacy-2026-09.csv"))}) |
| + d, _, err := readAll(t, src) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if d.Final || d.NotFinal != "invoice_not_issued" { |
| + t.Fatalf("Final=%v NotFinal=%q, want not final: invoice_not_issued", d.Final, d.NotFinal) |
| + } |
| +} |
| + |
| +func TestReadDeliveryNoS3LinesIsNotFinal(t *testing.T) { |
| + src := NewMemSource() |
| + fixture := string(readFixture(t, "cur2-2026-09.csv")) |
| + header, _, _ := strings.Cut(fixture, "\n") |
| + var ec2 string |
| + for _, row := range strings.Split(fixture, "\n") { |
| + if strings.HasPrefix(row, "id-ec2,") { |
| + ec2 = row |
| + } |
| + } |
| + putDelivery(t, src, "", "exec-2", map[string]string{"part-00001.csv": header + "\n" + ec2 + "\n"}) |
| + d, _, err := readAll(t, src) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if d.Final || d.NotFinal != "no_s3_lines" { |
| + t.Fatalf("Final=%v NotFinal=%q", d.Final, d.NotFinal) |
| + } |
| +} |
| + |
| +func TestReadDeliveryNoManifest(t *testing.T) { |
| + if _, _, err := readAll(t, NewMemSource()); !errors.Is(err, ErrNoDelivery) { |
| + t.Fatalf("err = %v, want ErrNoDelivery", err) |
| + } |
| +} |
| + |
| +// A period redelivered in the versioned layout: only the newest delivery |
| +// counts, and the superseded one is never summed with it. |
| +func TestReadDeliveryNewestSupersedes(t *testing.T) { |
| + src := NewMemSource() |
| + fixture := string(readFixture(t, "cur2-2026-09.csv")) |
| + header, rows, _ := strings.Cut(fixture, "\n") |
| + first, _, _ := strings.Cut(rows, "\n") |
| + older := header + "\n" + first + "\n" |
| + newer := header + "\n" + strings.Replace(first, ",2.5,0.09,0.225,", ",3,0.09,0.27,", 1) + "\n" |
| + putDelivery(t, src, "20260902T040000Z-aaa", "exec-old", map[string]string{"part-00001.csv": older}) |
| + putDelivery(t, src, "20260903T040000Z-bbb", "exec-new", map[string]string{"part-00001.csv": newer}) |
| + d, lines, err := readAll(t, src) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if d.AssemblyID != "exec-new" || len(lines) != 1 { |
| + t.Fatalf("delivery = %s with %d lines, want exec-new with 1", d.AssemblyID, len(lines)) |
| + } |
| + _, n, _, _ := ChargedS3Transfer(lines[0]) |
| + if n != 3_000_000_000 { |
| + t.Fatalf("bytes = %d, want the newest delivery's 3e9", n) |
| + } |
| +} |
| + |
| +func TestReadDeliveryAmbiguousOrder(t *testing.T) { |
| + src := NewMemSource() |
| + body := string(readFixture(t, "cur2-2026-09.csv")) |
| + putDelivery(t, src, "latest", "exec-a", map[string]string{"p.csv": body}) |
| + putDelivery(t, src, "20260903T040000Z-bbb", "exec-b", map[string]string{"p.csv": body}) |
| + if _, _, err := readAll(t, src); !errors.Is(err, ErrAmbiguousDelivery) { |
| + t.Fatalf("err = %v, want ErrAmbiguousDelivery", err) |
| + } |
| + src2 := NewMemSource() |
| + putDelivery(t, src2, "20260903T040000Z-aaa", "exec-a", map[string]string{"p.csv": body}) |
| + putDelivery(t, src2, "20260903T040000Z-bbb", "exec-b", map[string]string{"p.csv": body}) |
| + if _, _, err := readAll(t, src2); !errors.Is(err, ErrAmbiguousDelivery) { |
| + t.Fatalf("same stamp: err = %v, want ErrAmbiguousDelivery", err) |
| + } |
| +} |
| + |
| +// An identical row delivered twice (in two partitions) counts once; the |
| +// same identity with different values refuses the delivery. |
| +func TestReadDeliveryDedup(t *testing.T) { |
| + fixture := string(readFixture(t, "cur2-2026-09.csv")) |
| + header, rows, _ := strings.Cut(fixture, "\n") |
| + first, _, _ := strings.Cut(rows, "\n") |
| + |
| + src := NewMemSource() |
| + putDelivery(t, src, "", "exec-dup", map[string]string{ |
| + "part-00001.csv": header + "\n" + first + "\n", |
| + "part-00002.csv": header + "\n" + first + "\n", |
| + }) |
| + d, lines, err := readAll(t, src) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(lines) != 1 || d.Duplicates != 1 { |
| + t.Fatalf("lines=%d duplicates=%d, want 1 and 1", len(lines), d.Duplicates) |
| + } |
| + |
| + conflict := NewMemSource() |
| + putDelivery(t, conflict, "", "exec-conflict", map[string]string{ |
| + "part-00001.csv": header + "\n" + first + "\n", |
| + "part-00002.csv": header + "\n" + strings.Replace(first, ",2.5,", ",9,", 1) + "\n", |
| + }) |
| + if _, _, err := readAll(t, conflict); !errors.Is(err, ErrConflictingDuplicate) { |
| + t.Fatalf("err = %v, want ErrConflictingDuplicate", err) |
| + } |
| +} |
| + |
| +func TestReadDeliveryManifestRefusals(t *testing.T) { |
| + body := string(readFixture(t, "cur2-2026-09.csv")) |
| + mk := func(m map[string]any) *MemSource { |
| + src := NewMemSource() |
| + src.Put(DataPrefix(testPrefix, sept)+"p.csv", []byte(body)) |
| + raw, _ := json.Marshal(m) |
| + src.Put(MetadataPrefix(testPrefix, sept)+"x-Manifest.json", raw) |
| + return src |
| + } |
| + period := map[string]string{"start": "2026-09-01T00:00:00.000Z", "end": "2026-10-01T00:00:00.000Z"} |
| + good := "s3://b/" + DataPrefix(testPrefix, sept) + "p.csv" |
| + cases := map[string]map[string]any{ |
| + "no id": {"billingPeriod": period, "dataFiles": []string{good}}, |
| + "wrong period": {"executionId": "e", "billingPeriod": map[string]string{"start": "2026-08-01T00:00:00.000Z", "end": "2026-09-01T00:00:00.000Z"}, "dataFiles": []string{good}}, |
| + "no files": {"executionId": "e", "billingPeriod": period}, |
| + "outside data": {"executionId": "e", "billingPeriod": period, "dataFiles": []string{"s3://b/cur/other/secret.csv"}}, |
| + "dot-dot": {"executionId": "e", "billingPeriod": period, "dataFiles": []string{"s3://b/" + DataPrefix(testPrefix, sept) + "../../x.csv"}}, |
| + "listed twice": {"executionId": "e", "billingPeriod": period, "dataFiles": []string{good, good}}, |
| + "partial month": {"executionId": "e", "billingPeriod": map[string]string{"start": "2026-09-01T00:00:00.000Z", "end": "2026-09-15T00:00:00.000Z"}, "dataFiles": []string{good}}, |
| + } |
| + for name, m := range cases { |
| + if _, _, err := readAll(t, mk(m)); !errors.Is(err, ErrMalformed) { |
| + t.Errorf("%s: err = %v, want ErrMalformed", name, err) |
| + } |
| + } |
| +} |
| + |
| +// A line from another billing period inside this period's delivery |
| +// refuses the delivery. |
| +func TestReadDeliveryForeignPeriodLine(t *testing.T) { |
| + fixture := string(readFixture(t, "cur2-2026-09.csv")) |
| + header, rows, _ := strings.Cut(fixture, "\n") |
| + first, _, _ := strings.Cut(rows, "\n") |
| + shifted := strings.NewReplacer( |
| + "2026-09-01T00:00:00Z,2026-10-01T00:00:00Z", "2026-10-01T00:00:00Z,2026-11-01T00:00:00Z", |
| + "2026-09-03T10:00:00Z", "2026-10-03T10:00:00Z", |
| + "2026-09-03T11:00:00Z", "2026-10-03T11:00:00Z", |
| + ).Replace(first) |
| + src := NewMemSource() |
| + putDelivery(t, src, "", "e", map[string]string{"p.csv": header + "\n" + first + "\n" + shifted + "\n"}) |
| + if _, _, err := readAll(t, src); !errors.Is(err, ErrMalformed) { |
| + t.Fatalf("err = %v, want ErrMalformed", err) |
| + } |
| +} |
| + |
| +// The overwrite layout replaces files in place: a manifest that changes |
| +// while its files are read means the files may straddle two deliveries. |
| +func TestReadDeliveryManifestChangedDuringRead(t *testing.T) { |
| + src := NewMemSource() |
| + body := string(readFixture(t, "cur2-2026-09.csv")) |
| + manifestKey := putDelivery(t, src, "", "exec-1", map[string]string{"p.csv": body}) |
| + src.OnGet = func(key string) { |
| + if strings.HasSuffix(key, "p.csv") { |
| + src.OnGet = nil |
| + putDelivery(t, src, "", "exec-2", map[string]string{"p.csv": body}) |
| + } |
| + } |
| + _, _, err := readAll(t, src) |
| + if !errors.Is(err, ErrDeliveryChanged) { |
| + t.Fatalf("err = %v, want ErrDeliveryChanged (%s)", err, manifestKey) |
| + } |
| +} |
| + |
| +func TestReadDeliveryRefusesMalformedFile(t *testing.T) { |
| + src := NewMemSource() |
| + putDelivery(t, src, "", "e", map[string]string{"p.csv": "not,a,cur\n1,2,3\n"}) |
| + if _, _, err := readAll(t, src); !errors.Is(err, ErrMalformed) { |
| + t.Fatalf("err = %v, want ErrMalformed", err) |
| + } |
| +} |
| + |
| +func TestReadDeliveryRefusesMidMonthPeriod(t *testing.T) { |
| + _, err := ReadDelivery(context.Background(), NewMemSource(), testPrefix, sept.Add(time.Hour), func(Line) error { return nil }) |
| + if err == nil { |
| + t.Fatal("a period that is not a month start must be refused") |
| + } |
| +} |
diff --git a/internal/awsbilling/memsource.go b/internal/awsbilling/memsource.go| new file mode 100644 |
| index 000000000..07a46b0e9 |
| --- /dev/null |
| +++ b/internal/awsbilling/memsource.go |
| @@ -0,0 +1,58 @@ |
| +package awsbilling |
| + |
| +import ( |
| + "context" |
| + "os" |
| + "sort" |
| + "strings" |
| + "sync" |
| +) |
| + |
| +// MemSource is an in-memory Source for tests and local runs, in the same |
| +// spirit as objstore's MemProvisioner. The ETag is the body's digest. |
| +type MemSource struct { |
| + mu sync.Mutex |
| + objects map[string][]byte |
| + // OnGet, when set, runs before each Get with the key; tests use it to |
| + // change the bucket mid-read. |
| + OnGet func(key string) |
| +} |
| + |
| +// NewMemSource returns an empty MemSource. |
| +func NewMemSource() *MemSource { return &MemSource{objects: map[string][]byte{}} } |
| + |
| +// Put stores body at key, replacing any previous body. |
| +func (m *MemSource) Put(key string, body []byte) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + m.objects[key] = append([]byte(nil), body...) |
| +} |
| + |
| +// List returns every key under prefix, sorted. |
| +func (m *MemSource) List(_ context.Context, prefix string) ([]string, error) { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + var out []string |
| + for k := range m.objects { |
| + if strings.HasPrefix(k, prefix) { |
| + out = append(out, k) |
| + } |
| + } |
| + sort.Strings(out) |
| + return out, nil |
| +} |
| + |
| +// Get returns the body at key and its digest as the ETag, or an error |
| +// wrapping os.ErrNotExist. |
| +func (m *MemSource) Get(_ context.Context, key string) ([]byte, string, error) { |
| + if m.OnGet != nil { |
| + m.OnGet(key) |
| + } |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + b, ok := m.objects[key] |
| + if !ok { |
| + return nil, "", &os.PathError{Op: "get", Path: key, Err: os.ErrNotExist} |
| + } |
| + return append([]byte(nil), b...), `"` + digest(b)[:16] + `"`, nil |
| +} |
diff --git a/internal/awsbilling/testdata/cur2-2026-09.csv b/internal/awsbilling/testdata/cur2-2026-09.csv| new file mode 100644 |
| index 000000000..7d5a1faf2 |
| --- /dev/null |
| +++ b/internal/awsbilling/testdata/cur2-2026-09.csv |
| @@ -0,0 +1,11 @@ |
| +identity_line_item_id,identity_time_interval,bill_invoice_id,bill_billing_period_start_date,bill_billing_period_end_date,bill_payer_account_id,line_item_usage_account_id,line_item_line_item_type,line_item_usage_start_date,line_item_usage_end_date,line_item_product_code,line_item_usage_type,line_item_resource_id,line_item_usage_amount,line_item_unblended_rate,line_item_unblended_cost,line_item_line_item_description,product_servicecode,resource_tags_user_app |
| +id-internet,2026-09-03T10:00:00Z/2026-09-03T11:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z,AmazonS3,EUW1-DataTransfer-Out-Bytes,carlos-store-acc1-shop,2.5,0.09,0.225,$0.09 per GB - first 10 TB / month data transfer out beyond the global free tier,AmazonS3,shop |
| +id-regional,2026-09-03T11:00:00Z/2026-09-03T12:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-03T11:00:00Z,2026-09-03T12:00:00Z,AmazonS3,EUW1-DataTransfer-Regional-Bytes,carlos-store-acc1-shop,1,0.01,0.01,$0.01 per GB - regional data transfer,AmazonS3,shop |
| +id-interregion,2026-09-04T00:00:00Z/2026-09-04T01:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-04T00:00:00Z,2026-09-04T01:00:00Z,AmazonS3,EUW1-USE1-AWS-Out-Bytes,carlos-store-acc2-blog,0.5,0.02,0.01,$0.02 per GB - EU (Ireland) data transfer to US East (N. Virginia),AmazonS3,blog |
| +id-free,2026-09-02T00:00:00Z/2026-09-02T01:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-02T00:00:00Z,2026-09-02T01:00:00Z,AmazonS3,EUW1-DataTransfer-Out-Bytes,carlos-store-acc1-shop,40,0,0,$0.000 per GB - data transfer out under the global free tier,AmazonS3,shop |
| +id-cloudfront,2026-09-03T10:00:00Z/2026-09-03T11:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z,AmazonS3,EUW1-CloudFront-Out-Bytes,carlos-store-acc1-shop,7,0,0,$0.00 per GB - data transfer out to CloudFront,AmazonS3,shop |
| +id-inbound,2026-09-03T10:00:00Z/2026-09-03T11:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z,AmazonS3,EUW1-DataTransfer-In-Bytes,carlos-store-acc1-shop,9,0,0,$0.00 per GB - data transfer in,AmazonS3,shop |
| +id-requests,2026-09-03T10:00:00Z/2026-09-03T11:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z,AmazonS3,EUW1-Requests-Tier2,carlos-store-acc1-shop,1000,0.0000004,0.0004,$0.0004 per 1000 GET requests,AmazonS3,shop |
| +id-storage,2026-09-03T00:00:00Z/2026-09-04T00:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-03T00:00:00Z,2026-09-04T00:00:00Z,AmazonS3,EUW1-TimedStorage-ByteHrs,carlos-store-acc1-shop,0.1,0.023,0.0023,$0.023 per GB-Month of storage used,AmazonS3,shop |
| +id-ec2,2026-09-03T10:00:00Z/2026-09-03T11:00:00Z,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-03T10:00:00Z,2026-09-03T11:00:00Z,AmazonEC2,EUW1-DataTransfer-Out-Bytes,i-0abc,3,0.09,0.27,$0.09 per GB - data transfer out,AmazonEC2, |
| +id-tax,,EUINIE26-0001,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,271376211898,271376211898,Tax,2026-09-01T00:00:00Z,2026-10-01T00:00:00Z,AmazonS3,,,,,0.05,Tax for product code AmazonS3,AmazonS3, |
diff --git a/internal/awsbilling/testdata/legacy-2026-09.csv b/internal/awsbilling/testdata/legacy-2026-09.csv| new file mode 100644 |
| index 000000000..0b50e7ae6 |
| --- /dev/null |
| +++ b/internal/awsbilling/testdata/legacy-2026-09.csv |
| @@ -0,0 +1,3 @@ |
| +identity/LineItemId,identity/TimeInterval,bill/InvoiceId,bill/BillingPeriodStartDate,bill/PayerAccountId,lineItem/UsageAccountId,lineItem/LineItemType,lineItem/UsageStartDate,lineItem/UsageEndDate,lineItem/ProductCode,lineItem/UsageType,lineItem/ResourceId,lineItem/UsageAmount,lineItem/UnblendedCost,lineItem/LineItemDescription,product/servicecode,resourceTags/user:app |
| +legacy-1,2026-09-05T13:00:00Z/2026-09-05T14:00:00Z,,2026-09-01T00:00:00Z,271376211898,271376211898,Usage,2026-09-05T13:00:00Z,2026-09-05T14:00:00Z,AmazonS3,EUW1-DataTransfer-Out-Bytes,carlos-store-acc1-shop,1.5E-1,0.0135,$0.09 per GB - data transfer out,AmazonS3,shop |
| +legacy-2,2026-09-05T13:00:00Z/2026-09-05T14:00:00Z,,2026-09-01T00:00:00Z,271376211898,271376211898,Credit,2026-09-05T13:00:00Z,2026-09-05T14:00:00Z,AmazonS3,EUW1-DataTransfer-Out-Bytes,carlos-store-acc1-shop,0,-0.0135,Promotional credit,AmazonS3,shop |