package facade_imp

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"math"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/go-resty/resty/v2"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	aws2 "github.com/tidbcloud/serverless-service/internal/infra/aws"
	"github.com/tidbcloud/serverless-service/internal/metric"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	RetryCount    = 3
	RetryWaitTime = 200 * time.Millisecond

	/* only scrape the storage size that is larger than 40MiB, ignore the other data */
	TikvUsedBytesPromQL       = `quantile_over_time(0.5, sum by(keyspace_id) (avg by(keyspace_id, region_id) (tikv_store_size_bytes{namespace="tidb-serverless",component="tikv",type="used"}))[5m]) > 40*1024*1024`
	TiflashUsedBytesPromQL    = `quantile_over_time(0.5, sum by(keyspace_id) (avg by(keyspace_id, region_id) (tikv_store_size_bytes{namespace="tidb-serverless",component="tikv",type="tiflash_used"}))[5m]) > 40*1024*1024`
	TidbRequestUnitPromQL     = `((sum(increase(resource_manager_resource_unit_read_request_unit_sum{namespace="tidb-serverless",type=~"|tp"}[%dm])) by(name) or vector(0)) + (sum(increase(resource_manager_resource_unit_write_request_unit_sum{namespace="tidb-serverless",type=~"|tp"}[%dm])) by(name) or vector(0)) + (sum(increase(resource_manager_resource_unit_sql_layer_request_unit{namespace="tidb-serverless",type=~"|tp"}[%dm])) by(name) or vector(0)) + (sum(increase(resource_manager_resource_unit_sql_layer_request_unit_sum{namespace="tidb-serverless",type=~"|tp"}[%dm])) by(name) or vector(0))) > 0`
	LoadDataRequestUnitPromQL = `sum(increase(sum(tikv_worker_load_data_wru_cost_counter{namespace="tidb-serverless"}) by(keyspace_id) [%dm])) by(keyspace_id) > 0`
	TiflashRequestUnitPromQL  = `((sum(increase(resource_manager_resource_unit_read_request_unit_sum{namespace="tidb-serverless",type="ap"}[%dm])) by(name)) + (sum(increase(resource_manager_resource_unit_write_request_unit_sum{namespace="tidb-serverless",type="ap"}[%dm])) by(name))) > 0`
	PublicEgressBytesPromQL   = `sum(increase(gateway_connection_flow_bytes_total{namespace="tidb-admin",direction="out",cluster_id!="",service=~"gateway-metrics|gateway-metrics-privatelink"}[%dm])) by(cluster_id) > 0`
	ExportRequestUnitPromQL   = `((avg(resource_manager_resource_unit_read_request_unit_sum{namespace="tidb-serverless",name="%s",type=~"|tp"}) or vector(0)) + (avg(resource_manager_resource_unit_sql_layer_request_unit_sum{namespace="tidb-serverless",name="%s",type=~"|tp"}) or vector(0))) > 0`

	ClusterRowStoragebyRegionPromQL                    = `cluster:tikv_store_size_bytes:row_based_storage{keyspace_id="%d"}`
	ClusterColumnStoragebyRegionPromQL                 = `cluster:tikv_store_size_bytes:columnar_storage{keyspace_id="%d"}`
	BranchesRowStorageByRegionPromQL                   = `cluster:tikv_store_size_bytes:row_based_storage{keyspace_id=~"%s"}`
	BranchesColumnStorageByRegionPromQL                = `cluster:tikv_store_size_bytes:columnar_storage{keyspace_id=~"%s"}`
	BranchesRowStorageByRegionWithoutQuantilePromQL    = `sum by(keyspace_id) (avg by(keyspace_id, region_id) (tikv_store_size_bytes{namespace="tidb-serverless",component="tikv",type="used", keyspace_id=~"%s"}))`
	BranchesColumnStorageByRegionWithoutQuantilePromQL = `sum by(keyspace_id) (avg by(keyspace_id, region_id) (tikv_store_size_bytes{namespace="tidb-serverless",component="tikv",type="tiflash_used", keyspace_id=~"%s"}))`

	provisionedRCUPromQL = "max(max_over_time(serverless_metering_rcu_provisioned_capacity[1h])) by (cluster_id)"
)

