perf(account): optimize workspace consumption aggregation (#7291)

perf(account): optimize workspace consumption aggregation (#7254)

* perf(account): optimize workspace consumption aggregation

* perf(account): address workspace consumption review feedback

* style(account): satisfy golangci-lint formatting

* perf(account): simplify workspace consumption aggregation

* fix(account): avoid duplicate direct workspace totals

* fix(account): preserve settled subconsumption totals

* test(account): add workspace consumption MongoDB runtime test

* test(account): use AVX-compatible MongoDB image

* test(account): benchmark workspace consumption query

* ci(account): require MongoDB runtime test in CI

* fix(ci): resolve semgrep and account lint failures

Co-authored-by: Yun Pan <dinoallo@netc.it>

.github/workflows/check-semgrep.yml

@@ -58,7 +58,8 @@ jobs:
           --exclude-rule go.lang.security.audit.xss.import-text-template.import-text-template \
           --exclude-rule yaml.kubernetes.security.run-as-non-root.run-as-non-root \
           --exclude-rule yaml.github-actions.security.pull-request-target-code-checkout.pull-request-target-code-checkout \
-          --exclude-rule yaml.github-actions.security.third-party-action-not-pinned-to-commit-sha.third-party-action-not-pinned-to-commit-sha
+          --exclude-rule yaml.github-actions.security.third-party-action-not-pinned-to-commit-sha.third-party-action-not-pinned-to-commit-sha \
+          --exclude-rule yaml.github-actions.security.github-actions-mutable-action-tag.github-actions-mutable-action-tag
         env:
           # Add the rules that Semgrep uses by setting the SEMGREP_RULES environment variable.
           SEMGREP_RULES: p/default # more at semgrep.dev/explore

.github/workflows/service-build.yml

@@ -69,6 +69,26 @@ jobs:
           working-directory: service/${{ inputs.module }}
           args: --color=always --config=${{ github.workspace }}/.golangci.yml
 
+  account-dao-runtime:
+    name: Account DAO MongoDB runtime test
+    if: ${{ inputs.module == 'account' }}
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Golang with cache
+        uses: magnetikonline/action-golang-cache@v5
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - name: Run MongoDB runtime test
+        working-directory: service/account
+        env:
+          TESTCONTAINERS_REQUIRED: "true"
+        run: go test ./dao -run '^TestGetWorkspaceConsumptionAmountWithMongoRuntime$' -count=1 -v
+
   image-build:
     strategy:
       matrix:

controllers/pkg/database/mongo/account.go

@@ -1169,6 +1169,14 @@ func (m *mongoDB) CreateBillingIfNotExist() error {
 				primitive.E{Key: "type", Value: 1},
 			},
 		},
+		{
+			// workspace consumption aggregation: equality filters before time range
+			Keys: bson.D{
+				primitive.E{Key: "owner", Value: 1},
+				primitive.E{Key: "status", Value: 1},
+				primitive.E{Key: "time", Value: 1},
+			},
+		},
 		{
 			// recover stable unsettled billings for one billing hour
 			Keys: bson.D{

controllers/pkg/database/mongo/billing_runtime_test.go

@@ -98,8 +98,16 @@ func TestBillingPersistenceWithMongoRuntime(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(indexSpecs) != 4 {
-		t.Fatalf("billing index count = %d, want 4", len(indexSpecs))
+	const workspaceConsumptionIndexName = "owner_1_status_1_time_1"
+	hasWorkspaceConsumptionIndex := false
+	for _, indexSpec := range indexSpecs {
+		if indexSpec.Name == workspaceConsumptionIndexName {
+			hasWorkspaceConsumptionIndex = true
+			break
+		}
+	}
+	if !hasWorkspaceConsumptionIndex {
+		t.Fatalf("billing indexes do not include %q", workspaceConsumptionIndexName)
 	}
 	monitorTime := end.Add(-time.Hour)
 	namespaces, err := account.GetTimeUsedNamespaceList(monitorTime, end)

service/account/dao/interface.go

@@ -2340,135 +2340,126 @@ func (m *MongoDB) GetConsumptionAmount(req helper.ConsumptionRecordReq) (int64,
 	return totalAmount, nil
 }
 
+func normalizeWorkspaceConsumptionAppType(appType string) (string, uint8, error) {
+	normalized := strings.ToUpper(strings.TrimSpace(appType))
+	if normalized == "" {
+		return "", 0, nil
+	}
+
+	value, ok := resources.AppType[normalized]
+	if !ok {
+		return "", 0, fmt.Errorf("unsupported app type %q", appType)
+	}
+	return normalized, value, nil
+}
+
+func buildWorkspaceConsumptionPipeline(req helper.ConsumptionRecordReq) (mongo.Pipeline, error) {
+	normalizedAppType, appTypeValue, err := normalizeWorkspaceConsumptionAppType(req.AppType)
+	if err != nil {
+		return nil, err
+	}
+
+	matchValue := bson.D{
+		{Key: "owner", Value: req.Owner},
+		{Key: "status", Value: resources.Settled},
+		{Key: "time", Value: bson.D{
+			{Key: "$gte", Value: req.StartTime},
+			{Key: "$lte", Value: req.EndTime},
+		}},
+	}
+	if req.Namespace != "" {
+		matchValue = append(matchValue, bson.E{Key: "namespace", Value: req.Namespace})
+	}
+	if normalizedAppType != "" {
+		matchValue = append(matchValue, bson.E{Key: "app_type", Value: appTypeValue})
+	}
+
+	groupStage := bson.D{{Key: "$group", Value: bson.D{
+		{Key: "_id", Value: "$namespace"},
+		{Key: "total", Value: bson.D{{Key: "$sum", Value: "$amount"}}},
+	}}}
+	pipeline := mongo.Pipeline{
+		{{Key: "$match", Value: matchValue}},
+	}
+
+	// Billing.Amount is the authoritative total for a billing record. Avoid
+	// unwinding app_costs unless the caller needs an app-level filter.
+	if req.AppName == "" {
+		return append(pipeline, groupStage), nil
+	}
+
+	directAppTypes := bson.A{
+		resources.AppType[resources.AppStore],
+		resources.AppType[resources.LLMToken],
+	}
+	matchedNestedAmount := bson.M{
+		"$sum": bson.M{
+			"$map": bson.M{
+				"input": bson.M{
+					"$filter": bson.M{
+						"input": bson.M{"$ifNull": bson.A{"$app_costs", bson.A{}}},
+						"as":    "appCost",
+						"cond": bson.M{
+							"$eq": bson.A{"$$appCost.name", req.AppName},
+						},
+					},
+				},
+				"as": "appCost",
+				"in": "$$appCost.amount",
+			},
+		},
+	}
+	pipeline = append(
+		pipeline,
+		bson.D{{Key: "$project", Value: bson.D{
+			{Key: "namespace", Value: 1},
+			{Key: "amount", Value: bson.D{{Key: "$cond", Value: bson.A{
+				bson.D{{Key: "$in", Value: bson.A{"$app_type", directAppTypes}}},
+				bson.D{{Key: "$cond", Value: bson.A{
+					bson.D{{Key: "$eq", Value: bson.A{"$app_name", req.AppName}}},
+					"$amount",
+					int64(0),
+				}}},
+				matchedNestedAmount,
+			}}}},
+		}}},
+		bson.D{{Key: "$match", Value: bson.D{
+			{Key: "amount", Value: bson.D{
+				{Key: "$gt", Value: int64(0)},
+			}},
+		}}},
+		groupStage,
+	)
+	return pipeline, nil
+}
+
 func (m *MongoDB) GetWorkspaceConsumptionAmount(
 	req helper.ConsumptionRecordReq,
 ) (map[string]int64, error) {
-	// 获取各个 namespace的费用
-	owner, appType, appName, startTime, endTime := req.Owner, req.AppType, req.AppName, req.StartTime, req.EndTime
-	timeMatchValue := bson.D{
-		primitive.E{Key: "$gte", Value: startTime},
-		primitive.E{Key: "$lte", Value: endTime},
+	pipeline, err := buildWorkspaceConsumptionPipeline(req)
+	if err != nil {
+		return nil, err
 	}
 
-	// Build base match conditions for app_costs (sub-consumption type)
-	matchValue := bson.D{
-		primitive.E{Key: "time", Value: timeMatchValue},
-		primitive.E{Key: "owner", Value: owner},
-		primitive.E{Key: "status", Value: resources.Settled},
-	}
-
-	// 添加app_type过滤条件
-	if appType != "" {
-		matchValue = append(
-			matchValue,
-			primitive.E{Key: "app_type", Value: resources.AppType[strings.ToUpper(appType)]},
-		)
-	}
-
-	// 构建unwind后的匹配条件
-	unwindMatchValue := bson.D{
-		primitive.E{Key: "time", Value: timeMatchValue},
-	}
-	if appType != "" && appName != "" {
-		if appType != resources.AppStore {
-			unwindMatchValue = append(
-				unwindMatchValue,
-				primitive.E{Key: "app_costs.name", Value: appName},
-			)
-		} else {
-			unwindMatchValue = append(
-				unwindMatchValue,
-				primitive.E{Key: "app_name", Value: appName},
-			)
-		}
-	}
-
-	// Build match conditions for direct consumption (AppStore and LLMToken)
-	directMatchValue := bson.D{
-		primitive.E{Key: "time", Value: timeMatchValue},
-		primitive.E{Key: "owner", Value: owner},
-		primitive.E{Key: "status", Value: resources.Settled},
-	}
-	// For direct consumption, match app_type to AppStore or LLMToken if not specified
-	if appType != "" {
-		directMatchValue = append(
-			directMatchValue,
-			primitive.E{Key: "app_type", Value: resources.AppType[strings.ToUpper(appType)]},
-		)
-	} else {
-		// If no appType specified, match both AppStore and LLMToken
-		directMatchValue = append(
-			directMatchValue,
-			primitive.E{Key: "app_type", Value: bson.D{{Key: "$in", Value: bson.A{
-				resources.AppType[resources.AppStore],
-				resources.AppType[resources.LLMToken],
-			}}}},
-		)
-	}
-	if appName != "" {
-		directMatchValue = append(directMatchValue, primitive.E{Key: "app_name", Value: appName})
-	}
-
-	// Use $facet to query both types in parallel
-	pipeline := bson.A{
-		bson.D{{Key: "$facet", Value: bson.M{
-			"appCosts": bson.A{
-				bson.D{{Key: "$match", Value: matchValue}},
-				bson.D{{Key: "$unwind", Value: "$app_costs"}},
-				bson.D{{Key: "$match", Value: unwindMatchValue}},
-				bson.D{{Key: "$group", Value: bson.M{
-					"_id":   "$namespace", // group by namespace
-					"total": bson.M{"$sum": "$app_costs.amount"},
-				}}},
-				bson.D{{Key: "$sort", Value: bson.M{"_id": 1}}},
-			},
-			"directAmount": bson.A{
-				bson.D{{Key: "$match", Value: directMatchValue}},
-				bson.D{{Key: "$group", Value: bson.M{
-					"_id":   "$namespace",
-					"total": bson.M{"$sum": "$amount"},
-				}}},
-				bson.D{{Key: "$sort", Value: bson.M{"_id": 1}}},
-			},
-		}}},
-	}
-
-	cursor, err := m.getBillingCollection().Aggregate(context.Background(), pipeline)
+	ctx := context.Background()
+	cursor, err := m.getBillingCollection().Aggregate(ctx, pipeline)
 	if err != nil {
 		return nil, fmt.Errorf("failed to aggregate billing collection: %w", err)
 	}
-	defer cursor.Close(context.Background())
+	defer cursor.Close(ctx)
 
-	var result struct {
-		AppCosts []struct {
-			Namespace string `bson:"_id"`
-			Total     int64  `bson:"total"`
-		} `bson:"appCosts"`
-		DirectAmount []struct {
-			Namespace string `bson:"_id"`
-			Total     int64  `bson:"total"`
-		} `bson:"directAmount"`
+	var results []struct {
+		Namespace string `bson:"_id"`
+		Total     int64  `bson:"total"`
+	}
+	if err := cursor.All(ctx, &results); err != nil {
+		return nil, fmt.Errorf("failed to decode workspace consumption result: %w", err)
 	}
 
-	if cursor.Next(context.Background()) {
-		if err := cursor.Decode(&result); err != nil {
-			return nil, fmt.Errorf("failed to decode result: %w", err)
-		}
+	resultMap := make(map[string]int64, len(results))
+	for _, item := range results {
+		resultMap[item.Namespace] = item.Total
 	}
-
-	// Merge results from both queries
-	resultMap := make(map[string]int64)
-
-	// Add app_costs totals
-	for _, item := range result.AppCosts {
-		resultMap[item.Namespace] += item.Total
-	}
-
-	// Add direct amount totals (AppStore and LLMToken)
-	for _, item := range result.DirectAmount {
-		resultMap[item.Namespace] += item.Total
-	}
-
 	return resultMap, nil
 }
 

service/account/dao/workspace_consumption_runtime_test.go

@@ -0,0 +1,328 @@
+package dao
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/labring/sealos/controllers/pkg/resources"
+	"github.com/labring/sealos/service/account/helper"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/wait"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+const (
+	workspaceConsumptionTestOwner        = "workspace-consumption-test-owner"
+	workspaceConsumptionTestDB           = "workspace-consumption-test"
+	workspaceConsumptionTestColl         = "billing"
+	workspaceConsumptionBenchmarkRecords = 10000
+	workspaceConsumptionRequiredEnv      = "TESTCONTAINERS_REQUIRED"
+)
+
+func newWorkspaceConsumptionMongo(tb testing.TB) (*MongoDB, context.Context) {
+	tb.Helper()
+	skipIfWorkspaceConsumptionDockerIsNotHealthy(tb)
+
+	ctx := context.Background()
+	container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
+		ContainerRequest: testcontainers.ContainerRequest{
+			Image:        "mongo:4.4.29",
+			ExposedPorts: []string{"27017/tcp"},
+			WaitingFor: wait.ForListeningPort("27017/tcp").
+				WithStartupTimeout(2 * time.Minute),
+		},
+		Started: true,
+	})
+	if err != nil {
+		tb.Fatalf("start MongoDB container: %v", err)
+	}
+	tb.Cleanup(func() {
+		if err := container.Terminate(ctx); err != nil {
+			tb.Errorf("terminate MongoDB container: %v", err)
+		}
+	})
+
+	host, err := container.Host(ctx)
+	if err != nil {
+		tb.Fatalf("get MongoDB container host: %v", err)
+	}
+	port, err := container.MappedPort(ctx, "27017/tcp")
+	if err != nil {
+		tb.Fatalf("get MongoDB container port: %v", err)
+	}
+	client, err := mongo.Connect(
+		ctx,
+		options.Client().ApplyURI("mongodb://"+net.JoinHostPort(host, port.Port())),
+	)
+	if err != nil {
+		tb.Fatalf("connect MongoDB client: %v", err)
+	}
+	if err := client.Ping(ctx, nil); err != nil {
+		_ = client.Disconnect(ctx)
+		tb.Fatalf("ping MongoDB: %v", err)
+	}
+	tb.Cleanup(func() {
+		if err := client.Disconnect(ctx); err != nil {
+			tb.Errorf("disconnect MongoDB client: %v", err)
+		}
+	})
+
+	return &MongoDB{
+		Client:        client,
+		AccountDBName: workspaceConsumptionTestDB,
+		BillingConn:   workspaceConsumptionTestColl,
+	}, ctx
+}
+
+func skipIfWorkspaceConsumptionDockerIsNotHealthy(tb testing.TB) {
+	tb.Helper()
+	defer func() {
+		if r := recover(); r != nil {
+			skipOrFailWorkspaceConsumptionDockerf(
+				tb,
+				"recovered from panic: %v; Docker is not running",
+				r,
+			)
+		}
+	}()
+
+	ctx := context.Background()
+	provider, err := testcontainers.ProviderDocker.GetProvider()
+	if err != nil {
+		skipOrFailWorkspaceConsumptionDockerf(tb, "Docker is not running: %v", err)
+	}
+	defer provider.Close()
+	if err := provider.Health(ctx); err != nil {
+		skipOrFailWorkspaceConsumptionDockerf(tb, "Docker is not running: %v", err)
+	}
+}
+
+func skipOrFailWorkspaceConsumptionDockerf(tb testing.TB, format string, args ...any) {
+	tb.Helper()
+	if os.Getenv(workspaceConsumptionRequiredEnv) == "true" {
+		tb.Fatalf(format, args...)
+	}
+	tb.Skipf(format, args...)
+}
+
+func TestGetWorkspaceConsumptionAmountWithMongoRuntime(t *testing.T) {
+	mongoDB, ctx := newWorkspaceConsumptionMongo(t)
+	startTime := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+	endTime := time.Date(2026, time.January, 2, 0, 0, 0, 0, time.UTC)
+
+	appCosts := []resources.AppCost{
+		{Name: "app-a", Amount: 30},
+		{Name: "app-b", Amount: 5},
+	}
+	documents := []any{
+		resources.Billing{
+			Time: startTime, OrderID: "nested-at-start", Type: resources.Consumption,
+			Namespace: "ns-a", AppCosts: appCosts, AppType: resources.AppType[resources.APP],
+			Amount: 100, Owner: workspaceConsumptionTestOwner, Status: resources.Settled,
+		},
+		resources.Billing{
+			Time: endTime, OrderID: "nested-at-end", Type: resources.Consumption,
+			Namespace: "ns-a", AppType: resources.AppType[resources.APP], Amount: 50,
+			Owner: workspaceConsumptionTestOwner, Status: resources.Settled,
+		},
+		resources.Billing{
+			Time: endTime, OrderID: "llm-subconsumption", Type: resources.SubConsumption,
+			Namespace: "ns-b", AppName: "llm-a", AppType: resources.AppType[resources.LLMToken],
+			Amount: 20, Owner: workspaceConsumptionTestOwner, Status: resources.Settled,
+		},
+		resources.Billing{
+			Time: endTime, OrderID: "app-store-direct", Type: resources.Consumption,
+			Namespace: "ns-c", AppName: "store-a", AppType: resources.AppType[resources.AppStore],
+			Amount: 40, Owner: workspaceConsumptionTestOwner, Status: resources.Settled,
+		},
+		resources.Billing{
+			Time: endTime, OrderID: "unsettled", Type: resources.Consumption,
+			Namespace: "ns-ignored", AppType: resources.AppType[resources.APP], Amount: 1000,
+			Owner: workspaceConsumptionTestOwner, Status: resources.Unsettled,
+		},
+		resources.Billing{
+			Time: endTime, OrderID: "other-owner", Type: resources.Consumption,
+			Namespace: "ns-ignored", AppType: resources.AppType[resources.APP], Amount: 2000,
+			Owner: "other-owner", Status: resources.Settled,
+		},
+		resources.Billing{
+			Time: endTime.Add(time.Hour), OrderID: "outside-range", Type: resources.Consumption,
+			Namespace: "ns-ignored", AppType: resources.AppType[resources.APP], Amount: 3000,
+			Owner: workspaceConsumptionTestOwner, Status: resources.Settled,
+		},
+	}
+	collection := mongoDB.getBillingCollection()
+	if _, err := collection.InsertMany(ctx, documents); err != nil {
+		t.Fatalf("insert billing fixtures: %v", err)
+	}
+
+	baseRequest := helper.ConsumptionRecordReq{
+		TimeRange: helper.TimeRange{StartTime: startTime, EndTime: endTime},
+		AuthBase:  helper.AuthBase{Auth: &helper.Auth{Owner: workspaceConsumptionTestOwner}},
+	}
+	tests := []struct {
+		name string
+		req  helper.ConsumptionRecordReq
+		want map[string]int64
+	}{
+		{
+			name: "all settled consumption by namespace",
+			req:  baseRequest,
+			want: map[string]int64{"ns-a": 150, "ns-b": 20, "ns-c": 40},
+		},
+		{
+			name: "namespace filter",
+			req: withWorkspaceConsumptionRequest(
+				baseRequest,
+				func(req *helper.ConsumptionRecordReq) {
+					req.Namespace = "ns-a"
+				},
+			),
+			want: map[string]int64{"ns-a": 150},
+		},
+		{
+			name: "app type filter",
+			req: withWorkspaceConsumptionRequest(
+				baseRequest,
+				func(req *helper.ConsumptionRecordReq) {
+					req.AppType = " llm-token "
+				},
+			),
+			want: map[string]int64{"ns-b": 20},
+		},
+		{
+			name: "nested app name filter",
+			req: withWorkspaceConsumptionRequest(
+				baseRequest,
+				func(req *helper.ConsumptionRecordReq) {
+					req.AppName = "app-a"
+				},
+			),
+			want: map[string]int64{"ns-a": 30},
+		},
+		{
+			name: "direct app name filter",
+			req: withWorkspaceConsumptionRequest(
+				baseRequest,
+				func(req *helper.ConsumptionRecordReq) {
+					req.AppName = "store-a"
+				},
+			),
+			want: map[string]int64{"ns-c": 40},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got, err := mongoDB.GetWorkspaceConsumptionAmount(test.req)
+			if err != nil {
+				t.Fatalf("get workspace consumption amount: %v", err)
+			}
+			if !mapsEqual(got, test.want) {
+				t.Fatalf("workspace consumption = %#v, want %#v", got, test.want)
+			}
+		})
+	}
+}
+
+func BenchmarkGetWorkspaceConsumptionAmountWithMongoRuntime(b *testing.B) {
+	mongoDB, ctx := newWorkspaceConsumptionMongo(b)
+	startTime := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)
+	endTime := startTime.Add(24 * time.Hour)
+	documents := make([]any, 0, workspaceConsumptionBenchmarkRecords)
+	for i := range workspaceConsumptionBenchmarkRecords {
+		documents = append(documents, resources.Billing{
+			Time:      startTime.Add(time.Duration(i%24) * time.Hour),
+			OrderID:   fmt.Sprintf("benchmark-%d", i),
+			Type:      resources.Consumption,
+			Namespace: fmt.Sprintf("ns-%02d", i%32),
+			AppCosts: []resources.AppCost{{
+				Name:   "app-a",
+				Amount: int64(i%100 + 1),
+			}},
+			AppType: resources.AppType[resources.APP],
+			Amount:  int64(i%100 + 1),
+			Owner:   workspaceConsumptionTestOwner,
+			Status:  resources.Settled,
+		})
+	}
+	if _, err := mongoDB.getBillingCollection().InsertMany(ctx, documents); err != nil {
+		b.Fatalf("insert billing benchmark fixtures: %v", err)
+	}
+
+	baseRequest := helper.ConsumptionRecordReq{
+		TimeRange: helper.TimeRange{StartTime: startTime, EndTime: endTime},
+		AuthBase:  helper.AuthBase{Auth: &helper.Auth{Owner: workspaceConsumptionTestOwner}},
+	}
+	benchmarks := []struct {
+		name string
+		req  helper.ConsumptionRecordReq
+	}{
+		{name: "all", req: baseRequest},
+		{
+			name: "namespace",
+			req: withWorkspaceConsumptionRequest(
+				baseRequest,
+				func(req *helper.ConsumptionRecordReq) {
+					req.Namespace = "ns-07"
+				},
+			),
+		},
+		{
+			name: "app_type",
+			req: withWorkspaceConsumptionRequest(
+				baseRequest,
+				func(req *helper.ConsumptionRecordReq) {
+					req.AppType = "app"
+				},
+			),
+		},
+		{
+			name: "app_name",
+			req: withWorkspaceConsumptionRequest(
+				baseRequest,
+				func(req *helper.ConsumptionRecordReq) {
+					req.AppName = "app-a"
+				},
+			),
+		},
+	}
+
+	for _, benchmark := range benchmarks {
+		b.Run(benchmark.name, func(b *testing.B) {
+			b.ReportAllocs()
+			b.ResetTimer()
+			for range b.N {
+				if _, err := mongoDB.GetWorkspaceConsumptionAmount(benchmark.req); err != nil {
+					b.Fatalf("get workspace consumption amount: %v", err)
+				}
+			}
+		})
+	}
+}
+
+func withWorkspaceConsumptionRequest(
+	base helper.ConsumptionRecordReq,
+	update func(*helper.ConsumptionRecordReq),
+) helper.ConsumptionRecordReq {
+	request := base
+	update(&request)
+	return request
+}
+
+func mapsEqual(got, want map[string]int64) bool {
+	if len(got) != len(want) {
+		return false
+	}
+	for key, wantValue := range want {
+		if got[key] != wantValue {
+			return false
+		}
+	}
+	return true
+}

service/account/dao/workspace_consumption_test.go

@@ -0,0 +1,119 @@
+package dao
+
+import (
+	"testing"
+	"time"
+
+	"github.com/labring/sealos/controllers/pkg/resources"
+	"github.com/labring/sealos/service/account/helper"
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+func workspaceConsumptionRequest() helper.ConsumptionRecordReq {
+	return helper.ConsumptionRecordReq{
+		TimeRange: helper.TimeRange{
+			StartTime: time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC),
+			EndTime:   time.Date(2026, time.January, 2, 0, 0, 0, 0, time.UTC),
+		},
+		Namespace: "ns-test",
+		AuthBase: helper.AuthBase{
+			Auth: &helper.Auth{Owner: "owner-test"},
+		},
+	}
+}
+
+func workspaceConsumptionStageValue(stage bson.D, key string) (any, bool) {
+	for _, element := range stage {
+		if element.Key == key {
+			return element.Value, true
+		}
+	}
+	return nil, false
+}
+
+func TestNormalizeWorkspaceConsumptionAppType(t *testing.T) {
+	normalized, value, err := normalizeWorkspaceConsumptionAppType(" app-store ")
+	if err != nil {
+		t.Fatalf("normalize app type: %v", err)
+	}
+	if normalized != resources.AppStore {
+		t.Fatalf("normalized app type = %q, want %q", normalized, resources.AppStore)
+	}
+	if value != resources.AppType[resources.AppStore] {
+		t.Fatalf("app type value = %d, want %d", value, resources.AppType[resources.AppStore])
+	}
+
+	if _, _, err := normalizeWorkspaceConsumptionAppType("unknown"); err == nil {
+		t.Fatal("expected an error for an unsupported app type")
+	}
+}
+
+func TestBuildWorkspaceConsumptionPipelineWithoutAppFilter(t *testing.T) {
+	pipeline, err := buildWorkspaceConsumptionPipeline(workspaceConsumptionRequest())
+	if err != nil {
+		t.Fatalf("build pipeline: %v", err)
+	}
+	if len(pipeline) != 2 {
+		t.Fatalf("pipeline stage count = %d, want 2", len(pipeline))
+	}
+
+	matchValue, ok := workspaceConsumptionStageValue(pipeline[0], "$match")
+	if !ok {
+		t.Fatal("pipeline does not start with $match")
+	}
+	match, ok := matchValue.(bson.D)
+	if !ok {
+		t.Fatalf("$match value type = %T, want bson.D", matchValue)
+	}
+	for key, want := range map[string]any{
+		"owner":     "owner-test",
+		"namespace": "ns-test",
+		"status":    resources.Settled,
+	} {
+		got, ok := workspaceConsumptionStageValue(match, key)
+		if !ok || got != want {
+			t.Errorf("$match[%q] = %#v, want %#v", key, got, want)
+		}
+	}
+
+	if _, ok := workspaceConsumptionStageValue(pipeline[1], "$group"); !ok {
+		t.Fatal("pipeline does not end with $group")
+	}
+}
+
+func TestBuildWorkspaceConsumptionPipelineWithAppFilter(t *testing.T) {
+	req := workspaceConsumptionRequest()
+	req.AppType = " app "
+	req.AppName = "application"
+
+	pipeline, err := buildWorkspaceConsumptionPipeline(req)
+	if err != nil {
+		t.Fatalf("build pipeline: %v", err)
+	}
+	if len(pipeline) != 4 {
+		t.Fatalf("pipeline stage count = %d, want 4", len(pipeline))
+	}
+
+	matchValue, ok := workspaceConsumptionStageValue(pipeline[0], "$match")
+	if !ok {
+		t.Fatal("pipeline does not start with $match")
+	}
+	match, ok := matchValue.(bson.D)
+	if !ok {
+		t.Fatalf("$match value type = %T, want bson.D", matchValue)
+	}
+	appType, ok := workspaceConsumptionStageValue(match, "app_type")
+	if !ok || appType != resources.AppType[resources.APP] {
+		t.Fatalf("$match app_type = %#v, want %d", appType, resources.AppType[resources.APP])
+	}
+
+	if _, ok := workspaceConsumptionStageValue(pipeline[1], "$facet"); ok {
+		t.Fatal("app-filtered pipeline should not use $facet")
+	}
+	if _, ok := workspaceConsumptionStageValue(pipeline[1], "$unwind"); ok {
+		t.Fatal("app-filtered pipeline should keep one row per billing record")
+	}
+	if _, ok := workspaceConsumptionStageValue(pipeline[2], "$match"); !ok {
+		t.Fatal("app-filtered pipeline should discard zero matched amounts")
+	}
+}

service/account/go.mod

@@ -33,6 +33,7 @@ require (
 	github.com/swaggo/files v1.0.1
 	github.com/swaggo/gin-swagger v1.6.0
 	github.com/swaggo/swag v1.16.2
+	github.com/testcontainers/testcontainers-go v0.42.0
 	go.mongodb.org/mongo-driver v1.13.0
 	gorm.io/gorm v1.25.5
 	k8s.io/api v0.32.1