// 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" "go.uber.org/zap" "gorm.io/gorm" "github.com/pingcap-inc/clinic/pkg/config" "github.com/pingcap-inc/clinic/pkg/db" "github.com/pingcap-inc/clinic/pkg/log" "github.com/pingcap-inc/clinic/pkg/ngm/storage" ) const ( objectsTimeRangeGap = 4 * 60 // seconds itemsInObjectTimeMismatchGap = 5 // seconds detailLookupForwardRetryStep = objectsTimeRangeGap detailLookupMaxForwardRetries = 8 // HARDCODE: See pingcap/tidb-dashboard/pkg/apiserver/slowquery/queries.go#QuerySlowLogList defaultListLimit = 100 ) var ( ErrUnsupportedOrderBy = errors.New("unsupported order by") errSlowQueryDetailNotFound = errors.New("log for detail not found") ) // 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) // Analyze analyzes the slow logs of the specified range and writes result to victoria metrics. Analyze(ctx context.Context, params *AnalyzeParams) error // AvailableFields returns which fields are currently available in the slow log. AvailableFields(ctx context.Context) ([]string, error) Download(ctx context.Context, params *DownloadParams) (*DownloadData, 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"` TimeQueuedByRc float64 `json:"Time_queued_by_rc,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"` 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"` // RUv2 metrics (optional) RequestUnitV2 float64 `json:"Request_unit_v2,omitempty"` RequestUnitV2Detail string `json:"Request_unit_v2_detail,omitempty"` RequestUnitV2Metrics *RequestUnitV2Metrics `json:"Request_unit_v2_metrics,omitempty"` } // RequestUnitV2Metrics represents the parsed RUv2_metrics field. type RequestUnitV2Metrics struct { TotalRU float64 `json:"total_ru"` TiDBRU float64 `json:"tidb_ru"` TiKVRU float64 `json:"tikv_ru"` TiFlashRU float64 `json:"tiflash_ru"` // Optional detail fields (only present when non-zero) ResultChunkCells float64 `json:"result_chunk_cells,omitempty"` ExecutorL1 float64 `json:"executor_l1,omitempty"` ExecutorL2 float64 `json:"executor_l2,omitempty"` ExecutorL3 float64 `json:"executor_l3,omitempty"` ExecutorL5InsertRows float64 `json:"executor_l5_insert_rows,omitempty"` PlanCnt float64 `json:"plan_cnt,omitempty"` PlanDeriveStatsPaths float64 `json:"plan_derive_stats_paths,omitempty"` SessionParserTotal float64 `json:"session_parser_total,omitempty"` TxnCnt float64 `json:"txn_cnt,omitempty"` ResourceManagerReadCnt float64 `json:"resource_manager_read_cnt,omitempty"` ResourceManagerWriteCnt float64 `json:"resource_manager_write_cnt,omitempty"` TiKVKVEngineCacheMiss float64 `json:"tikv_kv_engine_cache_miss,omitempty"` TiKVCoprocessorExecutorIterations float64 `json:"tikv_coprocessor_executor_iterations,omitempty"` TiKVCoprocessorResponseBytes float64 `json:"tikv_coprocessor_response_bytes,omitempty"` TiKVRaftstoreStoreWriteTriggerWbBytes float64 `json:"tikv_raftstore_store_write_trigger_wb_bytes,omitempty"` TiKVStorageProcessedKeysBatchGet float64 `json:"tikv_storage_processed_keys_batch_get,omitempty"` TiKVStorageProcessedKeysGet float64 `json:"tikv_storage_processed_keys_get,omitempty"` TiKVCoprocessorExecutorWorkTotal map[string]float64 `json:"tikv_coprocessor_executor_work_total,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 { Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool NextGen bool StorageBucket string StoragePrefix string BeginTime int EndTime int Limit int OrderBy string Text string Digest string DB []string Plans []string IsDesc bool ShowInternal bool } // 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 { Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool NextGen bool StorageBucket string StoragePrefix string ConnID uint64 Digest string Timestamp float64 // e.g. 1704294935.1507144 1704294986.406164 ShowInternal bool } type AnalyzeParams struct { Provider string Region string TenantID string ProjectID string ClusterID string SharedTier bool StorageBucket string StoragePrefix string 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 isInternalSlowLog(sl *SlowLog, sharedTier bool) bool { if sl == nil { return false } // return sl.IsInternal || len(sl.User) == 0 || (sharedTier && sl.User == "cloud_admin") return sl.IsInternal } func (data *DownloadData) Close() error { return data.Reader.Close() } // NewSlowQuery creates a new SlowQuery implementation. func NewSlowQuery(env string, cfg *config.DBConfig, stores storage.StorageFactory) SlowQuery { return &slowQuery{ env: env, stores: stores, tenantDB: cfg, } } type slowQuery struct { env string stores storage.StorageFactory tenantDB *config.DBConfig } func (s *slowQuery) List(ctx context.Context, params *ListParams) (rs []*SlowLog, err error) { beginWithGap := params.BeginTime - objectsTimeRangeGap endWithGap := params.EndTime + objectsTimeRangeGap s3Bucket := slowQueryBucket(s.env, params.Provider, params.Region, params.SharedTier, params.NextGen, params.StorageBucket) s3Dir := slowQueryKeyPrefix(s.env, params.Provider, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.SharedTier, params.NextGen, params.StoragePrefix) s3ListPrefix := s3Dir + rangeTsPrefix(beginWithGap, endWithGap) log.Info(ctx, "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() { type listItem struct { Timestamp float64 `json:"timestamp"` ConnID uint64 `json:"conn_id"` Digest string `json:"digest"` } var ( firstTimestamp float64 lastTimestamp float64 samples []listItem ) if len(rs) > 0 { firstTimestamp = timeToFloat(rs[0].Time) lastTimestamp = timeToFloat(rs[len(rs)-1].Time) if params.Digest != "" { maxSample := 20 if len(rs) < maxSample { maxSample = len(rs) } samples = make([]listItem, 0, maxSample) for i := 0; i < maxSample; i++ { sl := rs[i] if sl == nil { continue } samples = append(samples, listItem{ Timestamp: timeToFloat(sl.Time), ConnID: sl.ConnID, Digest: sl.Digest, }) } } } fields := []zap.Field{ 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), zap.Int("result_count", len(rs)), zap.Float64("first_timestamp", firstTimestamp), zap.Float64("last_timestamp", lastTimestamp), } if len(rs) == 0 || err != nil { fields = append(fields, zap.String("bucket", s3Bucket), zap.String("dir", s3Dir), zap.String("prefix", s3ListPrefix), zap.Int("begin_time_with_gap", beginWithGap), zap.Int("end_time_with_gap", endWithGap), ) } if len(samples) > 0 { fields = append(fields, zap.Any("result_sample", samples)) } if err != nil { fields = append(fields, zap.Error(err)) } log.Info(ctx, "slow query list end", fields...) }() store, err := s.stores.Storage(ctx, params.Provider, params.Region) if err != nil { return nil, err } keys, err := s.findKeysWithinRange(ctx, store, params) if err != nil { return nil, err } if len(keys) == 0 { return nil, nil } sampleKeys := keys if len(sampleKeys) > 6 { sampleKeys = append(append([]string{}, keys[:3]...), keys[len(keys)-3:]...) } log.Info(ctx, "slow query got keys", zap.Int("len", len(keys)), zap.Strings("sample", sampleKeys)) // 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 { rs, err = s.findSlowLogsFromObjectsDesc(ctx, store, params, keys) return rs, err } rs, err = s.findSlowLogsFromObjectsAsc(ctx, store, params, keys) return rs, err 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. rs, err = s.findSlowLogsFromObjectsOrderBy(ctx, store, params, keys) return rs, err default: return nil, ErrUnsupportedOrderBy } } func (s *slowQuery) Detail(ctx context.Context, params *DetailParams) (*SlowLog, error) { store, err := s.stores.Storage(ctx, params.Provider, params.Region) if err != nil { return nil, err } seenKeys := make(map[string]struct{}) seenKeyList := make([]string, 0) searchTimes := detailLookupSearchTimes(params.Timestamp) var lastErr error = errSlowQueryDetailNotFound for attempt, searchTime := range searchTimes { listParams := &ListParams{ Provider: params.Provider, Region: params.Region, TenantID: params.TenantID, ProjectID: params.ProjectID, ClusterID: params.ClusterID, SharedTier: params.SharedTier, NextGen: params.NextGen, StorageBucket: params.StorageBucket, StoragePrefix: params.StoragePrefix, BeginTime: searchTime, EndTime: searchTime, } keys, err := s.findKeysWithinRange(ctx, store, listParams) if err != nil { return nil, err } newKeys := make([]string, 0, len(keys)) for _, key := range keys { if _, ok := seenKeys[key]; ok { continue } seenKeys[key] = struct{}{} seenKeyList = append(seenKeyList, key) newKeys = append(newKeys, key) } log.Info(ctx, "slow query detail got keys", 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.Bool("nextGen", params.NextGen), zap.Int("attempt", attempt), zap.Int("attempts", len(searchTimes)), zap.Int("search_time", searchTime), zap.Float64("timestamp", params.Timestamp), zap.Uint64("conn_id", params.ConnID), zap.String("digest", params.Digest), zap.Int("candidate_objects", len(keys)), zap.Int("new_candidate_objects", len(newKeys)), zap.Int("scanned_unique_objects", len(seenKeys)), zap.Strings("scanned_unique_objects_sample", sampleSlowQueryKeys(seenKeyList)), zap.Strings("candidate_objects_sample", sampleSlowQueryKeys(keys)), zap.Strings("new_candidate_objects_sample", sampleSlowQueryKeys(newKeys))) isLastAttempt := attempt == len(searchTimes)-1 sl, err := s.findSlowLogFromObjects(ctx, store, params, newKeys, detailLookupScanOptions{ LogNotFound: isLastAttempt, Attempt: attempt, Attempts: len(searchTimes), SearchTime: searchTime, MaxForwardRetries: detailLookupMaxForwardRetries, RetryStepSeconds: detailLookupForwardRetryStep, CandidateObjects: len(keys), NewCandidateObjects: len(newKeys), ScannedUniqueObjects: len(seenKeys), ScannedUniqueObjectsSample: sampleSlowQueryKeys(seenKeyList), AllowEarlyBreak: attempt == 0, }) if err == nil { return sl, nil } if !errors.Is(err, errSlowQueryDetailNotFound) { return nil, err } lastErr = err } return nil, lastErr } func detailLookupSearchTimes(timestamp float64) []int { baseTime := int(timestamp) searchTimes := make([]int, 0, detailLookupMaxForwardRetries+1) for retry := 0; retry <= detailLookupMaxForwardRetries; retry++ { searchTimes = append(searchTimes, baseTime+retry*detailLookupForwardRetryStep) } return searchTimes } func sampleSlowQueryKeys(keys []string) []string { if len(keys) <= 6 { return keys } sample := make([]string, 0, 6) sample = append(sample, keys[:3]...) sample = append(sample, keys[len(keys)-3:]...) return sample } func (s *slowQuery) Analyze(ctx context.Context, params *AnalyzeParams) error { log.Info(ctx, "analyzing slow logs begin", 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)) store, err := s.stores.Storage(ctx, params.Provider, params.Region) if err != nil { return err } listParams := &ListParams{ Provider: params.Provider, Region: params.Region, TenantID: params.TenantID, ProjectID: params.ProjectID, ClusterID: params.ClusterID, SharedTier: params.SharedTier, StorageBucket: params.StorageBucket, StoragePrefix: params.StoragePrefix, BeginTime: params.BeginTime, EndTime: params.EndTime, } keys, err := s.findKeysWithinRange(ctx, store, listParams) if err != nil { return err } if len(keys) == 0 { return nil } log.Info(ctx, "analyzing slow logs found keys", zap.Int("keys", len(keys)), 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)) var sls []*SlowLog for n := len(keys) - 1; n >= 0; n-- { err := s.findSlowLogsFromObject(ctx, store, listParams, keys[n], func(sl *SlowLog) bool { sls = append(sls, sl) return true }) if err != nil { return err } } log.Info(ctx, "analyzing slow logs found items", zap.Int("items", len(sls)), 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)) if len(sls) > 0 { return analyze(s.env, params.ClusterID, sls) } return nil } func (s *slowQuery) AvailableFields(_ context.Context) ([]string, error) { // HACK: We hardcoded here to avoid querying the database for // column names. So if new columns are added in the future, // we need to modify this additionally. return []string{ "digest", "query", "instance", "db", "connection_id", "success", "timestamp", "query_time", "parse_time", "compile_time", "rewrite_time", "preproc_subqueries_time", "optimize_time", "wait_ts", "cop_time", "lock_keys_time", "write_sql_response_total", "exec_retry_time", "memory_max", "disk_max", "txn_start_ts", "prev_stmt", "plan", "is_internal", "index_names", "stats", "backoff_types", "prepared", "plan_from_cache", "plan_from_binding", "user", "host", "process_time", "wait_time", "backoff_time", "get_commit_ts_time", "local_latch_wait_time", "resolve_lock_time", "prewrite_time", "wait_prewrite_binlog_time", "commit_time", "commit_backoff_time", "cop_proc_avg", "cop_proc_p90", "cop_proc_max", "cop_wait_avg", "cop_wait_p90", "cop_wait_max", "write_keys", "write_size", "prewrite_region", "txn_retry", "request_count", "process_keys", "total_keys", "cop_proc_addr", "cop_wait_addr", "rocksdb_delete_skipped_count", "rocksdb_key_skipped_count", "rocksdb_block_cache_hit_count", "rocksdb_block_read_count", "rocksdb_block_read_byte", "ru", "ru_v2", "ru_v2_detail", "ru_v2_metrics", "time_queued_by_rc", "ia_remote_read_segment_count", "ia_remote_read_segment_size", "ia_remote_read_segment_wait_time", "resource_group", }, nil } func (s *slowQuery) Download(ctx context.Context, params *DownloadParams) (*DownloadData, error) { isInternal, err := s.isInternalTenant(params.TenantID) if err != nil { return nil, err } if !isInternal { return nil, fmt.Errorf("only the slowlog of internal tenants can be downloaded") } // whether need to download the slowquery file of a total day totalDay := false t := time.Unix(int64(params.BeginTime), 0) date := t.Format("2006010215") if params.EndTime-params.BeginTime == 24*60*60 { totalDay = true date = t.Format("2006010200") } store, err := s.stores.Storage(ctx, params.Provider, params.Region) if err != nil { return nil, err } bucket := bucketName(s.env, params.Provider, params.Region, params.SharedTier, false) var key string prefix := fmt.Sprintf("diagnosis/data/%s/slowlog-stats", params.ClusterID) if totalDay { key = fmt.Sprintf("%s/24h/%s.db", prefix, date) } else { key = fmt.Sprintf("%s/1h/%s.db", prefix, date) } log.Info(ctx, "slow query download", 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.String("bucket", bucket), zap.String("key", key)) defer func() { log.Info(ctx, "slow query download 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.String("bucket", bucket), zap.String("key", key)) }() output, err := store.Object(ctx, bucket, key) if err != nil { return nil, err } k := strings.Split(key, "/") fname := fmt.Sprintf("%s_%s", params.ClusterID, k[len(k)-1]) extraHeaders := map[string]string{ "Content-Disposition": fmt.Sprintf(`attachment; filename="%s"`, fname), "Content-Length": fmt.Sprintf("%d", *output.ContentLength), } if output.ETag != nil { extraHeaders["ETag"] = *output.ETag } if output.ContentRange != nil { extraHeaders["Content-Range"] = *output.ContentRange } if output.AcceptRanges != nil { extraHeaders["Accept-Range"] = *output.AcceptRanges } data := &DownloadData{ ContentType: *output.ContentType, ContentLength: *output.ContentLength, Reader: output.Reader, Headers: extraHeaders, } return data, 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, store storage.Storage, 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 := slowQueryKeyPrefix(s.env, params.Provider, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.SharedTier, params.NextGen, params.StoragePrefix) bucket := slowQueryBucket(s.env, params.Provider, params.Region, params.SharedTier, params.NextGen, params.StorageBucket) prefix := dir + rangeTsPrefix(begin, end) metas, err := store.Objects(ctx, bucket, prefix) if err != nil { return nil, err } sort.Sort(&sortableKeys{ dir: dir, metas: metas, }) var rs []string 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) } if params.Digest != "" || params.BeginTime == params.EndTime { sampleKeys := rs if len(sampleKeys) > 6 { sampleKeys = append(append([]string{}, rs[:3]...), rs[len(rs)-3:]...) } log.Debug(ctx, "slow query find keys within range", 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.Bool("nextGen", params.NextGen), zap.Int("begin_time", params.BeginTime), zap.Int("end_time", params.EndTime), zap.Int("begin_time_with_gap", begin), zap.Int("end_time_with_gap", end), zap.String("bucket", bucket), zap.String("dir", dir), zap.String("prefix", prefix), zap.Int("metas", len(metas)), zap.Int("keys", len(rs)), zap.Strings("keys_sample", sampleKeys)) } 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 } type resultKeySample struct { Key string `json:"key"` Timestamp float64 `json:"timestamp"` ConnID uint64 `json:"conn_id"` Digest string `json:"digest"` } samples := make([]resultKeySample, 0, 20) var rs []*SlowLog for n := len(keys) - 1; n >= 0 && limit > 0; n-- { if limit <= 0 { break } key := keys[n] err := s.findSlowLogsFromObject(ctx, store, params, key, func(sl *SlowLog) bool { if isInternalSlowLog(sl, params.SharedTier) { // Internal slow queries are hidden unless explicitly requested. if !params.ShowInternal { return limit > 0 } } rs = append(rs, sl) if params.Digest != "" && len(samples) < cap(samples) { samples = append(samples, resultKeySample{ Key: key, Timestamp: timeToFloat(sl.Time), ConnID: sl.ConnID, Digest: sl.Digest, }) } limit-- return limit > 0 }) if err != nil { return nil, err } } if len(samples) > 0 { log.Debug(ctx, "slow query list result key sample", 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.String("digest", params.Digest), zap.Any("sample", samples)) } 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 } type resultKeySample struct { Key string `json:"key"` Timestamp float64 `json:"timestamp"` ConnID uint64 `json:"conn_id"` Digest string `json:"digest"` } samples := make([]resultKeySample, 0, 20) 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 isInternalSlowLog(logs[n], params.SharedTier) { // Internal slow queries are hidden unless explicitly requested. if !params.ShowInternal { continue } } rs = append(rs, logs[n]) if params.Digest != "" && len(samples) < cap(samples) { samples = append(samples, resultKeySample{ Key: key, Timestamp: timeToFloat(logs[n].Time), ConnID: logs[n].ConnID, Digest: logs[n].Digest, }) } limit-- } } if len(samples) > 0 { log.Debug(ctx, "slow query list result key sample", 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.String("digest", params.Digest), zap.Any("sample", samples)) } 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 isInternalSlowLog(sl, params.SharedTier) { // Internal slow queries are hidden unless explicitly requested. 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 := slowQueryBucket(s.env, params.Provider, params.Region, params.SharedTier, params.NextGen, params.StorageBucket) 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(ctx, "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(ctx, "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(ctx, "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 != "" { if !strings.Contains(sl.Query, strings.ToLower(params.Text)) && !strings.Contains(sl.Query, strings.ToUpper(params.Text)) { 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). type detailLookupScanOptions struct { LogNotFound bool AllowEarlyBreak bool Attempt int Attempts int SearchTime int MaxForwardRetries int RetryStepSeconds int CandidateObjects int NewCandidateObjects int ScannedUniqueObjects int ScannedUniqueObjectsSample []string } func (s *slowQuery) findSlowLogFromObjects( ctx context.Context, store storage.Storage, params *DetailParams, keys []string, opts detailLookupScanOptions, ) (*SlowLog, error) { type digestMatchSample struct { Time float64 `json:"time"` Delta float64 `json:"delta"` ConnID uint64 `json:"conn_id"` ConnMatch bool `json:"conn_match"` } type objectSummary struct { Key string `json:"key"` Entries int `json:"entries"` ParseErrors int `json:"parse_errors"` NilEntries int `json:"nil_entries"` EarlyBreak bool `json:"early_break"` DigestMatchEntries int `json:"digest_match_entries"` ConnMatchEntries int `json:"conn_match_entries"` DigestConnMatchEntry int `json:"digest_conn_match_entries"` FirstEntryTime float64 `json:"first_entry_time,omitempty"` LastEntryTime float64 `json:"last_entry_time,omitempty"` BestConnDigestDelta float64 `json:"best_conn_digest_delta,omitempty"` BestConnDigestTime float64 `json:"best_conn_digest_time,omitempty"` } bucket := slowQueryBucket(s.env, params.Provider, params.Region, params.SharedTier, params.NextGen, params.StorageBucket) if len(keys) == 0 && opts.LogNotFound { log.Warn(ctx, "slow query detail scan got no candidate objects", 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.Bool("nextGen", params.NextGen), zap.String("bucket", bucket), zap.Int("attempt", opts.Attempt), zap.Int("attempts", opts.Attempts), zap.Int("search_time", opts.SearchTime), zap.Bool("allow_early_break", opts.AllowEarlyBreak), zap.Int("max_forward_retries", opts.MaxForwardRetries), zap.Int("retry_step_seconds", opts.RetryStepSeconds), zap.Int("candidate_objects", opts.CandidateObjects), zap.Int("new_candidate_objects", opts.NewCandidateObjects), zap.Int("scanned_unique_objects", opts.ScannedUniqueObjects), zap.Strings("scanned_unique_objects_sample", opts.ScannedUniqueObjectsSample), zap.Float64("timestamp", params.Timestamp), zap.Uint64("conn_id", params.ConnID), zap.String("digest", params.Digest)) } sampleKeys := keys if len(sampleKeys) > 6 { sampleKeys = append(append([]string{}, keys[:3]...), keys[len(keys)-3:]...) } log.Debug(ctx, "slow query detail scan start", 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.Bool("nextGen", params.NextGen), zap.String("bucket", bucket), zap.Int("attempt", opts.Attempt), zap.Int("attempts", opts.Attempts), zap.Int("search_time", opts.SearchTime), zap.Bool("allow_early_break", opts.AllowEarlyBreak), zap.Int("candidate_objects", len(keys)), zap.Strings("candidate_objects_sample", sampleKeys), zap.Float64("timestamp", params.Timestamp), zap.Uint64("conn_id", params.ConnID), zap.String("digest", params.Digest)) targetTs := params.Timestamp totalEntries := 0 totalParseErrors := 0 totalNilEntries := 0 totalEarlyBreakObjects := 0 totalDigestMatch := 0 totalConnMatch := 0 totalDigestConnMatch := 0 minAbsDelta := math.MaxFloat64 closestKey := "" closestTime := 0.0 closestConnID := uint64(0) closestDigest := "" minAbsDeltaSameDigest := math.MaxFloat64 closestSameDigestKey := "" closestSameDigestTime := 0.0 closestSameDigestConnID := uint64(0) minAbsDeltaSameDigestConn := math.MaxFloat64 closestSameDigestConnKey := "" closestSameDigestConnTime := 0.0 const maxDigestMatchSamples = 200 digestMatchSamples := make([]digestMatchSample, 0, maxDigestMatchSamples) nearbySamples := make([]map[string]any, 0, 10) objectSummaries := make([]objectSummary, 0, len(keys)) for _, key := range keys { perEntries := 0 perParseErrors := 0 perNilEntries := 0 perEarlyBreak := false perDigestMatch := 0 perConnMatch := 0 perDigestConnMatch := 0 firstTm := 0.0 lastTm := 0.0 firstTmSet := false bestConnDigestDelta := math.MaxFloat64 bestConnDigestTime := 0.0 obj, err := 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(ctx, "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 { perParseErrors++ log.Warn(ctx, "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 { perNilEntries++ continue } perEntries++ curBlockL += len(lines) tm := timeToFloat(sl.Time) if !firstTmSet { firstTm = tm firstTmSet = true } lastTm = tm delta := math.Abs(tm - targetTs) if delta < minAbsDelta { minAbsDelta = delta closestKey = key closestTime = tm closestConnID = sl.ConnID closestDigest = sl.Digest } if sl.Digest == params.Digest { perDigestMatch++ if len(digestMatchSamples) < maxDigestMatchSamples { digestMatchSamples = append(digestMatchSamples, digestMatchSample{ Time: tm, Delta: delta, ConnID: sl.ConnID, ConnMatch: sl.ConnID == params.ConnID, }) } else { worstIndex := 0 worstDelta := digestMatchSamples[0].Delta for i := 1; i < len(digestMatchSamples); i++ { if digestMatchSamples[i].Delta > worstDelta { worstDelta = digestMatchSamples[i].Delta worstIndex = i } } if delta < worstDelta { digestMatchSamples[worstIndex] = digestMatchSample{ Time: tm, Delta: delta, ConnID: sl.ConnID, ConnMatch: sl.ConnID == params.ConnID, } } } if delta < minAbsDeltaSameDigest { minAbsDeltaSameDigest = delta closestSameDigestKey = key closestSameDigestTime = tm closestSameDigestConnID = sl.ConnID } } if sl.ConnID == params.ConnID { perConnMatch++ } if params.ConnID == sl.ConnID && params.Digest == sl.Digest && delta < bestConnDigestDelta { perDigestConnMatch++ if delta < minAbsDeltaSameDigestConn { minAbsDeltaSameDigestConn = delta closestSameDigestConnKey = key closestSameDigestConnTime = tm } bestConnDigestDelta = delta bestConnDigestTime = tm } if delta <= 0.1 && len(nearbySamples) < cap(nearbySamples) { nearbySamples = append(nearbySamples, map[string]any{ "time": tm, "conn_id": sl.ConnID, "digest": sl.Digest, }) } if opts.AllowEarlyBreak && params.Timestamp < (tm-itemsInObjectTimeMismatchGap) { perEarlyBreak = true break } if params.Timestamp == tm && params.ConnID == sl.ConnID && params.Digest == sl.Digest { if isInternalSlowLog(sl, params.SharedTier) && !params.ShowInternal { continue } _ = body.Close() return sl, nil } } _ = body.Close() totalEntries += perEntries totalParseErrors += perParseErrors totalNilEntries += perNilEntries if perEarlyBreak { totalEarlyBreakObjects++ } totalDigestMatch += perDigestMatch totalConnMatch += perConnMatch totalDigestConnMatch += perDigestConnMatch summary := objectSummary{ Key: key, Entries: perEntries, ParseErrors: perParseErrors, NilEntries: perNilEntries, EarlyBreak: perEarlyBreak, DigestMatchEntries: perDigestMatch, ConnMatchEntries: perConnMatch, DigestConnMatchEntry: perDigestConnMatch, } if firstTmSet { summary.FirstEntryTime = firstTm summary.LastEntryTime = lastTm } if bestConnDigestDelta != math.MaxFloat64 && bestConnDigestDelta > 0 { summary.BestConnDigestDelta = bestConnDigestDelta summary.BestConnDigestTime = bestConnDigestTime } if perDigestMatch > 0 || perConnMatch > 0 || perEarlyBreak || perParseErrors > 0 || perDigestConnMatch > 0 { objectSummaries = append(objectSummaries, summary) } if perEarlyBreak || perParseErrors > 0 { fields := []zap.Field{ 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.Float64("timestamp", targetTs), zap.Uint64("conn_id", params.ConnID), zap.String("digest", params.Digest), zap.Int("entries", perEntries), zap.Bool("allow_early_break", opts.AllowEarlyBreak), zap.Int("parse_errors", perParseErrors), zap.Int("nil_entries", perNilEntries), zap.Bool("early_break", perEarlyBreak), zap.Int("digest_match_entries", perDigestMatch), zap.Int("conn_match_entries", perConnMatch), zap.Int("digest_conn_match_entries", perDigestConnMatch), } if firstTmSet { fields = append(fields, zap.Float64("first_entry_time", firstTm), zap.Float64("last_entry_time", lastTm), zap.Float64("first_entry_time_delta", firstTm-targetTs), zap.Float64("last_entry_time_delta", lastTm-targetTs)) } if bestConnDigestDelta != math.MaxFloat64 && bestConnDigestDelta > 0 { fields = append(fields, zap.Float64("best_conn_digest_delta", bestConnDigestDelta), zap.Float64("best_conn_digest_time", bestConnDigestTime)) } log.Debug(ctx, "slow query detail scan object summary", fields...) } } if !opts.LogNotFound { return nil, errSlowQueryDetailNotFound } if len(digestMatchSamples) > 0 { sort.Slice(digestMatchSamples, func(i, j int) bool { return digestMatchSamples[i].Delta < digestMatchSamples[j].Delta }) } // Keep object summary details for "not found" only, and limit output size. const maxObjectSummaries = 20 sampledObjectSummaries := objectSummaries if len(sampledObjectSummaries) > maxObjectSummaries { sort.Slice(sampledObjectSummaries, func(i, j int) bool { ai, aj := sampledObjectSummaries[i], sampledObjectSummaries[j] if ai.DigestConnMatchEntry != aj.DigestConnMatchEntry { return ai.DigestConnMatchEntry > aj.DigestConnMatchEntry } if ai.DigestMatchEntries != aj.DigestMatchEntries { return ai.DigestMatchEntries > aj.DigestMatchEntries } if ai.ConnMatchEntries != aj.ConnMatchEntries { return ai.ConnMatchEntries > aj.ConnMatchEntries } if ai.ParseErrors != aj.ParseErrors { return ai.ParseErrors > aj.ParseErrors } if ai.EarlyBreak != aj.EarlyBreak { return ai.EarlyBreak } return ai.Entries > aj.Entries }) sampledObjectSummaries = sampledObjectSummaries[:maxObjectSummaries] } detailBeginTime := int(targetTs) detailEndTime := int(targetTs) detailBeginTimeWithGap := detailBeginTime - objectsTimeRangeGap detailEndTimeWithGap := detailEndTime + objectsTimeRangeGap log.Warn(ctx, "slow query detail not found after scanning candidate objects", 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.Bool("nextGen", params.NextGen), zap.String("bucket", bucket), zap.Int("attempt", opts.Attempt), zap.Int("attempts", opts.Attempts), zap.Int("search_time", opts.SearchTime), zap.Bool("allow_early_break", opts.AllowEarlyBreak), zap.Int("max_forward_retries", opts.MaxForwardRetries), zap.Int("retry_step_seconds", opts.RetryStepSeconds), zap.Int("candidate_objects", len(keys)), zap.Int("candidate_objects_in_attempt", opts.CandidateObjects), zap.Int("new_candidate_objects_in_attempt", opts.NewCandidateObjects), zap.Int("scanned_unique_objects", opts.ScannedUniqueObjects), zap.Strings("scanned_unique_objects_sample", opts.ScannedUniqueObjectsSample), zap.Strings("candidate_objects_sample", sampleKeys), zap.Int("detail_begin_time", detailBeginTime), zap.Int("detail_end_time", detailEndTime), zap.Int("detail_begin_time_with_gap", detailBeginTimeWithGap), zap.Int("detail_end_time_with_gap", detailEndTimeWithGap), zap.Float64("timestamp", targetTs), zap.Uint64("conn_id", params.ConnID), zap.String("digest", params.Digest), zap.Int("scanned_entries", totalEntries), zap.Int("parse_errors", totalParseErrors), zap.Int("nil_entries", totalNilEntries), zap.Int("early_break_objects", totalEarlyBreakObjects), zap.Int("digest_match_entries", totalDigestMatch), zap.Int("conn_match_entries", totalConnMatch), zap.Int("digest_conn_match_entries", totalDigestConnMatch), zap.Float64("closest_time_abs_delta", minAbsDelta), zap.String("closest_key", closestKey), zap.Float64("closest_time", closestTime), zap.Uint64("closest_conn_id", closestConnID), zap.String("closest_digest", closestDigest), zap.Float64("closest_same_digest_abs_delta", minAbsDeltaSameDigest), zap.String("closest_same_digest_key", closestSameDigestKey), zap.Float64("closest_same_digest_time", closestSameDigestTime), zap.Uint64("closest_same_digest_conn_id", closestSameDigestConnID), zap.Bool("closest_digest_match_samples_truncated", totalDigestMatch > len(digestMatchSamples)), zap.Any("closest_digest_match_samples", digestMatchSamples), zap.Float64("closest_same_digest_conn_abs_delta", minAbsDeltaSameDigestConn), zap.String("closest_same_digest_conn_key", closestSameDigestConnKey), zap.Float64("closest_same_digest_conn_time", closestSameDigestConnTime), zap.Bool("object_summaries_truncated", len(objectSummaries) > len(sampledObjectSummaries)), zap.Any("object_summaries", sampledObjectSummaries), zap.Any("nearby_samples", nearbySamples)) return nil, errSlowQueryDetailNotFound } 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 bucketName(env, provider, region string, sharedTier, nextGen bool) string { if nextGen { return fmt.Sprintf("o11y-%s-shared-%s", env, region) } if sharedTier { if (env == "dev" || env == "staging") && provider == "aws" && region == "us-east-1" { return fmt.Sprintf("o11y-prod-shared-%s-serverless", region) } if (env == "dev") && provider == "alicloud" { return fmt.Sprintf("o11y-dev-shared-%s", region) } return fmt.Sprintf("o11y-prod-shared-%s", region) } switch env { case "prod": return fmt.Sprintf("o11y-prod-shared-%s", region) default: if provider == "aws" && region == "us-west-2" { return fmt.Sprintf("o11y-prod-shared-%s-staging", region) } else { return fmt.Sprintf("o11y-test-shared-%s", region) } } } 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 { // todo: remove this special case when o11y api is fixed if tenant == "1372813089209292046" && project == "1372813089454610005" && cluster == "10661816924665497020" { project = "1372813089454738610" log.Info(context.Background(), "project changed for this shared tier", zap.String("project_id", "1372813089454610005"), zap.String("original_project_id", "1372813089454738610"), zap.String("cluster", cluster)) } 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", bucketName(env, provider, region, sharedTier, false), r) } return r } func slowQueryBucket(env, provider, region string, sharedTier, nextGen bool, override string) string { if override != "" { return override } return bucketName(env, provider, region, sharedTier, nextGen) } func slowQueryKeyPrefix(env, provider, region, tenant, project, cluster string, sharedTier, nextGen bool, override string) string { if override != "" { return ensureTrailingSlash(override) } return keyPrefix(env, provider, region, tenant, project, cluster, sharedTier, nextGen) } func ensureTrailingSlash(prefix string) string { if prefix == "" || strings.HasSuffix(prefix, "/") { return prefix } return prefix + "/" } 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) } func (s *slowQuery) isInternalTenant(id string) (bool, error) { cli, err := db.Connect(s.tenantDB) if err != nil { return false, err } tenant := &Tenant{} err = cli.Transaction(func(tx *gorm.DB) error { return tx.Where("id = ?", id).First(tenant).Error }) if err != nil { return false, fmt.Errorf("load tenant %s failed: %w", id, err) } kind := tenant.Kind if tenant.SalesForceKind != "" { kind = tenant.SalesForceKind } log.Info(context.Background(), "get tenant", zap.String("id", id), zap.String("kind", kind)) return kind == "internal", nil }