// Package slowquery provides implementation of query on slow log files. // Slow log files of the user TiDB cluster is written to S3 via tidbcloud/vector. package slowquery import ( "context" "errors" "fmt" "io" "math" "sort" "strconv" "strings" "sync" "time" "github.com/pingcap/log" "go.uber.org/zap" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/slowquery/slowstats" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage" ) const ( objectsTimeRangeGap = 4 * 60 // seconds itemsInObjectTimeMismatchGap = 5 // seconds // HARDCODE: See pingcap/tidb-dashboard/pkg/apiserver/slowquery/queries.go#QuerySlowLogList defaultListLimit = 100 ) var ErrUnsupportedOrderBy = errors.New("unsupported order by") // SlowQuery is an interface for querying slow logs. type SlowQuery interface { // List lists slow logs which match the conditions described in ListParams. List(ctx context.Context, params *ListParams) ([]*SlowLog, error) // Detail looks for a single slow log specified in DetailParams. Detail(ctx context.Context, params *DetailParams) (*SlowLog, error) // StatsMetric returns time-bucketed slow query metrics. // Returns timestamps and corresponding values for the specified metric. // Only implemented in DuckDB adapter. StatsMetric( ctx context.Context, cluster slowstats.ClusterInfo, beginHourTs int64, hours int, name string, ) ([][]int64, error) // Stats returns aggregated statistics about slow queries. // Only implemented in DuckDB adapter. Stats( ctx context.Context, cluster slowstats.ClusterInfo, beginHourTs int64, hours int, statementTypes []string, statementTypesNotIn bool, databases []string, internal bool, order string, limit int, ) ([]slowstats.SlowStats, error) // Databases returns a list of distinct database names from slow logs. // Only implemented in DuckDB adapter. Databases(ctx context.Context, cluster slowstats.ClusterInfo, beginHourTs int64, hours int, internal bool) ([]string, error) // AvailableFields returns which fields are currently available in the slow log. // For Delta Lake (DuckDB) backends, availability is determined from table schema. AvailableFields(ctx context.Context, params *AvailableFieldsParams) ([]string, error) } // SlowLog represents a structured slow log entry. Each field corresponds // to a row in the table below. // // (tidb v8.0.0) desc INFORMATION_SCHRMA.CLUSTER_SLOW_QUERY; // +-------------------------------+---------------------+------+------+---------+-------+ // | Field | Type | Null | Key | Default | Extra | // +-------------------------------+---------------------+------+------+---------+-------+ // | INSTANCE | varchar(64) | YES | | NULL | | // | Time | timestamp(6) | NO | PRI | NULL | | // | Txn_start_ts | bigint(20) unsigned | YES | | NULL | | // | User | varchar(64) | YES | | NULL | | // | Host | varchar(64) | YES | | NULL | | // | Conn_ID | bigint(20) unsigned | YES | | NULL | | // | Session_alias | varchar(64) | YES | | NULL | | // | Exec_retry_count | bigint(20) unsigned | YES | | NULL | | // | Exec_retry_time | double | YES | | NULL | | // | Query_time | double | YES | | NULL | | // | Parse_time | double | YES | | NULL | | // | Compile_time | double | YES | | NULL | | // | Rewrite_time | double | YES | | NULL | | // | Preproc_subqueries | bigint(20) unsigned | YES | | NULL | | // | Preproc_subqueries_time | double | YES | | NULL | | // | Optimize_time | double | YES | | NULL | | // | Wait_TS | double | YES | | NULL | | // | Prewrite_time | double | YES | | NULL | | // | Wait_prewrite_binlog_time | double | YES | | NULL | | // | Commit_time | double | YES | | NULL | | // | Get_commit_ts_time | double | YES | | NULL | | // | Commit_backoff_time | double | YES | | NULL | | // | Backoff_types | varchar(64) | YES | | NULL | | // | Resolve_lock_time | double | YES | | NULL | | // | Local_latch_wait_time | double | YES | | NULL | | // | Write_keys | bigint(22) | YES | | NULL | | // | Write_size | bigint(22) | YES | | NULL | | // | Prewrite_region | bigint(22) | YES | | NULL | | // | Txn_retry | bigint(22) | YES | | NULL | | // | Cop_time | double | YES | | NULL | | // | Process_time | double | YES | | NULL | | // | Wait_time | double | YES | | NULL | | // | Backoff_time | double | YES | | NULL | | // | LockKeys_time | double | YES | | NULL | | // | Request_count | bigint(20) unsigned | YES | | NULL | | // | Total_keys | bigint(20) unsigned | YES | | NULL | | // | Process_keys | bigint(20) unsigned | YES | | NULL | | // | Rocksdb_delete_skipped_count | bigint(20) unsigned | YES | | NULL | | // | Rocksdb_key_skipped_count | bigint(20) unsigned | YES | | NULL | | // | Rocksdb_block_cache_hit_count | bigint(20) unsigned | YES | | NULL | | // | Rocksdb_block_read_count | bigint(20) unsigned | YES | | NULL | | // | Rocksdb_block_read_byte | bigint(20) unsigned | YES | | NULL | | // | DB | varchar(64) | YES | | NULL | | // | Index_names | varchar(100) | YES | | NULL | | // | Is_internal | tinyint(1) | YES | | NULL | | // | Digest | varchar(64) | YES | | NULL | | // | Stats | varchar(512) | YES | | NULL | | // | Cop_proc_avg | double | YES | | NULL | | // | Cop_proc_p90 | double | YES | | NULL | | // | Cop_proc_max | double | YES | | NULL | | // | Cop_proc_addr | varchar(64) | YES | | NULL | | // | Cop_wait_avg | double | YES | | NULL | | // | Cop_wait_p90 | double | YES | | NULL | | // | Cop_wait_max | double | YES | | NULL | | // | Cop_wait_addr | varchar(64) | YES | | NULL | | // | Mem_max | bigint(20) | YES | | NULL | | // | Disk_max | bigint(20) | YES | | NULL | | // | KV_total | double | YES | | NULL | | // | PD_total | double | YES | | NULL | | // | Backoff_total | double | YES | | NULL | | // | Write_sql_response_total | double | YES | | NULL | | // | Result_rows | bigint(22) | YES | | NULL | | // | Warnings | longtext | YES | | NULL | | // | Backoff_Detail | varchar(4096) | YES | | NULL | | // | Prepared | tinyint(1) | YES | | NULL | | // | Succ | tinyint(1) | YES | | NULL | | // | IsExplicitTxn | tinyint(1) | YES | | NULL | | // | IsWriteCacheTable | tinyint(1) | YES | | NULL | | // | Plan_from_cache | tinyint(1) | YES | | NULL | | // | Plan_from_binding | tinyint(1) | YES | | NULL | | // | Has_more_results | tinyint(1) | YES | | NULL | | // | Resource_group | varchar(64) | YES | | NULL | | // | Request_unit_read | double | YES | | NULL | | // | Request_unit_write | double | YES | | NULL | | // | Time_queued_by_rc | double | YES | | NULL | | // | Plan | longtext | YES | | NULL | | // | Plan_digest | varchar(128) | YES | | NULL | | // | Binary_plan | longtext | YES | | NULL | | // | Prev_stmt | longtext | YES | | NULL | | // | Query | longtext | YES | | NULL | | // +-------------------------------+---------------------+------+------+---------+-------+ // 80 rows in set (0.09 sec) // // Serverless fields: // - Keyspace_name // - Keyspace_ID // - Serverless_tenant_ID // - Serverless_project_ID // - Serverless_cluster_ID // - Is_tiflash type SlowLog struct { Instance string `json:"INSTANCE,omitempty"` Time time.Time `json:"-"` // We'll deal with it separately TxnStartTs uint64 `json:"Txn_start_ts,omitempty"` User string `json:"User,omitempty"` Host string `json:"Host,omitempty"` ConnID uint64 `json:"Conn_ID,omitempty"` SessionAlias string `json:"Session_alias,omitempty"` ExecRetryCount uint64 `json:"Exec_retry_count,omitempty"` ExecRetryTime float64 `json:"Exec_retry_time,omitempty"` QueryTime float64 `json:"Query_time,omitempty"` ParseTime float64 `json:"Parse_time,omitempty"` CompileTime float64 `json:"Compile_time,omitempty"` RewriteTime float64 `json:"Rewrite_time,omitempty"` PreprocSubqueries uint64 `json:"Preproc_subqueries,omitempty"` PreprocSubqueriesTime float64 `json:"Preproc_subqueries_time,omitempty"` OptimizeTime float64 `json:"Optimize_time,omitempty"` WaitTs float64 `json:"Wait_TS,omitempty"` PrewriteTime float64 `json:"Prewrite_time,omitempty"` WaitPrewriteBinlogTime float64 `json:"Wait_prewrite_binlog_time,omitempty"` CommitTime float64 `json:"Commit_time,omitempty"` GetCommitTsTime float64 `json:"Get_commit_ts_time,omitempty"` CommitBackoffTime float64 `json:"Commit_backoff_time,omitempty"` BackoffTypes string `json:"Backoff_types,omitempty"` ResolveLockTime float64 `json:"Resolve_lock_time,omitempty"` LocalLatchWaitTime float64 `json:"Local_latch_wait_time,omitempty"` WriteKeys uint64 `json:"Write_keys,omitempty"` WriteSize uint64 `json:"Write_size,omitempty"` PrewriteRegion uint64 `json:"Prewrite_region,omitempty"` TxnRetry uint64 `json:"Txn_retry,omitempty"` CopTime float64 `json:"Cop_time,omitempty"` ProcessTime float64 `json:"Process_time,omitempty"` WaitTime float64 `json:"Wait_time,omitempty"` BackoffTime float64 `json:"Backoff_time,omitempty"` LockKeysTime float64 `json:"LockKeys_time,omitempty"` RequestCount uint64 `json:"Request_count,omitempty"` TotalKeys uint64 `json:"Total_keys,omitempty"` ProcessKeys uint64 `json:"Process_keys,omitempty"` RocksdbDeleteSkippedCount uint64 `json:"Rocksdb_delete_skipped_count,omitempty"` RocksdbKeySkippedCount uint64 `json:"Rocksdb_key_skipped_count,omitempty"` RocksdbBlockCacheHitCount uint64 `json:"Rocksdb_block_cache_hit_count,omitempty"` RocksdbBlockReadCount uint64 `json:"Rocksdb_block_read_count,omitempty"` RocksdbBlockReadByte uint64 `json:"Rocksdb_block_read_byte,omitempty"` DB string `json:"DB,omitempty"` IndexNames string `json:"Index_names,omitempty"` IsInternal bool `json:"Is_internal,omitempty"` Digest string `json:"Digest,omitempty"` Stats string `json:"Stats,omitempty"` CopProcAvg float64 `json:"Cop_proc_avg,omitempty"` CopProcP90 float64 `json:"Cop_proc_p90,omitempty"` CopProcMax float64 `json:"Cop_proc_max,omitempty"` CopProcAddr string `json:"Cop_proc_addr,omitempty"` CopWaitAvg float64 `json:"Cop_wait_avg,omitempty"` CopWaitP90 float64 `json:"Cop_wait_p90,omitempty"` CopWaitMax float64 `json:"Cop_wait_max,omitempty"` CopWaitAddr string `json:"Cop_wait_addr,omitempty"` MemMax uint64 `json:"Mem_max,omitempty"` DiskMax uint64 `json:"Disk_max,omitempty"` KVTotal float64 `json:"KV_total,omitempty"` PDTotal float64 `json:"PD_total,omitempty"` BackoffTotal float64 `json:"Backoff_total,omitempty"` WriteSQLResponseTotal float64 `json:"Write_sql_response_total,omitempty"` ResultRows uint64 `json:"Result_rows,omitempty"` Warnings string `json:"Warnings,omitempty"` BackoffDetail string `json:"Backoff_Detail,omitempty"` Prepared bool `json:"Prepared,omitempty"` Succ bool `json:"Succ,omitempty"` IsExplicitTxn bool `json:"IsExplicitTxn,omitempty"` IsWriteCacheTable bool `json:"IsWriteCacheTable,omitempty"` PlanFromCache bool `json:"Plan_from_cache,omitempty"` PlanFromBinding bool `json:"Plan_from_binding,omitempty"` HasMoreResults bool `json:"Has_more_results,omitempty"` ResourceGroup string `json:"Resource_group,omitempty"` RequestUnitRead float64 `json:"Request_unit_read,omitempty"` RequestUnitWrite float64 `json:"Request_unit_write,omitempty"` RequestUnitV2 float64 `json:"Request_unit_v2,omitempty"` TimeQueuedByRc float64 `json:"Time_queued_by_rc,omitempty"` TidbCPUTime float64 `json:"Tidb_cpu_time,omitempty"` TikvCPUTime float64 `json:"Tikv_cpu_time,omitempty"` IARemoteReadSegmentCount uint64 `json:"IA_remote_read_segment_count,omitempty"` IARemoteReadSegmentSize uint64 `json:"IA_remote_read_segment_size,omitempty"` IARemoteReadSegmentWaitTime float64 `json:"IA_remote_read_segment_wait_time,omitempty"` Plan string `json:"Plan,omitempty"` PlanDigest string `json:"Plan_digest,omitempty"` BinaryPlan string `json:"Binary_plan,omitempty"` ReadPoolTaskDetails string `json:"Read_pool_task_details,omitempty"` PrevStmt string `json:"Prev_stmt,omitempty"` Query string `json:"Query,omitempty"` // Serverless fields KeyspaceName string `json:"Keyspace_name,omitempty"` KeyspaceID string `json:"Keyspace_ID,omitempty"` ServerlessTenantID string `json:"Serverless_tenant_ID,omitempty"` ServerlessProjectID string `json:"Serverless_project_ID,omitempty"` ServerlessClusterID string `json:"Serverless_cluster_ID,omitempty"` IsTiflash bool `json:"Is_tiflash,omitempty"` RU float64 `json:"RU,omitempty"` RUv2 float64 `json:"RU_v2,omitempty"` } // ListParams is used to list a set of slow logs. // Its parameters are divided into two parts, one part is used to // specify a specific cluster, and the other part is used to specify // query conditions (such as LIMIT, ORDER BY, DESC, etc.) type ListParams struct { Env string Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool NextGen bool Byoc bool BeginTime int EndTime int Limit int OrderBy string Text string Digest string DB []string Plans []string IsDesc bool ShowInternal bool Fields string } // DetailParams is used to specify a specific slow log entry. // Its parameters are divided into two parts, one for specifying // a particular cluster and one for specifying that entry. type DetailParams struct { Env string Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool NextGen bool Byoc bool ConnID uint64 Digest string Timestamp float64 // e.g. 1704294935.1507144 1704294986.406164 } // AvailableFieldsParams identifies the cluster whose slowlog schema should be inspected. type AvailableFieldsParams struct { Env string Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool NextGen bool Byoc bool } type AnalyzeParams struct { Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool BeginTime int EndTime int } type DownloadParams struct { Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool BeginTime int EndTime int } type DownloadData struct { ContentType string ContentLength int64 Reader io.ReadCloser Headers map[string]string } func (data *DownloadData) Close() error { return data.Reader.Close() } // NewSlowQuery creates a new SlowQuery implementation. func NewSlowQuery(store storage.Storage) SlowQuery { return &slowQuery{ store: store, } } type slowQuery struct { store storage.Storage } // StatsMetric returns time-bucketed slow query metrics. // This implementation returns an error as it's only supported by the DuckDB adapter. func (s *slowQuery) StatsMetric( ctx context.Context, cluster slowstats.ClusterInfo, beginHourTs int64, hours int, name string, ) ([][]int64, error) { return nil, errors.New("StatsMetric is not supported - use DuckDB adapter") } // Stats returns aggregated statistics about slow queries. // This implementation returns an error as it's only supported by the DuckDB adapter. func (s *slowQuery) Stats( ctx context.Context, cluster slowstats.ClusterInfo, beginHourTs int64, hours int, statementTypes []string, statementTypesNotIn bool, databases []string, internal bool, order string, limit int, ) ([]slowstats.SlowStats, error) { return nil, errors.New("Stats is not supported - use DuckDB adapter") } // Databases returns a list of distinct database names from slow logs. // This implementation returns an error as it's only supported by the DuckDB adapter. func (s *slowQuery) Databases(ctx context.Context, cluster slowstats.ClusterInfo, beginHourTs int64, hours int, internal bool) ([]string, error) { return nil, errors.New("Databases is not supported - use DuckDB adapter") } func (s *slowQuery) List(ctx context.Context, params *ListParams) ([]*SlowLog, error) { log.Info("slow query list", 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.Int("limit", params.Limit), zap.String("orderBy", params.OrderBy), zap.String("text", params.Text), zap.String("digest", params.Digest), zap.String("db", fmt.Sprintf("%v", params.DB)), zap.String("plans", fmt.Sprintf("%v", params.Plans)), zap.Bool("desc", params.IsDesc)) defer func() { log.Info("slow query list 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.Int("limit", params.Limit), zap.String("orderBy", params.OrderBy), zap.String("text", params.Text), zap.String("digest", params.Digest), zap.String("db", fmt.Sprintf("%v", params.DB)), zap.String("plans", fmt.Sprintf("%v", params.Plans)), zap.Bool("desc", params.IsDesc)) }() keys, err := s.findKeysWithinRange(ctx, params) if err != nil { return nil, err } if len(keys) == 0 { return nil, nil } log.Info("slow query got keys", zap.Int("len", len(keys))) // Most queries are order by timestamp, so we can take advantage of slow log's time series // storage to avoid reading all records into memory for sorting (params.OrderBy). switch strings.ToLower(params.OrderBy) { case "", "timestamp": if params.IsDesc { return s.findSlowLogsFromObjectsDesc(ctx, s.store, params, keys) } return s.findSlowLogsFromObjectsAsc(ctx, s.store, params, keys) case "query_time", "memory_max", "total_keys", "process_keys": // We need to sort by other fields, so we need to read all records between BeginTime ~ EndTime // into memory. This is a completely different piece of logic. return s.findSlowLogsFromObjectsOrderBy(ctx, s.store, params, keys) default: return nil, ErrUnsupportedOrderBy } } func (s *slowQuery) Detail(ctx context.Context, params *DetailParams) (*SlowLog, error) { keys, err := s.findKeysWithinRange(ctx, &ListParams{ Env: params.Env, Provider: params.Provider, Region: params.Region, TenantID: params.TenantID, ProjectID: params.ProjectID, ClusterID: params.ClusterID, SharedTier: params.SharedTier, NextGen: params.NextGen, BeginTime: int(params.Timestamp), EndTime: int(params.Timestamp), }) if err != nil { return nil, err } return s.findSlowLogFromObjects(ctx, params, keys) } func (s *slowQuery) AvailableFields(_ context.Context, _ *AvailableFieldsParams) ([]string, error) { // Shared / raw-log path has no Delta Lake schema to inspect. return allSlowlogAvailableFieldNames(), nil } // findKeysWithinRange finds associated s3 objects by BeginTime and EndTime. // We can consider the returned keys to contain all slow log entries between BeginTime and EndTime. // // NOTE: The returned keys are in descending order by timestamp. func (s *slowQuery) findKeysWithinRange(ctx context.Context, params *ListParams) ([]string, error) { // Since multiple vector agents exist within a single cluster, we cannot just look for // objects within a strictly matching time range. // // BTW, multiple vector agents may have out-of-order problems when uploading s3 files. // // For example: // 1676430000-367551a1-b7d1-4eb8-a521-3e8e323cbd42.log.gz // 1676430002-178b7e68-1dc5-4124-82ed-f646f2941179.log.gz <--- here // 1676430001-f91b48d7-74c6-4fb4-9e76-4f2ec76a2d24.log.gz // 1676430060-a8f7800a-4ef4-4e69-9d41-ab7523795830.log.gz // 1676430060-3d910c3c-501b-457b-9e18-7e32d2286cca.log.gz // 1676430061-872dc6af-f53e-461f-9a77-a2d0059c6244.log.gz // // Although 1676430001 initiated the upload first, due to network problems, // 1676430002 reached s3 first. So we can't assume the keys are in order // and have to deal with this error. `objectsTimeRangeGap` can solve this // problem by the way. begin, end := params.BeginTime-objectsTimeRangeGap, params.EndTime+objectsTimeRangeGap dir := keyPrefix(params.Env, params.Provider, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.SharedTier, params.NextGen) bucket := storage.BucketName(params.Env, params.Region, params.SharedTier, params.NextGen) prefix := dir + rangeTsPrefix(begin, end) metas, err := s.store.Objects(ctx, bucket, prefix) if err != nil { return nil, err } sort.Sort(&sortableKeys{ dir: dir, metas: metas, }) var rs = make([]string, 0, len(metas)) for _, meta := range metas { ts := 0 _, err = fmt.Sscanf(strings.TrimPrefix(meta.Key, dir), "%d-", &ts) if err != nil || ts == 0 { continue } if ts > end { continue } if ts < begin { break } rs = append(rs, meta.Key) } return rs, nil } // findSlowLogsFromObjectsAsc query slow log entries based on ListParams // from the s3/gcs objects specified by keys. The results are sorted by // timestamp in ascending order. // // NOTE: We assume that keys are in descending order by timestamp. func (s *slowQuery) findSlowLogsFromObjectsAsc( ctx context.Context, store storage.Storage, params *ListParams, keys []string, ) ([]*SlowLog, error) { limit := params.Limit if limit == 0 { limit = defaultListLimit } var rs []*SlowLog for n := len(keys) - 1; n >= 0 && limit > 0; n-- { if limit <= 0 { break } err := s.findSlowLogsFromObject(ctx, store, params, keys[n], func(sl *SlowLog) bool { if len(sl.User) == 0 || (params.SharedTier && sl.User == "cloud_admin") { // Empty user means this is an internal query. We don't show // internal slow queries to the user. if !params.ShowInternal { return limit > 0 } } rs = append(rs, sl) limit-- return limit > 0 }) if err != nil { return nil, err } } return rs, nil } // findSlowLogsFromObjectsDesc query slow log entries based on ListParams // from the s3/gcs objects specified by keys. The results are sorted by // timestamp in descending order. // // NOTE: We assume that keys are in descending order by timestamp. func (s *slowQuery) findSlowLogsFromObjectsDesc( ctx context.Context, store storage.Storage, params *ListParams, keys []string, ) ([]*SlowLog, error) { limit := params.Limit if limit == 0 { limit = defaultListLimit } var rs []*SlowLog for _, key := range keys { if limit <= 0 { break } var logs []*SlowLog err := s.findSlowLogsFromObject(ctx, store, params, key, func(sl *SlowLog) bool { logs = append(logs, sl) return true }) if err != nil { return nil, err } for n := len(logs) - 1; n >= 0 && limit > 0; n-- { if len(logs[n].User) == 0 || (params.SharedTier && logs[n].User == "cloud_admin") { // Empty user means this is an internal query. We don't show // internal slow queries to the user. if !params.ShowInternal { continue } } rs = append(rs, logs[n]) limit-- } } return rs, nil } // findSlowLogsFromObjectsDesc query slow log entries based on ListParams // from the s3/gcs objects specified by keys. The results are sorted according // to ListParams.OrderBy and ListParams.IsDesc. // // NOTE: We assume that keys are in descending order by timestamp. func (s *slowQuery) findSlowLogsFromObjectsOrderBy( ctx context.Context, store storage.Storage, params *ListParams, keys []string, ) ([]*SlowLog, error) { limit := params.Limit if limit == 0 { limit = defaultListLimit } errCh := make(chan error) keyCh := make(chan string, len(keys)) logCh := make(chan *SlowLog) for _, key := range keys { keyCh <- key } var wg sync.WaitGroup for n := 0; n < 100; n++ { wg.Add(1) go func() { defer wg.Done() for { select { case key := <-keyCh: err := s.findSlowLogsFromObject(ctx, store, params, key, func(sl *SlowLog) bool { if len(sl.User) == 0 || (params.SharedTier && sl.User == "cloud_admin") { // Empty user means this is an internal query. We don't show // internal slow queries to the user. if !params.ShowInternal { return true } } logCh <- sl return true }) if err != nil { errCh <- err return } default: return } } }() } h := newSlowLogsHeap(params.OrderBy, params.IsDesc, limit) doneCh := make(chan struct{}) go func() { wg.Wait() doneCh <- struct{}{} }() for { select { case <-doneCh: return h.Slice(), nil case err := <-errCh: return nil, err case log := <-logCh: h.Push(log) } } } // findSlowLogsFromObject query slow log entries based on ListParams // from the s3/gcs object specified by key. The scan will end when f // returns false. func (s *slowQuery) findSlowLogsFromObject( ctx context.Context, store storage.Storage, params *ListParams, key string, f func(sl *SlowLog) bool, ) error { bucket := storage.BucketName(params.Env, params.Region, params.SharedTier, params.NextGen) obj, err := store.Object(ctx, bucket, key) if err != nil { return err } body := obj.Reader defer func() { _ = body.Close() }() curBlockL := 0 curBlockR := 0 reader := NewSlowLogReader(body) for { lines, err := reader.ReadNext() if err != nil { return err } if len(lines) == 0 { break } curBlockR += len(lines) // OPTIMIZE: Avoid full parsing here, parsing only the first line // can filter out many entries ahead of time. var sl *SlowLog func() { defer func() { if v := recover(); v != nil { log.Warn("recover from panic when parse slow log entry", 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.String("bucket", bucket), zap.String("key", key), zap.Int("linesBegin", curBlockL), zap.Int("linesEnd", curBlockR)) } }() sl, err = ParseSlowLog(lines) }() if err != nil { log.Warn("invalid slow log entry", 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.String("bucket", bucket), zap.String("key", key), zap.Int("linesBegin", curBlockL), zap.Int("linesEnd", curBlockR), zap.Error(err)) continue } if sl == nil { log.Warn("unexpected nil slow log entry") continue } curBlockL += len(lines) if sl.Time.Unix() < int64(params.BeginTime) { continue } if sl.Time.Unix() > int64(params.EndTime) { break } if len(params.DB) > 0 { matchDB := false for _, db := range params.DB { if sl.DB == db { matchDB = true break } } if !matchDB { continue } } if params.Text != "" { textLower := strings.ToLower(params.Text) if !strings.Contains(strings.ToLower(sl.Query), textLower) && !strings.Contains(strings.ToLower(sl.Digest), textLower) { continue } } if len(params.Digest) > 0 { if !strings.EqualFold(params.Digest, sl.Digest) { continue } } if len(params.Plans) > 0 { found := false for _, plan := range params.Plans { if plan == sl.PlanDigest { found = true break } } if !found { continue } } if !f(sl) { return nil } } return nil } // findSlowLogFromObjects finds the corresponding SlowLog entry from a // set of objects (keys). func (s *slowQuery) findSlowLogFromObjects( ctx context.Context, params *DetailParams, keys []string, ) (*SlowLog, error) { bucket := storage.BucketName(params.Env, params.Region, params.SharedTier, params.NextGen) for _, key := range keys { obj, err := s.store.Object(ctx, bucket, key) if err != nil { return nil, err } body := obj.Reader curBlockL := 0 curBlockR := 0 reader := NewSlowLogReader(body) for { lines, err := reader.ReadNext() if err != nil { _ = body.Close() return nil, err } if len(lines) == 0 { break } curBlockR += len(lines) // OPTIMIZE: Avoid full parsing here, parsing only the first line // can filter out many entries ahead of time. var sl *SlowLog func() { defer func() { if v := recover(); v != nil { log.Warn("recover from panic when parse slow log entry", 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.String("bucket", bucket), zap.String("key", key), zap.Int("linesBegin", curBlockL), zap.Int("linesEnd", curBlockR)) } }() sl, err = ParseSlowLog(lines) }() if err != nil { log.Warn("invalid slow log entry", 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.String("bucket", bucket), zap.String("key", key), zap.Int("linesBegin", curBlockL), zap.Int("linesEnd", curBlockR), zap.Error(err)) continue } if sl == nil { continue } curBlockL += len(lines) tm := timeToFloat(sl.Time) if params.Timestamp < (tm - itemsInObjectTimeMismatchGap) { break } if params.Timestamp == tm && params.ConnID == sl.ConnID && params.Digest == sl.Digest { _ = body.Close() return sl, nil } } _ = body.Close() } return nil, errors.New("log for detail not found") } type sortableKeys struct { dir string metas []storage.ObjectMeta } func (o sortableKeys) Len() int { return len(o.metas) } func (o sortableKeys) Less(i, j int) bool { var iTs, jTs int iKey := strings.TrimPrefix(o.metas[i].Key, o.dir) jKey := strings.TrimPrefix(o.metas[j].Key, o.dir) _, _ = fmt.Sscanf(iKey, "%d-", &iTs) _, _ = fmt.Sscanf(jKey, "%d-", &jTs) return iTs > jTs // desc } func (o sortableKeys) Swap(i, j int) { o.metas[i].Key, o.metas[j].Key = o.metas[j].Key, o.metas[i].Key } func keyPrefix(env, provider, region, tenant, project, cluster string, sharedTier, nextGen bool) string { var r string if nextGen { r = fmt.Sprintf("0/next-gen/%v/logs/slowlog/", cluster) } else { if sharedTier { r = fmt.Sprintf("0/%v-devtier/serverless/%v/%v/%v/logs/slowlog/", env, tenant, project, cluster) } else { r = fmt.Sprintf("0/%v/%v/%v/logs/slowlog/", tenant, project, cluster) } } switch provider { case "alicloud", "aliyun": r = fmt.Sprintf("%s/%s", storage.BucketName(env, region, sharedTier, nextGen), r) } return r } func rangeTsPrefix(beginTime, endTime int) string { if beginTime <= 0 || endTime <= 0 { return "" } for n := 2; n <= 9; n++ { base := beginTime - beginTime%int(math.Pow10(n)) if base <= beginTime && (base+int(math.Pow10(n)-1)) >= endTime { baseStr := strconv.Itoa(base) return baseStr[:len(baseStr)-n] } } return "" } func singleTsPrefix(ts int) string { if ts <= 0 { return "" } for n := 3; n <= 9; n++ { base := ts - ts%int(math.Pow10(n)) if base <= (ts-5*60) && (base+int(math.Pow10(n)-1)) >= (ts+5*60) { baseStr := strconv.Itoa(base) return baseStr[:len(baseStr)-n] } } return "" } func timeToFloat(t time.Time) float64 { // If time.Time is the empty value, UnixNano will return the farthest back // timestamp a float can represent, which is some large negative value. We // compromise and call it zero. if t.IsZero() { return 0 } return float64(t.UnixMicro()) / 1e6 } //lint:ignore U1000 Ignore unused function func timeFromFloat(ts float64) time.Time { secs := int64(ts) nsecs := int64((ts - float64(secs)) * 1e9) return time.Unix(secs, nsecs) }