var downtimePromQL = map[string]string{
	"down_rate": `sum(increase(gateway_connection_sla_total{ru_mode!="throttled",cluster_id!="",status=~"inner_error|tidb_error|cancel"}[1m])) by (tidb_pool, cluster_id)
/sum(increase(gateway_connection_sla_total{ru_mode!="throttled",cluster_id!="",status!="missmatch"}[1m])) by (tidb_pool, cluster_id) > 0`,
	"connections": `(sum(increase(gateway_connection_sla_total{ru_mode!="throttled",cluster_id!=""}[1m])) by (tidb_pool, cluster_id, status) and
sum(increase(gateway_connection_sla_total{ru_mode!="throttled",cluster_id!="",status=~"inner_error|tidb_error"}[1m])) by (tidb_pool, cluster_id, status)) > 0`,
}

var metricTypeToKey = map[constant.MetricType]string{
	constant.MetricTypeTikvStorage:                   "keyspace_id",
	constant.MetricTypeTiflashStorage:                "keyspace_id",
	constant.MetricTypeTidbRU:                        "name",
	constant.MetricTypeLoadDataRU:                    "keyspace_id",
	constant.MetricTypeTiflashRU:                     "name",
	constant.MetricTypePublicEgress:                  "cluster_id",
	constant.MetricTypeTikvStorageWithoutQuantile:    "keyspace_id",
	constant.MetricTypeTiflashStorageWithoutQuantile: "keyspace_id",
}

var sku2BranchPromQL = map[constant.MetricType]string{
	constant.MetricTypeTikvStorage:    BranchesRowStorageByRegionPromQL,
	constant.MetricTypeTiflashStorage: BranchesColumnStorageByRegionPromQL,
	// Don't use this metrics in billing, use BranchesRowStorageByRegionPromQL instead.
	constant.MetricTypeTikvStorageWithoutQuantile:    BranchesRowStorageByRegionWithoutQuantilePromQL,
	constant.MetricTypeTiflashStorageWithoutQuantile: BranchesColumnStorageByRegionWithoutQuantilePromQL,
}

var sku2ExportPromQL = map[constant.MetricType]string{
	constant.MetricTypeExportRU: ExportRequestUnitPromQL,
}

type promResponse struct {
	Status    string   `json:"status"`
	Data      promData `json:"data"`
	ErrorType string   `json:"errorType"`
	Error     string   `json:"error"`
}

type promData struct {
	ResultType string       `json:"resultType"`
	Result     []resultItem `json:"result"`
}

/* !!NOTE!!
 * resultItem uses one of `Values` and `Value`:
 *   `Values` is used for accessing endpoint `query_range`;
 *   `Value` is used for accessing endpoint `query`;
 */
type resultItem struct {
	Metric map[string]string `json:"metric"`
	Values [][]any           `json:"values"`
	Value  []any             `json:"value"`
}

func queryRangeMetrics(
	ctx context.Context,
	o11yUrl string,
	promQL string,
	start, end time.Time,
) (*promResponse, bizerr.BizError) {
	params := url.Values{
		"query": []string{promQL},
		"start": []string{start.Truncate(time.Minute).Format(time.RFC3339)},
		"end":   []string{end.Truncate(time.Minute).Format(time.RFC3339)},
		"step":  []string{"60"},
	}
	url := fmt.Sprintf("http://%s/api/v1/query_range?%s", o11yUrl, params.Encode())
	return callO11y(ctx, url)
}

func queryMetrics(ctx context.Context, o11yUrl, promQL string, t time.Time) (*promResponse, bizerr.BizError) {
	params := url.Values{
		"query": []string{promQL},
		"time":  []string{t.Truncate(time.Minute).Format(time.RFC3339)},
		"_":     []string{strconv.FormatInt(t.Unix(), 10)}, // a dummy value just avoid cache
	}
	url := fmt.Sprintf("http://%s/api/v1/query?%s", o11yUrl, params.Encode())
	return callO11y(ctx, url)
}

