package slowquery import ( "context" "encoding/json" "fmt" "net/http" "net/http/httputil" "net/url" "strconv" "strings" "time" gcpstorage "cloud.google.com/go/storage" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/service/s3" "github.com/gin-gonic/gin" // "github.com/pingcap/log" "github.com/pingcap/tidb/util/plancodec" "go.uber.org/zap" "github.com/pingcap-inc/clinic/pkg/config" "github.com/pingcap-inc/clinic/pkg/log" "github.com/pingcap-inc/clinic/pkg/ngm/slowquery/slowstats" "github.com/pingcap-inc/clinic/pkg/ngm/storage" "github.com/pingcap-inc/clinic/pkg/ngm/utils" "github.com/pingcap-inc/clinic/pkg/o11y" o11yurls "github.com/pingcap-inc/clinic/pkg/o11y/urls" ) const ( perClusterQPSLimit = 10.0 perClusterLimiterBucketSize = 10 maxListQueryTimeRange = 3*60*60*24 + 5*60 // 3d (+5min) defaultTimeout = time.Second * 180 defaultListFields = "query,timestamp,query_time,memory_max" slowQueryListPath = "/internal/diagnosis/api/v1/slow_query/list" slowQueryAvailableFieldsPath = "/internal/diagnosis/api/v1/slow_query/available_fields" ) func InitSlowQueryApi(g *gin.RouterGroup, s *Service) { g.GET("/slow_query/list", s.handleList) g.GET("/slow_query/detail", s.handleDetail) g.GET("/slow_query/databases", s.handleDatabases) g.GET("/slow_query/available_fields", s.handleAvailableFields) g.GET("/slow_query/stats", s.handleStats) g.GET("/slow_query/stats/metric", s.handleStatsMetric) g.GET("/slow_query/stats/databases", s.handleStatsDatabases) g.GET("/slow_query/stats/time_windows", s.handleStatsTimeWindows) g.GET("/slow_query/analyze", s.handleAnalyze) g.GET("/slow_query/files", s.handleDownload) g.GET("/slow_query/vm_query", s.handleVmQuery("/vm_query")) g.GET("/slow_query/vm_query_range", s.handleVmQuery("/vm_query_range")) } func InitSlowQueryApiForDbaas(g *gin.RouterGroup, s *Service) { g.GET("/slow_query/list", s.handleList) g.GET("/slow_query/detail", s.handleDetail) g.GET("/slow_query/available_fields", s.handleAvailableFields) g.GET("/slow_query/files", s.handleDownload) } func NewSlowQueryService(env, cacheDir string, cfg *config.DBConfig, stores storage.StorageFactory, o11yCli *o11y.Client) *Service { return &Service{ env: env, o11yCli: o11yCli, SlowQuery: newRateLimitedSlowQuery(NewSlowQuery(env, cfg, stores), perClusterQPSLimit, perClusterLimiterBucketSize), SlowQueryV2: newRateLimitedSlowQuery(NewSlowQueryV2(stores), perClusterQPSLimit, perClusterLimiterBucketSize), StatsQuery: slowstats.NewStatsQuery(cacheDir, stores), } } // Service is a SlowQuery wrapper for handling HTTP requests and responses. type Service struct { env string o11yCli *o11y.Client SlowQuery SlowQuery SlowQueryV2 SlowQuery StatsQuery *slowstats.StatsQuery } // errorResponse represents the error details returned to the web front. type errorResponse struct { Error bool `json:"error"` Message string `json:"message"` Code string `json:"code"` FullText string `json:"full_text"` } // slowLogRespItem is a SlowLog representation for HTTP transport. // // Ported from pingcap/tidb-dashboard/pkg/apiserver/slowquery/model.go type slowLogRespItem struct { Digest string `gorm:"column:Digest" json:"digest"` Query string `gorm:"column:Query" json:"query"` Instance string `gorm:"column:INSTANCE" json:"instance"` DB string `gorm:"column:DB" json:"db"` // TODO: Switch back to uint64 when modern browser as well as Swagger handles BigInt well. ConnectionID string `gorm:"column:Conn_ID" json:"connection_id"` Success int `gorm:"column:Succ" json:"success"` Timestamp float64 `gorm:"column:timestamp" proj:"(UNIX_TIMESTAMP(Time) + 0E0)" json:"timestamp"` // finish time QueryTime float64 `gorm:"column:Query_time" json:"query_time"` // latency ParseTime float64 `gorm:"column:Parse_time" json:"parse_time"` CompileTime float64 `gorm:"column:Compile_time" json:"compile_time"` RewriteTime float64 `gorm:"column:Rewrite_time" json:"rewrite_time"` PreprocSubqueriesTime float64 `gorm:"column:Preproc_subqueries_time" json:"preproc_subqueries_time"` OptimizeTime float64 `gorm:"column:Optimize_time" json:"optimize_time"` WaitTSTime float64 `gorm:"column:Wait_TS" json:"wait_ts"` CopTime float64 `gorm:"column:Cop_time" json:"cop_time"` LockKeysTime float64 `gorm:"column:LockKeys_time" json:"lock_keys_time"` WriteRespTime float64 `gorm:"column:Write_sql_response_total" json:"write_sql_response_total"` ExecRetryTime float64 `gorm:"column:Exec_retry_time" json:"exec_retry_time"` MemoryMax int `gorm:"column:Mem_max" json:"memory_max"` DiskMax int `gorm:"column:Disk_max" json:"disk_max"` // TODO: Switch back to uint64 when modern browser as well as Swagger handles BigInt well. TxnStartTS string `gorm:"column:Txn_start_ts" json:"txn_start_ts"` // Detail PrevStmt string `gorm:"column:Prev_stmt" json:"prev_stmt"` Plan string `gorm:"column:Plan" json:"plan"` // BinaryPlan string `gorm:"column:Binary_plan" json:"binary_plan"` // Basic IsInternal int `gorm:"column:Is_internal" json:"is_internal"` IndexNames string `gorm:"column:Index_names" json:"index_names"` Stats string `gorm:"column:Stats" json:"stats"` BackoffTypes string `gorm:"column:Backoff_types" json:"backoff_types"` Prepared int `json:"prepared"` PlanFromCache int `json:"plan_from_cache"` PlanFromBinding int `json:"plan_from_binding"` // Connection User string `gorm:"column:User" json:"user"` Host string `gorm:"column:Host" json:"host"` // Time ProcessTime float64 `gorm:"column:Process_time" json:"process_time"` WaitTime float64 `gorm:"column:Wait_time" json:"wait_time"` BackoffTime float64 `gorm:"column:Backoff_time" json:"backoff_time"` GetCommitTSTime float64 `gorm:"column:Get_commit_ts_time" json:"get_commit_ts_time"` LocalLatchWaitTime float64 `gorm:"column:Local_latch_wait_time" json:"local_latch_wait_time"` ResolveLockTime float64 `gorm:"column:Resolve_lock_time" json:"resolve_lock_time"` PrewriteTime float64 `gorm:"column:Prewrite_time" json:"prewrite_time"` WaitPreWriteBinlogTime float64 `gorm:"column:Wait_prewrite_binlog_time" json:"wait_prewrite_binlog_time"` CommitTime float64 `gorm:"column:Commit_time" json:"commit_time"` CommitBackoffTime float64 `gorm:"column:Commit_backoff_time" json:"commit_backoff_time"` CopProcAvg float64 `gorm:"column:Cop_proc_avg" json:"cop_proc_avg"` CopProcP90 float64 `gorm:"column:Cop_proc_p90" json:"cop_proc_p90"` CopProcMax float64 `gorm:"column:Cop_proc_max" json:"cop_proc_max"` CopWaitAvg float64 `gorm:"column:Cop_wait_avg" json:"cop_wait_avg"` CopWaitP90 float64 `gorm:"column:Cop_wait_p90" json:"cop_wait_p90"` CopWaitMax float64 `gorm:"column:Cop_wait_max" json:"cop_wait_max"` // Transaction WriteKeys int `gorm:"column:Write_keys" json:"write_keys"` WriteSize int `gorm:"column:Write_size" json:"write_size"` PrewriteRegion int `gorm:"column:Prewrite_region" json:"prewrite_region"` TxnRetry int `gorm:"column:Txn_retry" json:"txn_retry"` // Coprocessor RequestCount uint `gorm:"column:Request_count" json:"request_count"` ProcessKeys uint `gorm:"column:Process_keys" json:"process_keys"` TotalKeys uint `gorm:"column:Total_keys" json:"total_keys"` CopProcAddr string `gorm:"column:Cop_proc_addr" json:"cop_proc_addr"` CopWaitAddr string `gorm:"column:Cop_wait_addr" json:"cop_wait_addr"` // RocksDB RocksdbDeleteSkippedCount uint `gorm:"column:Rocksdb_delete_skipped_count" json:"rocksdb_delete_skipped_count"` RocksdbKeySkippedCount uint `gorm:"column:Rocksdb_key_skipped_count" json:"rocksdb_key_skipped_count"` RocksdbBlockCacheHitCount uint `gorm:"column:Rocksdb_block_cache_hit_count" json:"rocksdb_block_cache_hit_count"` RocksdbBlockReadCount uint `gorm:"column:Rocksdb_block_read_count" json:"rocksdb_block_read_count"` RocksdbBlockReadByte uint `gorm:"column:Rocksdb_block_read_byte" json:"rocksdb_block_read_byte"` IARemoteReadSegmentCount uint64 `gorm:"column:IA_remote_read_segment_count" json:"ia_remote_read_segment_count"` IARemoteReadSegmentSize uint64 `gorm:"column:IA_remote_read_segment_size" json:"ia_remote_read_segment_size"` IARemoteReadSegmentWaitTime float64 `gorm:"column:IA_remote_read_segment_wait_time" json:"ia_remote_read_segment_wait_time"` // Resource Control RU float64 `gorm:"column:RU" json:"ru" proj:"(Request_unit_write + Request_unit_read)" related:"Request_unit_write,Request_unit_read"` QueuedTime float64 `gorm:"column:Time_queued_by_rc" json:"time_queued_by_rc"` ResourceGroup string `gorm:"column:Resource_group" json:"resource_group"` Warnings string `gorm:"column:Warnings" json:"warnings"` // RUv2 metrics RequestUnitV2 float64 `json:"ru_v2,omitempty"` RequestUnitV2Detail string `json:"ru_v2_detail,omitempty"` RequestUnitV2Metrics *RequestUnitV2Metrics `json:"ru_v2_metrics,omitempty"` } type clusterHeader struct { Provider string `header:"x-provider" binding:"required"` Region string `header:"x-region" binding:"required"` TenantID string `header:"x-org-id"` ProjectID string `header:"x-project-id"` ClusterID string `header:"x-cluster-id" binding:"required"` DeployType string `header:"x-deploy-type"` // "dedicated" or "shared" Env string `header:"x-env"` // Use default env if emprty } func isServerlessSlowQueryDeployType(deployType string) bool { return strings.EqualFold(deployType, o11y.SharedType) || strings.EqualFold(deployType, o11y.ServerlessType) } func (s *Service) resolveStorageTarget(ctx context.Context, header clusterHeader, category string) (*o11yurls.StorageTarget, error) { return o11yurls.NewClusterURLsResolver(s.o11yCli).ResolveStorageTarget(ctx, o11yurls.StorageTargetParam{ Provider: header.Provider, Region: header.Region, TenantID: header.TenantID, ClusterID: header.ClusterID, DeployType: header.DeployType, }, category) } type listRequest struct { BeginTime int `json:"begin_time" form:"begin_time"` EndTime int `json:"end_time" form:"end_time" binding:"required"` DB []string `json:"db" form:"db"` Limit int `json:"limit" form:"limit"` Text string `json:"text" form:"text"` OrderBy string `json:"orderBy" form:"orderBy"` IsDesc bool `json:"desc" form:"desc"` ShowInternal bool `json:"show_internal" form:"show_internal"` // for showing slow queries in the statement detail page Plans []string `json:"plans" form:"plans"` Digest string `json:"digest" form:"digest"` Fields string `json:"fields" form:"fields" binding:"required"` // example: "query,timestamp,query_time,memory_max,digest" } // @Summary Get list of slow query // @Tags slowquery // @Description Get list of slow query // @Produce json // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param x-deploy-type header string true "dedicated/shared" // @Param begin_time query int false "begin time in unix seconds" // @Param end_time query int true "end time in unix seconds" // @Param fields query string false "fields to return, separated by commas" // @Param db query []string false "database names to search" // @Param limit query int false "result count limit" // @Param text query string false "text to search" // @Param orderBy query string false "order by which field" // @Param desc query bool false "descending order" // @Param plans query []string false "plan digests to search" // @Param digest query string false "sql digest to search" // @Param show_internal query bool false "show internal slow queries" // @Success 200 {object} []slowLogRespItem // @Router /slow_query/list [get] func (s *Service) handleList(c *gin.Context) { ctx := c.Request.Context() f := recordHttpMetric(c, "/list") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req listRequest if err := c.ShouldBindQuery(&req); err != nil { log.Error(ctx, "invalid params", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if err := s.sanitizeListRequestFields(ctx, c, header, &req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to query available fields", FullText: err.Error(), }) return } if s.shouldProxySlowQueryToO11y(header.DeployType) { // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/list", header.ClusterID) // log.Info("update byoc slowquery list path", zap.String("path", path)) // } err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, slowQueryListPath) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } sharedTier := isServerlessSlowQueryDeployType(header.DeployType) nextGen := (strings.ToLower(header.DeployType) == "nextgen") || (strings.ToLower(header.DeployType) == "premium") params := &ListParams{ Provider: header.Provider, Region: header.Region, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, SharedTier: sharedTier, NextGen: nextGen, BeginTime: req.BeginTime, EndTime: req.EndTime, Limit: req.Limit, OrderBy: req.OrderBy, Text: req.Text, Digest: req.Digest, DB: req.DB, Plans: req.Plans, IsDesc: req.IsDesc, ShowInternal: req.ShowInternal, } if sharedTier { target, err := s.resolveStorageTarget(ctx, header, o11y.StorageCategorySlowQuery) if err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to resolve slow query storage target", FullText: err.Error(), }) return } params.StorageBucket = target.BucketName params.StoragePrefix = target.ObjectPath } var err error var slowLogs []*SlowLog if nextGen { slowLogs, err = s.SlowQuery.List(c.Request.Context(), params) } else { if !sharedTier && s.env == "prod" { slowLogs, err = s.SlowQueryV2.List(c.Request.Context(), params) } else { slowLogs, err = s.SlowQuery.List(c.Request.Context(), params) } } if err != nil { if e, ok := err.(awserr.Error); ok && e.Code() == s3.ErrCodeNoSuchBucket { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("aws region '%s' not deployed", header.Region), FullText: err.Error(), }) } else if err == gcpstorage.ErrBucketNotExist { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("gcp region '%s' not deployed", header.Region), FullText: err.Error(), }) } else if err == ErrUnsupportedOrderBy { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("unsupported order by '%s'", req.OrderBy), FullText: err.Error(), }) } else { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to query slow log list", FullText: err.Error(), }) } return } fields := strings.Split(req.Fields, ",") rs := make([]*slowLogRespItem, 0, len(slowLogs)) for _, sl := range slowLogs { rs = append(rs, slowLogToRespItemByFields(ctx, sl, fields)) } c.JSON(http.StatusOK, rs) } func (s *Service) sanitizeListRequestFields(ctx context.Context, c *gin.Context, header clusterHeader, req *listRequest) error { availableFields, err := s.availableFieldsForList(ctx, c, header) if err != nil { return err } req.Fields = filterListFields(req.Fields, availableFields) setRequestQueryParam(c.Request, "fields", req.Fields) return nil } func filterListFields(fields string, availableFields []string) string { available := make(map[string]string, len(availableFields)) for _, field := range availableFields { field = strings.TrimSpace(field) if field == "" { continue } available[strings.ToLower(field)] = field } filtered := make([]string, 0, len(strings.Split(fields, ","))) for _, field := range strings.Split(fields, ",") { field = strings.TrimSpace(field) if field == "" { continue } if canonicalField, ok := available[strings.ToLower(field)]; ok { filtered = append(filtered, canonicalField) } } if len(filtered) == 0 { return defaultListFields } return strings.Join(filtered, ",") } func setRequestQueryParam(r *http.Request, key, value string) { q := r.URL.Query() q.Set(key, value) r.URL.RawQuery = q.Encode() } func (s *Service) availableFieldsForList(ctx context.Context, c *gin.Context, header clusterHeader) ([]string, error) { if s.shouldProxySlowQueryToO11y(header.DeployType) { return fetchO11ySlowQueryAvailableFields(s, ctx, c, header) } return s.SlowQuery.AvailableFields(ctx) } var fetchO11ySlowQueryAvailableFields = func(s *Service, ctx context.Context, c *gin.Context, header clusterHeader) ([]string, error) { return s.fetchO11ySlowQueryAvailableFields(ctx, c, header) } func (s *Service) shouldProxySlowQueryToO11y(deployType string) bool { return !strings.EqualFold(deployType, o11y.SharedType) && !strings.EqualFold(deployType, o11y.ServerlessType) } func (s *Service) fetchO11ySlowQueryAvailableFields(ctx context.Context, c *gin.Context, header clusterHeader) ([]string, error) { o11yURL, o11yCluster, bizType, err := s.o11ySlowQueryURL(ctx, header, slowQueryAvailableFieldsPath) if err != nil { return nil, err } if o11yCluster == nil && bizType == "byoc" { return nil, fmt.Errorf("o11y cluster is nil") } req, err := http.NewRequestWithContext(ctx, http.MethodGet, o11yURL, nil) if err != nil { return nil, err } req.Header["x-env"] = []string{s.env} for k, v := range c.Request.Header { if strings.EqualFold(k, "Accept-Encoding") { continue } req.Header[k] = v } if bizType != "" { req.Header["x-biz-type"] = []string{bizType} } if bizType == "byoc" && o11yCluster != nil { req.Header["X-Org-Id"] = []string{o11yCluster.TenantID} req.Header["X-Region-Id"] = []string{fmt.Sprintf("%s-%s", o11yCluster.Vendor, o11yCluster.Region)} req.Header["X-Application-Id"] = []string{o11yCluster.ApplicationID} } resp, err := http.DefaultClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode >= 300 { return nil, fmt.Errorf("got abnormal status %d from o11y available fields", resp.StatusCode) } var fields []string if err := json.NewDecoder(resp.Body).Decode(&fields); err != nil { return nil, err } return fields, nil } func (s *Service) o11ySlowQueryURL(ctx context.Context, header clusterHeader, path string) (string, *o11y.ClusterInfo, string, error) { if s.o11yCli == nil { return "", nil, "", fmt.Errorf("o11y client is nil") } bizType := slowQueryO11yBizType(header.DeployType, s.env) target, err := utils.ResolveO11yInternalReadTarget(ctx, s.o11yCli, header.DeployType, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID) if err != nil { return "", nil, "", fmt.Errorf("failed to resolve o11y internal read target: %w", err) } o11yURL, err := url.Parse(target.InternalReadURL) if err != nil { return "", nil, "", fmt.Errorf("failed to parse o11y internal read uri: %w", err) } o11yURL.Path = path if header.DeployType == o11y.ByocType { if byocDSEndpoint := s.o11yCli.GetByocDSEndpoint(); byocDSEndpoint != "" { o11yURL.Host = byocDSEndpoint } if target.O11yCluster == nil { return "", nil, "", fmt.Errorf("byoc o11y cluster is empty") } o11yURL.Path = "/internal/" + target.O11yCluster.ID + strings.TrimPrefix(o11yURL.Path, "/internal") } return o11yURL.String(), target.O11yCluster, bizType, nil } func slowQueryO11yBizType(deployType, env string) string { return o11y.InternalReadBizType(env, deployType) } type detailRequest struct { Digest string `json:"digest" form:"digest" binding:"required"` Timestamp float64 `json:"timestamp" form:"timestamp" binding:"required"` // TODO: Switch back to uint64 when modern browser as well as Swagger handles BigInt well. ConnectID string `json:"connect_id" form:"connect_id" binding:"required"` ShowInternal bool `json:"show_internal" form:"show_internal"` } // @Summary Get detail of slow query // @Tags slowquery // @Description Get detail of slow query // @Produce json // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param x-deploy-type header string true "dedicated/shared" // @Param digest query string true "sql digest to match slow log item" // @Param timestamp query float64 true "timestamp to match slow log item" // @Param connect_id query string true "connection id to match slow log item" // @Param show_internal query bool false "show internal slow queries" // @Success 200 {object} slowLogRespItem // @Router /slow_query/detail [get] func (s *Service) handleDetail(c *gin.Context) { f := recordHttpMetric(c, "/detail") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req detailRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } connID, err := strconv.ParseInt(req.ConnectID, 10, 64) if err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("invalid connection id format: %s", req.ConnectID), FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if s.shouldProxySlowQueryToO11y(header.DeployType) { path := "/internal/diagnosis/api/v1/slow_query/detail" // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/detail", header.ClusterID) // log.Info("update byoc slowquery detail path", zap.String("path", path)) // } err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, path) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } sharedTier := isServerlessSlowQueryDeployType(header.DeployType) nextGen := (strings.ToLower(header.DeployType) == "nextgen") || (strings.ToLower(header.DeployType) == "premium") params := &DetailParams{ Provider: header.Provider, Region: header.Region, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, SharedTier: sharedTier, NextGen: nextGen, ConnID: uint64(connID), Digest: req.Digest, Timestamp: req.Timestamp, ShowInternal: req.ShowInternal, } if sharedTier { target, err := s.resolveStorageTarget(ctx, header, o11y.StorageCategorySlowQuery) if err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to resolve slow query storage target", FullText: err.Error(), }) return } params.StorageBucket = target.BucketName params.StoragePrefix = target.ObjectPath } var sl *SlowLog if nextGen { sl, err = s.SlowQuery.Detail(c.Request.Context(), params) } else { if !sharedTier && s.env == "prod" { sl, err = s.SlowQueryV2.Detail(c.Request.Context(), params) } else { sl, err = s.SlowQuery.Detail(c.Request.Context(), params) } } if err != nil { log.Warn(ctx, "failed to query slow log detail", zap.String("provider", header.Provider), zap.String("region", header.Region), zap.String("tenant", header.TenantID), zap.String("project", header.ProjectID), zap.String("cluster", header.ClusterID), zap.String("deploy_type", header.DeployType), zap.Bool("shared_tier", sharedTier), zap.Bool("next_gen", nextGen), zap.String("raw_connect_id", req.ConnectID), zap.Int64("parsed_conn_id", connID), zap.String("digest", req.Digest), zap.Float64("timestamp", req.Timestamp), zap.Error(err)) if e, ok := err.(awserr.Error); ok && e.Code() == s3.ErrCodeNoSuchBucket { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("aws region '%s' not deployed", header.Region), FullText: err.Error(), }) } else if err == gcpstorage.ErrBucketNotExist { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("gcp region '%s' not deployed", header.Region), FullText: err.Error(), }) } else { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to query slow log detail", FullText: err.Error(), }) } return } c.JSON(http.StatusOK, slowLogToRespItem(ctx, sl)) } type databasesRequest struct { BeginTime int `json:"begin_time" form:"begin_time" binding:"required"` EndTime int `json:"end_time" form:"end_time" binding:"required"` } // @Summary Get database list of slow log // @Tags slowquery // @Description Get database list of slow log // @Produce json // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param x-deploy-type header string true "dedicated/shared" // @Param begin_time query int false "begin time in unix seconds" // @Param end_time query int true "end time in unix seconds" // @Router /slow_query/databases [get] func (s *Service) handleDatabases(c *gin.Context) { f := recordHttpMetric(c, "/databases") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req databasesRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if s.shouldProxySlowQueryToO11y(header.DeployType) { path := "/internal/diagnosis/api/v1/slow_query/databases" // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/databases", header.ClusterID) // log.Info("update byoc slowquery databases path", zap.String("path", path)) // } err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, path) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } env := header.Env if env == "" { env = s.env } beginHourTs := req.BeginTime - 72*3600 - req.BeginTime%(72*3600) databases, err := s.StatsQuery.QueryDatabases(c.Request.Context(), slowstats.ClusterInfo{ Provider: header.Provider, Region: header.Region, Env: env, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, }, int64(beginHourTs), 72) if err != nil { c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "failed to query databases", FullText: err.Error(), }) return } c.JSON(http.StatusOK, databases) } // @Summary Get available fields of slow log // @Tags slowquery // @Description Get available fields of slow log // @Produce json // @Success 200 json []string // @Router /slow_query/available_fields [get] func (s *Service) handleAvailableFields(c *gin.Context) { f := recordHttpMetric(c, "/available_fields") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if s.shouldProxySlowQueryToO11y(header.DeployType) { // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/available_fields", header.ClusterID) // log.Info("update byoc slowquery available fields path", zap.String("path", path)) // } err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, slowQueryAvailableFieldsPath) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } fields, err := s.SlowQuery.AvailableFields(c.Request.Context()) if err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to query available fields", FullText: err.Error(), }) return } c.JSON(http.StatusOK, fields) } type statsRequest struct { BeginTime int64 `json:"begin_time" form:"begin_time" binding:"required"` Hours int `json:"hours" form:"hours" binding:"required"` StatementTypes []string `json:"statement_types" form:"statement_types"` StatementTypesNotIn bool `json:"statement_types_not_in" form:"statement_types_not_in"` Database string `json:"database" form:"database"` // deprecated Databases []string `json:"databases" form:"databases"` Internal bool `json:"internal" form:"internal"` OrderBy string `json:"order_by" form:"order_by"` Limit int `json:"limit" form:"limit"` } // @Summary Get slow query statistics // @Tags slowquery // @Description Get slow query statistics // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param begin_time query int true "hour begin time in unix seconds" // @Param hours query int true "n hours (3/6/12/24/)" // @Param statement_types query []string false "statement types, default empty array for all" // @Param statement_types_not_in query bool false "statement types not in, default false" // @Param database query string false "database name, default empty string for all" // @Param databases query []string false "database names, default empty array for all" // @Param order_by query string false "order by field, default empty string for count" // @Param limit query int false "top limit, default 10" // @Produce json // @Success 200 json []slowstats.SlowStats // @Router /slow_query/stats [get] func (s *Service) handleStats(c *gin.Context) { f := recordHttpMetric(c, "/stats") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req statsRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if s.shouldProxySlowQueryToO11y(header.DeployType) { path := "/internal/diagnosis/api/v1/slow_query/stats" // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/stats", header.ClusterID) // log.Info("update byoc slowquery stats path", zap.String("path", path)) // } err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, path) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } env := header.Env if env == "" { env = s.env } if req.Database != "" { found := false for _, db := range req.Databases { if db == req.Database { found = true } } if !found { req.Databases = append(req.Databases, req.Database) } } rs, err := s.StatsQuery.QueryStats(c.Request.Context(), slowstats.ClusterInfo{ Provider: header.Provider, Region: header.Region, Env: env, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, }, req.BeginTime, req.Hours, req.StatementTypes, req.StatementTypesNotIn, req.Databases, req.Internal, req.OrderBy, req.Limit) if err != nil { c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "failed to query stats", FullText: err.Error(), }) return } c.JSON(http.StatusOK, rs) } type statsMetricRequest struct { BeginTime int64 `json:"begin_time" form:"begin_time" binding:"required"` Hours int `json:"hours" form:"hours" binding:"required"` MetricName string `json:"metric_name" form:"metric_name"` } // @Summary Get slow query statistics metric // @Tags slowquery // @Description Get slow query statistics metric // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param begin_time query int true "hour begin time in unix seconds" // @Param hours query int true "n hours (3/6/12/24/)" // @Param metric_name query string false "e.g. count_per_minute (default)" // @Produce json // @Success 200 json []slowstats.SlowStats // @Router /slow_query/stats/metric [get] func (s *Service) handleStatsMetric(c *gin.Context) { f := recordHttpMetric(c, "/stats/metric") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req statsMetricRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if s.shouldProxySlowQueryToO11y(header.DeployType) { path := "/internal/diagnosis/api/v1/slow_query/stats/metric" // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/stats/metric", header.ClusterID) // log.Info("update byoc slowquery metric path", zap.String("path", path)) // } err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, path) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } env := header.Env if env == "" { env = s.env } r, err := s.StatsQuery.QueryMetric(c.Request.Context(), slowstats.ClusterInfo{ Provider: header.Provider, Region: header.Region, Env: env, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, }, req.BeginTime, req.Hours, req.MetricName) if err != nil { c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "failed to query stats", FullText: err.Error(), }) return } c.JSON(http.StatusOK, r) } type statsDatabasesRequest struct { BeginTime int64 `json:"begin_time" form:"begin_time" binding:"required"` Hours int `json:"hours" form:"hours" binding:"required"` } // @Summary Get slow query statistics database list // @Tags slowquery // @Description Get slow query statistics database list // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param begin_time query int true "hour begin time in unix seconds" // @Param hours query int true "n hours (3/6/12/24/)" // @Produce json // @Success 200 json []string // @Router /slow_query/stats/databases [get] func (s *Service) handleStatsDatabases(c *gin.Context) { f := recordHttpMetric(c, "/stats/databases") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req statsDatabasesRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if s.shouldProxySlowQueryToO11y(header.DeployType) { path := "/internal/diagnosis/api/v1/slow_query/stats/databases" // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/stats/databases", header.ClusterID) // log.Info("update byoc slowquery databases path", zap.String("path", path)) // } err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, path) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } env := header.Env if env == "" { env = s.env } rs, err := s.StatsQuery.QueryDatabases(c.Request.Context(), slowstats.ClusterInfo{ Provider: header.Provider, Region: header.Region, Env: env, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, }, req.BeginTime, req.Hours) if err != nil { c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "failed to query stats", FullText: err.Error(), }) return } c.JSON(http.StatusOK, rs) } type statsTimeWindowsRequest struct { BeginTime int64 `json:"begin_time" form:"begin_time" binding:"required"` EndTime int64 `json:"end_time" form:"end_time" binding:"required"` Hours int64 `json:"hours" form:"hours" binding:"required"` } func (s *Service) handleStatsTimeWindows(c *gin.Context) { f := recordHttpMetric(c, "/stats/time_windows") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req statsTimeWindowsRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout) defer cancel() if s.shouldProxySlowQueryToO11y(header.DeployType) { path := "/internal/diagnosis/api/v1/slow_query/stats/time_windows" // if strings.EqualFold(header.DeployType, o11y.ByocType) { // path = fmt.Sprintf("/internal/%s/diagnosis/api/v1/slow_query/stats/time_windows", header.ClusterID) // log.Info("update byoc slowquery time_windows path", zap.String("path", path)) // } // @todo upate byoc url err := utils.ProxyO11yByQueryCluster(ctx, c, s.o11yCli, header.DeployType, s.env, header.Provider, header.Region, header.TenantID, header.ProjectID, header.ClusterID, path) if err != nil { log.Error(ctx, "failed to proxy to o11y", zap.Error(err)) c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to proxy to o11y", FullText: err.Error(), }) return } return } windows := slowstats.GetTimeWindows(req.BeginTime, req.EndTime, req.Hours) var newWindows slowstats.TimeWindows now := time.Now().Unix() for _, w := range windows { if w.BeginTime < (now - 14*24*3600) { // 14d ago continue } newWindows = append(newWindows, w) } c.JSON(http.StatusOK, newWindows) } type analyzeRequest struct { BeginTime int `json:"begin_time" form:"begin_time"` EndTime int `json:"end_time" form:"end_time" binding:"required"` } // @Summary Analyze slow logs and write result to victoria metrics // @Tags slowquery // @Description Analyze slow logs and write result to victoria metrics // @Produce json // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param x-deploy-type header string true "dedicated/shared" // @Param begin_time query int false "begin time in unix seconds" // @Param end_time query int true "end time in unix seconds" // @Router /slow_query/analyze [get] func (s *Service) handleAnalyze(c *gin.Context) { f := recordHttpMetric(c, "/analyze") ctx := c.Request.Context() defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req analyzeRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } params := &AnalyzeParams{ Provider: header.Provider, Region: header.Region, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, SharedTier: isServerlessSlowQueryDeployType(header.DeployType), BeginTime: req.BeginTime, EndTime: req.EndTime, } if params.SharedTier { target, err := s.resolveStorageTarget(c.Request.Context(), header, o11y.StorageCategorySlowQuery) if err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to resolve slow query storage target", FullText: err.Error(), }) return } params.StorageBucket = target.BucketName params.StoragePrefix = target.ObjectPath } err := s.SlowQuery.Analyze(c.Request.Context(), params) log.Info(ctx, "analyzing slow logs end", zap.String("provider", params.Provider), zap.String("region", params.Region), zap.String("tenant", params.TenantID), zap.String("project", params.ProjectID), zap.String("cluster", params.ClusterID), zap.Bool("sharedTier", params.SharedTier), zap.Int("beginTime", params.BeginTime), zap.Int("endTime", params.EndTime), zap.Error(err)) if err != nil { if e, ok := err.(awserr.Error); ok && e.Code() == s3.ErrCodeNoSuchBucket { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("aws region '%s' not deployed", header.Region), FullText: err.Error(), }) } else if err == gcpstorage.ErrBucketNotExist { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: fmt.Sprintf("gcp region '%s' not deployed", header.Region), FullText: err.Error(), }) } else { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to analyze slow log", FullText: err.Error(), }) } return } c.Status(http.StatusOK) } type downloadRequest struct { BeginTime int `json:"begin_time" form:"begin_time"` EndTime int `json:"end_time" form:"end_time" binding:"required"` } // @Summary Download slow query file // @Tags slowquery // @Description Download slow query file // @Produce json // @Param x-provider header string true "provider aws/gcp" // @Param x-region header string true "region" // @Param x-org-id header string true "organization id from dbaas" // @Param x-project-id header string true "project id from dbaas" // @Param x-cluster-id header string true "cluster id from dbaas" // @Param x-deploy-type header string true "dedicated/shared" // @Param begin_time query int false "begin time in unix seconds" // @Param end_time query int true "end time in unix seconds" // @Success 200 {file} file // @Router /slow_query/files [get] func (s *Service) handleDownload(c *gin.Context) { f := recordHttpMetric(c, "/files") defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var req downloadRequest if err := c.ShouldBindQuery(&req); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid params", FullText: err.Error(), }) return } sharedTier := strings.ToLower(header.DeployType) == "shared" if sharedTier { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "not supported for serverless tier cluster", FullText: "not supported for serverless tier cluster", }) return } validTime := isValidTimePeriod(req.BeginTime, req.EndTime) if !validTime { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "time range should be a specific hour or day", FullText: "time range should be a specific hour or day", }) return } slowLogs, err := s.SlowQuery.Download(c.Request.Context(), &DownloadParams{ Provider: header.Provider, Region: header.Region, TenantID: header.TenantID, ProjectID: header.ProjectID, ClusterID: header.ClusterID, SharedTier: sharedTier, BeginTime: req.BeginTime, EndTime: req.EndTime, }) if err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "failed to download slow log", FullText: err.Error(), }) return } c.DataFromReader(http.StatusOK, slowLogs.ContentLength, slowLogs.ContentType, slowLogs.Reader, slowLogs.Headers) _ = slowLogs.Close() } func (s *Service) handleVmQuery(path string) gin.HandlerFunc { return func(c *gin.Context) { f := recordHttpMetric(c, path) defer f() var header clusterHeader if err := c.ShouldBindHeader(&header); err != nil { c.JSON(http.StatusBadRequest, errorResponse{ Error: true, Message: "invalid header", FullText: err.Error(), }) return } var urlPath string switch path { case "/vm_query": urlPath = fmt.Sprintf("/internal/clinic-vmselect/select/%v/prometheus/api/v1/query", header.ClusterID) case "/vm_query_range": urlPath = fmt.Sprintf("/internal/clinic-vmselect/select/%v/prometheus/api/v1/query_range", header.ClusterID) default: panic("invalid vm query path: " + path) } u, err := url.Parse(vmAddr(s.env)) if err != nil { c.JSON(http.StatusInternalServerError, errorResponse{ Error: true, Message: "failed to parse vm url", FullText: err.Error(), }) return } proxy := httputil.NewSingleHostReverseProxy(u) proxy.Director = func(r *http.Request) { r.URL.Scheme = u.Scheme r.URL.Host = u.Host r.URL.Path = urlPath r.Host = u.Host } proxy.ServeHTTP(c.Writer, c.Request) } } func parsePlan(ctx context.Context, plan, binaryPlan string) string { var r string var err error if binaryPlan != "" && strings.HasPrefix(binaryPlan, "tidb_decode_binary_plan('") { binaryPlan = strings.TrimPrefix(binaryPlan, "tidb_decode_binary_plan('") binaryPlan = strings.TrimSuffix(binaryPlan, "')") r, err = plancodec.DecodeBinaryPlan(binaryPlan) } else if plan != "" && strings.HasPrefix(plan, "tidb_decode_plan('") { plan = strings.TrimPrefix(plan, "tidb_decode_plan('") plan = strings.TrimSuffix(plan, "')") r, err = plancodec.DecodePlan(plan) } else { r = plan } if err != nil { log.Error(ctx, "decode plan in slow log failed", zap.String("plan", plan), zap.String("binaryPlan", binaryPlan), zap.Error(err)) } return r } func vmAddr(env string) string { switch strings.ToLower(env) { case "staging": return "http://www.ds.us-west-2.aws.observability-staging.pingcap.cloud" case "prod": return "http://www.ds.us-west-2.aws.observability.tidbcloud.com" case "dev": fallthrough default: return "http://www.ds.us-west-2.aws.observability-dev.pingcap.cloud" } } func slowLogToRespItem(ctx context.Context, sl *SlowLog) *slowLogRespItem { resp := &slowLogRespItem{ Digest: sl.Digest, Query: sl.Query, Instance: sl.Instance, DB: sl.DB, ConnectionID: fmt.Sprintf("%d", sl.ConnID), Success: boolToInt(sl.Succ), Timestamp: timeToFloat(sl.Time), QueryTime: sl.QueryTime, ParseTime: sl.ParseTime, CompileTime: sl.CompileTime, RewriteTime: sl.RewriteTime, PreprocSubqueriesTime: sl.PreprocSubqueriesTime, OptimizeTime: sl.OptimizeTime, WaitTSTime: sl.WaitTs, CopTime: sl.CopTime, LockKeysTime: sl.LockKeysTime, WriteRespTime: sl.WriteSQLResponseTotal, ExecRetryTime: sl.ExecRetryTime, MemoryMax: int(sl.MemMax), DiskMax: int(sl.DiskMax), TxnStartTS: fmt.Sprintf("%d", sl.TxnStartTs), PrevStmt: sl.PrevStmt, Plan: parsePlan(ctx, sl.Plan, sl.BinaryPlan), IsInternal: boolToInt(sl.IsInternal), IndexNames: sl.IndexNames, Stats: sl.Stats, BackoffTypes: sl.BackoffTypes, Prepared: boolToInt(sl.Prepared), PlanFromCache: boolToInt(sl.PlanFromCache), PlanFromBinding: boolToInt(sl.PlanFromBinding), User: sl.User, Host: sl.Host, ProcessTime: sl.ProcessTime, WaitTime: sl.WaitTime, BackoffTime: sl.BackoffTime, GetCommitTSTime: sl.GetCommitTsTime, LocalLatchWaitTime: sl.LocalLatchWaitTime, ResolveLockTime: sl.ResolveLockTime, PrewriteTime: sl.PrewriteTime, WaitPreWriteBinlogTime: sl.WaitPrewriteBinlogTime, CommitTime: sl.CommitTime, CommitBackoffTime: sl.CommitBackoffTime, CopProcAvg: sl.CopProcAvg, CopProcP90: sl.CopProcP90, CopProcMax: sl.CopProcMax, CopWaitAvg: sl.CopWaitAvg, CopWaitP90: sl.CopWaitP90, CopWaitMax: sl.CopWaitMax, WriteKeys: int(sl.WriteKeys), WriteSize: int(sl.WriteSize), PrewriteRegion: int(sl.PrewriteRegion), TxnRetry: int(sl.TxnRetry), RequestCount: uint(sl.RequestCount), ProcessKeys: uint(sl.ProcessKeys), TotalKeys: uint(sl.TotalKeys), CopProcAddr: sl.CopProcAddr, CopWaitAddr: sl.CopWaitAddr, RocksdbDeleteSkippedCount: uint(sl.RocksdbDeleteSkippedCount), RocksdbKeySkippedCount: uint(sl.RocksdbKeySkippedCount), RocksdbBlockCacheHitCount: uint(sl.RocksdbBlockCacheHitCount), RocksdbBlockReadCount: uint(sl.RocksdbBlockReadCount), RocksdbBlockReadByte: uint(sl.RocksdbBlockReadByte), RU: sl.RequestUnitRead + sl.RequestUnitWrite, QueuedTime: sl.TimeQueuedByRc, IARemoteReadSegmentCount: sl.IARemoteReadSegmentCount, IARemoteReadSegmentSize: sl.IARemoteReadSegmentSize, IARemoteReadSegmentWaitTime: sl.IARemoteReadSegmentWaitTime, ResourceGroup: sl.ResourceGroup, Warnings: sl.Warnings, RequestUnitV2: sl.RequestUnitV2, RequestUnitV2Detail: sl.RequestUnitV2Detail, RequestUnitV2Metrics: sl.RequestUnitV2Metrics, } // resp.BinaryPlan, _ = binaryplan.GenerateBinaryPlanJSON(resp.BinaryPlan) return resp } func slowLogToRespItemByFields(ctx context.Context, sl *SlowLog, fields []string) *slowLogRespItem { item := &slowLogRespItem{ Digest: sl.Digest, ConnectionID: fmt.Sprintf("%d", sl.ConnID), Timestamp: timeToFloat(sl.Time), } for _, field := range fields { switch field { // case "digest": // item.Digest = sl.Digest case "query": item.Query = sl.Query case "instance": item.Instance = sl.Instance case "db": item.DB = sl.DB // case "connection_id": // item.ConnectionID = fmt.Sprintf("%d", sl.ConnID) case "success": item.Success = boolToInt(sl.Succ) // case "timestamp": // item.Timestamp = timeToFloat(sl.Time) case "query_time": item.QueryTime = sl.QueryTime case "parse_time": item.ParseTime = sl.ParseTime case "compile_time": item.CompileTime = sl.CompileTime case "rewrite_time": item.RewriteTime = sl.RewriteTime case "preproc_subqueries_time": item.PreprocSubqueriesTime = sl.PreprocSubqueriesTime case "optimize_time": item.OptimizeTime = sl.OptimizeTime case "wait_ts": item.WaitTSTime = sl.WaitTs case "cop_time": item.CopTime = sl.CopTime case "lock_keys_time": item.LockKeysTime = sl.LockKeysTime case "write_sql_response_total": item.WriteRespTime = sl.WriteSQLResponseTotal case "exec_retry_time": item.ExecRetryTime = sl.ExecRetryTime case "memory_max": item.MemoryMax = int(sl.MemMax) case "disk_max": item.DiskMax = int(sl.DiskMax) case "txn_start_ts": item.TxnStartTS = fmt.Sprintf("%d", sl.TxnStartTs) case "prev_stmt": item.PrevStmt = sl.PrevStmt case "plan": item.Plan = parsePlan(ctx, sl.Plan, sl.BinaryPlan) case "is_internal": item.IsInternal = boolToInt(sl.IsInternal) case "index_names": item.IndexNames = sl.IndexNames case "stats": item.Stats = sl.Stats case "backoff_types": item.BackoffTypes = sl.BackoffTypes case "prepared": item.Prepared = boolToInt(sl.Prepared) case "plan_from_cache": item.PlanFromCache = boolToInt(sl.PlanFromCache) case "plan_from_biniding": item.PlanFromBinding = boolToInt(sl.PlanFromBinding) case "user": item.User = sl.User case "host": item.Host = sl.Host case "process_time": item.ProcessTime = sl.ProcessTime case "wait_time": item.WaitTime = sl.WaitTime case "backoff_time": item.BackoffTime = sl.BackoffTime case "get_commit_ts_time": item.GetCommitTSTime = sl.GetCommitTsTime case "local_latch_wait_time": item.LocalLatchWaitTime = sl.LocalLatchWaitTime case "resolve_lock_time": item.ResolveLockTime = sl.ResolveLockTime case "prewrite_time": item.PrewriteTime = sl.PrewriteTime case "wait_prewrite_binlog_time": item.WaitPreWriteBinlogTime = sl.WaitPrewriteBinlogTime case "commit_time": item.CommitTime = sl.CommitTime case "commit_backoff_time": item.CommitBackoffTime = sl.CommitBackoffTime case "cop_proc_avg": item.CopProcAvg = sl.CopProcAvg case "cop_proc_p90": item.CopProcP90 = sl.CopProcP90 case "cop_proc_max": item.CopProcMax = sl.CopProcMax case "cop_wait_avg": item.CopWaitAvg = sl.CopWaitAvg case "cop_wait_p90": item.CopWaitP90 = sl.CopWaitP90 case "cop_wait_max": item.CopWaitMax = sl.CopWaitMax case "write_keys": item.WriteKeys = int(sl.WriteKeys) case "write_size": item.WriteSize = int(sl.WriteSize) case "prewrite_region": item.PrewriteRegion = int(sl.PrewriteRegion) case "txn_retry": item.TxnRetry = int(sl.TxnRetry) case "request_count": item.RequestCount = uint(sl.RequestCount) case "process_keys": item.ProcessKeys = uint(sl.ProcessKeys) case "total_keys": item.TotalKeys = uint(sl.TotalKeys) case "cop_proc_addr": item.CopProcAddr = sl.CopProcAddr case "cop_wait_addr": item.CopWaitAddr = sl.CopWaitAddr case "rocksdb_delete_skipped_count": item.RocksdbDeleteSkippedCount = uint(sl.RocksdbDeleteSkippedCount) case "rocksdb_key_skipped_count": item.RocksdbKeySkippedCount = uint(sl.RocksdbKeySkippedCount) case "rocksdb_block_cache_hit_count": item.RocksdbBlockCacheHitCount = uint(sl.RocksdbBlockCacheHitCount) case "rocksdb_block_read_count": item.RocksdbBlockReadCount = uint(sl.RocksdbBlockReadCount) case "rocksdb_block_read_byte": item.RocksdbBlockReadByte = uint(sl.RocksdbBlockReadByte) case "ru": item.RU = sl.RequestUnitRead + sl.RequestUnitWrite case "time_queued_by_rc": item.QueuedTime = sl.TimeQueuedByRc case "ia_remote_read_segment_count": item.IARemoteReadSegmentCount = sl.IARemoteReadSegmentCount case "ia_remote_read_segment_size": item.IARemoteReadSegmentSize = sl.IARemoteReadSegmentSize case "ia_remote_read_segment_wait_time": item.IARemoteReadSegmentWaitTime = sl.IARemoteReadSegmentWaitTime case "resource_group": item.ResourceGroup = sl.ResourceGroup case "warnings": item.Warnings = sl.Warnings case "ru_v2": item.RequestUnitV2 = sl.RequestUnitV2 case "ru_v2_detail": item.RequestUnitV2Detail = sl.RequestUnitV2Detail case "ru_v2_metrics": item.RequestUnitV2Metrics = sl.RequestUnitV2Metrics } } return item } func boolToInt(v bool) int { if v { return 1 } return 0 } func isValidTimePeriod(beginTime, endTime int) bool { t1 := time.Unix(int64(beginTime), 0) t2 := time.Unix(int64(endTime), 0) if t1.Minute() == 0 && t1.Second() == 0 { if t2.Minute() == 0 && t2.Second() == 0 { if endTime-beginTime == 60*60 { return true } else if endTime-beginTime == 24*60*60 { if t1.Hour() == 0 && t2.Hour() == 0 { return true } } } } return false }