energylink/pkg/tsdb/mysql/mysql.go

package mysql

import (
	"context"
	"fmt"
	"strings"
	"sync"

	"github.com/go-xorm/core"
	"github.com/go-xorm/xorm"
	"github.com/grafana/grafana/pkg/log"
	"github.com/grafana/grafana/pkg/models"
	"github.com/grafana/grafana/pkg/tsdb"
)

type MysqlExecutor struct {
	*models.DataSource
	engine *xorm.Engine
	log    log.Logger
}

type engineCacheType struct {
	cache    map[int64]*xorm.Engine
	versions map[int64]int
	sync.Mutex
}

var engineCache = engineCacheType{
	cache:    make(map[int64]*xorm.Engine),
	versions: make(map[int64]int),
}

func NewMysqlExecutor(datasource *models.DataSource) (tsdb.Executor, error) {
	engine, err := getEngineFor(datasource)
	if err != nil {
		return nil, err
	}

	return &MysqlExecutor{
		log:    log.New("tsdb.mysql"),
		engine: engine,
	}, nil
}

func getEngineFor(ds *models.DataSource) (*xorm.Engine, error) {
	engineCache.Lock()
	defer engineCache.Unlock()

	if engine, present := engineCache.cache[ds.Id]; present {
		if version, _ := engineCache.versions[ds.Id]; version == ds.Version {
			return engine, nil
		}
	}

	cnnstr := fmt.Sprintf("%s:%s@%s(%s)/%s?charset=utf8mb4", ds.User, ds.Password, "tcp", ds.Url, ds.Database)
	engine, err := xorm.NewEngine("mysql", cnnstr)
	engine.SetMaxConns(10)
	engine.SetMaxIdleConns(10)
	if err != nil {
		return nil, err
	}

	engineCache.cache[ds.Id] = engine
	return engine, nil
}

func init() {
	tsdb.RegisterExecutor("graphite", NewMysqlExecutor)
}

func (e *MysqlExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice, context *tsdb.QueryContext) *tsdb.BatchResult {
	result := &tsdb.BatchResult{}

	session := engine.NewSession()
	defer session.Close()

	db := session.DB()
	result, err := getData(db, &req)

	if err != nil {
		return
	}
}

func getData(db *core.DB, req *sqlDataRequest) (interface{}, error) {
	queries := strings.Split(req.Query, ";")

	data := dataStruct{}
	data.Results = make([]resultsStruct, 1)
	data.Results[0].Series = make([]seriesStruct, 0)

	for i := range queries {
		if queries[i] == "" {
			continue
		}

		rows, err := db.Query(queries[i])
		if err != nil {
			return nil, err
		}
		defer rows.Close()

		name := fmt.Sprintf("table_%d", i+1)
		series, err := arrangeResult(rows, name)
		if err != nil {
			return nil, err
		}
		data.Results[0].Series = append(data.Results[0].Series, series.(seriesStruct))
	}

	return data, nil
}

func arrangeResult(rows *core.Rows, name string) (interface{}, error) {
	columnNames, err := rows.Columns()

	series := seriesStruct{}
	series.Columns = columnNames
	series.Name = name

	for rows.Next() {
		columnValues := make([]interface{}, len(columnNames))

		err = rows.ScanSlice(&columnValues)
		if err != nil {
			return nil, err
		}

		// bytes -> string
		for i := range columnValues {
			switch columnValues[i].(type) {
			case []byte:
				columnValues[i] = fmt.Sprintf("%s", columnValues[i])
			}
		}

		series.Values = append(series.Values, columnValues)
	}

	return series, err
}

type sqlDataRequest struct {
	Query string `json:"query"`
	Body  []byte `json:"-"`
}

type seriesStruct struct {
	Columns []string        `json:"columns"`
	Name    string          `json:"name"`
	Values  [][]interface{} `json:"values"`
}

type resultsStruct struct {
	Series []seriesStruct `json:"series"`
}

type dataStruct struct {
	Results []resultsStruct `json:"results"`
}