func callO11y(ctx context.Context, url string) (*promResponse, bizerr.BizError) {
	req := resty.New().
		SetRetryCount(RetryCount).
		SetRetryWaitTime(RetryWaitTime).
		AddRetryAfterErrorCondition().
		AddRetryHook(func(resp *resty.Response, err error) {
			log.WithContextL(ctx).Info("retry call o11y",
				zap.String("url", url), zap.Int("code", resp.StatusCode()), zap.Error(err))
		}).R().EnableTrace().SetContext(ctx)
	resp, err := req.Get(url)
	log.WithContextL(ctx).Info("call o11y", zap.String("url", url), zap.Any("resp", resp), zap.Error(err))
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeHTTP, "http request failed: %s", err)
	}

	respBody := resp.Body()
	if resp.StatusCode() != http.StatusOK {
		return nil, bizerr.New(bizerr.ErrCodeHTTP, " http status code(%d) is not ok", resp.StatusCode())
	}

	promResp := &promResponse{}
	err = json.Unmarshal(respBody, promResp)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "failed to unmarshal prometheus response: %s", err)
	}

	if promResp.Status != "success" {
		return nil, bizerr.New(bizerr.ErrCodeHTTP, "prometheus response status != success")
	}

	return promResp, nil
}

type o11yFacadeImp struct {
	s3ClientProvider *aws2.S3ClientProvider
}

func NewO11yFacade(s3ClientProvider *aws2.S3ClientProvider) facade.O11yFacade {
	return &o11yFacadeImp{s3ClientProvider: s3ClientProvider}
}

func (f *o11yFacadeImp) GetClusterSkuUsageByRegion(
	ctx context.Context,
	clusterId string,
	keyspaceId uint32,
	o11yAddr string,
) (_ map[string]float64, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "o11yFacade.GetClusterSkuUsageByRegion")
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", clusterId), zap.Uint32("keyspace_id", keyspaceId))
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("o11y", "GetClusterSkuUsageByRegion", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("o11y", "GetClusterSkuUsageByRegion", status).
			Observe(time.Since(start).Seconds())
	}()

	sku2PromQLbyRegionTemplate := make(map[constant.SkuType]string)
	sku2PromQLbyRegionTemplate[constant.SkuTypeRowStorage] = fmt.Sprintf(ClusterRowStoragebyRegionPromQL, keyspaceId)
	sku2PromQLbyRegionTemplate[constant.SkuTypeColumnStorage] = fmt.Sprintf(
		ClusterColumnStoragebyRegionPromQL,
		keyspaceId,
	)
	usages := make(map[string]float64)
	for sku, promql := range sku2PromQLbyRegionTemplate {
		// resp, err := queryRangeMetrics(o11yAddr, promql, start, start) #by yys, disable query_range
		resp, err := queryMetrics(ctx, o11yAddr, promql, start)
		if err != nil {
			return nil, err
		}
		for _, r := range resp.Data.Result {
			// v, ok := r.Values[0][1].(string) #by yys, disable query_range
			v, ok := r.Value[1].(string)
			if !ok {
				// logger.Error("failed to get metric value", zap.Any("value", r.Values[0][1])) #by yys, disable query_range
				logger.Error("failed to get metric value", zap.Any("value", r.Value[1]))
				continue
			}
			value, err := strconv.ParseFloat(v, 64)
			if err != nil {
				logger.Error("failed to parse value", zap.Error(err))
				continue
			}
			if sku == constant.SkuTypeRowStorage ||
				sku == constant.SkuTypeColumnStorage {
				// storage to MiB
				usages[sku.String()] = value / 1024 / 1024
			} else {
				usages[sku.String()] = value
			}
		}
	}
	logger.Info(
		"GetClusterSkuUsageByRegion",
		zap.Any("result", usages),
		zap.String("o11y_addr", o11yAddr),
		zap.String("duration", time.Since(start).String()),
	)
	return usages, nil
}

