|
|
@@ -5,10 +5,9 @@ import (
|
|
|
"context"
|
|
|
"database/sql"
|
|
|
"fmt"
|
|
|
- "strconv"
|
|
|
"strings"
|
|
|
|
|
|
- _ "time"
|
|
|
+ "time"
|
|
|
|
|
|
_ "github.com/denisenkom/go-mssqldb"
|
|
|
"github.com/go-xorm/core"
|
|
|
@@ -69,6 +68,10 @@ func (e MssqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows,
|
|
|
return err
|
|
|
}
|
|
|
|
|
|
+ rowLimit := 1000000
|
|
|
+ rowCount := 0
|
|
|
+ timeIndex := -1
|
|
|
+
|
|
|
table := &tsdb.Table{
|
|
|
Columns: make([]tsdb.TableColumn, columnCount),
|
|
|
Rows: make([]tsdb.RowValues, 0),
|
|
|
@@ -76,6 +79,12 @@ func (e MssqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows,
|
|
|
|
|
|
for i, name := range columnNames {
|
|
|
table.Columns[i].Text = name
|
|
|
+
|
|
|
+ // check if there is a column named time
|
|
|
+ switch name {
|
|
|
+ case "time":
|
|
|
+ timeIndex = i
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
columnTypes, err := rows.ColumnTypes()
|
|
|
@@ -83,9 +92,6 @@ func (e MssqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows,
|
|
|
return err
|
|
|
}
|
|
|
|
|
|
- rowLimit := 1000000
|
|
|
- rowCount := 0
|
|
|
-
|
|
|
for ; rows.Next(); rowCount++ {
|
|
|
if rowCount > rowLimit {
|
|
|
return fmt.Errorf("MsSQL query row limit exceeded, limit %d", rowLimit)
|
|
|
@@ -96,6 +102,15 @@ func (e MssqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows,
|
|
|
return err
|
|
|
}
|
|
|
|
|
|
+ // convert column named time to unix timestamp to make
|
|
|
+ // native datetime mssql types work in annotation queries
|
|
|
+ if timeIndex != -1 {
|
|
|
+ switch value := values[timeIndex].(type) {
|
|
|
+ case time.Time:
|
|
|
+ values[timeIndex] = float64(value.Unix())
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
table.Rows = append(table.Rows, values)
|
|
|
}
|
|
|
|
|
|
@@ -123,42 +138,107 @@ func (e MssqlQueryEndpoint) getTypedRowData(types []*sql.ColumnType, rows *core.
|
|
|
func (e MssqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
|
|
|
pointsBySeries := make(map[string]*tsdb.TimeSeries)
|
|
|
seriesByQueryOrder := list.New()
|
|
|
+
|
|
|
columnNames, err := rows.Columns()
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
|
|
|
+ columnTypes, err := rows.ColumnTypes()
|
|
|
if err != nil {
|
|
|
return err
|
|
|
}
|
|
|
|
|
|
- rowData := NewStringStringScan(columnNames)
|
|
|
- rowLimit := 1000
|
|
|
+ rowLimit := 1000000
|
|
|
rowCount := 0
|
|
|
+ timeIndex := -1
|
|
|
+ metricIndex := -1
|
|
|
+
|
|
|
+ // check columns of resultset: a column named time is mandatory
|
|
|
+ // the first text column is treated as metric name unless a column named metric is present
|
|
|
+ for i, col := range columnNames {
|
|
|
+ switch col {
|
|
|
+ case "time":
|
|
|
+ timeIndex = i
|
|
|
+ case "metric":
|
|
|
+ metricIndex = i
|
|
|
+ default:
|
|
|
+ if metricIndex == -1 {
|
|
|
+ switch columnTypes[i].DatabaseTypeName() {
|
|
|
+ case "VARCHAR", "CHAR", "NVARCHAR", "NCHAR":
|
|
|
+ metricIndex = i
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if timeIndex == -1 {
|
|
|
+ return fmt.Errorf("Found no column named time")
|
|
|
+ }
|
|
|
+
|
|
|
+ for rows.Next() {
|
|
|
+ var timestamp float64
|
|
|
+ var value null.Float
|
|
|
+ var metricColVal string
|
|
|
+ var metric string
|
|
|
|
|
|
- for ; rows.Next(); rowCount++ {
|
|
|
if rowCount > rowLimit {
|
|
|
- return fmt.Errorf("MsSQL query row limit exceeded, limit %d", rowLimit)
|
|
|
+ return fmt.Errorf("MSSQL query row limit exceeded, limit %d", rowLimit)
|
|
|
}
|
|
|
|
|
|
- err := rowData.Update(rows.Rows)
|
|
|
+ values, err := e.getTypedRowData(columnTypes, rows)
|
|
|
if err != nil {
|
|
|
- e.log.Error("MsSQL response parsing", "error", err)
|
|
|
- return fmt.Errorf("MsSQL response parsing error %v", err)
|
|
|
+ return err
|
|
|
}
|
|
|
|
|
|
- if rowData.metric == "" {
|
|
|
- rowData.metric = "Unknown"
|
|
|
+ switch columnValue := values[timeIndex].(type) {
|
|
|
+ case int64:
|
|
|
+ timestamp = float64(columnValue * 1000)
|
|
|
+ case float64:
|
|
|
+ timestamp = columnValue * 1000
|
|
|
+ case time.Time:
|
|
|
+ timestamp = (float64(columnValue.Unix()) * 1000) + float64(columnValue.Nanosecond()/1e6) // in case someone is trying to map times beyond 2262 :D
|
|
|
+ default:
|
|
|
+ return fmt.Errorf("Invalid type for column time, must be of type timestamp or unix timestamp")
|
|
|
}
|
|
|
|
|
|
- if !rowData.time.Valid {
|
|
|
- return fmt.Errorf("Found row with no time value")
|
|
|
+ if metricIndex >= 0 {
|
|
|
+ if columnValue, ok := values[metricIndex].(string); ok == true {
|
|
|
+ metricColVal = columnValue
|
|
|
+ } else {
|
|
|
+ return fmt.Errorf("Column metric must be of type CHAR, VARCHAR, NCHAR or NVARCHAR. metric column name: %s type: %s but datatype is %T", columnNames[metricIndex], columnTypes[metricIndex].DatabaseTypeName(), values[metricIndex])
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
- if series, exist := pointsBySeries[rowData.metric]; exist {
|
|
|
- series.Points = append(series.Points, tsdb.TimePoint{rowData.value, rowData.time})
|
|
|
- } else {
|
|
|
- series := &tsdb.TimeSeries{Name: rowData.metric}
|
|
|
- series.Points = append(series.Points, tsdb.TimePoint{rowData.value, rowData.time})
|
|
|
- pointsBySeries[rowData.metric] = series
|
|
|
- seriesByQueryOrder.PushBack(rowData.metric)
|
|
|
+ for i, col := range columnNames {
|
|
|
+ if i == timeIndex || i == metricIndex {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+
|
|
|
+ switch columnValue := values[i].(type) {
|
|
|
+ case int64:
|
|
|
+ value = null.FloatFrom(float64(columnValue))
|
|
|
+ case float64:
|
|
|
+ value = null.FloatFrom(columnValue)
|
|
|
+ case nil:
|
|
|
+ value.Valid = false
|
|
|
+ default:
|
|
|
+ return fmt.Errorf("Value column must have numeric datatype, column: %s type: %T value: %v", col, columnValue, columnValue)
|
|
|
+ }
|
|
|
+
|
|
|
+ // construct the metric name
|
|
|
+ // if there is more than 3 columns (more than one value) and there is
|
|
|
+ // a metric column, join them to make the metric name
|
|
|
+ if metricIndex == -1 {
|
|
|
+ metric = col
|
|
|
+ } else if len(columnNames) > 3 {
|
|
|
+ metric = metricColVal + " - " + col
|
|
|
+ } else {
|
|
|
+ metric = metricColVal
|
|
|
+ }
|
|
|
+
|
|
|
+ e.appendTimePoint(pointsBySeries, seriesByQueryOrder, metric, timestamp, value)
|
|
|
+ rowCount++
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -171,61 +251,14 @@ func (e MssqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.
|
|
|
return nil
|
|
|
}
|
|
|
|
|
|
-type stringStringScan struct {
|
|
|
- rowPtrs []interface{}
|
|
|
- rowValues []string
|
|
|
- columnNames []string
|
|
|
- columnCount int
|
|
|
-
|
|
|
- time null.Float
|
|
|
- value null.Float
|
|
|
- metric string
|
|
|
-}
|
|
|
-
|
|
|
-func NewStringStringScan(columnNames []string) *stringStringScan {
|
|
|
- s := &stringStringScan{
|
|
|
- columnCount: len(columnNames),
|
|
|
- columnNames: columnNames,
|
|
|
- rowPtrs: make([]interface{}, len(columnNames)),
|
|
|
- rowValues: make([]string, len(columnNames)),
|
|
|
+func (e MssqlQueryEndpoint) appendTimePoint(pointsBySeries map[string]*tsdb.TimeSeries, seriesByQueryOrder *list.List, metric string, timestamp float64, value null.Float) {
|
|
|
+ if series, exist := pointsBySeries[metric]; exist {
|
|
|
+ series.Points = append(series.Points, tsdb.TimePoint{value, null.FloatFrom(timestamp)})
|
|
|
+ } else {
|
|
|
+ series := &tsdb.TimeSeries{Name: metric}
|
|
|
+ series.Points = append(series.Points, tsdb.TimePoint{value, null.FloatFrom(timestamp)})
|
|
|
+ pointsBySeries[metric] = series
|
|
|
+ seriesByQueryOrder.PushBack(metric)
|
|
|
}
|
|
|
-
|
|
|
- for i := 0; i < s.columnCount; i++ {
|
|
|
- s.rowPtrs[i] = new(sql.RawBytes)
|
|
|
- }
|
|
|
-
|
|
|
- return s
|
|
|
-}
|
|
|
-
|
|
|
-func (s *stringStringScan) Update(rows *sql.Rows) error {
|
|
|
- if err := rows.Scan(s.rowPtrs...); err != nil {
|
|
|
- return err
|
|
|
- }
|
|
|
-
|
|
|
- s.time = null.FloatFromPtr(nil)
|
|
|
- s.value = null.FloatFromPtr(nil)
|
|
|
-
|
|
|
- for i := 0; i < s.columnCount; i++ {
|
|
|
- if rb, ok := s.rowPtrs[i].(*sql.RawBytes); ok {
|
|
|
- s.rowValues[i] = string(*rb)
|
|
|
-
|
|
|
- switch s.columnNames[i] {
|
|
|
- case "time_sec":
|
|
|
- if sec, err := strconv.ParseInt(s.rowValues[i], 10, 64); err == nil {
|
|
|
- s.time = null.FloatFrom(float64(sec * 1000))
|
|
|
- }
|
|
|
- case "value":
|
|
|
- if value, err := strconv.ParseFloat(s.rowValues[i], 64); err == nil {
|
|
|
- s.value = null.FloatFrom(value)
|
|
|
- }
|
|
|
- case "metric":
|
|
|
- s.metric = s.rowValues[i]
|
|
|
- }
|
|
|
-
|
|
|
- *rb = nil // reset pointer to discard current value to avoid a bug
|
|
|
- } else {
|
|
|
- return fmt.Errorf("Cannot convert index %d column %s to type *sql.RawBytes", i, s.columnNames[i])
|
|
|
- }
|
|
|
- }
|
|
|
- return nil
|
|
|
+ e.log.Debug("Rows", "metric", metric, "time", timestamp, "value", value)
|
|
|
}
|