package nextgenmetering import ( "context" "encoding/json" "fmt" "os" "path/filepath" "sort" "strconv" "strings" "sync/atomic" "testing" "time" sdkcommon "github.com/pingcap/metering_sdk/common" sdkconfig "github.com/pingcap/metering_sdk/config" metawriter "github.com/pingcap/metering_sdk/writer/meta" meteringwriter "github.com/pingcap/metering_sdk/writer/metering" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/tidbcloud/metering/pkg/common/config" "github.com/tidbcloud/metering/pkg/common/constants" "github.com/tidbcloud/metering/pkg/common/models" "github.com/tidbcloud/metering/pkg/common/utils" "github.com/tidbcloud/metering/pkg/common/utils/deltautils" "go.uber.org/zap" ) var meteringTestCounter int64 // getMeteringUniqueTimestamp generates a unique timestamp for each metering test func getMeteringUniqueTimestamp() int64 { counter := atomic.AddInt64(&meteringTestCounter, 1) // Start from a far future timestamp and add counter to ensure uniqueness return time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).Unix() + counter*120 } // testWriters holds writers for tests that need to write data type testWriters struct { meteringWriter *meteringwriter.MeteringWriter metaWriter *metawriter.MetaWriter } // newTestWriters creates writers for test data using the loader's storage func newTestWriters(ng *NextGenMetering) *testWriters { sdkCfg := sdkconfig.DefaultConfig() sdkCfg = sdkCfg.WithLogger(ng.log) return &testWriters{ meteringWriter: meteringwriter.NewMeteringWriter(ng.loader.storage, sdkCfg), metaWriter: metawriter.NewMetaWriter(ng.loader.storage, sdkCfg), } } // WriteMeteringData writes metering data for testing func (w *testWriters) WriteMeteringData(ctx context.Context, data *sdkcommon.MeteringData) error { return w.meteringWriter.Write(ctx, data) } // WriteMetaData writes metadata for testing func (w *testWriters) WriteMetaData(ctx context.Context, clusterID string, metaType sdkcommon.MetaType, timestamp int64, data *sdkcommon.MetaData) error { data.Type = metaType return w.metaWriter.Write(ctx, data) } // Close closes the writers func (w *testWriters) Close() error { var errs []error if err := w.meteringWriter.Close(); err != nil { errs = append(errs, err) } if err := w.metaWriter.Close(); err != nil { errs = append(errs, err) } if len(errs) > 0 { return fmt.Errorf("errors during close: %v", errs) } return nil } // setupTestNextGenMetering creates a test NextGenMetering instance func setupTestNextGenMetering(t *testing.T) *NextGenMetering { // Reset global table manager to ensure clean state for each test deltautils.ResetGlobalTableManager() logger := zap.NewNop() // Create temporary directory for test delta table tmpDir := t.TempDir() deltaTablePath := filepath.Join(tmpDir, "test_delta_table") //create deltaTablePath err := os.MkdirAll(deltaTablePath, 0755) require.NoError(t, err) // Use test name to create unique prefix testName := strings.ReplaceAll(t.Name(), "/", "_") testPrefix := fmt.Sprintf("test_%s_%d/", testName, time.Now().UnixNano()) // cleanup metering sdk local path os.RemoveAll("/tmp/nextgen-metering-data") conf := &config.ServiceConfig{ Vendor: "local", LogLevel: "info", NextGenMeteringConfig: &config.NextGenMeteringConfig{ MeteringPrefix: testPrefix, DeltaTablePath: deltaTablePath, EnableRecompute: false, StorageOptions: make(map[string]string), }, } ng, err := NewNextGenMetering(conf, logger) require.NoError(t, err) require.NotNil(t, ng) return ng } // setupTestNextGenMeteringWithWriters creates a test NextGenMetering instance with test writers func setupTestNextGenMeteringWithWriters(t *testing.T) (*NextGenMetering, *testWriters) { ng := setupTestNextGenMetering(t) writers := newTestWriters(ng) return ng, writers } func TestPrepareAlicloudStorageOptions(t *testing.T) { t.Setenv(constants.AlicloudAccessKeyIDEnv, "ak-test") t.Setenv(constants.AlicloudAccessKeySecretEnv, "sk-test") conf := &config.ServiceConfig{ Vendor: constants.VendorAlicloud, Region: "oss-cn-test", Bucket: "my-test-bucket", NextGenMeteringConfig: &config.NextGenMeteringConfig{ BucketEndpoint: "https://oss-cn-test.aliyuncs.com", StorageOptions: map[string]string{ "AWS_S3_FORCE_PATH_STYLE": "false", "custom_option": "keep", }, }, } err := deltautils.PrepareStorageOptions(conf) require.NoError(t, err) opts := conf.NextGenMeteringConfig.StorageOptions assert.Equal(t, "ak-test", opts["AWS_ACCESS_KEY_ID"]) assert.Equal(t, "sk-test", opts["AWS_SECRET_ACCESS_KEY"]) assert.Equal(t, "https://my-test-bucket.oss-cn-test.aliyuncs.com", opts["AWS_ENDPOINT_URL"]) assert.Equal(t, conf.Region, opts["region"]) assert.Equal(t, conf.Region, opts["AWS_REGION"]) assert.Equal(t, "false", opts["AWS_S3_FORCE_PATH_STYLE"]) assert.Equal(t, "keep", opts["custom_option"]) assert.Equal(t, "https://oss-cn-test.aliyuncs.com", conf.NextGenMeteringConfig.BucketEndpoint) } func TestPrepareAlicloudStorageOptionsMissingEndpoint(t *testing.T) { t.Setenv(constants.AlicloudAccessKeyIDEnv, "ak-test") t.Setenv(constants.AlicloudAccessKeySecretEnv, "sk-test") conf := &config.ServiceConfig{ Vendor: constants.VendorAlicloud, NextGenMeteringConfig: &config.NextGenMeteringConfig{}, } err := deltautils.PrepareStorageOptions(conf) assert.Error(t, err) assert.Contains(t, err.Error(), "bucketEndpoint") } func TestNextGenMetering_Constructor(t *testing.T) { logger := zap.NewNop() tests := []struct { name string conf *config.ServiceConfig wantErr bool errMsg string setup func(t *testing.T) func() }{ { name: "valid config", conf: &config.ServiceConfig{ Vendor: "local", LogLevel: "info", NextGenMeteringConfig: &config.NextGenMeteringConfig{ MeteringPrefix: "test/", DeltaTablePath: "/tmp/test-delta", EnableRecompute: false, }, }, wantErr: false, }, { name: "missing NextGenMeteringConfig", conf: &config.ServiceConfig{ Vendor: "local", LogLevel: "info", }, wantErr: true, errMsg: "NextGenMeteringConfig is required", }, { name: "valid config with AWS vendor", conf: &config.ServiceConfig{ Vendor: "aws", LogLevel: "info", NextGenMeteringConfig: &config.NextGenMeteringConfig{ MeteringPrefix: "test/", DeltaTablePath: "/tmp/test-delta-aws", EnableRecompute: true, }, }, wantErr: false, setup: func(t *testing.T) func() { t.Helper() prevAccessKey, accessExists := os.LookupEnv("AWS_ACCESS_KEY_ID") prevSecretKey, secretExists := os.LookupEnv("AWS_SECRET_ACCESS_KEY") if err := os.Setenv("AWS_ACCESS_KEY_ID", "fake-access-key"); err != nil { t.Fatalf("failed to set AWS_ACCESS_KEY_ID: %v", err) } if err := os.Setenv("AWS_SECRET_ACCESS_KEY", "fake-secret-key"); err != nil { t.Fatalf("failed to set AWS_SECRET_ACCESS_KEY: %v", err) } return func() { if accessExists { _ = os.Setenv("AWS_ACCESS_KEY_ID", prevAccessKey) } else { _ = os.Unsetenv("AWS_ACCESS_KEY_ID") } if secretExists { _ = os.Setenv("AWS_SECRET_ACCESS_KEY", prevSecretKey) } else { _ = os.Unsetenv("AWS_SECRET_ACCESS_KEY") } } }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { var cleanup func() if tt.setup != nil { cleanup = tt.setup(t) if cleanup != nil { defer cleanup() } } ng, err := NewNextGenMetering(tt.conf, logger) if tt.wantErr { assert.Error(t, err) assert.Contains(t, err.Error(), tt.errMsg) assert.Nil(t, ng) } else { assert.NoError(t, err) assert.NotNil(t, ng) assert.Equal(t, tt.conf, ng.conf) assert.NotNil(t, ng.loader) assert.NotNil(t, ng.log) assert.Equal(t, "nextgen-metering", ng.taskName) assert.NotNil(t, ng.skuGenerator) // Clean up if ng.loader != nil { ng.loader.Close() } } }) } } func TestNextGenMetering_CreateDeltaTable(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() // Test creating a new delta table err := ng.CreateDeltaTable() assert.NoError(t, err) assert.NotNil(t, ng.tableManager) assert.True(t, ng.tableManager.IsOpen()) // Verify table exists exists, err := ng.TableExists(context.Background()) assert.NoError(t, err) assert.True(t, exists) // Test creating table when it already exists should fail err = ng.CreateDeltaTable() assert.Error(t, err) assert.Contains(t, err.Error(), "delta table already opened") // Reset global table manager after this test t.Cleanup(func() { deltautils.ResetGlobalTableManager() }) } func TestNextGenMetering_OpenDeltaTable(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() // Clean up after test t.Cleanup(func() { deltautils.ResetGlobalTableManager() }) // Test opening non-existent table should fail err := ng.openDeltaTable() assert.Error(t, err) assert.Contains(t, err.Error(), "failed to open delta table") // Create the table first err = ng.CreateDeltaTable() require.NoError(t, err) // Close table to test reopening ng.tableManager.Close() // Test opening existing table should succeed err = ng.openDeltaTable() assert.NoError(t, err) assert.True(t, ng.tableManager.IsOpen()) // Test opening already opened table should succeed (no-op) err = ng.openDeltaTable() assert.NoError(t, err) assert.True(t, ng.tableManager.IsOpen()) } func TestNextGenMetering_Close(t *testing.T) { ng := setupTestNextGenMetering(t) // Create delta table err := ng.CreateDeltaTable() require.NoError(t, err) // Close should succeed (note: Close does not close the global table manager) err = ng.Close() assert.NoError(t, err) assert.Nil(t, ng.loader) // loader should be set to nil // Calling close again should not error err = ng.Close() assert.NoError(t, err) // Clean up global table manager t.Cleanup(func() { deltautils.ResetGlobalTableManager() }) } func TestNextGenMetering_TableExists(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() // Clean up after test t.Cleanup(func() { deltautils.ResetGlobalTableManager() }) // Test non-existent table - returns (false, nil), not an error exists, err := ng.TableExists(context.Background()) assert.NoError(t, err) assert.False(t, exists) // Create table err = ng.CreateDeltaTable() require.NoError(t, err) // Test existing table exists, err = ng.TableExists(context.Background()) assert.NoError(t, err) assert.True(t, exists) } func TestNextGenMetering_GetTableVersion(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() // Clean up after test t.Cleanup(func() { deltautils.ResetGlobalTableManager() }) // Test with no table opened - should return error (table doesn't exist or not opened) _, err := ng.GetTableVersion(context.Background()) assert.Error(t, err) // Create table err = ng.CreateDeltaTable() require.NoError(t, err) // Test with opened table version, err := ng.GetTableVersion(context.Background()) assert.NoError(t, err) assert.Equal(t, int64(0), version) // Initial version should be 0 } func TestNextGenMetering_QueryMaxTimestampWithPartitionFilter(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() // Clean up after test t.Cleanup(func() { deltautils.ResetGlobalTableManager() }) // Create table first to avoid "delta table not opened" error err := ng.CreateDeltaTable() require.NoError(t, err) timestamp1 := time.Now().Add(-2 * time.Hour).UTC().Unix() timestamp2 := time.Now().Add(-1 * time.Hour).UTC().Unix() records := []models.BillingReportV2Deltalake{ { SchemeVersion: "V1", TenantID: 1, CostID: "cost-1", CostTimestamp: timestamp1, Unit: "unit-1", UnitValue: "1", Cost: 100, Labels: "{}", CreatedAt: timestamp1, SKUType: models.NextGenSKUTypePublic, HourDimension: utils.GenerateHourDimension(timestamp1), }, { SchemeVersion: "V1", TenantID: 1, CostID: "cost-2", CostTimestamp: timestamp2, Unit: "unit-1", UnitValue: "1", Cost: 200, Labels: "{}", CreatedAt: timestamp2, SKUType: models.NextGenSKUTypePublic, HourDimension: utils.GenerateHourDimension(timestamp2), }, } err = ng.WriteToDeltaTable(records, "Append") require.NoError(t, err) // Query with partition time before first record - should find both records maxTimestamp, err := ng.QueryMaxTimestampWithPartitionFilter(context.Background(), timestamp1-3600) require.NoError(t, err) assert.Equal(t, timestamp2, maxTimestamp) // Query with partition time after second record // Due to -30 day offset, it will still capture existing data if within 30 days maxTimestamp, err = ng.QueryMaxTimestampWithPartitionFilter(context.Background(), timestamp2+3600) require.NoError(t, err) // Should still find timestamp2 since it's within the 30-day window assert.Equal(t, timestamp2, maxTimestamp) // Query with partition time 31 days in the future - should return 0 // This is beyond the 30-day lookback window, so no data should be found futureTime := time.Now().Add(31 * 24 * time.Hour).UTC().Unix() maxTimestamp, err = ng.QueryMaxTimestampWithPartitionFilter(context.Background(), futureTime) require.NoError(t, err) assert.Equal(t, int64(0), maxTimestamp) } func TestNextGenMetering_ProcessMinute(t *testing.T) { ng, writers := setupTestNextGenMeteringWithWriters(t) defer ng.Close() defer writers.Close() ctx := context.Background() timestamp := time.Now().UnixNano()/1000000 + 7200000 // Add 2 hours in milliseconds, then convert to seconds // Create delta table first err := ng.CreateDeltaTable() require.NoError(t, err) // Test with no metering data _, err = ng.ProcessMinute(ctx, timestamp) assert.NoError(t, err) // Should succeed but do nothing // Write some test metering data first meteringData := ng.loader.CreateMeteringData( "testserverprocessmin", models.NextGenCategoryResourceManager, timestamp, []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "test_cluster_processmin", models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, "memory_gb": 16, }, }, ) err = writers.WriteMeteringData(ctx, meteringData) require.NoError(t, err) // Allow time for write time.Sleep(100 * time.Millisecond) // Test processing with data (this might fail due to missing metadata) _, err = ng.ProcessMinute(ctx, timestamp) // This is expected to fail due to missing metadata in test environment if err != nil { t.Logf("ProcessMinute failed as expected due to missing metadata: %v", err) assert.Contains(t, err.Error(), "failed to generate billing records") } } func TestNextGenMetering_GenerateBillingRecords(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() tests := []struct { name string meteringData []*sdkcommon.MeteringData expectError bool expectEmpty bool }{ { name: "empty metering data", meteringData: []*sdkcommon.MeteringData{}, expectError: false, expectEmpty: true, }, { name: "data with cluster_id", meteringData: []*sdkcommon.MeteringData{ { SelfID: "testserver", Category: models.NextGenCategoryResourceManager, Timestamp: timestamp, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "test-cluster", models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, }, }, }, }, expectError: false, // Should succeed but return empty due to missing metadata expectEmpty: true, }, { name: "data without cluster_id (shared pool)", meteringData: []*sdkcommon.MeteringData{ { SelfID: "testserver", SharedPoolID: "shared-pool-123", Category: models.NextGenCategoryResourceManager, Timestamp: timestamp, Data: []map[string]interface{}{ { models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, }, }, }, }, expectError: false, // Should succeed but return empty due to missing metadata expectEmpty: true, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { records, err := ng.generateBillingRecords(ctx, tt.meteringData, timestamp) if tt.expectError { assert.Error(t, err) } else { assert.NoError(t, err) if tt.expectEmpty { assert.Empty(t, records) } else { assert.NotEmpty(t, records) } } }) } t.Run("byoc mode skips non-worker-node categories", func(t *testing.T) { previousPlan := ng.conf.NextGenMeteringConfig.Plan ng.conf.NextGenMeteringConfig.Plan = models.PlanBYOC defer func() { ng.conf.NextGenMeteringConfig.Plan = previousPlan }() meteringData := []*sdkcommon.MeteringData{ { SelfID: "byoc-server", Category: models.NextGenCategoryResourceManager, Timestamp: timestamp, Data: []map[string]interface{}{ {models.NextGenMeteringSourceName: "tikv"}, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) assert.NoError(t, err) assert.Empty(t, records) }) t.Run("reserved node category aggregates data points", func(t *testing.T) { meteringData := []*sdkcommon.MeteringData{ { SelfID: "reserved-server", Category: models.NextGenCategoryReservedNode, Timestamp: timestamp, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "reserved-cluster", "usage": map[string]interface{}{ "value": uint64(0), "unit": "vCPU", }, }, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) assert.NoError(t, err) assert.Empty(t, records) }) } func TestNextGenMetering_GenerateBillingRecords_ManualBackup(t *testing.T) { ng, writers := setupTestNextGenMeteringWithWriters(t) defer ng.Close() defer writers.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() clusterID := "manual-backup-cluster" sharedPoolID := "manual-backup-pool" metadata := map[string]interface{}{ models.NextGenMetaTenantID: "12345", models.NextGenMeteringRegion: "ap-south-1", models.NextGenMetaPlan: models.PlanPremium, models.NextGenMetaRUVersion: "1", models.NextGenMetaMinRCU: "5000", models.NextGenMetaMaxRCU: "20000", models.NextGenMetaDiskType: "gp3", models.NextGenMetaStorageEncryptionEnabled: false, } metaData := ng.loader.CreateMetaData(clusterID, timestamp, metadata) require.NoError(t, writers.WriteMetaData(ctx, clusterID, sdkcommon.MetaTypeLogic, timestamp, metaData)) time.Sleep(100 * time.Millisecond) meteringData := []*sdkcommon.MeteringData{ { SelfID: "manual-backup-self", Timestamp: timestamp, Category: models.NextGenCategoryManualBackup, SharedPoolID: sharedPoolID, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: clusterID, models.NextGenMeteringBackupID: "backup-1", models.NextGenMeteringBackupSizeGB: map[string]interface{}{ "value": uint64(128), "unit": "GB", }, }, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) require.NoError(t, err) require.Len(t, records, 1) record := records[0] assert.Equal(t, "backup-storage:local:ap-south-1:premium:manual", record.Unit) assert.Equal(t, "128", record.UnitValue) assert.Equal(t, uint64(12345), record.TenantID) var labels map[string]string require.NoError(t, json.Unmarshal([]byte(record.Labels), &labels)) assert.Equal(t, "gb", labels[models.NextGenSKULabelKeySourceUnit]) assert.Equal(t, models.NextGenBackupTypeManual, labels[models.NextGenSKULabelKeyBackupType]) assert.Equal(t, "backup-1", labels[models.NextGenSKULabelKeyBackupID]) assert.Equal(t, clusterID, labels[models.NextGenSKULabelKeyClusterID]) assert.Equal(t, sharedPoolID, labels[models.NextGenSKULabelKeySharedPoolID]) } func TestNextGenMetering_GenerateBillingRecords_FlexBackup(t *testing.T) { ng, writers := setupTestNextGenMeteringWithWriters(t) defer ng.Close() defer writers.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() clusterID := "flex-backup-cluster" sharedPoolID := "flex-backup-pool" metadata := map[string]interface{}{ models.NextGenMetaTenantID: "12345", models.NextGenMeteringRegion: "ap-northeast-1", models.NextGenMetaPlan: models.PlanPremium, models.NextGenMetaRUVersion: "1", models.NextGenMetaMinRCU: "5000", models.NextGenMetaMaxRCU: "20000", models.NextGenMetaDiskType: "gp3", models.NextGenMetaStorageEncryptionEnabled: false, } metaData := ng.loader.CreateMetaData(clusterID, timestamp, metadata) require.NoError(t, writers.WriteMetaData(ctx, clusterID, sdkcommon.MetaTypeLogic, timestamp, metaData)) time.Sleep(100 * time.Millisecond) meteringData := []*sdkcommon.MeteringData{ { SelfID: "flex-backup-self", Timestamp: timestamp, Category: models.NextGenCategoryFlexBackup, SharedPoolID: sharedPoolID, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: clusterID, models.NextGenMeteringBackupID: "flex-backup-1", models.NextGenMeteringBackupSizeGB: map[string]interface{}{ "value": uint64(64), "unit": "GB", }, }, { models.NextGenMeteringLogicalClusterID: clusterID, models.NextGenMeteringBackupID: "flex-backup-2", models.NextGenMeteringBackupSizeGB: map[string]interface{}{ "value": uint64(200), "unit": "GB", }, }, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) require.NoError(t, err) require.Len(t, records, 2) sort.Slice(records, func(i, j int) bool { ai, _ := strconv.ParseUint(records[i].UnitValue, 10, 64) aj, _ := strconv.ParseUint(records[j].UnitValue, 10, 64) return ai < aj }) first := records[0] assert.Equal(t, "backup-storage:local:ap-northeast-1:premium:daily", first.Unit) assert.Equal(t, "64", first.UnitValue) assert.Equal(t, int64(1), first.Cost) assert.Equal(t, uint64(12345), first.TenantID) var firstLabels map[string]string require.NoError(t, json.Unmarshal([]byte(first.Labels), &firstLabels)) assert.Equal(t, "gb", firstLabels[models.NextGenSKULabelKeySourceUnit]) assert.Equal(t, models.NextGenBackupTypeFlex, firstLabels[models.NextGenSKULabelKeyBackupType]) assert.Equal(t, "flex-backup-1", firstLabels[models.NextGenSKULabelKeyBackupID]) assert.Equal(t, clusterID, firstLabels[models.NextGenSKULabelKeyClusterID]) assert.Equal(t, sharedPoolID, firstLabels[models.NextGenSKULabelKeySharedPoolID]) second := records[1] assert.Equal(t, "backup-storage:local:ap-northeast-1:premium:daily", second.Unit) assert.Equal(t, "200", second.UnitValue) } func TestNextGenMetering_GenerateBillingRecordsForCluster(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() clusterID := "test-cluster-123" category := models.NextGenCategoryResourceManager dataPoints := []map[string]interface{}{ { models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, "memory_gb": 16, }, } // Test will return empty records due to missing metadata, which is expected in test environment records, err := ng.generateBillingRecordsForCluster(ctx, clusterID, "", timestamp, dataPoints, category) assert.NoError(t, err) // Should not error but return empty records assert.Empty(t, records) // No records due to missing metadata } func TestNextGenMetering_DealSKU(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() clusterID := "test-cluster-123" category := models.NextGenCategoryResourceManager dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, "memory_gb": 16, } // Test with cluster ID (will return empty due to missing metadata) records, err := ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, category) assert.NoError(t, err) // Should not error but return empty records assert.Empty(t, records) // No records due to missing metadata // Test with shared pool (will return empty due to missing metadata) records, err = ng.dealSKU(ctx, "", "shared-pool-123", timestamp, dataPoint, category) assert.NoError(t, err) // Should not error but return empty records assert.Empty(t, records) // No records due to missing metadata // Test with system ID (will return empty due to missing metadata) records, err = ng.dealSKU(ctx, models.NextGenMeteringSystemID, "shared-pool-123", timestamp, dataPoint, category) assert.NoError(t, err) // Should not error but return empty records assert.Empty(t, records) // No records due to missing metadata } func TestNextGenMetering_WorkerNodeCategory(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() timestamp := getMeteringUniqueTimestamp() now := time.Now() t.Run("worker-node category generates vCPU SKU records", func(t *testing.T) { dataPoint := map[string]interface{}{ "usage": map[string]interface{}{ "value": uint64(16), "unit": "vCPU", }, } records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryWorkerNode, "", // clusterID - empty for BYOC "shared-pool-byoc", // sharedPoolID 12345, // tenantID timestamp, "aws", "us-west-2", "byoc", map[string]interface{}{}, nil, now, ) require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, "vcpu:aws:us-west-2:byoc:total", records[0].Unit) assert.Equal(t, uint64(12345), records[0].TenantID) assert.Equal(t, "16", records[0].UnitValue) assert.Equal(t, models.NextGenSKUTypePublic, records[0].SKUType) }) t.Run("worker-node with no vCPU data returns empty", func(t *testing.T) { dataPoint := map[string]interface{}{} records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryWorkerNode, "", "pool-empty", 54321, timestamp, "gcp", "us-central1", "premium", map[string]interface{}{}, nil, now, ) require.NoError(t, err) assert.Empty(t, records) }) } func TestNextGenMetering_BYOCPlan_Filtering(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() ng.conf.NextGenMeteringConfig.Plan = "byoc" t.Run("non-worker-node categories are skipped", func(t *testing.T) { meteringData := []*sdkcommon.MeteringData{ { SelfID: "testserver", Category: models.NextGenCategoryResourceManager, Timestamp: timestamp, Data: []map[string]interface{}{ {models.NextGenMeteringLogicalClusterID: "cluster-1"}, }, }, { SelfID: "testserver", Category: models.NextGenCategoryProxy, Timestamp: timestamp, Data: []map[string]interface{}{ {models.NextGenMeteringLogicalClusterID: "cluster-2"}, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) assert.NoError(t, err) assert.Empty(t, records, "BYOC mode should skip non-worker-node categories") }) t.Run("worker-node category is processed", func(t *testing.T) { meteringData := []*sdkcommon.MeteringData{ { SelfID: "testserver", SharedPoolID: "pool-1", Category: models.NextGenCategoryWorkerNode, Timestamp: timestamp, Data: []map[string]interface{}{ { "usage": map[string]interface{}{ "value": uint64(8), "unit": "vCPU", }, }, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) assert.NoError(t, err) t.Logf("BYOC mode: worker-node generated %d records", len(records)) }) } func TestNextGenMetering_DealSKU_DeletedMetadataFeatureGate(t *testing.T) { ng, writers := setupTestNextGenMeteringWithWriters(t) defer ng.Close() defer writers.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() clusterID := "test-cluster-deleted-gate" if ng.conf.NextGenMeteringConfig.FeatureGateFilter == nil { ng.conf.NextGenMeteringConfig.FeatureGateFilter = make(map[string]config.FeatureGateFilter) } metadata := map[string]interface{}{ models.NextGenMetaTenantID: "12345", models.NextGenMeteringRegion: "us-west-2", models.NextGenMetaMaxRCU: "1000", models.NextGenMetaDeleted: true, } metaData := ng.loader.CreateMetaData(clusterID, timestamp, metadata) err := writers.WriteMetaData(ctx, clusterID, sdkcommon.MetaTypeLogic, timestamp, metaData) require.NoError(t, err) dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(100), "unit": "RU", }, } ng.conf.NextGenMeteringConfig.FeatureGateFilter[constants.FeatureGateNextGenSkipDeletedMetadata] = config.FeatureGateFilter{ FullSwitchTime: timestamp - 60, } records, err := ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) require.NoError(t, err) assert.Empty(t, records) ng.conf.NextGenMeteringConfig.FeatureGateFilter[constants.FeatureGateNextGenSkipDeletedMetadata] = config.FeatureGateFilter{ FullSwitchTime: timestamp + 3600, } records, err = ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) require.NoError(t, err) assert.NotEmpty(t, records) } // TestNextGenMetering_DealSKU_ComprehensiveLogic tests detailed dealSKU logic scenarios func TestNextGenMetering_DealSKU_ComprehensiveLogic(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() t.Run("cluster ID routing logic", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, } // Test with normal cluster ID - should try to load cluster metadata records, err := ng.dealSKU(ctx, "test-cluster-123", "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Expected to be empty due to missing metadata // Test with empty cluster ID - should try to load shared pool metadata records, err = ng.dealSKU(ctx, "", "shared-pool-456", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Expected to be empty due to missing metadata // Test with system ID - should try to load shared pool metadata records, err = ng.dealSKU(ctx, models.NextGenMeteringSystemID, "system-pool", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Expected to be empty due to missing metadata }) t.Run("different category routing", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, } // Test resource manager category records, err := ng.dealSKU(ctx, "test-cluster", "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test proxy category proxyDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", "connections": 100, } records, err = ng.dealSKU(ctx, "test-cluster", "", timestamp, proxyDataPoint, models.NextGenCategoryProxy) assert.NoError(t, err) assert.Empty(t, records) // Test DXF category dxfDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", models.NextGenMeteringTaskType: "import-into", } records, err = ng.dealSKU(ctx, "test-cluster", "", timestamp, dxfDataPoint, models.NextGenCategoryDXF) assert.NoError(t, err) assert.Empty(t, records) // Test DFS stats category dfsDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "rfengine", } records, err = ng.dealSKU(ctx, "test-cluster", "", timestamp, dfsDataPoint, models.NextGenCategoryDFSStats) assert.NoError(t, err) assert.Empty(t, records) // Test worker-node category (BYOC) workerNodeDataPoint := map[string]interface{}{ "usage": map[string]interface{}{ "value": uint64(8), "unit": "vCPU", }, } records, err = ng.dealSKU(ctx, "test-cluster", "", timestamp, workerNodeDataPoint, models.NextGenCategoryWorkerNode) assert.NoError(t, err) assert.Empty(t, records) // Empty because no metadata // Test unknown category records, err = ng.dealSKU(ctx, "test-cluster", "", timestamp, dataPoint, "unknown-category") assert.NoError(t, err) assert.Empty(t, records) // Should return empty for unknown category }) t.Run("edge cases and boundary conditions", func(t *testing.T) { // Test with empty dataPoint emptyDataPoint := map[string]interface{}{} records, err := ng.dealSKU(ctx, "test-cluster", "", timestamp, emptyDataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test with nil dataPoint - this should still work records, err = ng.dealSKU(ctx, "test-cluster", "", timestamp, nil, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test with large timestamp values futureTimestamp := timestamp + 365*24*3600 // One year in the future dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, } records, err = ng.dealSKU(ctx, "test-cluster", "", futureTimestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test with negative timestamp negativeTimestamp := int64(-1000) records, err = ng.dealSKU(ctx, "test-cluster", "", negativeTimestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) }) t.Run("concurrent access simulation", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, } // Simulate multiple concurrent calls to dealSKU done := make(chan bool, 10) for i := 0; i < 10; i++ { go func(index int) { defer func() { done <- true }() clusterID := fmt.Sprintf("concurrent-cluster-%d", index) records, err := ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) }(i) } // Wait for all goroutines to complete for i := 0; i < 10; i++ { <-done } }) t.Run("parameter validation behavior", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, } // Test with very long cluster ID longClusterID := strings.Repeat("a", 1000) records, err := ng.dealSKU(ctx, longClusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test with special characters in cluster ID specialClusterID := "cluster-with-特殊字符-and-symbols!@#$%" records, err = ng.dealSKU(ctx, specialClusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test with very long shared pool ID longSharedPoolID := strings.Repeat("b", 1000) records, err = ng.dealSKU(ctx, "", longSharedPoolID, timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) }) t.Run("data point content variations", func(t *testing.T) { // Test with complex nested data structures complexDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "nested_object": map[string]interface{}{ "inner_value": 123, "inner_array": []interface{}{1, 2, 3}, }, "array_field": []interface{}{"a", "b", "c"}, "null_field": nil, } records, err := ng.dealSKU(ctx, "complex-cluster", "", timestamp, complexDataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test with very large data point largeDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", } for i := 0; i < 100; i++ { largeDataPoint[fmt.Sprintf("field_%d", i)] = fmt.Sprintf("value_%d", i) } records, err = ng.dealSKU(ctx, "large-data-cluster", "", timestamp, largeDataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) // Test with data point containing only required fields for each category minimalResourceManager := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", } records, err = ng.dealSKU(ctx, "minimal-rm-cluster", "", timestamp, minimalResourceManager, models.NextGenCategoryResourceManager) assert.NoError(t, err) assert.Empty(t, records) minimalProxy := map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", } records, err = ng.dealSKU(ctx, "minimal-proxy-cluster", "", timestamp, minimalProxy, models.NextGenCategoryProxy) assert.NoError(t, err) assert.Empty(t, records) minimalDXF := map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", models.NextGenMeteringTaskType: "export", } records, err = ng.dealSKU(ctx, "minimal-dxf-cluster", "", timestamp, minimalDXF, models.NextGenCategoryDXF) assert.NoError(t, err) assert.Empty(t, records) }) } func TestNextGenMetering_GenerateSKURecords(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() timestamp := getMeteringUniqueTimestamp() now := time.Now() clusterID := "test-cluster-123" tenantID := uint64(12345) cloudProvider := "aws" region := "us-west-2" plan := "premium" enrichmentData := map[string]interface{}{ "version": "1.0", models.NextGenMetaDiskType: "gp3", models.NextGenMetaColumnDataDiskType: "gp3", models.NextGenMetaDiskIOPS: "8000", models.NextGenMetaDiskThroughput: "400", } tests := []struct { name string dataPoint map[string]interface{} category string wantErr bool }{ { name: "resource manager category", dataPoint: map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, }, category: models.NextGenCategoryResourceManager, wantErr: false, }, { name: "proxy category", dataPoint: map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", "connections": 100, }, category: models.NextGenCategoryProxy, wantErr: false, }, { name: "dxf category", dataPoint: map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", "queries": 50, }, category: models.NextGenCategoryDXF, wantErr: false, }, { name: "storage size category", dataPoint: map[string]interface{}{ models.NextGenMeteringRowStorageSize: map[string]interface{}{ "value": uint64(1024 * 1024 * 1024), "unit": "Bytes", }, models.NextGenMeteringColumnStorageSize: map[string]interface{}{ "value": uint64(512 * 1024 * 1024), "unit": "Bytes", }, models.NextGenMeteringSourceName: "pd", models.NextGenMeteringVersion: "1", }, category: models.NextGenCategoryStorageSize, wantErr: false, }, { name: "control-plane category", dataPoint: map[string]interface{}{ models.NextGenMeteringSourceName: "control-planel", models.NextGenMeteringTiKVDataDiskCount: map[string]interface{}{"value": uint64(3), "unit": "count"}, models.NextGenMeteringTiKVDataDiskIOPS: map[string]interface{}{"value": uint64(8000), "unit": "iops"}, models.NextGenMeteringTiKVDataDiskTP: map[string]interface{}{"value": uint64(400), "unit": "mb/s"}, models.NextGenMeteringTiKVDataDiskType: "gp3", models.NextGenMeteringTiFlashComputeDiskCount: map[string]interface{}{"value": uint64(1), "unit": "count"}, models.NextGenMeteringTiFlashComputeDiskIOPS: map[string]interface{}{"value": uint64(6000), "unit": "iops"}, models.NextGenMeteringTiFlashComputeDiskTP: map[string]interface{}{"value": uint64(625), "unit": "mb/s"}, models.NextGenMeteringTiFlashComputeDiskType: "gp3", models.NextGenMeteringTiFlashWriteDiskCount: map[string]interface{}{"value": uint64(1), "unit": "count"}, models.NextGenMeteringTiFlashWriteDiskIOPS: map[string]interface{}{"value": uint64(6000), "unit": "iops"}, models.NextGenMeteringTiFlashWriteDiskTP: map[string]interface{}{"value": uint64(625), "unit": "mb/s"}, models.NextGenMeteringTiFlashWriteDiskType: "gp3", }, category: models.NextGenCategoryControlPlane, wantErr: false, }, { name: "unknown category", dataPoint: map[string]interface{}{ models.NextGenMeteringSourceName: "unknown", "value": 100, }, category: "unknown-category", wantErr: false, // Should not error but return empty records }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { records, err := ng.generateSKURecords( tt.dataPoint, tt.category, clusterID, "", tenantID, timestamp, cloudProvider, region, plan, enrichmentData, nil, now, ) if tt.wantErr { assert.Error(t, err) } else { assert.NoError(t, err) // Records should not be nil, but can be empty if records == nil { // This might happen if SKU generator returns nil for some categories t.Logf("SKU generator returned nil records for category %s, which is acceptable", tt.category) } else { assert.NotNil(t, records) if tt.category == "unknown-category" { assert.Empty(t, records) // Unknown categories return empty records } else if tt.category == models.NextGenCategoryStorageSize { require.Len(t, records, 2) expectedUnits := map[string]bool{ "disk-storage-size:aws:us-west-2:premium:row:standard:data:gp3": false, "disk-storage-size:aws:us-west-2:premium:column:standard:data:gp3": false, } for _, rec := range records { expectedUnits[rec.Unit] = true assert.NotContains(t, rec.Unit, models.NextGenSKUDiskIOPS) } assert.True(t, expectedUnits["disk-storage-size:aws:us-west-2:premium:row:standard:data:gp3"]) assert.True(t, expectedUnits["disk-storage-size:aws:us-west-2:premium:column:standard:data:gp3"]) } else if tt.category == models.NextGenCategoryControlPlane { require.Len(t, records, 4) expectedUnits := map[string]struct { cost int64 }{ "disk-iops:aws:us-west-2:premium:row:standard:data:gp3": {cost: 3}, "disk-throughput:aws:us-west-2:premium:row:standard:data:gp3": {cost: 3}, "disk-iops:aws:us-west-2:premium:column:standard:data:gp3": {cost: 2}, "disk-throughput:aws:us-west-2:premium:column:standard:data:gp3": {cost: 2}, } for _, rec := range records { exp, ok := expectedUnits[rec.Unit] assert.True(t, ok, "unexpected unit: %s", rec.Unit) if !ok { continue } assert.Equal(t, exp.cost, rec.Cost) if strings.Contains(rec.Unit, "tiflash") { var labels map[string]string require.NoError(t, json.Unmarshal([]byte(rec.Labels), &labels)) assert.Equal(t, "gp3", labels[models.NextGenSKULabelKeyTiFlashWriteDiskType]) _, hasWriteIOPS := labels[models.NextGenSKULabelKeyTiFlashWriteIOPS] _, hasWriteIOPSUnit := labels[models.NextGenSKULabelKeyTiFlashWriteIOPSUnit] _, hasWriteThroughput := labels[models.NextGenSKULabelKeyTiFlashWriteThroughput] _, hasWriteThroughputUnit := labels[models.NextGenSKULabelKeyTiFlashWriteThroughputUnit] if strings.Contains(rec.Unit, models.NextGenSKUDiskIOPS) { assert.True(t, hasWriteIOPS) assert.True(t, hasWriteIOPSUnit) assert.False(t, hasWriteThroughput) assert.False(t, hasWriteThroughputUnit) } else if strings.Contains(rec.Unit, models.NextGenSKUDiskThroughput) { assert.True(t, hasWriteThroughput) assert.True(t, hasWriteThroughputUnit) assert.False(t, hasWriteIOPS) assert.False(t, hasWriteIOPSUnit) } } } } // Known categories might return empty or non-empty depending on data } } }) } } func TestNextGenMetering_WriteToDeltalTable(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() // Create delta table first err := ng.CreateDeltaTable() require.NoError(t, err) // Create test billing records now := time.Now() records := []models.BillingReportV2Deltalake{ { SchemeVersion: "v2", TenantID: 12345, CostID: "test-cost-id-1", CostTimestamp: now.Unix(), Unit: "cores", UnitValue: "4.0", Cost: 100, Labels: `{"cluster_id": "test-cluster"}`, CreatedAt: now.Unix(), SKUType: "compute", HourDimension: fmt.Sprintf("%04d%02d%02d%02d", now.Year(), now.Month(), now.Day(), now.Hour()), }, } // Test append mode err = ng.WriteToDeltaTable(records, "Append") assert.NoError(t, err) // Test overwrite mode err = ng.WriteToDeltaTable(records, "Overwrite") assert.NoError(t, err) // Test default mode (should default to append) err = ng.WriteToDeltaTable(records, "InvalidMode") assert.NoError(t, err) // Test with empty records err = ng.WriteToDeltaTable([]models.BillingReportV2Deltalake{}, "Append") assert.NoError(t, err) } // TestNextGenMetering_Integration is an integration test func TestNextGenMetering_Integration(t *testing.T) { // Skip this test in CI or mark as integration if os.Getenv("SKIP_INTEGRATION") == "true" { t.Skip("Skipping integration test") } ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() // 1. Create delta table err := ng.CreateDeltaTable() require.NoError(t, err) // 2. Check table exists exists, err := ng.TableExists(ctx) require.NoError(t, err) assert.True(t, exists) // 3. Get initial version version, err := ng.GetTableVersion(ctx) require.NoError(t, err) assert.Equal(t, int64(0), version) // 4. Write some test records directly to table now := time.Now() records := []models.BillingReportV2Deltalake{ { SchemeVersion: "v2", TenantID: 12345, CostID: "integration-test-cost-id", CostTimestamp: timestamp, Unit: "cores", UnitValue: "2.0", Cost: 50, Labels: `{"cluster_id": "integration-test-cluster"}`, CreatedAt: now.Unix(), SKUType: "compute", HourDimension: fmt.Sprintf("%04d%02d%02d%02d", now.Year(), now.Month(), now.Day(), now.Hour()), }, } err = ng.WriteToDeltaTable(records, "Append") require.NoError(t, err) // 5. Check version increased newVersion, err := ng.GetTableVersion(ctx) require.NoError(t, err) assert.Greater(t, newVersion, version) t.Logf("Integration test completed successfully. Initial version: %d, Final version: %d", version, newVersion) } // TestNextGenMetering_DealSKU_ComprehensiveScenarios tests various dealSKU scenarios func TestNextGenMetering_DealSKU_ComprehensiveScenarios(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() t.Run("cluster with metadata not found", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, } records, err := ng.dealSKU(ctx, "non-existent-cluster", "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) // Should not error but return empty assert.Empty(t, records) // No metadata found }) t.Run("shared pool with metadata not found", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", "connections": 100, } records, err := ng.dealSKU(ctx, "", "non-existent-pool", timestamp, dataPoint, models.NextGenCategoryProxy) assert.NoError(t, err) // Should not error but return empty assert.Empty(t, records) // No metadata found }) t.Run("system ID with shared pool", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", "task_type": "import-into", } records, err := ng.dealSKU(ctx, models.NextGenMeteringSystemID, "system-pool", timestamp, dataPoint, models.NextGenCategoryDXF) assert.NoError(t, err) // Should not error but return empty assert.Empty(t, records) // No metadata found }) t.Run("empty cluster ID triggers shared pool logic", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 2, } records, err := ng.dealSKU(ctx, "", "shared-pool-123", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) // Should not error but return empty assert.Empty(t, records) // No metadata found }) } // TestNextGenMetering_ProcessMinute_DetailedScenarios tests processMinute with various data scenarios func TestNextGenMetering_ProcessMinute_DetailedScenarios(t *testing.T) { ng, writers := setupTestNextGenMeteringWithWriters(t) defer ng.Close() defer writers.Close() ctx := context.Background() // Create delta table first err := ng.CreateDeltaTable() require.NoError(t, err) t.Run("no metering data available", func(t *testing.T) { // Use unique timestamp to avoid conflicts timestamp := getMeteringUniqueTimestamp() + 7200 // Add 2 hours _, err := ng.ProcessMinute(ctx, timestamp) assert.NoError(t, err) // Should succeed with no data }) t.Run("metering data with uninitialized loader", func(t *testing.T) { // Create a new NextGenMetering with uninitialized loader tmpDir := t.TempDir() deltaTablePath := filepath.Join(tmpDir, "test_delta_error") conf := &config.ServiceConfig{ Vendor: "local", LogLevel: "info", NextGenMeteringConfig: &config.NextGenMeteringConfig{ MeteringPrefix: "test-error/", DeltaTablePath: deltaTablePath, EnableRecompute: false, }, } ngErr, err := NewNextGenMetering(conf, zap.NewNop()) require.NoError(t, err) defer ngErr.Close() // Don't create delta table, just test with uninitialized state timestamp := getMeteringUniqueTimestamp() + 10800 // Add 3 hours _, err = ngErr.ProcessMinute(ctx, timestamp) // This should succeed but do nothing since no data is available assert.NoError(t, err) }) t.Run("multiple categories with data", func(t *testing.T) { timestamp := getMeteringUniqueTimestamp() + 14400 // Add 4 hours // Write test data for multiple categories resourceManagerData := ng.loader.CreateMeteringData( fmt.Sprintf("testedserverrmdetailed_%d", timestamp), models.NextGenCategoryResourceManager, timestamp, []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "cluster_rm_detailed", models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, }, }, ) err := writers.WriteMeteringData(ctx, resourceManagerData) require.NoError(t, err) proxyData := ng.loader.CreateMeteringData( fmt.Sprintf("testedserverproxydetailed_%d", timestamp), models.NextGenCategoryProxy, timestamp, []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "cluster_proxy_detailed", models.NextGenMeteringSourceName: "proxy", "connections": 200, }, }, ) err = writers.WriteMeteringData(ctx, proxyData) require.NoError(t, err) // Allow time for write time.Sleep(100 * time.Millisecond) // Process data - will fail due to missing metadata, but that's expected _, err = ng.ProcessMinute(ctx, timestamp) // This might fail due to missing metadata, which is expected in test environment if err != nil { t.Logf("ProcessMinute failed as expected due to missing metadata: %v", err) assert.Contains(t, err.Error(), "failed to generate billing records") } }) t.Run("recompute mode enabled", func(t *testing.T) { ng.conf.NextGenMeteringConfig.EnableRecompute = true timestamp := getMeteringUniqueTimestamp() + 18000 // Add 5 hours _, err := ng.ProcessMinute(ctx, timestamp) // Should succeed regardless of data presence assert.NoError(t, err) }) } // TestNextGenMetering_GenerateBillingRecords_EdgeCases tests edge cases for generateBillingRecords func TestNextGenMetering_GenerateBillingRecords_EdgeCases(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() t.Run("metering data with invalid cluster ID", func(t *testing.T) { meteringData := []*sdkcommon.MeteringData{ { SelfID: "testserver", Category: models.NextGenCategoryResourceManager, Timestamp: timestamp, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "", // Empty cluster ID models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, }, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) assert.NoError(t, err) assert.Empty(t, records) // No valid cluster ID }) t.Run("metering data with mixed valid and invalid entries", func(t *testing.T) { meteringData := []*sdkcommon.MeteringData{ { SelfID: "testserver1", Category: models.NextGenCategoryResourceManager, Timestamp: timestamp, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "valid-cluster", models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, }, }, }, { SelfID: "testserver2", Category: "invalid-category", // Invalid category Timestamp: timestamp, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "another-cluster", models.NextGenMeteringSourceName: "pd", "cpu_cores": 2, }, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) assert.NoError(t, err) assert.Empty(t, records) // Should process but return empty due to missing metadata }) t.Run("data point with both cluster ID and shared pool", func(t *testing.T) { meteringData := []*sdkcommon.MeteringData{ { SelfID: "testserver", SharedPoolID: "shared-pool-123", Category: models.NextGenCategoryResourceManager, Timestamp: timestamp, Data: []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "cluster-with-pool", models.NextGenMeteringSourceName: "tikv", "cpu_cores": 8, }, }, }, } records, err := ng.generateBillingRecords(ctx, meteringData, timestamp) assert.NoError(t, err) assert.Empty(t, records) // Should process cluster ID path }) } // TestNextGenMetering_GenerateSKURecords_DetailedCoverage tests generateSKURecords with different scenarios func TestNextGenMetering_GenerateSKURecords_DetailedCoverage(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() timestamp := getMeteringUniqueTimestamp() now := time.Now() t.Run("resource manager category with valid data", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", "cpu_cores": 4, "memory_gb": 8, } records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryResourceManager, "test-cluster", "", 12345, timestamp, "aws", "us-east-1", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // Records might be nil if no valid SKU data is found, which is OK if records != nil { t.Logf("Generated %d records for resource manager", len(records)) } }) t.Run("resource manager backup storage v2 with metadata", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: models.NextGenSourceNamePD, models.NextGenMeteringIncrementalWriteByte: map[string]interface{}{ "value": uint64(4096), "unit": "bytes", }, } backupMeta := &sdkcommon.MetaData{ Metadata: map[string]interface{}{ models.NextGenMetaBackupStorageV2: "T2", }, } records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryResourceManager, "backup-cluster", "", 12345, timestamp, "aws", "us-east-1", "premium", map[string]interface{}{}, backupMeta.Metadata, now, ) require.NoError(t, err) require.Len(t, records, 2) var backupRecord *models.BillingReportV2Deltalake for i := range records { if strings.HasPrefix(records[i].Unit, models.NextGenSKUBackupStorage+":") { backupRecord = &records[i] break } } require.NotNil(t, backupRecord) assert.Equal(t, "backup-storage:aws:us-east-1:premium:T2", backupRecord.Unit) assert.Equal(t, "4096", backupRecord.UnitValue) var labels map[string]string require.NoError(t, json.Unmarshal([]byte(backupRecord.Labels), &labels)) assert.Equal(t, "bytes", labels[models.NextGenSKULabelKeySourceUnit]) assert.Equal(t, "backup-cluster", labels[models.NextGenSKULabelKeyClusterID]) }) t.Run("proxy category with valid data", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", "connections": 150, "requests": 5000, } records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryProxy, "proxy-cluster", "proxy-pool", 54321, timestamp, "gcp", "us-central1", models.PlanPremium, map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // Records might be nil if no valid SKU data is found, which is OK if records != nil { t.Logf("Generated %d records for proxy", len(records)) } }) t.Run("dxf category with task data", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", models.NextGenMeteringTaskType: "export", models.NextGenMeteringTaskID: "task-001", "processing_time": 3600, } records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryDXF, "dxf-cluster", "", 98765, timestamp, "azure", "eastus", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // DXF category should generate records if records != nil { t.Logf("Generated %d records for DXF", len(records)) assert.NotEmpty(t, records) } }) t.Run("unknown category", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "unknown", "some_metric": 100, } records, err := ng.generateSKURecords( dataPoint, "unknown-category", "test-cluster", "", 11111, timestamp, "aws", "us-west-2", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) assert.Empty(t, records) // Unknown categories should return empty }) t.Run("empty data point", func(t *testing.T) { dataPoint := map[string]interface{}{} records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryResourceManager, "empty-cluster", "", 22222, timestamp, "aws", "us-west-1", models.PlanPremium, map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // Changed: Now always reports RCU even if totalRU is 0 assert.Len(t, records, 1) assert.Contains(t, records[0].Unit, "all:total") assert.Contains(t, records[0].Labels, "\"intervalValue\":\"0\"") }) t.Run("manual backup category with gb size", func(t *testing.T) { dataPoint := map[string]interface{}{ models.NextGenMeteringLogicalClusterID: "manual-backup-cluster", models.NextGenMeteringBackupID: "backup-1", models.NextGenMeteringBackupSizeGB: map[string]interface{}{ "value": uint64(64), "unit": "GB", }, } records, err := ng.generateSKURecords( dataPoint, models.NextGenCategoryManualBackup, "manual-backup-cluster", "manual-backup-pool", 12345, timestamp, "aws", "ap-south-1", "premium", map[string]interface{}{}, nil, now, ) require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, "backup-storage:aws:ap-south-1:premium:manual", records[0].Unit) assert.Equal(t, "64", records[0].UnitValue) }) } // TestNextGenMetering_GenerateSKURecords_EdgeCases tests generateSKURecords with edge cases and error conditions func TestNextGenMetering_GenerateSKURecords_EdgeCases(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() timestamp := getMeteringUniqueTimestamp() now := time.Now() t.Run("invalid input parameters", func(t *testing.T) { // Test with empty category records, err := ng.generateSKURecords( map[string]interface{}{"test": "value"}, "", // empty category "test-cluster", "", 12345, timestamp, "aws", "us-west-2", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // Empty category won't match any generator, so should return empty assert.Empty(t, records) // Test with nil enrichmentData records, err = ng.generateSKURecords( map[string]interface{}{"test": "value"}, models.NextGenCategoryResourceManager, "test-cluster", "", 12345, timestamp, "aws", "us-west-2", "premium", nil, // nil enrichmentData nil, now, ) assert.NoError(t, err) // Changed: Now generates RCU record even with no RU data (totalRU=0) assert.Len(t, records, 1) }) t.Run("zero and negative values", func(t *testing.T) { // Test with zero tenantID records, err := ng.generateSKURecords( map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(100), "unit": "RU", }, }, models.NextGenCategoryResourceManager, "test-cluster", "", 0, // zero tenantID timestamp, "aws", "us-west-2", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) if len(records) > 0 { assert.Equal(t, uint64(0), records[0].TenantID) } // Test with zero timestamp records, err = ng.generateSKURecords( map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(100), "unit": "RU", }, }, models.NextGenCategoryResourceManager, "test-cluster", "", 12345, 0, // zero timestamp "aws", "us-west-2", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) if len(records) > 0 { assert.Equal(t, int64(0), records[0].CostTimestamp) } }) t.Run("special characters and unicode in parameters", func(t *testing.T) { records, err := ng.generateSKURecords( map[string]interface{}{ models.NextGenMeteringSourceName: "tikv-特殊字符", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(100), "unit": "RU", }, }, models.NextGenCategoryResourceManager, "cluster-特殊字符-🚀", // cluster with special chars and emoji "pool-测试", // shared pool with Chinese chars 12345, timestamp, "aws-中国", // cloud provider with Chinese "region-测试", // region with Chinese "plan-高级版", // plan with Chinese map[string]interface{}{ "special_key_特殊": "special_value_特殊", }, nil, now, ) assert.NoError(t, err) // Should handle unicode characters gracefully if len(records) > 0 { assert.Contains(t, records[0].Unit, "aws-中国") assert.Contains(t, records[0].Unit, "region-测试") } }) t.Run("all categories with comprehensive data", func(t *testing.T) { // Test ResourceManager category with all possible fields rmDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(1000), "unit": "RU", }, models.NextGenMeteringOLTPRU: map[string]interface{}{ "value": uint64(2000), "unit": "RU", }, models.NextGenMeteringCrossAZTraffic: map[string]interface{}{ "value": uint64(1024000), "unit": "bytes", }, } records, err := ng.generateSKURecords( rmDataPoint, models.NextGenCategoryResourceManager, "rm-cluster", "rm-pool", 12345, timestamp, "aws", "us-west-2", "premium", map[string]interface{}{ models.NextGenMetaMinRCU: "500", models.NextGenMetaMaxRCU: "5000", }, nil, now, ) assert.NoError(t, err) // Should potentially generate multiple records (OLAP, OLTP, CrossAZ) t.Logf("ResourceManager generated %d records", len(records)) // Test Proxy category with all possible fields proxyDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", models.NextGenMeteringOutBoundBytes: map[string]interface{}{ "value": uint64(2048000), "unit": "bytes", }, models.NextGenMeteringCrossZoneBytes: map[string]interface{}{ "value": uint64(512000), "unit": "bytes", }, } records, err = ng.generateSKURecords( proxyDataPoint, models.NextGenCategoryProxy, "proxy-cluster", "", 54321, timestamp, "gcp", "us-central1", "basic", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // Should potentially generate multiple records (OutBound, CrossZone) t.Logf("Proxy generated %d records", len(records)) // Test DXF category with all possible fields dxfDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", models.NextGenMeteringTaskType: "import-into", models.NextGenMeteringTaskID: "task-comprehensive-12345", models.NextGenMeteringVersion: "2.0", models.NextGenMeteringGetExtRequests: 100, models.NextGenMeteringPutExtRequests: 200, models.NextGenMeteringReadDataBytes: map[string]interface{}{ "value": uint64(10240000), "unit": "bytes", }, models.NextGenMeteringWriteDataBytes: map[string]interface{}{ "value": uint64(5120000), "unit": "bytes", }, } records, err = ng.generateSKURecords( dxfDataPoint, models.NextGenCategoryDXF, "dxf-cluster", "dxf-pool", 98765, timestamp, "azure", "eastus2", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // DXF should generate task-ru records t.Logf("DXF generated %d records", len(records)) if len(records) > 0 { assert.Contains(t, records[0].Unit, "task-ru") assert.Contains(t, records[0].Unit, "import-into") assert.Contains(t, records[0].Labels, "task-comprehensive-12345") } }) t.Run("error conditions and recovery", func(t *testing.T) { // Test with malformed metering values malformedDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: "not-a-valid-metering-value", // Should be map with value/unit } records, err := ng.generateSKURecords( malformedDataPoint, models.NextGenCategoryResourceManager, "malformed-cluster", "", 12345, timestamp, "aws", "us-west-2", "premium", map[string]interface{}{}, nil, now, ) assert.NoError(t, err) // Should handle malformed data gracefully t.Logf("Malformed data generated %d records", len(records)) }) } func TestNextGenMetering_GenerateSKURecords_TopologyDrivenReservedMode(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() timestamp := getMeteringUniqueTimestamp() now := time.Now() enrichmentData := map[string]interface{}{ models.NextGenMetaCapacityMode: "TOPOLOGY_DRIVEN", } t.Run("resource manager keeps non-rcu with reserved plan", func(t *testing.T) { records, err := ng.generateSKURecords( map[string]interface{}{ models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(1000), "unit": "RU", }, models.NextGenMeteringOLTPRU: map[string]interface{}{ "value": uint64(2000), "unit": "RU", }, models.NextGenMeteringCrossAZTraffic: map[string]interface{}{ "value": uint64(2048), "unit": "bytes", }, }, models.NextGenCategoryResourceManager, "topology-cluster", "", 12345, timestamp, "aws", "us-west-2", "premium", enrichmentData, nil, now, ) require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, "data-transfer-cross-az:aws:us-west-2:reserved:pd", records[0].Unit) for _, record := range records { assert.NotContains(t, record.Unit, "rcu:") } }) t.Run("storage size uses reserved plan", func(t *testing.T) { records, err := ng.generateSKURecords( map[string]interface{}{ models.NextGenMeteringRowStorageSize: map[string]interface{}{ "value": uint64(1024 * 1024 * 1024), "unit": "Bytes", }, models.NextGenMeteringSourceName: "pd", }, models.NextGenCategoryStorageSize, "topology-cluster", "", 12345, timestamp, "aws", "us-west-2", "premium_reserved", map[string]interface{}{ models.NextGenMetaCapacityMode: "TOPOLOGY_DRIVEN", models.NextGenMetaDiskType: "gp3", models.NextGenMetaColumnDataDiskType: "gp3", }, nil, now, ) require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, "disk-storage-size:aws:us-west-2:reserved:row:standard:data:gp3", records[0].Unit) }) t.Run("proxy is kept with reserved plan", func(t *testing.T) { records, err := ng.generateSKURecords( map[string]interface{}{ models.NextGenMeteringSourceName: models.NextGenSourceNameProxy, models.NextGenMeteringOutBoundBytes: map[string]interface{}{ "value": uint64(1024), "unit": "bytes", }, }, models.NextGenCategoryProxy, "topology-cluster", "", 12345, timestamp, "aws", "us-west-2", "premium", enrichmentData, nil, now, ) require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, "data-transfer-out:aws:us-west-2:reserved:proxy", records[0].Unit) }) t.Run("reserved node uses reserved plan", func(t *testing.T) { records, err := ng.generateSKURecords( map[string]interface{}{ "usage": map[string]interface{}{ "value": uint64(2000), "unit": "vCPU-minutes", }, models.NextGenMeteringNodeType: "tidb", }, models.NextGenCategoryReservedNode, "topology-cluster", "shared-pool-123", 12345, timestamp, "aws", "us-west-2", "premium", enrichmentData, nil, now, ) require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, "vcpu:aws:us-west-2:reserved:all", records[0].Unit) }) } func TestNextGenMetering_AggregateReservedNodeDataPoints(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() t.Run("aggregate by cluster with all node type", func(t *testing.T) { dataPoints := []map[string]interface{}{ { models.NextGenMeteringLogicalClusterID: "cluster-a", models.NextGenMeteringNodeType: "tikv", "usage": map[string]interface{}{ "value": uint64(3), "unit": "vCPU", }, }, { models.NextGenMeteringLogicalClusterID: "cluster-a", models.NextGenMeteringNodeType: "tidb", "usage": map[string]interface{}{ "value": uint64(5), "unit": "vCPU", }, }, { models.NextGenMeteringLogicalClusterID: "cluster-b", models.NextGenMeteringNodeType: "tiflash", "usage": map[string]interface{}{ "value": uint64(7), "unit": "vCPU", }, }, } aggregated := ng.aggregateReservedNodeDataPoints(dataPoints) require.Len(t, aggregated, 2) assert.Equal(t, "cluster-a", aggregated[0].clusterID) usageA, ok := aggregated[0].dataPoint["usage"].(map[string]interface{}) require.True(t, ok) assert.Equal(t, uint64(8), usageA["value"]) assert.Equal(t, "vCPU", usageA["unit"]) assert.Equal(t, "all", aggregated[0].dataPoint[models.NextGenMeteringNodeType]) assert.Equal(t, "cluster-b", aggregated[1].clusterID) usageB, ok := aggregated[1].dataPoint["usage"].(map[string]interface{}) require.True(t, ok) assert.Equal(t, uint64(7), usageB["value"]) assert.Equal(t, "vCPU", usageB["unit"]) assert.Equal(t, "all", aggregated[1].dataPoint[models.NextGenMeteringNodeType]) }) } // TestNextGenMetering_DealSKU_MetadataHandling tests dealSKU with different metadata scenarios func TestNextGenMetering_DealSKU_MetadataHandling(t *testing.T) { ng := setupTestNextGenMetering(t) defer ng.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() // Note: In real scenarios, metadata would be loaded from storage // For testing purposes, we focus on the dealSKU logic paths sampleDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(500), "unit": "RU", }, } t.Run("cluster_with_valid_metadata_processing", func(t *testing.T) { // Test cluster metadata processing paths records, err := ng.dealSKU( ctx, "test-cluster-123", // clusterID "", // sharedPoolID (empty for cluster) timestamp, sampleDataPoint, models.NextGenCategoryResourceManager, ) assert.NoError(t, err) // Even without persistent metadata, should handle gracefully // In real scenarios with metadata, this would generate records t.Logf("Cluster metadata test generated %d records", len(records)) }) t.Run("shared_pool_with_valid_metadata_processing", func(t *testing.T) { // Test shared pool metadata processing paths records, err := ng.dealSKU( ctx, "", // clusterID (empty for shared pool) "test-shared-pool-456", // sharedPoolID timestamp, sampleDataPoint, models.NextGenCategoryResourceManager, ) assert.NoError(t, err) // Even without persistent metadata, should handle gracefully t.Logf("Shared pool metadata test generated %d records", len(records)) }) t.Run("system_cluster_id_triggers_shared_pool_metadata_logic", func(t *testing.T) { // Test that system cluster ID triggers shared pool logic records, err := ng.dealSKU( ctx, models.NextGenMeteringSystemID, // special system ID "system-shared-pool", // sharedPoolID timestamp, sampleDataPoint, models.NextGenCategoryResourceManager, ) assert.NoError(t, err) // Should use shared pool logic for system ID t.Logf("System cluster ID test generated %d records", len(records)) }) t.Run("metadata_tenant_id_parsing_logic", func(t *testing.T) { // Test various tenant ID parsing scenarios testCases := []struct { name string clusterID string sharedID string }{ {"regular_cluster", "cluster-tenant-test", ""}, {"shared_pool", "", "shared-tenant-test"}, {"system_cluster", models.NextGenMeteringSystemID, "system-shared-test"}, } for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { records, err := ng.dealSKU( ctx, tc.clusterID, tc.sharedID, timestamp, sampleDataPoint, models.NextGenCategoryResourceManager, ) assert.NoError(t, err) t.Logf("Tenant ID parsing test '%s' generated %d records", tc.name, len(records)) }) } }) t.Run("metadata_enrichment_data_extraction", func(t *testing.T) { // Test enrichment data extraction from metadata complexDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(1500), "unit": "RU", }, models.NextGenMeteringOLTPRU: map[string]interface{}{ "value": uint64(2500), "unit": "RU", }, "customField": "customValue", } records, err := ng.dealSKU( ctx, "enrichment-test-cluster", "", timestamp, complexDataPoint, models.NextGenCategoryResourceManager, ) assert.NoError(t, err) // Should handle complex data points with enrichment t.Logf("Enrichment data test generated %d records", len(records)) }) t.Run("cloud_provider_and_region_extraction", func(t *testing.T) { // Test cloud provider and region extraction from config and metadata testDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringCrossAZTraffic: map[string]interface{}{ "value": uint64(1024000), "unit": "bytes", }, } records, err := ng.dealSKU( ctx, "region-test-cluster", "", timestamp, testDataPoint, models.NextGenCategoryResourceManager, ) assert.NoError(t, err) // Should extract provider and region correctly t.Logf("Cloud provider/region test generated %d records", len(records)) }) t.Run("all_category_types_with_metadata_paths", func(t *testing.T) { // Test all categories to ensure metadata paths work for each categories := []string{ models.NextGenCategoryResourceManager, models.NextGenCategoryProxy, models.NextGenCategoryDXF, } for _, category := range categories { t.Run(fmt.Sprintf("category_%s", category), func(t *testing.T) { var categoryDataPoint map[string]interface{} switch category { case models.NextGenCategoryResourceManager: categoryDataPoint = map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(800), "unit": "RU", }, } case models.NextGenCategoryProxy: categoryDataPoint = map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", models.NextGenMeteringOutBoundBytes: map[string]interface{}{ "value": uint64(1048576), "unit": "bytes", }, } case models.NextGenCategoryDXF: categoryDataPoint = map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", models.NextGenMeteringTaskType: "import-into", models.NextGenMeteringTaskID: fmt.Sprintf("task-metadata-%d", time.Now().UnixNano()), } } records, err := ng.dealSKU( ctx, fmt.Sprintf("metadata-test-%s", category), "", timestamp, categoryDataPoint, category, ) assert.NoError(t, err) t.Logf("Category %s with metadata generated %d records", category, len(records)) }) } }) t.Run("error_handling_in_metadata_processing", func(t *testing.T) { // Test error conditions in metadata processing errorDataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "error-test", } // Test with very old timestamp that might cause metadata issues oldTimestamp := int64(1) records, err := ng.dealSKU( ctx, "error-test-cluster", "", oldTimestamp, errorDataPoint, models.NextGenCategoryResourceManager, ) // Should handle metadata errors gracefully if err != nil { t.Logf("Expected error handling: %v", err) } else { t.Logf("Error handling test generated %d records", len(records)) } }) } // TestNextGenMetering_DealSKU_WithRealMetadata tests dealSKU with real metadata written to local storage func TestNextGenMetering_DealSKU_WithRealMetadata(t *testing.T) { ng, writers := setupTestNextGenMeteringWithWriters(t) defer ng.Close() defer writers.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() t.Run("cluster_metadata_processing_with_all_paths", func(t *testing.T) { clusterID := "test-cluster-with-metadata" // Create and write cluster metadata clusterMetadata := map[string]interface{}{ models.NextGenMetaTenantID: "12345", models.NextGenMeteringRegion: "us-west-2", models.NextGenMetaMinRCU: "1000", models.NextGenMetaMaxRCU: "10000", "clusterType": "dedicated", } metaData := ng.loader.CreateMetaData(clusterID, timestamp, clusterMetadata) // Write metadata to storage using the SDK err := writers.WriteMetaData(ctx, clusterID, sdkcommon.MetaTypeLogic, timestamp, metaData) require.NoError(t, err, "Failed to write cluster metadata") // Test data point for ResourceManager category dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(2000), "unit": "RU", }, models.NextGenMeteringOLTPRU: map[string]interface{}{ "value": uint64(1500), "unit": "RU", }, } // Call dealSKU with real metadata records, err := ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err, "dealSKU should succeed with real metadata") // Should generate actual SKU records with real metadata if len(records) > 0 { t.Logf("Generated %d billing records with real cluster metadata", len(records)) // Verify the records have correct tenant ID from metadata for _, record := range records { assert.Equal(t, uint64(12345), record.TenantID, "TenantID should be extracted from metadata") assert.Contains(t, record.Unit, "us-west-2", "Region should be extracted from metadata") assert.NotEmpty(t, record.Labels, "Labels should be populated") } } else { t.Logf("No records generated - this tests the metadata loading path") } }) t.Run("storage_size_with_cmek_metadata_generates_dual_encryption_skus", func(t *testing.T) { clusterID := "test-cluster-with-cmek" previousVendor := ng.conf.Vendor ng.conf.Vendor = "aws" defer func() { ng.conf.Vendor = previousVendor }() if ng.conf.NextGenMeteringConfig.FeatureGateFilter == nil { ng.conf.NextGenMeteringConfig.FeatureGateFilter = make(map[string]config.FeatureGateFilter) } ng.conf.NextGenMeteringConfig.FeatureGateFilter[constants.FeatureGateNextGenCMEKStorageSKU] = config.FeatureGateFilter{ Clusters: []config.ClusterFilter{{ ClusterId: clusterID, SwitchTime: timestamp - 60, }}, } clusterMetadata := map[string]interface{}{ models.NextGenMetaTenantID: "12345", models.NextGenMeteringRegion: "ap-northeast-1", models.NextGenMetaDiskType: "gp3", models.NextGenMetaColumnDataDiskType: "gp3", models.NextGenMetaStorageEncryptionEnabled: true, } metaData := ng.loader.CreateMetaData(clusterID, timestamp, clusterMetadata) err := writers.WriteMetaData(ctx, clusterID, sdkcommon.MetaTypeLogic, timestamp, metaData) require.NoError(t, err, "Failed to write cluster metadata with CMEK flag") dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "pd", models.NextGenMeteringVersion: "1", models.NextGenMeteringRowStorageSize: map[string]interface{}{ "value": uint64(2 * 1024 * 1024 * 1024), "unit": "Bytes", }, models.NextGenMeteringColumnStorageSize: map[string]interface{}{ "value": uint64(1024 * 1024 * 1024), "unit": "Bytes", }, } records, err := ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, models.NextGenCategoryStorageSize) require.NoError(t, err, "dealSKU should succeed with CMEK metadata") require.Len(t, records, 2) expectedUnits := map[string]bool{ "disk-storage-size-dual-encryption:aws:ap-northeast-1:premium:row:standard:data:gp3": false, "disk-storage-size:aws:ap-northeast-1:premium:column:standard:data:gp3": false, } for _, record := range records { _, ok := expectedUnits[record.Unit] require.Truef(t, ok, "unexpected unit %s", record.Unit) expectedUnits[record.Unit] = true assert.Equal(t, uint64(12345), record.TenantID) assert.Equal(t, int64(1), record.Cost) } for unit, seen := range expectedUnits { assert.Truef(t, seen, "expected unit %s not generated", unit) } }) t.Run("shared_pool_metadata_processing_with_all_paths", func(t *testing.T) { sharedPoolID := "test-shared-pool-with-metadata" // Create and write shared pool metadata sharedPoolMetadata := map[string]interface{}{ models.NextGenMetaTenantID: "67890", models.NextGenMeteringRegion: "us-east-1", "poolCapacity": "5000", "poolType": "shared", } metaData := ng.loader.CreateMetaData(sharedPoolID, timestamp, sharedPoolMetadata) // Write shared pool metadata to storage err := writers.WriteMetaData(ctx, sharedPoolID, sdkcommon.MetaTypeSharedpool, timestamp, metaData) require.NoError(t, err, "Failed to write shared pool metadata") // Test data point for Proxy category dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "proxy", models.NextGenMeteringOutBoundBytes: map[string]interface{}{ "value": uint64(2048000), "unit": "bytes", }, models.NextGenMeteringCrossZoneBytes: map[string]interface{}{ "value": uint64(1024000), "unit": "bytes", }, } // Call dealSKU with empty clusterID to trigger shared pool logic records, err := ng.dealSKU(ctx, "", sharedPoolID, timestamp, dataPoint, models.NextGenCategoryProxy) assert.NoError(t, err, "dealSKU should succeed with real shared pool metadata") // Should generate actual SKU records with shared pool metadata if len(records) > 0 { t.Logf("Generated %d billing records with real shared pool metadata", len(records)) // Verify the records have correct tenant ID from shared pool metadata for _, record := range records { assert.Equal(t, uint64(67890), record.TenantID, "TenantID should be extracted from shared pool metadata") assert.Contains(t, record.Unit, "us-east-1", "Region should be extracted from shared pool metadata") } } else { t.Logf("No records generated - this tests the shared pool metadata loading path") } }) t.Run("system_cluster_id_with_shared_pool_metadata", func(t *testing.T) { systemSharedPoolID := "system-shared-pool-with-metadata" // Create and write system shared pool metadata systemMetadata := map[string]interface{}{ models.NextGenMetaTenantID: "99999", models.NextGenMeteringRegion: "eu-west-1", "systemPool": "true", "priority": "high", } metaData := ng.loader.CreateMetaData(systemSharedPoolID, timestamp, systemMetadata) // Write system shared pool metadata err := writers.WriteMetaData(ctx, systemSharedPoolID, sdkcommon.MetaTypeSharedpool, timestamp, metaData) require.NoError(t, err, "Failed to write system shared pool metadata") // Test data point for DXF category dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "dxf", models.NextGenMeteringTaskType: "import-into", models.NextGenMeteringTaskID: "system-task-123", models.NextGenMeteringVersion: "2.0", models.NextGenMeteringGetExtRequests: 50, models.NextGenMeteringPutExtRequests: 100, } // Call dealSKU with system cluster ID - should use shared pool logic records, err := ng.dealSKU(ctx, models.NextGenMeteringSystemID, systemSharedPoolID, timestamp, dataPoint, models.NextGenCategoryDXF) assert.NoError(t, err, "dealSKU should succeed with system cluster ID and shared pool metadata") // Should generate actual SKU records using system shared pool metadata if len(records) > 0 { t.Logf("Generated %d billing records with system shared pool metadata", len(records)) // Verify the records have correct tenant ID from system metadata for _, record := range records { assert.Equal(t, uint64(99999), record.TenantID, "TenantID should be extracted from system shared pool metadata") assert.Contains(t, record.Unit, "eu-west-1", "Region should be extracted from system shared pool metadata") // For DXF category, should have task information in labels if record.Labels != "" { assert.Contains(t, record.Labels, "system-task-123", "Task ID should be in labels") } } } else { t.Logf("No records generated - this tests the system cluster ID with shared pool metadata path") } }) t.Run("metadata_with_different_tenant_id_formats", func(t *testing.T) { // Test different tenant ID formats and edge cases testCases := []struct { name string tenantID interface{} expectedID uint64 shouldSucceed bool }{ {"string_numeric", "54321", 54321, true}, {"string_zero", "0", 0, true}, {"string_large_number", "9223372036854775807", 9223372036854775807, true}, {"numeric_int", 12345, 0, false}, // Non-string should not parse {"string_non_numeric", "invalid", 0, false}, {"empty_string", "", 0, false}, } for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { clusterID := fmt.Sprintf("tenant-test-%s", tc.name) // Create metadata with specific tenant ID format metadata := map[string]interface{}{ models.NextGenMetaTenantID: tc.tenantID, models.NextGenMeteringRegion: "us-central1", } metaData := ng.loader.CreateMetaData(clusterID, timestamp, metadata) err := writers.WriteMetaData(ctx, clusterID, sdkcommon.MetaTypeLogic, timestamp, metaData) require.NoError(t, err) dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(100), "unit": "RU", }, } records, err := ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) if len(records) > 0 && tc.shouldSucceed { assert.Equal(t, tc.expectedID, records[0].TenantID, "Tenant ID should be parsed correctly") t.Logf("Tenant ID parsing test '%s': expected %d, got %d", tc.name, tc.expectedID, records[0].TenantID) } else { t.Logf("Tenant ID parsing test '%s': generated %d records", tc.name, len(records)) } }) } }) t.Run("metadata_enrichment_data_flow", func(t *testing.T) { clusterID := "enrichment-test-cluster" // Create metadata with rich enrichment data enrichmentMetadata := map[string]interface{}{ models.NextGenMetaTenantID: "88888", models.NextGenMeteringRegion: "ap-southeast-1", models.NextGenMetaMinRCU: "2000", models.NextGenMetaMaxRCU: "20000", "customField1": "customValue1", "customField2": 123, "customField3": map[string]interface{}{ "nested": "value", }, } metaData := ng.loader.CreateMetaData(clusterID, timestamp, enrichmentMetadata) err := writers.WriteMetaData(ctx, clusterID, sdkcommon.MetaTypeLogic, timestamp, metaData) require.NoError(t, err) // Test data point that should use enrichment data dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(3000), "unit": "RU", }, models.NextGenMeteringCrossAZTraffic: map[string]interface{}{ "value": uint64(5242880), "unit": "bytes", }, } records, err := ng.dealSKU(ctx, clusterID, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err) if len(records) > 0 { t.Logf("Generated %d billing records with enrichment metadata", len(records)) // Verify enrichment data is used in SKU generation for _, record := range records { assert.Equal(t, uint64(88888), record.TenantID) assert.Contains(t, record.Unit, "ap-southeast-1") // Check if enrichment data flows through to labels if record.Labels != "" { t.Logf("Record labels: %s", record.Labels) } } } else { t.Logf("No records generated - this tests enrichment data flow") } }) t.Run("metadata_error_conditions_and_fallbacks", func(t *testing.T) { // Test error conditions and fallback behavior // Case 1: Metadata exists but has parsing errors clusterID1 := "error-metadata-cluster" invalidMetadata := map[string]interface{}{ models.NextGenMetaTenantID: []string{"invalid", "format"}, // Invalid format models.NextGenMeteringRegion: 123, // Wrong type } metaData1 := ng.loader.CreateMetaData(clusterID1, timestamp, invalidMetadata) err := writers.WriteMetaData(ctx, clusterID1, sdkcommon.MetaTypeLogic, timestamp, metaData1) require.NoError(t, err) dataPoint := map[string]interface{}{ models.NextGenMeteringSourceName: "tikv", models.NextGenMeteringOLAPRU: map[string]interface{}{ "value": uint64(100), "unit": "RU", }, } records1, err := ng.dealSKU(ctx, clusterID1, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err, "Should handle invalid metadata gracefully") t.Logf("Invalid metadata test generated %d records", len(records1)) // Case 2: Metadata is nil/empty clusterID2 := "empty-metadata-cluster" emptyMetadata := map[string]interface{}{} metaData2 := ng.loader.CreateMetaData(clusterID2, timestamp, emptyMetadata) err = writers.WriteMetaData(ctx, clusterID2, sdkcommon.MetaTypeLogic, timestamp, metaData2) require.NoError(t, err) records2, err := ng.dealSKU(ctx, clusterID2, "", timestamp, dataPoint, models.NextGenCategoryResourceManager) assert.NoError(t, err, "Should handle empty metadata gracefully") t.Logf("Empty metadata test generated %d records", len(records2)) }) } func TestNextGenMetering_DealSKU_WorkerNode(t *testing.T) { ng, writers := setupTestNextGenMeteringWithWriters(t) defer ng.Close() defer writers.Close() ctx := context.Background() timestamp := getMeteringUniqueTimestamp() ng.conf.NextGenMeteringConfig.Plan = "byoc" t.Run("worker-node with shared pool metadata", func(t *testing.T) { sharedPoolID := "byoc-pool-with-meta" sharedPoolMetadata := map[string]interface{}{ models.NextGenMetaTenantID: "99999", models.NextGenMeteringRegion: "us-east-1", } metaData := ng.loader.CreateMetaData(sharedPoolID, timestamp, sharedPoolMetadata) err := writers.WriteMetaData(ctx, sharedPoolID, sdkcommon.MetaTypeSharedpool, timestamp, metaData) require.NoError(t, err) dataPoint := map[string]interface{}{ "usage": map[string]interface{}{ "value": uint64(32), "unit": "vCPU", }, } records, err := ng.dealSKU(ctx, "", sharedPoolID, timestamp, dataPoint, models.NextGenCategoryWorkerNode) require.NoError(t, err) require.Len(t, records, 1) assert.Equal(t, uint64(99999), records[0].TenantID) assert.Contains(t, records[0].Unit, "us-east-1") assert.Equal(t, "vcpu:local:us-east-1:byoc:total", records[0].Unit) var labels map[string]string require.NoError(t, json.Unmarshal([]byte(records[0].Labels), &labels)) assert.Equal(t, sharedPoolID, labels[models.NextGenSKULabelKeySharedPoolID]) assert.Equal(t, "32", labels[models.NextGenSKULabelKeyIntervalValue]) }) } func TestConvertToInt64(t *testing.T) { tests := []struct { name string input interface{} expect int64 wantErr bool }{ {name: "int64", input: int64(123), expect: 123}, {name: "int32", input: int32(456), expect: 456}, {name: "int", input: 789, expect: 789}, {name: "float64", input: float64(910), expect: 910}, {name: "float32", input: float32(11), expect: 11}, {name: "uint64", input: uint64(22), expect: 22}, {name: "uint32", input: uint32(33), expect: 33}, {name: "string valid", input: "444", expect: 444}, {name: "string invalid", input: "not-a-number", wantErr: true}, {name: "unsupported type", input: []int{1, 2}, wantErr: true}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { got, err := convertToInt64(tt.input) if tt.wantErr { assert.Error(t, err) return } require.NoError(t, err) assert.Equal(t, tt.expect, got) }) } }