# Azure Log Backup Size Billing Bug — Analysis & Fix Plan

## 1. Executive Summary

**Bug**: Azure Dedicated clusters with PiTR (log backup) enabled are never billed for log backup storage because the `CalculateLogBackupSize` worker fails to compute sizes for Azure Blob Storage paths.

**Root Cause**: `NewLogBackupSizeHandler()` only handles `s3://` and `gcs://`/`gs://` URL schemes. Azure log backup paths use `https://` scheme (e.g. `https://<account>.blob.core.windows.net/<container>/<prefix>`), which falls into the `default` case returning `"unsupported schema https"`.

**Impact**: All Azure clusters with PiTR have **zero log billing records**, resulting in **revenue leakage** since Azure Dedicated launched.

**Severity**: Critical (P1) — active revenue loss.

---

## 2. Detailed Root Cause Analysis

### 2.1 Code Flow

```
Worker cron (10 * * * *)
  → CalculateLogBackupSize()
    → logBackupSizeCalculatorImpl.CalculateAll()
      → countSizeForOne() [for each active log backup]
        → countLogBackupSize()
          → generateLogBackupSizeHandler()
            → NewLogBackupSizeHandler(ctx, endTime, startTime, truncateUntil, logBackup.Storage, s3Option)
```

### 2.2 Bug Location

**File**: `app/models/backup/impl/log_backup_size_calculator/log_backup_size.go` lines 37-55

```go
func NewLogBackupSizeHandler(ctx context.Context,
    currentReadTime, lastReadTime, truncateUntil time.Time,
    logBackupPath string, s3Option *internalS3.S3Option,
) (LogBackupSizeHandler, error) {
    u, err := url.Parse(logBackupPath)
    if err != nil {
        return nil, errors.Annotate(err, "parse log backup path")
    }
    switch u.Scheme {
    case "gcs":
        return newLogBackupSizeGCSHandler(...)
    case "gs":
        return newLogBackupSizeGCSHandler(...)
    case "s3":
        return newLogBackupSizeS3Handler(...)
    default:
        return nil, errors.Errorf("unsupported schema %s of log backup path", u.Scheme) // ← BUG: Azure "https" hits this
    }
}
```

### 2.3 Why Azure Uses `https://` Scheme

The `logBackup.Storage` field is populated from the Kubernetes CR's `Status.BackupPath` (see `app/models/backup/impl/sync_log_backup_from_infra.go:139`). For Azure clusters, TiDB Operator produces Azure Blob Storage URLs in the native format:

```
https://476cc5554d960110404b9fdd.blob.core.windows.net/dbaas-nightly-eastus2-nightly-ms-3199728/...
```

This is the standard Azure Blob Storage URL format, unlike S3/GCS which use their own scheme prefixes.

### 2.4 Secondary Bug: Permanent Masking

When an Azure log backup gets deleted, the flow is:
1. `countSizeForOne()` → `logBackup.IsDeleted()` = true
2. `logBackupSize == nil` (no previous size record ever computed)
3. → `markLogBackupFinishSizeCalculator()` sets `SizeCalculatorFinished = true`
4. The log backup is now permanently excluded from future calculations

This means **even if we fix the handler, already-deleted Azure log backups will never be retroactively billed**. For active log backups, they will start generating billing once fixed.

### 2.5 `generateLogBackupSizeHandler()` Gap

**File**: `app/models/backup/impl/log_backup_size_calculator/log_backup_size_calculator.go` lines 168-210

```go
func (l *logBackupSizeCalculatorImpl) generateLogBackupSizeHandler(...) {
    bucket, err := l.bucketRepo.Load(ctx, uint64(logBackup.BucketID))
    if bucket.InfraProviderType == entity.InfraProviderTypeAWS ||
        bucket.InfraProviderType == entity.InfraProviderTypeEKS {
        // AWS path: passes S3Option with role ARN
        handler, err := NewLogBackupSizeHandler(ctx, ..., logBackup.Storage, &internalS3.S3Option{...})
        return handler, nil
    }
    // Everything else (GCP, Azure) → no S3Option
    handler, iErr := NewLogBackupSizeHandler(ctx, ..., logBackup.Storage, nil)
    return handler, nil
}
```