func (f *o11yFacadeImp) GetBranchesSkuUsagesByRegion(
	ctx context.Context,
	keyspaceIds []uint32,
	metricType constant.MetricType,
	o11yAddr string,
) (_ map[string]float64, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "o11yFacade.GetBranchesSkuUsagesByRegion")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("o11y", "GetBranchesSkuUsagesByRegion", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("o11y", "GetBranchesSkuUsagesByRegion", status).
			Observe(time.Since(start).Seconds())
	}()

	logger := log.WithContextL(ctx).With(
		zap.Any("keyspace_ids", keyspaceIds),
	)

	var promql string
	if temp, exist := sku2BranchPromQL[metricType]; exist {
		var builder strings.Builder
		for i, num := range keyspaceIds {
			if i > 0 {
				builder.WriteString("|")
			}
			builder.WriteString(strconv.FormatUint(uint64(num), 10))
		}
		joined := builder.String()
		promql = fmt.Sprintf(temp, joined)
	} else {
		return nil, errors.Errorf("not supported metricType type %s for query branches", metricType)
	}

	usages := make(map[string]float64)
	resp, err := queryMetrics(ctx, o11yAddr, promql, start)
	if err != nil {
		return nil, err
	}

	metricKey, exist := metricTypeToKey[metricType]
	if !exist {
		return nil, errors.Errorf("unknown metricType type %s to get metrics key", metricType)
	}

	for _, r := range resp.Data.Result {
		id, exist := r.Metric[metricKey]
		if !exist {
			logger.Error("failed to get metric label", zap.String("key", metricKey), zap.Error(err))
			continue
		}

		v, ok := r.Value[1].(string)
		if !ok {
			logger.Error("failed to get metric value", zap.Any("value", r.Value[1]))
			continue
		}
		value, err := strconv.ParseFloat(v, 64)
		if err != nil {
			logger.Error("failed to parse value", zap.Error(err))
			continue
		}
		if metricType == constant.MetricTypeTikvStorage ||
			metricType == constant.MetricTypeTiflashStorage {
			// storage to MiB
			usages[id] = value / 1024 / 1024
		} else {
			usages[id] = value
		}
	}
	logger.Info(
		"GetBranchesSkuUsagesByRegion",
		zap.Any("result", usages),
		zap.String("o11y_addr", o11yAddr),
		zap.String("duration", time.Since(start).String()),
	)
	return usages, nil
}

func (f *o11yFacadeImp) GetDowntimes(
	ctx context.Context,
	o11yAddr string,
) (result map[string]facade.DowntimeRes, err error) {
	now := time.Now()
	ctx, span := Tracer.Start(ctx, "o11yFacade.GetDowntimeRate")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("o11y", "GetDowntimeRate", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("o11y", "GetDowntimeRate", status).
			Observe(time.Since(now).Seconds())
	}()
	logger := log.WithContextL(ctx)

	result = make(map[string]facade.DowntimeRes)
	for types, promql := range downtimePromQL {
		resp, err := queryRangeMetrics(ctx, o11yAddr, promql, now.Add(-time.Minute*10), now)
		if err != nil {
			return nil, err
		}
		for _, r := range resp.Data.Result {
			for _, v := range r.Values {
				timestamp, ok := v[0].(float64)
				if !ok {
					logger.Error("failed to parser timestamp", zap.Any("value", v[0]))
					continue
				}
				tmp, ok := v[1].(string)
				if !ok {
					logger.Error("failed to parser downtime rate", zap.Any("value", v[1]))
					continue
				}
				value, err := strconv.ParseFloat(tmp, 64)
				if err != nil {
					logger.Error("failed to parse value = "+tmp, zap.Error(err))
					continue
				}
				if math.IsInf(value, 0) || math.IsNaN(value) {
					logger.Info(
						"failed to parser value",
						zap.String("value", tmp),
						zap.String("cluster_id", r.Metric["cluster_id"]),
					)
					continue
				}
				key := r.Metric["cluster_id"] + ":" + fmt.Sprint(int64(timestamp)) + ":" + r.Metric["tidb_pool"]
				if _, ok := result[key]; ok {
					res := result[key]
					if types == "connections" {
						res.Connections = value
					} else {
						res.DownRate = value
					}
					result[key] = res
					continue
				}
				res := facade.DowntimeRes{}
				if types == "connections" {
					res.Connections = value
				} else {
					res.DownRate = value
				}
				result[key] = res
			}
		}
	}
	return result, nil
}

