Statement API and integration (#34)

pkg/apiserver/apiserver.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/pingcap-incubator/tidb-dashboard/pkg/apiserver/foo"
 	"github.com/pingcap-incubator/tidb-dashboard/pkg/apiserver/info"
+	"github.com/pingcap-incubator/tidb-dashboard/pkg/apiserver/statement"
 	"github.com/pingcap-incubator/tidb-dashboard/pkg/apiserver/user"
 	"github.com/pingcap-incubator/tidb-dashboard/pkg/apiserver/utils"
 	"github.com/pingcap-incubator/tidb-dashboard/pkg/config"
@@ -59,6 +60,7 @@ func Handler(apiPrefix string, config *config.Config, services *Services) http.H
 	foo.NewService(config).Register(endpoint, auth)
 	info.NewService(config, services.TiDBForwarder, services.Store).Register(endpoint, auth)
 	services.KeyVisual.Register(endpoint, auth)
+	statement.NewService(config, services.TiDBForwarder).Register(endpoint, auth)
 
 	return r
 }

pkg/apiserver/statement/models.go

@@ -0,0 +1,57 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package statement
+
+// TimeRange represents a range of time
+type TimeRange struct {
+	BeginTime string `json:"begin_time"`
+	EndTime   string `json:"end_time"`
+}
+
+// Overview represents the overview of a statement
+type Overview struct {
+	SchemaName         string `json:"schema_name"`
+	Digest             string `json:"digest"`
+	DigestText         string `json:"digest_text"`
+	AggSumLatency      int    `json:"sum_latency"`
+	AggAvgLatency      int    `json:"avg_latency"`
+	AggExecCount       int    `json:"exec_count"`
+	AggAvgAffectedRows int    `json:"avg_affected_rows"`
+	AggAvgMem          int    `json:"avg_mem"`
+}
+
+// Detail represents the detail of a statement
+type Detail struct {
+	SchemaName         string `json:"schema_name"`
+	Digest             string `json:"digest"`
+	DigestText         string `json:"digest_text"`
+	AggSumLatency      int    `json:"sum_latency"`
+	AggExecCount       int    `json:"exec_count"`
+	AggAvgAffectedRows int    `json:"avg_affected_rows"`
+	AggAvgTotalKeys    int    `json:"avg_total_keys"`
+
+	QuerySampleText string `json:"query_sample_text"`
+	LastSeen        string `json:"last_seen"`
+}
+
+// Node represents the statement in each node
+type Node struct {
+	Address         string `json:"address"`
+	SumLatency      int    `json:"sum_latency"`
+	ExecCount       int    `json:"exec_count"`
+	AvgLatency      int    `json:"avg_latency"`
+	MaxLatency      int    `json:"max_latency"`
+	AvgMem          int    `json:"avg_mem"`
+	SumBackoffTimes int    `json:"sum_backoff_times"`
+}

pkg/apiserver/statement/queries.go

