app

type Client ¶

type Client struct {
	// contains filtered or unexported fields
}

func NewClient(
	cpuTimeRecordStream tipb.TopSQLAgent_ReportCPUTimeRecordsClient,
	sqlMetaStream tipb.TopSQLAgent_ReportSQLMetaClient,
	planMetaStream tipb.TopSQLAgent_ReportPlanMetaClient,
) *Client

func (s *Client) Start()

type DiskWAL ¶

type DiskWAL struct {
}

func NewDiskWAL() *DiskWAL

func (wal *DiskWAL) ReadNext() interface{}

func (wal *DiskWAL) WriteMulti(data []interface{})

type MemStore ¶

type MemStore struct {
	// contains filtered or unexported fields
}

MemStore is for testing purpose

func NewMemStore() *MemStore

func (s *MemStore) InitSchema() error

func (s *MemStore) WriteCPUTimeRecord(record *tipb.CPUTimeRecord, instanceID string) error

func (s *MemStore) WritePlanMeta(meta *tipb.PlanMeta) error

func (s *MemStore) WriteSQLMeta(meta *tipb.SQLMeta) error

type MemWAL ¶

type MemWAL struct {
	// contains filtered or unexported fields
}

func NewMemWAL() *MemWAL

func (wal *MemWAL) Acknowledge(count uint32)

func (wal *MemWAL) ReadMulti(count uint32) interface{}

func (wal *MemWAL) WriteMulti(data interface{})

type Prefetcher ¶

type Prefetcher struct {
	// contains filtered or unexported fields
}

func NewPrefetcher(wal WAL, config PrefetcherConfig) *Prefetcher

func (p *Prefetcher) ReadOneCPUTimeRecord() *tipb.CPUTimeRecord

func (p *Prefetcher) ReadOnePlanMetaOrNil() *tipb.PlanMeta

func (p *Prefetcher) ReadOneSQLMetaOrNil() *tipb.SQLMeta

func (p *Prefetcher) WriteOneCPUTimeRecordOrDrop(record *tipb.CPUTimeRecord)

func (p *Prefetcher) WriteOnePlanMetaOrDrop(meta *tipb.PlanMeta)

func (p *Prefetcher) WriteOneSQLMetaOrDrop(meta *tipb.SQLMeta)

type PrefetcherConfig ¶

type PrefetcherConfig struct {
	CPUTimeRecordCount uint32
	SQLMetaCount       uint32
	PlanMetaCount      uint32
}

type Receiver ¶

type Receiver struct {
	// contains filtered or unexported fields
}

TODO: add back pressure to ensure memory not growing to da moon

func NewReceiver(wal WAL, prefetcher *Prefetcher) *Receiver

type Sender ¶

type Sender struct {
	// contains filtered or unexported fields
}

func NewSender(prefetcher *Prefetcher, store Store) *Sender

func (s *Sender) Start()

type Store ¶

type Store interface {
	WriteCPUTimeRecord(record *tipb.CPUTimeRecord, instanceID string) error
	WriteSQLMeta(*tipb.SQLMeta) error
	WritePlanMeta(*tipb.PlanMeta) error
	InitSchema() error
}

type TiDBStore ¶

type TiDBStore struct {
	// contains filtered or unexported fields
}

TiDBStore uses TiDB as the storage bakend

func NewTiDBStore(dsn string, clusterID uint64) *TiDBStore

func (s *TiDBStore) InitSchema() error

func (s *TiDBStore) WriteCPUTimeRecord(record *tipb.CPUTimeRecord, instanceID string) error

func (s *TiDBStore) WritePlanMeta(meta *tipb.PlanMeta) error

func (s *TiDBStore) WriteSQLMeta(meta *tipb.SQLMeta) error

type TopSQLAgentServer ¶

type TopSQLAgentServer struct {
	// contains filtered or unexported fields
}

TopSQLAgentServer is the main struct which controls the Top SQL metrics collecting and reporting.

In general, there are several parts in the agent which are working together:

• gRPC server receives raw data from tidb-server/tikv-server
• receiver writes gRPC data to WAL, which typically resides in the disk. There're also in-memory implementations which is used in testing.
• WAL-buffer-prefetcher prefetches WAL data to an in-memory buffer, which is a sliding-window-like in-memory view of the WAL data. The prefetch buffer has an upper limit, which can be seemed as the max window size of the view.
• sender reads data from prefetched buffer, and asynchronously sends it to the remote store.

When the data is receiving and sending quickly enough, the data will only be written to WAL once, and served directly to the sender from receiver in memory, which is IO efficient. When the agent process crashes, it will re-construct the prefetch buffer, and restarts sending data to the store as usual. After all, the prefetched buffer is the only source for data from which to get out of the agent process.

func NewAgentServer(wal WAL, store Store) *TopSQLAgentServer

func StartGrpcServer(addr string, store Store) *TopSQLAgentServer

func (as *TopSQLAgentServer) ReportCPUTimeRecords(stream tipb.TopSQLAgent_ReportCPUTimeRecordsServer) error

func (as *TopSQLAgentServer) ReportPlanMeta(stream tipb.TopSQLAgent_ReportPlanMetaServer) error

func (as *TopSQLAgentServer) ReportSQLMeta(stream tipb.TopSQLAgent_ReportSQLMetaServer) error

func (as *TopSQLAgentServer) Start()

type Type ¶

type Type uint8

const (
	// TODO: this needs to be rearranged to fit any collecting interval larger than 1s
	TypeCPUTimeRecord Type = iota
	TypeSQLMeta
	TypePlanMeta
)

type WAL ¶

type WAL interface {
	WriteMulti(interface{})
	ReadMulti(count uint32) interface{}
	Acknowledge(count uint32)
}