For Azure, the code reaches `NewLogBackupSizeHandler` with `logBackup.Storage = "https://..."` and `s3Option = nil`. Since there's no `"https"` case in the scheme switch, it errors out.

---

## 3. Fix Design

### 3.1 Approach: Add Azure Blob Storage Handler

Create a new `LogBackupSizeAzBlobHandler` that follows the same pattern as S3/GCS handlers, using the existing `pkg/azblob` package.

### 3.2 Changes Required

#### 3.2.1 New File: `log_backup_size_azblob.go`

Create `app/models/backup/impl/log_backup_size_calculator/log_backup_size_azblob.go`:

- `LogBackupSizeAzBlobHandler` struct embedding `LogBackupSizeBaseHandler`
- Uses Azure SDK's `container.Client` for blob operations
- Implements `ExecuteStatistic()`, `PutSizeFile()` using Azure Blob flat/hierarchy listing
- Key operations:
  - **Get size file**: `NewBlockBlobClient(path).DownloadStream()`
  - **Put size file**: `NewBlockBlobClient(path).Upload()`
  - **List subfolders**: `NewListBlobsHierarchyPager(delimiter="/", prefix)` — returns `CommonPrefixes`
  - **Count folder size**: `NewListBlobsFlatPager(prefix)` — sums `ContentLength`

#### 3.2.2 Modify: `log_backup_size.go`

Add `"https"` case to `NewLogBackupSizeHandler`:

```go
case "https":
    return newLogBackupSizeAzBlobHandler(ctx, currentReadTime, lastReadTime, truncateUntil, logBackupPath, sasToken)
```

The function signature needs to accept a SAS token parameter (or an option struct that can carry it).

#### 3.2.3 Modify: `log_backup_size_calculator.go`

In `generateLogBackupSizeHandler()`:
- Detect Azure infra provider type (`InfraProviderTypeAKS` / `InfraProviderTypeAzure`)
- Retrieve SAS token from bucket via `bucket.GetSASToken(encryptor)`
- Pass the SAS token to `NewLogBackupSizeHandler`

This requires the `logBackupSizeCalculatorImpl` to have access to the encryptor. We need to inject `encryptor_wrapper.Encryptor` into the struct.

#### 3.2.4 Modify: `container.go`

Pass `container.Encryptor` to `NewLogBackupSizeCalculator`.

### 3.3 Detailed Implementation

#### A) `NewLogBackupSizeHandler` Signature Change

Option 1 (Minimal): Add a generic options struct:
```go
type LogBackupSizeOption struct {
    S3Option *internalS3.S3Option
    SASToken string // for Azure
}
```


#### B) Azure Handler Implementation

The handler needs to:
1. Parse Azure URL to extract account, container, prefix (using `pkg/azblob.ParseAzblobURL`)
2. Create Azure client using `pkg/azblob.NewAzureBlobStorage`
3. Implement folder listing with hierarchy pager (delimiter `/`)
4. Implement flat listing for size counting

However, the existing `pkg/azblob.AzureBlobStorage` interface doesn't expose:
- Hierarchical listing (only flat `WalkDir`)
- File writing (`PutBlob`/`Upload`)
- Reading a specific named file relative to the prefix

We need to use the Azure SDK's `container.Client` directly from the `azureBlobStorageImpl`. Or we can use the underlying container client. Let's check what's available:

- `ReadFile(ctx, name, startOffset, endOffset)` — can read the size file (by passing name relative to prefix)
- `WalkDir(ctx, opt, fn)` — walks all blobs flat, returns path and size
- No hierarchical listing
- No write capability

**Decision**: We'll create the Azure handler using the Azure SDK `container.Client` directly (similar to how S3 handler uses `s3.S3` client directly). This gives us full control for:
- Hierarchical blob listing (`NewListBlobsHierarchyPager`)
- Flat blob listing for size sum (`NewListBlobsFlatPager`)
- Reading/writing size file via `BlockBlobClient`

#### C) `generateLogBackupSizeHandler()` Update