@@ -0,0 +1,154 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package statement
+
+import (
+	"fmt"
+	"regexp"
+	"sort"
+	"strings"
+
+	"github.com/jinzhu/gorm"
+)
+
+func QuerySchemas(db *gorm.DB) ([]string, error) {
+	sql := `SHOW DATABASES`
+
+	var schemas []string
+	err := db.Raw(sql).Pluck("Database", &schemas).Error
+	if err != nil {
+		return nil, err
+	}
+
+	for i, v := range schemas {
+		schemas[i] = strings.ToLower(v)
+	}
+	sort.Strings(schemas)
+	return schemas, nil
+}
+
+func QueryTimeRanges(db *gorm.DB) (result []*TimeRange, err error) {
+	err = db.
+		Select(`
+			DISTINCT
+			summary_begin_time AS begin_time,
+			summary_end_time AS end_time
+		`).
+		Table("PERFORMANCE_SCHEMA.cluster_events_statements_summary_by_digest_history").
+		Order("summary_begin_time DESC").
+		Find(&result).Error
+	return result, err
+}
+
+// Sample params:
+// schemas: ["tpcc", "test"]
+// beginTime: "2020-02-13 10:30:00"
+// endTime: "2020-02-13 11:00:00"
+func QueryStatementsOverview(db *gorm.DB, schemas []string, beginTime, endTime string) (result []*Overview, err error) {
+	query := db.
+		Select(`
+			schema_name,
+			digest,
+			digest_text,
+			sum(sum_latency) AS agg_sum_latency,
+			sum(exec_count) AS agg_exec_count,
+			round(sum(exec_count*avg_affected_rows)/sum(exec_count)) AS agg_avg_affected_rows,
+			round(sum(exec_count*avg_latency)/sum(exec_count)) AS agg_avg_latency,
+			round(sum(exec_count*avg_mem)/sum(exec_count)) AS agg_avg_mem
+		`).
+		Table("PERFORMANCE_SCHEMA.cluster_events_statements_summary_by_digest_history").
+		Where("summary_begin_time = ? AND summary_end_time = ?", beginTime, endTime).
+		Group("schema_name, digest, digest_text").
+		Order("agg_sum_latency DESC")
+
+	if len(schemas) > 0 {
+		regex := make([]string, 0, len(schemas))
+		for _, schema := range schemas {
+			regex = append(regex, fmt.Sprintf("\\b%s\\.", regexp.QuoteMeta(schema)))
+		}
+		regexAll := strings.Join(regex, "|")
+		query = query.Where("table_names REGEXP ?", regexAll)
+	}
+
+	err = query.Find(&result).Error
+	return result, err
+}
+
+// Sample params:
+// schemas: "tpcc"
+// beginTime: "2020-02-13 10:30:00"
+// endTime: "2020-02-13 11:00:00"
+// digest: "bcaa7bdb37e24d03fb48f20cc32f4ff3f51c0864dc378829e519650df5c7b923"
+func QueryStatementDetail(db *gorm.DB, schema, beginTime, endTime, digest string) (*Detail, error) {
+	result := &Detail{}
+
+	query := db.
+		Select(`
+			schema_name,
+			digest,
+			digest_text,
+			sum(sum_latency) AS agg_sum_latency,
+			sum(exec_count) AS agg_exec_count,
+			round(sum(exec_count*avg_affected_rows)/sum(exec_count)) AS agg_avg_affected_rows,
+			round(sum(exec_count*avg_total_keys)/sum(exec_count)) AS agg_avg_total_keys
+		`).
+		Table("PERFORMANCE_SCHEMA.cluster_events_statements_summary_by_digest_history").
+		Where("schema_name = ?", schema).
+		Where("summary_begin_time = ? AND summary_end_time = ?", beginTime, endTime).
+		Where("digest = ?", digest).
+		Group("digest, digest_text, schema_name")
+
+	if err := query.Scan(&result).Error; err != nil {
+		return nil, err
+	}
+
+	query = db.
+		Select(`query_sample_text, last_seen`).
+		Table("PERFORMANCE_SCHEMA.cluster_events_statements_summary_by_digest_history").
+		Where("schema_name = ?", schema).
+		Where("summary_begin_time = ? AND summary_end_time = ?", beginTime, endTime).
+		Where("digest = ?", digest).
+		Order("last_seen DESC")
+
+	if err := query.First(&result).Error; err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}
+
+// Sample params:
+// schemas: "tpcc"
+// beginTime: "2020-02-13 10:30:00"
+// endTime: "2020-02-13 11:00:00"
+// digest: "bcaa7bdb37e24d03fb48f20cc32f4ff3f51c0864dc378829e519650df5c7b923"
+func QueryStatementNodes(db *gorm.DB, schema, beginTime, endTime, digest string) (result []*Node, err error) {
+	err = db.
+		Select(`
+			address,
+			sum_latency,
+			exec_count,
+			avg_latency,
+			max_latency,
+			avg_mem,
+			sum_backoff_times
+		`).
+		Table("PERFORMANCE_SCHEMA.cluster_events_statements_summary_by_digest_history").
+		Where("schema_name = ?", schema).
+		Where("summary_begin_time = ? AND summary_end_time = ?", beginTime, endTime).
+		Where("digest = ?", digest).
+		Order("sum_latency DESC").
+		Find(&result).Error
+	return result, err
+}

pkg/apiserver/statement/statement.go