func (f *o11yFacadeImp) GetProvisionedRCU(ctx context.Context, o11yAddr string) (map[uint64]uint64, error) {
	resp, err := queryMetrics(ctx, o11yAddr, provisionedRCUPromQL, time.Now())
	if err != nil {
		return nil, err
	}
	logger := log.WithContextL(ctx)
	result := make(map[uint64]uint64)
	for _, r := range resp.Data.Result {
		clusterId, err := strconv.ParseUint(r.Metric["cluster_id"], 10, 64)
		if err != nil {
			logger.Error("failed to get metric cluster_id label", zap.Error(err))
			continue
		}
		v, ok := r.Value[1].(string)
		if !ok {
			logger.Error("failed to get metric value", zap.Any("value", r.Value[1]))
			continue
		}
		value, err := strconv.ParseUint(v, 10, 64)
		if err != nil {
			logger.Error("failed to parse value", zap.Error(err))
			continue
		}
		result[clusterId] = value
	}
	return result, nil
}

// GetExportUsage returns the latest usage of the export.
func (f *o11yFacadeImp) GetExportUsage(ctx context.Context,
	exportID, o11yAddr string, completedAt time.Time, metricType constant.MetricType,
) (_ int64, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "o11yFacade.GetExportUsages")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("o11y", "GetExportUsages", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("o11y", "GetExportUsages", status).
			Observe(time.Since(start).Seconds())
	}()
	logger := log.WithContextL(ctx).With(zap.String("metric", metricType.String()))

	promQL, ok := sku2ExportPromQL[metricType]
	if !ok {
		return 0, bizerr.New(bizerr.ErrCodeUnknown, "invalid metric type")
	}
	promQL = fmt.Sprintf(promQL, exportID, exportID)
	// query range metrics [completedAt-5m, completedAt+5m]
	resp, err := queryRangeMetrics(
		ctx,
		o11yAddr,
		promQL,
		completedAt.Add(-time.Minute*5),
		completedAt.Add(time.Minute*5),
	)
	if err != nil {
		return 0, err
	}

	if len(resp.Data.Result) == 0 {
		return 0, bizerr.New(bizerr.ErrCodeUnknown, "empty result")
	}
	for _, r := range resp.Data.Result {
		// first column is timestamp, the second column is value. Get the latest value.
		v, ok := r.Values[len(r.Values)-1][1].(string)
		if !ok {
			logger.Error("failed to get metric value", zap.Any("value", r.Values[len(r.Values)-1][1]))
			continue
		}
		value, err := strconv.ParseFloat(v, 64)
		if err != nil {
			logger.Error("failed to parse value", zap.Error(err))
			continue
		}
		return int64(value), nil
	}
	return 0, bizerr.New(bizerr.ErrCodeUnknown, "get result failed")
}

var (
	bucketPrefix    = "o11y-prod-shared-" // + "us-east-1"
	keyPrefix       = "diagnosis/data/serverless/"
	statementPrefix = keyPrefix + "statements/168h/"
	slowlogPrefix   = keyPrefix + "slowlog-stats/168h/"
)