```go
func (l *logBackupSizeCalculatorImpl) generateLogBackupSizeHandler(...) {
    bucket, err := l.bucketRepo.Load(ctx, uint64(logBackup.BucketID))
    ...
    switch {
    case bucket.InfraProviderType == entity.InfraProviderTypeAWS ||
         bucket.InfraProviderType == entity.InfraProviderTypeEKS:
        // existing S3 path
    case bucket.InfraProviderType == entity.InfraProviderTypeAKS ||
         bucket.InfraProviderType == entity.InfraProviderTypeAzure:
        sasToken, err := bucket.GetSASToken(l.encryptor)
        if err != nil {
            return nil, bizerr_wrapper.NewInternalErr(...)
        }
        handler, err := NewLogBackupSizeHandler(ctx, endTime, *startTime, truncateUntil,
            logBackup.Storage, nil, sasToken)
        return handler, nil
    default:
        // GCP path (unchanged)
    }
}
```

---

## 4. Implementation Steps

### Step 1: Modify `logBackupSizeCalculatorImpl` struct to include encryptor

**File**: `app/models/backup/impl/log_backup_size_calculator/log_backup_size_calculator.go`

Add `encryptor encryptor_wrapper.Encryptor` field, update `NewLogBackupSizeCalculator` constructor.

### Step 2: Update DI container

**File**: `app/container/container.go`

Pass `container.Encryptor` when constructing `LogBackupSizeCalculator`.

### Step 3: Update `NewLogBackupSizeHandler` signature

**File**: `app/models/backup/impl/log_backup_size_calculator/log_backup_size.go`

Add `sasToken string` parameter, add `"https"` case to switch.

### Step 4: Create Azure handler

**File**: `app/models/backup/impl/log_backup_size_calculator/log_backup_size_azblob.go`

Implement `LogBackupSizeAzBlobHandler` using Azure SDK.

### Step 5: Update `generateLogBackupSizeHandler()`

**File**: `app/models/backup/impl/log_backup_size_calculator/log_backup_size_calculator.go`

Add Azure branch that extracts SAS token and passes it.

### Step 6: Add unit tests

### Step 7: Build verification

---

## 5. Test Plan

### 5.1 Unit Tests

#### 5.1.1 `log_backup_size_azblob_test.go` — Azure Handler Tests

| Test Case | Description | Input | Expected |
|-----------|-------------|-------|----------|
| `TestNewLogBackupSizeAzBlobHandler_Success` | Valid Azure URL + SAS token creates handler | `https://acc.blob.core.windows.net/container/prefix`, valid SAS | Handler created, logBackupName = trimmed prefix |
| `TestNewLogBackupSizeAzBlobHandler_InvalidURL` | Invalid URL fails | `://invalid` | Error |
| `TestNewLogBackupSizeAzBlobHandler_EmptySASToken` | Empty SAS token fails | valid URL, empty SAS | Error |
| `TestAzBlobHandler_ExecuteStatistic` | Mocked Azure listing correctly sums folder sizes | Pre-set blobs with known sizes | foldersSizeMap populated correctly |
| `TestAzBlobHandler_PutSizeFile` | Size file uploaded correctly | Known foldersSizeMap state | Upload called with correct JSON content |
| `TestAzBlobHandler_GetSizeFile_NotExist` | Size file doesn't exist returns empty | BlobNotFound error | Empty content, no error |
| `TestAzBlobHandler_GetSizeFile_Exists` | Size file exists returns content | Pre-existing size file | Content returned |
| `TestAzBlobHandler_CountHourSubFoldersSize` | Hierarchical listing finds subfolders | Blobs under `prefix/v1/20240101/10/1/`, `prefix/v1/20240101/10/2/` | 2 subfolders counted |
| `TestAzBlobHandler_CountFolderSize` | Flat listing sums sizes | 3 blobs in folder, sizes 100, 200, 300 | Total = 600 |

#### 5.1.2 `log_backup_size_calculator_test.go` — Integration Tests

