[usage] Fetch sessions from usage report instead of from RPC arguments

components/usage/pkg/apiv1/billing.go

@@ -6,7 +6,7 @@ package apiv1
 
 import (
 	"context"
-	"fmt"
+	"github.com/gitpod-io/gitpod/usage/pkg/contentservice"
 	"math"
 	"time"
 
@@ -34,11 +34,22 @@ type BillingService struct {
 	stripeClient       *stripe.Client
 	billInstancesAfter time.Time
 
+	contentService contentservice.Interface
+
 	v1.UnimplementedBillingServiceServer
 }
 
 func (s *BillingService) UpdateInvoices(ctx context.Context, in *v1.UpdateInvoicesRequest) (*v1.UpdateInvoicesResponse, error) {
-	credits, err := s.creditSummaryForTeams(in.GetSessions())
+	if in.GetReportId() == "" {
+		return nil, status.Errorf(codes.InvalidArgument, "Missing report ID")
+	}
+
+	report, err := s.contentService.DownloadUsageReport(ctx, in.GetReportId())
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "Failed to download usage report with ID: %s", in.GetReportId())
+	}
+
+	credits, err := s.creditSummaryForTeams(report)
 	if err != nil {
 		log.Log.WithError(err).Errorf("Failed to compute credit summary.")
 		return nil, status.Errorf(codes.InvalidArgument, "failed to compute credit summary")
@@ -89,20 +100,15 @@ func (s *BillingService) GetUpcomingInvoice(ctx context.Context, in *v1.GetUpcom
 	}, nil
 }
 