func (f *o11yFacadeImp) GetSQLiteDBFileFromS3(
	ctx context.Context,
	region string,
	keySuffix string,
) (_, _ string, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "o11yFacade.GetSQLiteDBFileFromS3")
	logger := log.WithContextL(ctx).With(zap.String("GetSQLiteDBFileFromS3", region))
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
			logger.Error("download drhouse diagnosis sqlite file failed", zap.Error(err))
		}
		metric.FacadeRequestedCounter.WithLabelValues("o11y", "GetSQLiteDBFileFromS3", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("o11y", "GetSQLiteDBFileFromS3", status).
			Observe(time.Since(start).Seconds())
	}()

	s3Client, err := f.s3ClientProvider.GetO11yS3Client(region)
	if err != nil {
		return "", "", err
	}

	bucket := bucketPrefix + region
	statmentsFile := region + "_statements_168h_" + keySuffix
	if _, err := os.Stat(statmentsFile); err != nil {
		if dbFiles, err := filepath.Glob("*.db"); err == nil {
			for _, file := range dbFiles {
				if strings.HasSuffix(file, keySuffix) {
					continue
				}
				// remove old files
				os.Remove(file)
			}
		}
		if err := downloadS3Object(ctx, s3Client, bucket, statementPrefix+keySuffix, statmentsFile); err != nil {
			return "", "", err
		}
	}

	slowlogFile := region + "_slowlog-stats_168h_" + keySuffix
	if _, err := os.Stat(slowlogFile); err != nil {
		if dbFiles, err := filepath.Glob("*.db"); err == nil {
			for _, file := range dbFiles {
				if strings.HasSuffix(file, keySuffix) {
					continue
				}
				os.Remove(file)
			}
		}
		if err := downloadS3Object(ctx, s3Client, bucket, slowlogPrefix+keySuffix, slowlogFile); err != nil {
			return "", "", err
		}
	}
	logger.Info("download drhouse diagnosis sqlite file success",
		zap.String("statements", statmentsFile), zap.String("slow-stats", slowlogFile))
	return statmentsFile, slowlogFile, nil
}

func downloadS3Object(ctx context.Context, s3Client *s3.Client, bucket, key, fileName string) error {
	params := &s3.GetObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	}
	resp, err := s3Client.GetObject(ctx, params)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	size := int(*resp.ContentLength)
	buffer := make([]byte, size)
	var bbuffer bytes.Buffer
	for {
		num, rerr := resp.Body.Read(buffer)
		if num > 0 {
			bbuffer.Write(buffer[:num])
		} else if rerr == io.EOF || rerr != nil {
			break
		}
	}
	return os.WriteFile(fileName, bbuffer.Bytes(), 0o644) //nolint:gosec
}

func (f *o11yFacadeImp) GetClusterRowStorage(ctx context.Context, cluster *tidbmgmtservicev1.Cluster) (int64, error) {
	clusterId := cluster.ClusterId
	keyspaceId := cluster.GetServerlessCluster().GetCluster().GetKeyspaceId()
	provider := cluster.CloudProvider
	region := cluster.Region
	kubernetes := cluster.GetServerlessCluster().GetCluster().GetKubernetes()
	logger := log.WithContextL(ctx).
		With(zap.Uint64("cluster_id", clusterId), zap.String("provider", provider), zap.String("region", region), zap.String("kubernetes", kubernetes))

	o11yAddr := config.GetO11yUrl(provider, region, kubernetes)
	if o11yAddr == "" {
		logger.Error("can't get o11y regional url")
		return 0, errors.New("can't get o11y regional url")
	}

	rowStorageUsage, err := f.GetClusterSkuUsageByRegion(ctx, strconv.FormatUint(clusterId, 10), keyspaceId, o11yAddr)
	if err != nil {
		logger.Error(
			"GetClusterSkuUsageByRegion failed",
			zap.Error(err),
			zap.Uint32("keyspace_id", keyspaceId),
			zap.String("addr", o11yAddr),
		)
		return 0, errors.New("GetClusterRowStorage failed")
	}

	// Return in bytes
	return int64(rowStorageUsage[string(constant.SkuTypeRowStorage)] * 1024 * 1024), nil
}