@@ -0,0 +1,163 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package statement
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/jinzhu/gorm"
+
+	"github.com/pingcap-incubator/tidb-dashboard/pkg/apiserver/user"
+	"github.com/pingcap-incubator/tidb-dashboard/pkg/apiserver/utils"
+	"github.com/pingcap-incubator/tidb-dashboard/pkg/config"
+	"github.com/pingcap-incubator/tidb-dashboard/pkg/tidb"
+)
+
+type Service struct {
+	config        *config.Config
+	tidbForwarder *tidb.Forwarder
+}
+
+func NewService(config *config.Config, tidbForwarder *tidb.Forwarder) *Service {
+	return &Service{config: config, tidbForwarder: tidbForwarder}
+}
+
+func (s *Service) Register(r *gin.RouterGroup, auth *user.AuthService) {
+	endpoint := r.Group("/statements")
+	endpoint.Use(auth.MWAuthRequired())
+	endpoint.Use(utils.MWConnectTiDB(s.tidbForwarder))
+	endpoint.GET("/schemas", s.schemasHandler)
+	endpoint.GET("/time_ranges", s.timeRangesHandler)
+	endpoint.GET("/overviews", s.overviewsHandler)
+	endpoint.GET("/detail", s.detailHandler)
+	endpoint.GET("/nodes", s.nodesHandler)
+}
+
+// @Summary TiDB databases
+// @Description Get all databases of TiDB
+// @Produce json
+// @Success 200 {array} string
+// @Router /statements/schemas [get]
+// @Security JwtAuth
+// @Failure 401 {object} utils.APIError "Unauthorized failure"
+func (s *Service) schemasHandler(c *gin.Context) {
+	db := c.MustGet(utils.TiDBConnectionKey).(*gorm.DB)
+	schemas, err := QuerySchemas(db)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+	c.JSON(http.StatusOK, schemas)
+}
+
+// @Summary Statement time ranges
+// @Description Get all time ranges of the statements
+// @Produce json
+// @Success 200 {array} statement.TimeRange
+// @Router /statements/time_ranges [get]
+// @Security JwtAuth
+// @Failure 401 {object} utils.APIError "Unauthorized failure"
+func (s *Service) timeRangesHandler(c *gin.Context) {
+	db := c.MustGet(utils.TiDBConnectionKey).(*gorm.DB)
+	timeRanges, err := QueryTimeRanges(db)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+	c.JSON(http.StatusOK, timeRanges)
+}
+
+// @Summary Statements overview
+// @Description Get statements overview
+// @Produce json
+// @Param schemas query string false "Target schemas"
+// @Param begin_time query string true "Statement begin time"
+// @Param end_time query string true "Statement end time"
+// @Success 200 {array} statement.Overview
+// @Router /statements/overviews [get]
+// @Security JwtAuth
+// @Failure 401 {object} utils.APIError "Unauthorized failure"
+func (s *Service) overviewsHandler(c *gin.Context) {
+	var schemas []string
+	schemasQuery := c.Query("schemas")
+	if schemasQuery != "" {
+		schemas = strings.Split(schemasQuery, ",")
+	}
+	beginTime := c.Query("begin_time")
+	endTime := c.Query("end_time")
+	if beginTime == "" || endTime == "" {
+		_ = c.Error(fmt.Errorf("invalid begin_time or end_time"))
+		return
+	}
+	db := c.MustGet(utils.TiDBConnectionKey).(*gorm.DB)
+	overviews, err := QueryStatementsOverview(db, schemas, beginTime, endTime)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+	c.JSON(http.StatusOK, overviews)
+}
+
+// @Summary Statement detail
+// @Description Get statement detail
+// @Produce json
+// @Param schema query string true "Statement schema"
+// @Param begin_time query string true "Statement begin time"
+// @Param end_time query string true "Statement end time"
+// @Param digest query string true "Statement digest"
+// @Success 200 {object} statement.Detail
+// @Router /statements/detail [get]
+// @Security JwtAuth
+// @Failure 401 {object} utils.APIError "Unauthorized failure"
+func (s *Service) detailHandler(c *gin.Context) {
+	db := c.MustGet(utils.TiDBConnectionKey).(*gorm.DB)
+	schema := c.Query("schema")
+	beginTime := c.Query("begin_time")
+	endTime := c.Query("end_time")
+	digest := c.Query("digest")
+	detail, err := QueryStatementDetail(db, schema, beginTime, endTime, digest)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+	c.JSON(http.StatusOK, detail)
+}
+
+// @Summary Statement nodes
+// @Description Get statement in each node
+// @Produce json
+// @Param schema query string true "Statement schema"
+// @Param begin_time query string true "Statement begin time"
+// @Param end_time query string true "Statement end time"
+// @Param digest query string true "Statement digest"
+// @Success 200 {array} statement.Node
+// @Router /statements/nodes [get]
+// @Security JwtAuth
+// @Failure 401 {object} utils.APIError "Unauthorized failure"
+func (s *Service) nodesHandler(c *gin.Context) {
+	db := c.MustGet(utils.TiDBConnectionKey).(*gorm.DB)
+	schema := c.Query("schema")
+	beginTime := c.Query("begin_time")
+	endTime := c.Query("end_time")
+	digest := c.Query("digest")
+	nodes, err := QueryStatementNodes(db, schema, beginTime, endTime, digest)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+	c.JSON(http.StatusOK, nodes)
+}

ui/.github_release_version

@@ -1,3 +1,3 @@
 # This file contains a version number which will be used to release assets to
 # GitHub. To trigger a new asset release, simply increase this version number.
-20200218_1
+20200225_1