-func (s *BillingService) creditSummaryForTeams(sessions []*v1.BilledSession) (map[string]int64, error) {
+func (s *BillingService) creditSummaryForTeams(sessions db.UsageReport) (map[string]int64, error) {
 	creditsPerTeamID := map[string]float64{}
 
 	for _, session := range sessions {
-		if session.StartTime.AsTime().Before(s.billInstancesAfter) {
+		if session.StartedAt.Before(s.billInstancesAfter) {
 			continue
 		}
 
-		attributionID, err := db.ParseAttributionID(session.AttributionId)
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse attribution ID: %w", err)
-		}
-
-		entity, id := attributionID.Values()
+		entity, id := session.AttributionID.Values()
 		if entity != db.AttributionEntity_Team {
 			continue
 		}
@@ -111,7 +117,7 @@ func (s *BillingService) creditSummaryForTeams(sessions []*v1.BilledSession) (ma
 			creditsPerTeamID[id] = 0
 		}
 
-		creditsPerTeamID[id] += session.GetCredits()
+		creditsPerTeamID[id] += session.CreditsUsed
 	}
 
 	rounded := map[string]int64{}

components/usage/pkg/apiv1/billing_test.go

@@ -8,12 +8,10 @@ import (
 	"testing"
 	"time"
 
-	v1 "github.com/gitpod-io/gitpod/usage-api/v1"
 	"github.com/gitpod-io/gitpod/usage/pkg/db"
 	"github.com/gitpod-io/gitpod/usage/pkg/stripe"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
-	timestamppb "google.golang.org/protobuf/types/known/timestamppb"
 	"gorm.io/gorm"
 )
 
@@ -23,39 +21,39 @@ func TestCreditSummaryForTeams(t *testing.T) {
 
 	scenarios := []struct {
 		Name              string
-		Sessions          []*v1.BilledSession
+		Sessions          db.UsageReport
 		BillSessionsAfter time.Time
 		Expected          map[string]int64
 	}{
 		{
 			Name:              "no instances in report, no summary",
 			BillSessionsAfter: time.Time{},
-			Sessions:          []*v1.BilledSession{},
+			Sessions:          nil,
 			Expected:          map[string]int64{},
 		},
 		{
 			Name:              "skips user attributions",
 			BillSessionsAfter: time.Time{},
-			Sessions: []*v1.BilledSession{
+			Sessions: []db.WorkspaceInstanceUsage{
 				{
-					AttributionId: string(db.NewUserAttributionID(uuid.New().String())),
+					AttributionID: db.NewUserAttributionID(uuid.New().String()),
 				},
 			},
 			Expected: map[string]int64{},
 		},
 		{
 			Name:              "two workspace instances",
 			BillSessionsAfter: time.Time{},
-			Sessions: []*v1.BilledSession{
+			Sessions: []db.WorkspaceInstanceUsage{
 				{
 					// has 1 day and 23 hours of usage
-					AttributionId: string(teamAttributionID_A),
-					Credits:       (24 + 23) * 10,
+					AttributionID: teamAttributionID_A,
+					CreditsUsed:   (24 + 23) * 10,
 				},
 				{
 					// has 1 hour of usage
-					AttributionId: string(teamAttributionID_A),
-					Credits:       10,
+					AttributionID: teamAttributionID_A,
+					CreditsUsed:   10,
 				},
 			},
 			Expected: map[string]int64{
@@ -66,16 +64,16 @@ func TestCreditSummaryForTeams(t *testing.T) {
 		{
 			Name:              "multiple teams",
 			BillSessionsAfter: time.Time{},
-			Sessions: []*v1.BilledSession{
+			Sessions: []db.WorkspaceInstanceUsage{
 				{
 					// has 12 hours of usage
-					AttributionId: string(teamAttributionID_A),
-					Credits:       (12) * 10,
+					AttributionID: teamAttributionID_A,
+					CreditsUsed:   (12) * 10,
 				},
 				{
 					// has 1 day of usage
-					AttributionId: string(teamAttributionID_B),
-					Credits:       (24) * 10,
+					AttributionID: teamAttributionID_B,
+					CreditsUsed:   (24) * 10,
 				},
 			},
 			Expected: map[string]int64{
@@ -87,18 +85,18 @@ func TestCreditSummaryForTeams(t *testing.T) {
 		{
 			Name:              "two instances, same team, one of which started too early to be considered",
 			BillSessionsAfter: time.Now().AddDate(0, 0, -2),
-			Sessions: []*v1.BilledSession{
+			Sessions: []db.WorkspaceInstanceUsage{
 				{
 					// has 12 hours of usage, started yesterday
-					AttributionId: string(teamAttributionID_A),
-					Credits:       (12) * 10,
-					StartTime:     timestamppb.New(time.Now().AddDate(0, 0, -1)),
+					AttributionID: teamAttributionID_A,
+					CreditsUsed:   (12) * 10,
+					StartedAt:     time.Now().AddDate(0, 0, -1),
 				},
 				{
 					// has 1 day of usage, but started three days ago
-					AttributionId: string(teamAttributionID_A),
-					Credits:       (24) * 10,
-					StartTime:     timestamppb.New(time.Now().AddDate(0, 0, -3)),
+					AttributionID: teamAttributionID_A,
+					CreditsUsed:   (24) * 10,
+					StartedAt:     time.Now().AddDate(0, 0, -3),
 				},
 			},
 			Expected: map[string]int64{

components/usage/pkg/apiv1/usage.go

@@ -125,13 +125,7 @@ func (s *UsageService) ReconcileUsage(ctx context.Context, req *v1.ReconcileUsag
 		return nil, status.Error(codes.Internal, "failed to persist usage report to content service")
 	}
 
-	var sessions []*v1.BilledSession
-	for _, instance := range report.UsageRecords {
-		sessions = append(sessions, usageRecordToBilledUsageProto(instance))
-	}
-
 	return &v1.ReconcileUsageResponse{
-		Sessions: sessions,
 		ReportId: filename,
 	}, nil
 
@@ -175,26 +169,6 @@ func NewUsageService(conn *gorm.DB, reportGenerator *ReportGenerator, contentSvc
 	}
 }
 
-func usageRecordToBilledUsageProto(usageRecord db.WorkspaceInstanceUsage) *v1.BilledSession {
-	var endTime *timestamppb.Timestamp
-	if usageRecord.StoppedAt.Valid {
-		endTime = timestamppb.New(usageRecord.StoppedAt.Time)
-	}
-	return &v1.BilledSession{
-		AttributionId:  string(usageRecord.AttributionID),
-		UserId:         usageRecord.UserID.String(),
-		WorkspaceId:    usageRecord.WorkspaceID,
-		TeamId:         "",
-		WorkspaceType:  string(usageRecord.WorkspaceType),
-		ProjectId:      usageRecord.ProjectID,
-		InstanceId:     usageRecord.InstanceID.String(),
-		WorkspaceClass: usageRecord.WorkspaceClass,
-		StartTime:      timestamppb.New(usageRecord.StartedAt),
-		EndTime:        endTime,
-		Credits:        usageRecord.CreditsUsed,
-	}
-}
-
 func instancesToUsageRecords(instances []db.WorkspaceInstanceForUsage, pricer *WorkspacePricer, now time.Time) []db.WorkspaceInstanceUsage {
 	var usageRecords []db.WorkspaceInstanceUsage
 

components/usage/pkg/contentservice/client.go

@@ -31,7 +31,7 @@ func New(service api.UsageReportServiceClient) *Client {
 }
 
 func (c *Client) UploadUsageReport(ctx context.Context, filename string, report db.UsageReport) error {
-	uploadURLResp, err := c.service.UploadURL(ctx, &api.UsageReportUploadURLRequest{Name: key})
+	uploadURLResp, err := c.service.UploadURL(ctx, &api.UsageReportUploadURLRequest{Name: filename})
 	if err != nil {
 		return fmt.Errorf("failed to get upload URL from usage report service: %w", err)
 	}
@@ -61,7 +61,7 @@ func (c *Client) UploadUsageReport(ctx context.Context, filename string, report
 		return fmt.Errorf("failed to make http request: %w", err)
 	}
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("unexpected http response code: %s", uploadURLResp.Status)
+		return fmt.Errorf("unexpected http response code: %s", resp.Status)
 	}
 	log.Info("Upload complete")
 
@@ -76,7 +76,17 @@ func (c *Client) DownloadUsageReport(ctx context.Context, filename string) (db.U
 		return nil, fmt.Errorf("failed to get download URL: %w", err)
 	}
 
-	resp, err := http.Get(downloadURlResp.GetUrl())
+	req, err := http.NewRequest(http.MethodGet, downloadURlResp.GetUrl(), nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to construct request: %w", err)
+	}
+
+	// We want to receive it as gzip, this disables transcoding of the response
+	req.Header.Set("Accept-Encoding", "gzip")
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Content-Encoding", "gzip")
+
+	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("failed to make request to download usage report: %w", err)
 	}
@@ -88,13 +98,13 @@ func (c *Client) DownloadUsageReport(ctx context.Context, filename string) (db.U
 	body := resp.Body
 	defer body.Close()
 
-	decomressor, err := gzip.NewReader(body)
+	decompressor, err := gzip.NewReader(body)
 	if err != nil {
 		return nil, fmt.Errorf("failed to construct gzip decompressor from response: %w", err)
 	}
-	defer decomressor.Close()
+	defer decompressor.Close()
 
-	decoder := json.NewDecoder(decomressor)
+	decoder := json.NewDecoder(body)
 	var records []db.WorkspaceInstanceUsage
 	if err := decoder.Decode(&records); err != nil {
 		return nil, fmt.Errorf("failed to deserialize report: %w", err)

components/usage/pkg/contentservice/noop.go

@@ -6,12 +6,19 @@ package contentservice
 
 import (
 	"context"
+	"errors"
 
 	"github.com/gitpod-io/gitpod/usage/pkg/db"
 )
 
+var notImplementedError = errors.New("not implemented")
+
 type NoOpClient struct{}
 
-func (c *NoOpClient) UploadUsageReport(ctx context.Context, filename string, report []db.WorkspaceInstanceUsage) error {
-	return nil
+func (c *NoOpClient) UploadUsageReport(ctx context.Context, filename string, report db.UsageReport) error {
+	return notImplementedError
+}
+
+func (c *NoOpClient) DownloadUsageReport(ctx context.Context, filename string) (db.UsageReport, error) {
+	return nil, notImplementedError
 }

components/usage/pkg/controller/reconciler.go

@@ -57,15 +57,11 @@ func (r *UsageAndBillingReconciler) Reconcile() (err error) {
 		return fmt.Errorf("failed to reconcile usage: %w", err)
 	}
 
-	sessions := usageResp.GetSessions()
-	reportSessionsRetrievedTotal(len(sessions))
-
 	reportID := usageResp.GetReportId()
 
 	_, err = r.billingClient.UpdateInvoices(ctx, &v1.UpdateInvoicesRequest{
 		StartTime: timestamppb.New(startOfCurrentMonth),
 		EndTime:   timestamppb.New(startOfNextMonth),
-		Sessions:  sessions,
 		ReportId:  reportID,
 	})
 	if err != nil {

components/usage/pkg/controller/reporter.go

@@ -30,20 +30,12 @@ var (
 		Help:      "Histogram of reconcile duration",
 		Buckets:   prometheus.LinearBuckets(30, 30, 10), // every 30 secs, starting at 30secs
 	}, []string{"outcome"})
-
-	sessionsRetrievedTotal = prometheus.NewGauge(prometheus.GaugeOpts{
-		Namespace: namespace,
-		Subsystem: subsystem,
-		Name:      "usage_records_retrieved_total",
-		Help:      "Number of usage records retrieved during usage collection run",
-	})
 )
 
 func RegisterMetrics(reg *prometheus.Registry) error {
 	metrics := []prometheus.Collector{
 		reconcileStartedTotal,
 		reconcileStartedDurationSeconds,
-		sessionsRetrievedTotal,
 	}
 	for _, metric := range metrics {
 		err := reg.Register(metric)
@@ -66,7 +58,3 @@ func reportUsageReconcileFinished(duration time.Duration, err error) {
 	}
 	reconcileStartedDurationSeconds.WithLabelValues(outcome).Observe(duration.Seconds())
 }
-
-func reportSessionsRetrievedTotal(count int) {
-	sessionsRetrievedTotal.Set(float64(count))
-}