handled paginated usage and daily cost windows (#4060)

Signed-off-by: Kush Agarwal <agrawalkush783@gmail.com>

pkg/cloud/oracle/usageapiintegration.go

@@ -18,12 +18,20 @@ type UsageApiIntegration struct {
 	ConnectionStatus cloud.ConnectionStatus
 }
 
+type usageAPIClient interface {
+	RequestSummarizedUsages(context.Context, usageapi.RequestSummarizedUsagesRequest) (usageapi.RequestSummarizedUsagesResponse, error)
+}
+
 func (uai *UsageApiIntegration) GetCloudCost(start time.Time, end time.Time) (*opencost.CloudCostSetRange, error) {
 	client, err := uai.GetUsageApiClient()
 	if err != nil {
 		uai.ConnectionStatus = cloud.FailedConnection
 		return nil, fmt.Errorf("getting oracle usage api client: %s", err.Error())
 	}
+	return uai.getCloudCost(context.Background(), client, start, end)
+}
+
+func (uai *UsageApiIntegration) getCloudCost(ctx context.Context, client usageAPIClient, start time.Time, end time.Time) (*opencost.CloudCostSetRange, error) {
 
 	req := usageapi.RequestSummarizedUsagesRequest{
 		RequestSummarizedUsagesDetails: usageapi.RequestSummarizedUsagesDetails{
@@ -38,112 +46,134 @@ func (uai *UsageApiIntegration) GetCloudCost(start time.Time, end time.Time) (*o
 		Limit: common.Int(500),
 	}
 
-	resp, err := client.RequestSummarizedUsages(context.Background(), req)
-	if err != nil {
-		uai.ConnectionStatus = cloud.FailedConnection
-		return nil, fmt.Errorf("failed to query usage: %w", err)
-	}
-
 	ccsr, err := opencost.NewCloudCostSetRange(start, end, opencost.AccumulateOptionDay, uai.Key())
 	if err != nil {
 		return nil, err
 	}
 
-	// Set status to missing data if query comes back empty and the status isn't already successful
-	if len(resp.Items) == 0 && uai.ConnectionStatus != cloud.SuccessfulConnection {
-		uai.ConnectionStatus = cloud.MissingData
-		return ccsr, nil
-	}
-
-	for _, item := range resp.Items {
-		resourceId := ""
-		if item.ResourceId != nil {
-			resourceId = *item.ResourceId
+	hasItems := false
+	seenPageTokens := map[string]struct{}{}
+	for page := 1; ; page++ {
+		resp, err := client.RequestSummarizedUsages(ctx, req)
+		if err != nil {
+			uai.ConnectionStatus = cloud.FailedConnection
+			return nil, fmt.Errorf("failed to query usage: %w", err)
 		}
+		log.Debugf("UsageApiIntegration[%s]: received %d usage items from page %d", uai.Key(), len(resp.Items), page)
 
-		tenantName := ""
-		if item.TenantName != nil {
-			tenantName = *item.TenantName
+		if len(resp.Items) > 0 {
+			hasItems = true
 		}
 
-		subscriptionId := ""
-		if item.SubscriptionId != nil {
-			subscriptionId = *item.SubscriptionId
+		for _, item := range resp.Items {
+			if item.TimeUsageStarted == nil || item.TimeUsageEnded == nil {
+				log.Warnf("UsageApiIntegration[%s]: skipping usage item without a usage window", uai.Key())
+				continue
+			}
+
+			cc, err := uai.usageSummaryToCloudCost(item)
+			if err != nil {
+				return nil, err
+			}
+			ccsr.LoadCloudCost(cc)
 		}
 
-		service := ""
-		if item.Service != nil {
-			service = *item.Service
+		if resp.OpcNextPage == nil || *resp.OpcNextPage == "" {
+			break
+		}
+		if _, ok := seenPageTokens[*resp.OpcNextPage]; ok {
+			uai.ConnectionStatus = cloud.FailedConnection
+			return nil, fmt.Errorf("received a repeated OCI usage API page token")
 		}
+		seenPageTokens[*resp.OpcNextPage] = struct{}{}
+		req.Page = resp.OpcNextPage
+	}
 
-		category := SelectOCICategory(service)
+	// Set status to missing data if every response page was empty and the status isn't already successful.
+	if !hasItems && uai.ConnectionStatus != cloud.SuccessfulConnection {
+		uai.ConnectionStatus = cloud.MissingData
+		return ccsr, nil
+	}
 
-		// Iterate through the slice of tags, assigning
-		// keys and values to the map of labels
-		labels := opencost.CloudCostLabels{}
-		for _, tag := range item.Tags {
-			if tag.Key == nil || tag.Value == nil {
-				continue
-			}
-			labels[*tag.Key] = *tag.Value
-		}
+	uai.ConnectionStatus = cloud.SuccessfulConnection
+	return ccsr, nil
+}
 
-		properties := &opencost.CloudCostProperties{
-			ProviderID:      resourceId,
-			Provider:        opencost.OracleProvider,
-			AccountID:       uai.TenancyID,
-			AccountName:     tenantName,
-			InvoiceEntityID: subscriptionId,
-			RegionID:        uai.Region,
-			Service:         service,
-			Category:        category,
-			Labels:          labels,
-		}
+func (uai *UsageApiIntegration) usageSummaryToCloudCost(item usageapi.UsageSummary) (*opencost.CloudCost, error) {
+	resourceID := ""
+	if item.ResourceId != nil {
+		resourceID = *item.ResourceId
+	}
 
-		winStart := item.TimeUsageStarted.Time
-		winEnd := start.AddDate(0, 0, 1)
+	tenantName := ""
+	if item.TenantName != nil {
+		tenantName = *item.TenantName
+	}
 
-		listRate := 0.0
-		if item.ListRate != nil {
-			listRate = float64(*item.ListRate)
-		}
+	subscriptionID := ""
+	if item.SubscriptionId != nil {
+		subscriptionID = *item.SubscriptionId
+	}
 
-		attrCost, err := parseAttributedCost(item.AttributedCost)
-		if err != nil {
-			return nil, err
-		}
+	service := ""
+	if item.Service != nil {
+		service = *item.Service
+	}
 
-		computedAmt := 0.0
-		if item.ComputedAmount != nil {
-			computedAmt = float64(*item.ComputedAmount)
+	labels := opencost.CloudCostLabels{}
+	for _, tag := range item.Tags {
+		if tag.Key == nil || tag.Value == nil {
+			continue
 		}
+		labels[*tag.Key] = *tag.Value
+	}
 
-		cc := &opencost.CloudCost{
-			Properties: properties,
-			Window:     opencost.NewWindow(&winStart, &winEnd),
-			//todo: which returned costs go where?
-			ListCost: opencost.CostMetric{
-				Cost: listRate,
-			},
-			NetCost: opencost.CostMetric{
-				Cost: computedAmt,
-			},
-			AmortizedNetCost: opencost.CostMetric{
-				Cost: attrCost,
-			},
-			AmortizedCost: opencost.CostMetric{
-				Cost: attrCost,
-			},
-			InvoicedCost: opencost.CostMetric{
-				Cost: computedAmt,
-			},
-		}
+	listRate := 0.0
+	if item.ListRate != nil {
+		listRate = float64(*item.ListRate)
+	}
 
-		ccsr.LoadCloudCost(cc)
+	attributedCost, err := parseAttributedCost(item.AttributedCost)
+	if err != nil {
+		return nil, err
 	}
 
-	uai.ConnectionStatus = cloud.SuccessfulConnection
-	return ccsr, nil
+	computedAmount := 0.0
+	if item.ComputedAmount != nil {
+		computedAmount = float64(*item.ComputedAmount)
+	}
+
+	winStart := item.TimeUsageStarted.Time
+	winEnd := item.TimeUsageEnded.Time
+	return &opencost.CloudCost{
+		Properties: &opencost.CloudCostProperties{
+			ProviderID:      resourceID,
+			Provider:        opencost.OracleProvider,
+			AccountID:       uai.TenancyID,
+			AccountName:     tenantName,
+			InvoiceEntityID: subscriptionID,
+			RegionID:        uai.Region,
+			Service:         service,
+			Category:        SelectOCICategory(service),
+			Labels:          labels,
+		},
+		Window: opencost.NewWindow(&winStart, &winEnd),
+		ListCost: opencost.CostMetric{
+			Cost: listRate,
+		},
+		NetCost: opencost.CostMetric{
+			Cost: computedAmount,
+		},
+		AmortizedNetCost: opencost.CostMetric{
+			Cost: attributedCost,
+		},
+		AmortizedCost: opencost.CostMetric{
+			Cost: attributedCost,
+		},
+		InvoicedCost: opencost.CostMetric{
+			Cost: computedAmount,
+		},
+	}, nil
 }
 
 func (uai *UsageApiIntegration) GetStatus() cloud.ConnectionStatus {

pkg/cloud/oracle/usageapiintegration_test.go

@@ -1,12 +1,17 @@
 package oracle
 
 import (
+	"context"
 	"encoding/json"
+	"fmt"
 	"os"
 	"testing"
 	"time"
 
 	"github.com/opencost/opencost/core/pkg/util/timeutil"
+	"github.com/opencost/opencost/pkg/cloud"
+	"github.com/oracle/oci-go-sdk/v65/common"
+	"github.com/oracle/oci-go-sdk/v65/usageapi"
 )
 
 func TestParseAttributedCost(t *testing.T) {
@@ -87,3 +92,205 @@ func TestUsageAPIIntegration_GetCloudCost(t *testing.T) {
 		})
 	}
 }
+
+type fakeUsageAPIClient struct {
+	responses []usageapi.RequestSummarizedUsagesResponse
+	requests  []usageapi.RequestSummarizedUsagesRequest
+}
+
+func (f *fakeUsageAPIClient) RequestSummarizedUsages(_ context.Context, request usageapi.RequestSummarizedUsagesRequest) (usageapi.RequestSummarizedUsagesResponse, error) {
+	f.requests = append(f.requests, request)
+	if len(f.responses) == 0 {
+		return usageapi.RequestSummarizedUsagesResponse{}, fmt.Errorf("unexpected request")
+	}
+
+	response := f.responses[0]
+	f.responses = f.responses[1:]
+	return response, nil
+}
+
+func TestUsageAPIIntegrationGetCloudCostLoadsEachDailyItem(t *testing.T) {
+	start := time.Date(2025, time.January, 1, 0, 0, 0, 0, time.UTC)
+	client := &fakeUsageAPIClient{
+		responses: []usageapi.RequestSummarizedUsagesResponse{
+			{
+				UsageAggregation: usageapi.UsageAggregation{
+					Items: []usageapi.UsageSummary{
+						testUsageSummary(start, "resource-1", 1),
+						testUsageSummary(start.AddDate(0, 0, 1), "resource-2", 2),
+						testUsageSummary(start.AddDate(0, 0, 2), "resource-3", 3),
+					},
+				},
+			},
+		},
+	}
+	integration := &UsageApiIntegration{
+		UsageApiConfiguration: UsageApiConfiguration{
+			TenancyID: "tenancy-id",
+			Region:    "region",
+		},
+	}
+
+	ccsr, err := integration.getCloudCost(context.Background(), client, start, start.AddDate(0, 0, 3))
+	if err != nil {
+		t.Fatalf("getCloudCost() error = %v", err)
+	}
+
+	if len(ccsr.CloudCostSets) != 3 {
+		t.Fatalf("expected 3 daily CloudCostSets, got %d", len(ccsr.CloudCostSets))
+	}
+
+	for i, ccs := range ccsr.CloudCostSets {
+		if len(ccs.CloudCosts) != 1 {
+			t.Fatalf("day %d: expected 1 CloudCost, got %d", i+1, len(ccs.CloudCosts))
+		}
+		for _, cloudCost := range ccs.CloudCosts {
+			wantCost := float64(i + 1)
+			if cloudCost.NetCost.Cost != wantCost {
+				t.Errorf("day %d: NetCost = %v, want %v", i+1, cloudCost.NetCost.Cost, wantCost)
+			}
+		}
+	}
+}
+
+func TestUsageAPIIntegrationGetCloudCostFollowsPagination(t *testing.T) {
+	start := time.Date(2025, time.January, 1, 0, 0, 0, 0, time.UTC)
+	firstPageItems := make([]usageapi.UsageSummary, 500)
+	for i := range firstPageItems {
+		firstPageItems[i] = testUsageSummary(start, fmt.Sprintf("resource-%d", i), 1)
+	}
+
+	client := &fakeUsageAPIClient{
+		responses: []usageapi.RequestSummarizedUsagesResponse{
+			{
+				UsageAggregation: usageapi.UsageAggregation{Items: firstPageItems},
+				OpcNextPage:      common.String("next-page"),
+			},
+			{
+				UsageAggregation: usageapi.UsageAggregation{
+					Items: []usageapi.UsageSummary{testUsageSummary(start, "resource-500", 1)},
+				},
+			},
+		},
+	}
+	integration := &UsageApiIntegration{
+		UsageApiConfiguration: UsageApiConfiguration{
+			TenancyID: "tenancy-id",
+			Region:    "region",
+		},
+	}
+
+	ccsr, err := integration.getCloudCost(context.Background(), client, start, start.AddDate(0, 0, 1))
+	if err != nil {
+		t.Fatalf("getCloudCost() error = %v", err)
+	}
+
+	if len(client.requests) != 2 {
+		t.Fatalf("expected 2 OCI requests, got %d", len(client.requests))
+	}
+	if client.requests[0].Page != nil {
+		t.Errorf("first request page = %q, want nil", *client.requests[0].Page)
+	}
+	if client.requests[1].Page == nil || *client.requests[1].Page != "next-page" {
+		t.Errorf("second request page = %v, want next-page", client.requests[1].Page)
+	}
+	if client.requests[1].Limit == nil || *client.requests[1].Limit != 500 {
+		t.Errorf("second request limit = %v, want 500", client.requests[1].Limit)
+	}
+
+	if got := len(ccsr.CloudCostSets[0].CloudCosts); got != 501 {
+		t.Errorf("expected 501 CloudCosts from both pages, got %d", got)
+	}
+}
+
+func TestUsageAPIIntegrationGetCloudCostSkipsUsageSummaryWithoutTimeWindow(t *testing.T) {
+	start := time.Date(2025, time.January, 1, 0, 0, 0, 0, time.UTC)
+	item := testUsageSummary(start, "resource-1", 1)
+	item.TimeUsageEnded = nil
+	client := &fakeUsageAPIClient{
+		responses: []usageapi.RequestSummarizedUsagesResponse{
+			{
+				UsageAggregation: usageapi.UsageAggregation{Items: []usageapi.UsageSummary{item}},
+			},
+		},
+	}
+	integration := &UsageApiIntegration{
+		UsageApiConfiguration: UsageApiConfiguration{
+			TenancyID: "tenancy-id",
+			Region:    "region",
+		},
+	}
+
+	ccsr, err := integration.getCloudCost(context.Background(), client, start, start.AddDate(0, 0, 1))
+	if err != nil {
+		t.Fatalf("getCloudCost() error = %v", err)
+	}
+	if !ccsr.IsEmpty() {
+		t.Error("expected usage summary without a time window to be skipped")
+	}
+}
+
+func TestUsageAPIIntegrationGetCloudCostRejectsRepeatedPageToken(t *testing.T) {
+	start := time.Date(2025, time.January, 1, 0, 0, 0, 0, time.UTC)
+	client := &fakeUsageAPIClient{
+		responses: []usageapi.RequestSummarizedUsagesResponse{
+			{
+				UsageAggregation: usageapi.UsageAggregation{Items: []usageapi.UsageSummary{testUsageSummary(start, "resource-1", 1)}},
+				OpcNextPage:      common.String("page-token"),
+			},
+			{
+				UsageAggregation: usageapi.UsageAggregation{Items: []usageapi.UsageSummary{testUsageSummary(start, "resource-2", 1)}},
+				OpcNextPage:      common.String("page-token"),
+			},
+		},
+	}
+	integration := &UsageApiIntegration{
+		UsageApiConfiguration: UsageApiConfiguration{
+			TenancyID: "tenancy-id",
+			Region:    "region",
+		},
+	}
+
+	_, err := integration.getCloudCost(context.Background(), client, start, start.AddDate(0, 0, 1))
+	if err == nil {
+		t.Fatal("expected error for repeated OCI page token")
+	}
+	if got := len(client.requests); got != 2 {
+		t.Errorf("expected 2 OCI requests before repeated token error, got %d", got)
+	}
+	if integration.ConnectionStatus != cloud.FailedConnection {
+		t.Errorf("ConnectionStatus = %s, want %s", integration.ConnectionStatus, cloud.FailedConnection)
+	}
+}
+
+func TestUsageSummaryToCloudCostUsesOCIUsageWindow(t *testing.T) {
+	start := time.Date(2025, time.January, 1, 0, 0, 0, 0, time.UTC)
+	end := start.Add(12 * time.Hour)
+	item := testUsageSummary(start, "resource-1", 1)
+	item.TimeUsageEnded = &common.SDKTime{Time: end}
+	integration := &UsageApiIntegration{
+		UsageApiConfiguration: UsageApiConfiguration{
+			TenancyID: "tenancy-id",
+			Region:    "region",
+		},
+	}
+
+	cloudCost, err := integration.usageSummaryToCloudCost(item)
+	if err != nil {
+		t.Fatalf("usageSummaryToCloudCost() error = %v", err)
+	}
+	if got := cloudCost.Window.End(); !got.Equal(end) {
+		t.Errorf("CloudCost window end = %s, want %s", got, end)
+	}
+}
+
+func testUsageSummary(start time.Time, resourceID string, computedAmount float32) usageapi.UsageSummary {
+	return usageapi.UsageSummary{
+		TimeUsageStarted: &common.SDKTime{Time: start},
+		TimeUsageEnded:   &common.SDKTime{Time: start.AddDate(0, 0, 1)},
+		ResourceId:       common.String(resourceID),
+		Service:          common.String("Compute"),
+		ComputedAmount:   common.Float32(computedAmount),
+		AttributedCost:   common.String(fmt.Sprintf("%v", computedAmount)),
+	}
+}