| Test Case | Description | Expected |
|-----------|-------------|----------|
| `TestGenerateLogBackupSizeHandler_Azure` | Azure bucket type dispatches to Azure handler | Azure handler created with SAS token |
| `TestGenerateLogBackupSizeHandler_Azure_NoSASToken` | Azure bucket without SAS token | Error returned |
| `TestGenerateLogBackupSizeHandler_AWS` | AWS bucket still works as before | S3 handler created |
| `TestGenerateLogBackupSizeHandler_GCP` | GCP bucket still works as before | GCS handler created |

#### 5.1.3 `log_backup_size_test.go` — Factory Tests

| Test Case | Description | Expected |
|-----------|-------------|----------|
| `TestNewLogBackupSizeHandler_AzureScheme` | `https://` URL with SAS token | Azure handler returned |
| `TestNewLogBackupSizeHandler_S3Scheme` | `s3://` URL unchanged | S3 handler returned |
| `TestNewLogBackupSizeHandler_GCSScheme` | `gcs://` URL unchanged | GCS handler returned |
| `TestNewLogBackupSizeHandler_UnknownScheme` | `ftp://...` | Error: unsupported schema |

### 5.2 Integration Verification

After implementation, run:
```bash
make app-unit-test-local
```

### 5.3 Manual Verification (Pre-merge)

1. Deploy to staging environment with Azure clusters that have active PiTR log backups
2. Wait for the worker cron cycle (runs at :10 of each hour)
3. Verify:
   - No `"unsupported schema https"` error in logs
   - `log_backup_sizes` table has new records for Azure log backups
   - `total_size` and `increased_size` are non-zero and reasonable

### 5.4 E2E Test (Post-merge)

Existing BR E2E tests for PiTR should cover the happy path. Confirm with the test matrix that Azure PiTR scenarios are included.

### 5.5 Regression Safety

- S3 and GCS paths are unchanged (pass `""` for sasToken in non-Azure calls)
- The `default` case in the switch still catches truly unsupported schemes
- Existing tests for S3/GCS continue to pass

---

## 6. Risk Assessment

| Risk | Mitigation |
|------|-----------|
| SAS token decryption failure blocks all Azure billing | Graceful error handling: log error and skip that backup, don't crash the entire batch |
| Azure SDK version incompatibility | Use same version already imported in `pkg/azblob` |
| Hierarchy listing API behavior difference from S3 | Azure `NewListBlobsHierarchyPager` with `/` delimiter produces `BlobPrefixes` (equivalent to S3's CommonPrefixes) — well-documented |
| Performance: Azure listing may be slower | Same concurrency model as S3/GCS; worker pool already caps concurrency |
| SAS token expiry mid-operation | SAS tokens for backup buckets have long TTLs (typically 1+ year); if expired, error is logged per-backup and retried next cycle |

---

## 7. Files Changed Summary

| File | Change |
|------|--------|
| `app/models/backup/impl/log_backup_size_calculator/log_backup_size_azblob.go` | **NEW** — Azure handler implementation |
| `app/models/backup/impl/log_backup_size_calculator/log_backup_size.go` | Add `sasToken` param, `"https"` case |
| `app/models/backup/impl/log_backup_size_calculator/log_backup_size_calculator.go` | Add encryptor field, Azure branch in `generateLogBackupSizeHandler()` |
| `app/models/backup/impl/log_backup_size_calculator/log_backup_size_s3.go` | Update `newLogBackupSizeS3Handler` call sites (no functional change) |
| `app/models/backup/impl/log_backup_size_calculator/log_backup_size_gcs.go` | Update `newLogBackupSizeGCSHandler` call sites (no functional change) |
| `app/container/container.go` | Pass `Encryptor` to `NewLogBackupSizeCalculator` |
| `app/models/backup/impl/log_backup_size_calculator/log_backup_size_azblob_test.go` | **NEW** — Unit tests |

---

## 8. Post-Fix Actions

1. **Backfill investigation**: For Azure log backups already deleted (permanently masked by `SizeCalculatorFinished = true`), assess revenue impact and determine if manual billing record creation is needed.
2. **Monitoring alert**: Add an alert on `log_backup_sizes` record count per cloud provider to catch future gaps.
3. **Existing log backup recovery**: For active Azure log backups, the fix will start generating records from the next cycle. The first record will have `IncreasedSize = TotalSize` (since there's no prior size file).