Merge branch 'master' into mssql_datasource

pkg/tsdb/cloudwatch/cloudwatch.go

@@ -152,8 +152,6 @@ func (e *CloudWatchExecutor) executeQuery(ctx context.Context, parameters *simpl
 		MetricName: aws.String(query.MetricName),
 		Dimensions: query.Dimensions,
 		Period:     aws.Int64(int64(query.Period)),
-		StartTime:  aws.Time(startTime),
-		EndTime:    aws.Time(endTime),
 	}
 	if len(query.Statistics) > 0 {
 		params.Statistics = query.Statistics
@@ -162,15 +160,36 @@ func (e *CloudWatchExecutor) executeQuery(ctx context.Context, parameters *simpl
 		params.ExtendedStatistics = query.ExtendedStatistics
 	}
 
-	if setting.Env == setting.DEV {
-		plog.Debug("CloudWatch query", "raw query", params)
+	// 1 minutes resolutin metrics is stored for 15 days, 15 * 24 * 60 = 21600
+	if query.HighResolution && (((endTime.Unix() - startTime.Unix()) / int64(query.Period)) > 21600) {
+		return nil, errors.New("too long query period")
 	}
+	var resp *cloudwatch.GetMetricStatisticsOutput
+	for startTime.Before(endTime) {
+		params.StartTime = aws.Time(startTime)
+		if query.HighResolution {
+			startTime = startTime.Add(time.Duration(1440*query.Period) * time.Second)
+		} else {
+			startTime = endTime
+		}
+		params.EndTime = aws.Time(startTime)
 
-	resp, err := client.GetMetricStatisticsWithContext(ctx, params, request.WithResponseReadTimeout(10*time.Second))
-	if err != nil {
-		return nil, err
+		if setting.Env == setting.DEV {
+			plog.Debug("CloudWatch query", "raw query", params)
+		}
+
+		partResp, err := client.GetMetricStatisticsWithContext(ctx, params, request.WithResponseReadTimeout(10*time.Second))
+		if err != nil {
+			return nil, err
+		}
+		if resp != nil {
+			resp.Datapoints = append(resp.Datapoints, partResp.Datapoints...)
+		} else {
+			resp = partResp
+
+		}
+		metrics.M_Aws_CloudWatch_GetMetricStatistics.Inc()
 	}
-	metrics.M_Aws_CloudWatch_GetMetricStatistics.Inc()
 
 	queryRes, err := parseResponse(resp, query)
 	if err != nil {
@@ -274,6 +293,8 @@ func parseQuery(model *simplejson.Json) (*CloudWatchQuery, error) {
 		alias = "{{metric}}_{{stat}}"
 	}
 
+	highResolution := model.Get("highResolution").MustBool(false)
+
 	return &CloudWatchQuery{
 		Region:             region,
 		Namespace:          namespace,
@@ -283,6 +304,7 @@ func parseQuery(model *simplejson.Json) (*CloudWatchQuery, error) {
 		ExtendedStatistics: aws.StringSlice(extendedStatistics),
 		Period:             period,
 		Alias:              alias,
+		HighResolution:     highResolution,
 	}, nil
 }
 

pkg/tsdb/cloudwatch/cloudwatch_test.go

@@ -31,6 +31,7 @@ func TestCloudWatch(t *testing.T) {
 						"p90.00"
 					],
 					"period": "60",
+					"highResolution": false,
 					"alias": "{{metric}}_{{stat}}"
 				}
 			`

pkg/tsdb/cloudwatch/metric_find_query.go

@@ -3,6 +3,7 @@ package cloudwatch
 import (
 	"context"
 	"errors"
+	"fmt"
 	"reflect"
 	"sort"
 	"strings"
@@ -37,6 +38,7 @@ var customMetricsDimensionsMap map[string]map[string]map[string]*CustomMetricsCa
 
 func init() {
 	metricsMap = map[string][]string{
+		"AWS/AmazonMQ":       {"CpuUtilization", "HeapUsage", "NetworkIn", "NetworkOut", "TotalMessageCount", "ConsumerCount", "EnqueueCount", "EnqueueTime", "ExpiredCount", "InflightCount", "DispatchCount", "DequeueCount", "MemoryUsage", "ProducerCount", "QueueSize"},
 		"AWS/ApiGateway":     {"4XXError", "5XXError", "CacheHitCount", "CacheMissCount", "Count", "IntegrationLatency", "Latency"},
 		"AWS/ApplicationELB": {"ActiveConnectionCount", "ClientTLSNegotiationErrorCount", "HealthyHostCount", "HTTPCode_ELB_4XX_Count", "HTTPCode_ELB_5XX_Count", "HTTPCode_Target_2XX_Count", "HTTPCode_Target_3XX_Count", "HTTPCode_Target_4XX_Count", "HTTPCode_Target_5XX_Count", "IPv6ProcessedBytes", "IPv6RequestCount", "NewConnectionCount", "ProcessedBytes", "RejectedConnectionCount", "RequestCount", "RequestCountPerTarget", "TargetConnectionErrorCount", "TargetResponseTime", "TargetTLSNegotiationErrorCount", "UnHealthyHostCount"},
 		"AWS/AutoScaling":    {"GroupMinSize", "GroupMaxSize", "GroupDesiredCapacity", "GroupInServiceInstances", "GroupPendingInstances", "GroupStandbyInstances", "GroupTerminatingInstances", "GroupTotalInstances"},
@@ -96,16 +98,19 @@ func init() {
 		"AWS/SES":              {"Bounce", "Complaint", "Delivery", "Reject", "Send"},
 		"AWS/SNS":              {"NumberOfMessagesPublished", "PublishSize", "NumberOfNotificationsDelivered", "NumberOfNotificationsFailed"},
 		"AWS/SQS":              {"NumberOfMessagesSent", "SentMessageSize", "NumberOfMessagesReceived", "NumberOfEmptyReceives", "NumberOfMessagesDeleted", "ApproximateAgeOfOldestMessage", "ApproximateNumberOfMessagesDelayed", "ApproximateNumberOfMessagesVisible", "ApproximateNumberOfMessagesNotVisible"},
+		"AWS/States":           {"ExecutionTime", "ExecutionThrottled", "ExecutionsAborted", "ExecutionsFailed", "ExecutionsStarted", "ExecutionsSucceeded", "ExecutionsTimedOut", "ActivityRunTime", "ActivityScheduleTime", "ActivityTime", "ActivitiesFailed", "ActivitiesHeartbeatTimedOut", "ActivitiesScheduled", "ActivitiesScheduled", "ActivitiesSucceeded", "ActivitiesTimedOut", "LambdaFunctionRunTime", "LambdaFunctionScheduleTime", "LambdaFunctionTime", "LambdaFunctionsFailed", "LambdaFunctionsHeartbeatTimedOut", "LambdaFunctionsScheduled", "LambdaFunctionsStarted", "LambdaFunctionsSucceeded", "LambdaFunctionsTimedOut"},
 		"AWS/StorageGateway": {"CacheHitPercent", "CachePercentUsed", "CachePercentDirty", "CloudBytesDownloaded", "CloudDownloadLatency", "CloudBytesUploaded", "UploadBufferFree", "UploadBufferPercentUsed", "UploadBufferUsed", "QueuedWrites", "ReadBytes", "ReadTime", "TotalCacheSize", "WriteBytes", "WriteTime", "TimeSinceLastRecoveryPoint", "WorkingStorageFree", "WorkingStoragePercentUsed", "WorkingStorageUsed",
 			"CacheHitPercent", "CachePercentUsed", "CachePercentDirty", "ReadBytes", "ReadTime", "WriteBytes", "WriteTime", "QueuedWrites"},
 		"AWS/SWF": {"DecisionTaskScheduleToStartTime", "DecisionTaskStartToCloseTime", "DecisionTasksCompleted", "StartedDecisionTasksTimedOutOnClose", "WorkflowStartToCloseTime", "WorkflowsCanceled", "WorkflowsCompleted", "WorkflowsContinuedAsNew", "WorkflowsFailed", "WorkflowsTerminated", "WorkflowsTimedOut",
 			"ActivityTaskScheduleToCloseTime", "ActivityTaskScheduleToStartTime", "ActivityTaskStartToCloseTime", "ActivityTasksCanceled", "ActivityTasksCompleted", "ActivityTasksFailed", "ScheduledActivityTasksTimedOutOnClose", "ScheduledActivityTasksTimedOutOnStart", "StartedActivityTasksTimedOutOnClose", "StartedActivityTasksTimedOutOnHeartbeat"},
 		"AWS/VPN":        {"TunnelState", "TunnelDataIn", "TunnelDataOut"},
+		"Rekognition":    {"SuccessfulRequestCount", "ThrottledCount", "ResponseTime", "DetectedFaceCount", "DetectedLabelCount", "ServerErrorCount", "UserErrorCount"},
 		"WAF":            {"AllowedRequests", "BlockedRequests", "CountedRequests"},
 		"AWS/WorkSpaces": {"Available", "Unhealthy", "ConnectionAttempt", "ConnectionSuccess", "ConnectionFailure", "SessionLaunchTime", "InSessionLatency", "SessionDisconnect"},
 		"KMS":            {"SecondsUntilKeyMaterialExpiration"},
 	}
 	dimensionsMap = map[string][]string{
+		"AWS/AmazonMQ":         {"Broker", "Topic", "Queue"},
 		"AWS/ApiGateway":       {"ApiName", "Method", "Resource", "Stage"},
 		"AWS/ApplicationELB":   {"LoadBalancer", "TargetGroup", "AvailabilityZone"},
 		"AWS/AutoScaling":      {"AutoScalingGroupName"},
@@ -142,9 +147,11 @@ func init() {
 		"AWS/SES":              {},
 		"AWS/SNS":              {"Application", "Platform", "TopicName"},
 		"AWS/SQS":              {"QueueName"},
+		"AWS/States":           {"StateMachineArn", "ActivityArn", "LambdaFunctionArn"},
 		"AWS/StorageGateway":   {"GatewayId", "GatewayName", "VolumeId"},
 		"AWS/SWF":              {"Domain", "WorkflowTypeName", "WorkflowTypeVersion", "ActivityTypeName", "ActivityTypeVersion"},
 		"AWS/VPN":              {"VpnId", "TunnelIpAddress"},
+		"Rekognition":          {},
 		"WAF":                  {"Rule", "WebACL"},
 		"AWS/WorkSpaces":       {"DirectoryId", "WorkspaceId"},
 		"KMS":                  {"KeyId"},
@@ -185,18 +192,6 @@ func (e *CloudWatchExecutor) executeMetricFindQuery(ctx context.Context, queryCo
 		data, err = e.handleGetEbsVolumeIds(ctx, parameters, queryContext)
 		break
 	case "ec2_instance_attribute":
-		region := parameters.Get("region").MustString()
-		dsInfo := e.getDsInfo(region)
-		cfg, err := e.getAwsConfig(dsInfo)
-		if err != nil {
-			return nil, errors.New("Failed to call ec2:DescribeInstances")
-		}
-		sess, err := session.NewSession(cfg)
-		if err != nil {
-			return nil, errors.New("Failed to call ec2:DescribeInstances")
-		}
-		e.ec2Svc = ec2.New(sess, cfg)
-
 		data, err = e.handleGetEc2InstanceAttribute(ctx, parameters, queryContext)
 		break
 	}
@@ -224,6 +219,21 @@ func transformToTable(data []suggestData, result *tsdb.QueryResult) {
 	result.Meta.Set("rowCount", len(data))
 }
 
+func parseMultiSelectValue(input string) []string {
+	trimmedInput := strings.TrimSpace(input)
+
+	if strings.HasPrefix(trimmedInput, "{") {
+		values := strings.Split(strings.TrimRight(strings.TrimLeft(trimmedInput, "{"), "}"), ",")
+		trimValues := make([]string, len(values))
+		for i, v := range values {
+			trimValues[i] = strings.TrimSpace(v)
+		}
+		return trimValues
+	} else {
+		return []string{trimmedInput}
+	}
+}
+
 // Whenever this list is updated, frontend list should also be updated.
 // Please update the region list in public/app/plugins/datasource/cloudwatch/partials/config.html
 func (e *CloudWatchExecutor) handleGetRegions(ctx context.Context, parameters *simplejson.Json, queryContext *tsdb.TsdbQuery) ([]suggestData, error) {
@@ -362,19 +372,44 @@ func (e *CloudWatchExecutor) handleGetDimensionValues(ctx context.Context, param
 	return result, nil
 }
 
+func (e *CloudWatchExecutor) ensureClientSession(region string) error {
+	if e.ec2Svc == nil {
+		dsInfo := e.getDsInfo(region)
+		cfg, err := e.getAwsConfig(dsInfo)
+		if err != nil {
+			return fmt.Errorf("Failed to call ec2:getAwsConfig, %v", err)
+		}
+		sess, err := session.NewSession(cfg)
+		if err != nil {
+			return fmt.Errorf("Failed to call ec2:NewSession, %v", err)
+		}
+		e.ec2Svc = ec2.New(sess, cfg)
+	}
+	return nil
+}
+
 func (e *CloudWatchExecutor) handleGetEbsVolumeIds(ctx context.Context, parameters *simplejson.Json, queryContext *tsdb.TsdbQuery) ([]suggestData, error) {
 	region := parameters.Get("region").MustString()
 	instanceId := parameters.Get("instanceId").MustString()
 
-	instanceIds := []*string{aws.String(instanceId)}
+	err := e.ensureClientSession(region)
+	if err != nil {
+		return nil, err
+	}
+
+	instanceIds := aws.StringSlice(parseMultiSelectValue(instanceId))
 	instances, err := e.ec2DescribeInstances(region, nil, instanceIds)
 	if err != nil {
 		return nil, err
 	}
 
 	result := make([]suggestData, 0)
-	for _, mapping := range instances.Reservations[0].Instances[0].BlockDeviceMappings {
-		result = append(result, suggestData{Text: *mapping.Ebs.VolumeId, Value: *mapping.Ebs.VolumeId})
+	for _, reservation := range instances.Reservations {
+		for _, instance := range reservation.Instances {
+			for _, mapping := range instance.BlockDeviceMappings {
+				result = append(result, suggestData{Text: *mapping.Ebs.VolumeId, Value: *mapping.Ebs.VolumeId})
+			}
+		}
 	}
 
 	return result, nil
@@ -401,6 +436,11 @@ func (e *CloudWatchExecutor) handleGetEc2InstanceAttribute(ctx context.Context,
 		}
 	}
 
+	err := e.ensureClientSession(region)
+	if err != nil {
+		return nil, err
+	}
+
 	instances, err := e.ec2DescribeInstances(region, filters, nil)
 	if err != nil {
 		return nil, err
@@ -476,7 +516,7 @@ func (e *CloudWatchExecutor) cloudwatchListMetrics(region string, namespace stri
 			return !lastPage
 		})
 	if err != nil {
-		return nil, errors.New("Failed to call cloudwatch:ListMetrics")
+		return nil, fmt.Errorf("Failed to call cloudwatch:ListMetrics, %v", err)
 	}
 
 	return &resp, nil

pkg/tsdb/cloudwatch/metric_find_query_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/cloudwatch"
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
+	"github.com/bmizerany/assert"
 	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/tsdb"
 	. "github.com/smartystreets/goconvey/convey"
@@ -114,4 +115,85 @@ func TestCloudWatchMetrics(t *testing.T) {
 			So(result[0].Text, ShouldEqual, "i-12345678")
 		})
 	})
+
+	Convey("When calling handleGetEbsVolumeIds", t, func() {
+
+		executor := &CloudWatchExecutor{
+			ec2Svc: mockedEc2{Resp: ec2.DescribeInstancesOutput{
+				Reservations: []*ec2.Reservation{
+					{
+						Instances: []*ec2.Instance{
+							{
+								InstanceId: aws.String("i-1"),
+								BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-1-1")}},
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-1-2")}},
+								},
+							},
+							{
+								InstanceId: aws.String("i-2"),
+								BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-2-1")}},
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-2-2")}},
+								},
+							},
+						},
+					},
+					{
+						Instances: []*ec2.Instance{
+							{
+								InstanceId: aws.String("i-3"),
+								BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-3-1")}},
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-3-2")}},
+								},
+							},
+							{
+								InstanceId: aws.String("i-4"),
+								BlockDeviceMappings: []*ec2.InstanceBlockDeviceMapping{
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-4-1")}},
+									{Ebs: &ec2.EbsInstanceBlockDevice{VolumeId: aws.String("vol-4-2")}},
+								},
+							},
+						},
+					},
+				},
+			}},
+		}
+
+		json := simplejson.New()
+		json.Set("region", "us-east-1")
+		json.Set("instanceId", "{i-1, i-2, i-3, i-4}")
+		result, _ := executor.handleGetEbsVolumeIds(context.Background(), json, &tsdb.TsdbQuery{})
+
+		Convey("Should return all 8 VolumeIds", func() {
+			So(len(result), ShouldEqual, 8)
+			So(result[0].Text, ShouldEqual, "vol-1-1")
+			So(result[1].Text, ShouldEqual, "vol-1-2")
+			So(result[2].Text, ShouldEqual, "vol-2-1")
+			So(result[3].Text, ShouldEqual, "vol-2-2")
+			So(result[4].Text, ShouldEqual, "vol-3-1")
+			So(result[5].Text, ShouldEqual, "vol-3-2")
+			So(result[6].Text, ShouldEqual, "vol-4-1")
+			So(result[7].Text, ShouldEqual, "vol-4-2")
+		})
+	})
+}
+
+func TestParseMultiSelectValue(t *testing.T) {
+
+	var values []string
+
+	values = parseMultiSelectValue(" i-someInstance ")
+	assert.Equal(t, []string{"i-someInstance"}, values)
+
+	values = parseMultiSelectValue("{i-05}")
+	assert.Equal(t, []string{"i-05"}, values)
+
+	values = parseMultiSelectValue(" {i-01, i-03, i-04} ")
+	assert.Equal(t, []string{"i-01", "i-03", "i-04"}, values)
+
+	values = parseMultiSelectValue("i-{01}")
+	assert.Equal(t, []string{"i-{01}"}, values)
+
 }

pkg/tsdb/cloudwatch/types.go

@@ -13,4 +13,5 @@ type CloudWatchQuery struct {
 	ExtendedStatistics []*string
 	Period             int
 	Alias              string
+	HighResolution     bool
 }

pkg/tsdb/influxdb/query.go

@@ -70,7 +70,7 @@ func (query *Query) renderTags() []string {
 		} else if tag.Operator == "<" || tag.Operator == ">" {
 			textValue = tag.Value
 		} else {
-			textValue = fmt.Sprintf("'%s'", tag.Value)
+			textValue = fmt.Sprintf("'%s'", strings.Replace(tag.Value, `\`, `\\`, -1))
 		}
 
 		res = append(res, fmt.Sprintf(`%s"%s" %s %s`, str, tag.Key, tag.Operator, textValue))

pkg/tsdb/influxdb/query_test.go

@@ -170,6 +170,12 @@ func TestInfluxdbQueryBuilder(t *testing.T) {
 			So(strings.Join(query.renderTags(), ""), ShouldEqual, `"key" = 'value'`)
 		})
 
+		Convey("can escape backslashes when rendering string tags", func() {
+			query := &Query{Tags: []*Tag{{Operator: "=", Value: `C:\test\`, Key: "key"}}}
+
+			So(strings.Join(query.renderTags(), ""), ShouldEqual, `"key" = 'C:\\test\\'`)
+		})
+
 		Convey("can render regular measurement", func() {
 			query := &Query{Measurement: `apa`, Policy: "policy"}
 

pkg/tsdb/models.go

@@ -14,9 +14,7 @@ type TsdbQuery struct {
 type Query struct {
 	RefId         string
 	Model         *simplejson.Json
-	Depends       []string
 	DataSource    *models.DataSource
-	Results       []*TimeSeries
 	MaxDataPoints int64
 	IntervalMs    int64
 }

pkg/tsdb/mssql/macros.go

@@ -14,19 +14,25 @@ const sExpr = `\$` + rsIdentifier + `\(([^\)]*)\)`
 
 type MsSqlMacroEngine struct {
 	TimeRange *tsdb.TimeRange
+	Query     *tsdb.Query
 }
 
 func NewMssqlMacroEngine() tsdb.SqlMacroEngine {
 	return &MsSqlMacroEngine{}
 }
 
-func (m *MsSqlMacroEngine) Interpolate(timeRange *tsdb.TimeRange, sql string) (string, error) {
+func (m *MsSqlMacroEngine) Interpolate(query *tsdb.Query, timeRange *tsdb.TimeRange, sql string) (string, error) {
 	m.TimeRange = timeRange
+	m.Query = query
 	rExp, _ := regexp.Compile(sExpr)
 	var macroError error
 
 	sql = replaceAllStringSubmatchFunc(rExp, sql, func(groups []string) string {
-		res, err := m.evaluateMacro(groups[1], strings.Split(groups[2], ","))
+		args := strings.Split(groups[2], ",")
+		for i, arg := range args {
+			args[i] = strings.Trim(arg, " ")
+		}
+		res, err := m.evaluateMacro(groups[1], args)
 		if err != nil && macroError == nil {
 			macroError = err
 			return "macro_error()"

pkg/tsdb/mssql/macros_test.go

@@ -11,72 +11,73 @@ func TestMacroEngine(t *testing.T) {
 	Convey("MacroEngine", t, func() {
 		engine := &MsSqlMacroEngine{}
 		timeRange := &tsdb.TimeRange{From: "5m", To: "now"}
+		query := &tsdb.Query{}
 
 		Convey("interpolate __time function", func() {
-			sql, err := engine.Interpolate(nil, "select $__time(time_column)")
+			sql, err := engine.Interpolate(query, nil, "select $__time(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select time_column AS time")
 		})
 
 		Convey("interpolate __utcTime function", func() {
-			sql, err := engine.Interpolate(nil, "select $__utcTime(time_column)")
+			sql, err := engine.Interpolate(query, nil, "select $__utcTime(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select DATEADD(second, DATEDIFF(second,GETDATE(),GETUTCDATE()), time_column) AS time")
 		})
 
 		Convey("interpolate __timeEpoch function", func() {
-			sql, err := engine.Interpolate(nil, "select $__timeEpoch(time_column)")
+			sql, err := engine.Interpolate(query, nil, "select $__timeEpoch(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select DATEDIFF(second, {d '1970-01-01'}, DATEADD(second, DATEDIFF(second,GETDATE(),GETUTCDATE()), time_column) ) AS time")
 		})
 
 		Convey("interpolate __timeEpoch function wrapped in aggregation", func() {
-			sql, err := engine.Interpolate(nil, "select min($__timeEpoch(time_column))")
+			sql, err := engine.Interpolate(query, nil, "select min($__timeEpoch(time_column))")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select min(DATEDIFF(second, {d '1970-01-01'}, DATEADD(second, DATEDIFF(second,GETDATE(),GETUTCDATE()), time_column) ) AS time)")
 		})
 
 		Convey("interpolate __timeFilter function", func() {
-			sql, err := engine.Interpolate(timeRange, "WHERE $__timeFilter(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "WHERE $__timeFilter(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "WHERE time_column >= DATEADD(s, 18446744066914186738+DATEDIFF(second,GETUTCDATE(),GETDATE()), '1970-01-01') AND time_column <= DATEADD(s, 18446744066914187038+DATEDIFF(second,GETUTCDATE(),GETDATE()), '1970-01-01')")
 		})
 
 		Convey("interpolate __timeFrom function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__timeFrom(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeFrom(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select DATEADD(second, 18446744066914186738+DATEDIFF(second,GETUTCDATE(),GETDATE()), '1970-01-01')")
 		})
 
 		Convey("interpolate __timeTo function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__timeTo(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeTo(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select DATEADD(second, 18446744066914187038+DATEDIFF(second,GETUTCDATE(),GETDATE()), '1970-01-01')")
 		})
 
 		Convey("interpolate __unixEpochFilter function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFilter(18446744066914186738)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFilter(18446744066914186738)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914186738 >= 18446744066914186738 AND 18446744066914186738 <= 18446744066914187038")
 		})
 
 		Convey("interpolate __unixEpochFrom function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFrom()")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFrom()")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914186738")
 		})
 
 		Convey("interpolate __unixEpochTo function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochTo()")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochTo()")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914187038")

pkg/tsdb/mssql/mssql.go

@@ -60,7 +60,7 @@ func (e *MssqlQueryEndpoint) Query(ctx context.Context, dsInfo *models.DataSourc
 	return e.sqlEngine.Query(ctx, dsInfo, tsdbQuery, e.transformToTimeSeries, e.transformToTable)
 }
 
-func (e MssqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+func (e MssqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult, tsdbQuery *tsdb.TsdbQuery) error {
 	columnNames, err := rows.Columns()
 	columnCount := len(columnNames)
 
@@ -135,7 +135,7 @@ func (e MssqlQueryEndpoint) getTypedRowData(types []*sql.ColumnType, rows *core.
 	return values, nil
 }
 
-func (e MssqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+func (e MssqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult, tsdbQuery *tsdb.TsdbQuery) error {
 	pointsBySeries := make(map[string]*tsdb.TimeSeries)
 	seriesByQueryOrder := list.New()
 

pkg/tsdb/mysql/macros.go

@@ -3,6 +3,7 @@ package mysql
 import (
 	"fmt"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -15,19 +16,25 @@ const sExpr = `\$` + rsIdentifier + `\(([^\)]*)\)`
 
 type MySqlMacroEngine struct {
 	TimeRange *tsdb.TimeRange
+	Query     *tsdb.Query
 }
 
 func NewMysqlMacroEngine() tsdb.SqlMacroEngine {
 	return &MySqlMacroEngine{}
 }
 
-func (m *MySqlMacroEngine) Interpolate(timeRange *tsdb.TimeRange, sql string) (string, error) {
+func (m *MySqlMacroEngine) Interpolate(query *tsdb.Query, timeRange *tsdb.TimeRange, sql string) (string, error) {
 	m.TimeRange = timeRange
+	m.Query = query
 	rExp, _ := regexp.Compile(sExpr)
 	var macroError error
 
 	sql = replaceAllStringSubmatchFunc(rExp, sql, func(groups []string) string {
-		res, err := m.evaluateMacro(groups[1], strings.Split(groups[2], ","))
+		args := strings.Split(groups[2], ",")
+		for i, arg := range args {
+			args[i] = strings.Trim(arg, " ")
+		}
+		res, err := m.evaluateMacro(groups[1], args)
 		if err != nil && macroError == nil {
 			macroError = err
 			return "macro_error()"
@@ -76,13 +83,26 @@ func (m *MySqlMacroEngine) evaluateMacro(name string, args []string) (string, er
 	case "__timeTo":
 		return fmt.Sprintf("FROM_UNIXTIME(%d)", uint64(m.TimeRange.GetToAsMsEpoch()/1000)), nil
 	case "__timeGroup":
-		if len(args) != 2 {
+		if len(args) < 2 {
 			return "", fmt.Errorf("macro %v needs time column and interval", name)
 		}
-		interval, err := time.ParseDuration(strings.Trim(args[1], `'" `))
+		interval, err := time.ParseDuration(strings.Trim(args[1], `'"`))
 		if err != nil {
 			return "", fmt.Errorf("error parsing interval %v", args[1])
 		}
+		if len(args) == 3 {
+			m.Query.Model.Set("fill", true)
+			m.Query.Model.Set("fillInterval", interval.Seconds())
+			if args[2] == "NULL" {
+				m.Query.Model.Set("fillNull", true)
+			} else {
+				floatVal, err := strconv.ParseFloat(args[2], 64)
+				if err != nil {
+					return "", fmt.Errorf("error parsing fill value %v", args[2])
+				}
+				m.Query.Model.Set("fillValue", floatVal)
+			}
+		}
 		return fmt.Sprintf("cast(cast(UNIX_TIMESTAMP(%s)/(%.0f) as signed)*%.0f as signed)", args[0], interval.Seconds(), interval.Seconds()), nil
 	case "__unixEpochFilter":
 		if len(args) == 0 {

pkg/tsdb/mysql/macros_test.go

@@ -10,31 +10,32 @@ import (
 func TestMacroEngine(t *testing.T) {
 	Convey("MacroEngine", t, func() {
 		engine := &MySqlMacroEngine{}
+		query := &tsdb.Query{}
 		timeRange := &tsdb.TimeRange{From: "5m", To: "now"}
 
 		Convey("interpolate __time function", func() {
-			sql, err := engine.Interpolate(nil, "select $__time(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__time(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select UNIX_TIMESTAMP(time_column) as time_sec")
 		})
 
 		Convey("interpolate __time function wrapped in aggregation", func() {
-			sql, err := engine.Interpolate(nil, "select min($__time(time_column))")
+			sql, err := engine.Interpolate(query, timeRange, "select min($__time(time_column))")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select min(UNIX_TIMESTAMP(time_column) as time_sec)")
 		})
 
 		Convey("interpolate __timeFilter function", func() {
-			sql, err := engine.Interpolate(timeRange, "WHERE $__timeFilter(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "WHERE $__timeFilter(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "WHERE time_column >= FROM_UNIXTIME(18446744066914186738) AND time_column <= FROM_UNIXTIME(18446744066914187038)")
 		})
 
 		Convey("interpolate __timeFrom function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__timeFrom(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeFrom(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select FROM_UNIXTIME(18446744066914186738)")
@@ -42,35 +43,43 @@ func TestMacroEngine(t *testing.T) {
 
 		Convey("interpolate __timeGroup function", func() {
 
-			sql, err := engine.Interpolate(timeRange, "GROUP BY $__timeGroup(time_column,'5m')")
+			sql, err := engine.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column,'5m')")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "GROUP BY cast(cast(UNIX_TIMESTAMP(time_column)/(300) as signed)*300 as signed)")
+		})
+
+		Convey("interpolate __timeGroup function with spaces around arguments", func() {
+
+			sql, err := engine.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column , '5m')")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "GROUP BY cast(cast(UNIX_TIMESTAMP(time_column)/(300) as signed)*300 as signed)")
 		})
 
 		Convey("interpolate __timeTo function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__timeTo(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeTo(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select FROM_UNIXTIME(18446744066914187038)")
 		})
 
 		Convey("interpolate __unixEpochFilter function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFilter(18446744066914186738)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFilter(18446744066914186738)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914186738 >= 18446744066914186738 AND 18446744066914186738 <= 18446744066914187038")
 		})
 
 		Convey("interpolate __unixEpochFrom function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFrom()")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFrom()")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914186738")
 		})
 
 		Convey("interpolate __unixEpochTo function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochTo()")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochTo()")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914187038")

pkg/tsdb/mysql/mysql.go

@@ -5,8 +5,9 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"math"
+	"reflect"
 	"strconv"
-
 	"time"
 
 	"github.com/go-sql-driver/mysql"
@@ -56,7 +57,7 @@ func (e *MysqlQueryEndpoint) Query(ctx context.Context, dsInfo *models.DataSourc
 	return e.sqlEngine.Query(ctx, dsInfo, tsdbQuery, e.transformToTimeSeries, e.transformToTable)
 }
 
-func (e MysqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+func (e MysqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult, tsdbQuery *tsdb.TsdbQuery) error {
 	columnNames, err := rows.Columns()
 	columnCount := len(columnNames)
 
@@ -73,24 +74,36 @@ func (e MysqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows,
 		table.Columns[i].Text = name
 	}
 
-	columnTypes, err := rows.ColumnTypes()
-	if err != nil {
-		return err
-	}
-
 	rowLimit := 1000000
 	rowCount := 0
+	timeIndex := -1
+
+	// check if there is a column named time
+	for i, col := range columnNames {
+		switch col {
+		case "time_sec":
+			timeIndex = i
+		}
+	}
 
 	for ; rows.Next(); rowCount++ {
 		if rowCount > rowLimit {
 			return fmt.Errorf("MySQL query row limit exceeded, limit %d", rowLimit)
 		}
 
-		values, err := e.getTypedRowData(columnTypes, rows)
+		values, err := e.getTypedRowData(rows)
 		if err != nil {
 			return err
 		}
 
+		// for annotations, convert to epoch
+		if timeIndex != -1 {
+			switch value := values[timeIndex].(type) {
+			case time.Time:
+				values[timeIndex] = float64(value.UnixNano() / 1e9)
+			}
+		}
+
 		table.Rows = append(table.Rows, values)
 	}
 
@@ -99,60 +112,20 @@ func (e MysqlQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows,
 	return nil
 }
 
-func (e MysqlQueryEndpoint) getTypedRowData(types []*sql.ColumnType, rows *core.Rows) (tsdb.RowValues, error) {
+func (e MysqlQueryEndpoint) getTypedRowData(rows *core.Rows) (tsdb.RowValues, error) {
+	types, err := rows.ColumnTypes()
+	if err != nil {
+		return nil, err
+	}
+
 	values := make([]interface{}, len(types))
 
-	for i, stype := range types {
-		e.log.Debug("type", "type", stype)
-		switch stype.DatabaseTypeName() {
-		case mysql.FieldTypeNameTiny:
-			values[i] = new(int8)
-		case mysql.FieldTypeNameInt24:
-			values[i] = new(int32)
-		case mysql.FieldTypeNameShort:
-			values[i] = new(int16)
-		case mysql.FieldTypeNameVarString:
-			values[i] = new(string)
-		case mysql.FieldTypeNameVarChar:
-			values[i] = new(string)
-		case mysql.FieldTypeNameLong:
-			values[i] = new(int)
-		case mysql.FieldTypeNameLongLong:
-			values[i] = new(int64)
-		case mysql.FieldTypeNameDouble:
-			values[i] = new(float64)
-		case mysql.FieldTypeNameDecimal:
-			values[i] = new(float32)
-		case mysql.FieldTypeNameNewDecimal:
-			values[i] = new(float64)
-		case mysql.FieldTypeNameFloat:
-			values[i] = new(float64)
-		case mysql.FieldTypeNameTimestamp:
-			values[i] = new(time.Time)
-		case mysql.FieldTypeNameDateTime:
-			values[i] = new(time.Time)
-		case mysql.FieldTypeNameTime:
-			values[i] = new(string)
-		case mysql.FieldTypeNameYear:
-			values[i] = new(int16)
-		case mysql.FieldTypeNameNULL:
-			values[i] = nil
-		case mysql.FieldTypeNameBit:
+	for i := range values {
+		scanType := types[i].ScanType()
+		values[i] = reflect.New(scanType).Interface()
+
+		if types[i].DatabaseTypeName() == "BIT" {
 			values[i] = new([]byte)
-		case mysql.FieldTypeNameBLOB:
-			values[i] = new(string)
-		case mysql.FieldTypeNameTinyBLOB:
-			values[i] = new(string)
-		case mysql.FieldTypeNameMediumBLOB:
-			values[i] = new(string)
-		case mysql.FieldTypeNameLongBLOB:
-			values[i] = new(string)
-		case mysql.FieldTypeNameString:
-			values[i] = new(string)
-		case mysql.FieldTypeNameDate:
-			values[i] = new(string)
-		default:
-			return nil, fmt.Errorf("Database type %s not supported", stype.DatabaseTypeName())
 		}
 	}
 
@@ -160,14 +133,54 @@ func (e MysqlQueryEndpoint) getTypedRowData(types []*sql.ColumnType, rows *core.
 		return nil, err
 	}
 
+	for i := 0; i < len(types); i++ {
+		typeName := reflect.ValueOf(values[i]).Type().String()
+
+		switch typeName {
+		case "*sql.RawBytes":
+			values[i] = string(*values[i].(*sql.RawBytes))
+		case "*mysql.NullTime":
+			sqlTime := (*values[i].(*mysql.NullTime))
+			if sqlTime.Valid {
+				values[i] = sqlTime.Time
+			} else {
+				values[i] = nil
+			}
+		case "*sql.NullInt64":
+			nullInt64 := (*values[i].(*sql.NullInt64))
+			if nullInt64.Valid {
+				values[i] = nullInt64.Int64
+			} else {
+				values[i] = nil
+			}
+		case "*sql.NullFloat64":
+			nullFloat64 := (*values[i].(*sql.NullFloat64))
+			if nullFloat64.Valid {
+				values[i] = nullFloat64.Float64
+			} else {
+				values[i] = nil
+			}
+		}
+
+		if types[i].DatabaseTypeName() == "DECIMAL" {
+			f, err := strconv.ParseFloat(values[i].(string), 64)
+
+			if err == nil {
+				values[i] = f
+			} else {
+				values[i] = nil
+			}
+		}
+	}
+
 	return values, nil
 }
 
-func (e MysqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+func (e MysqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult, tsdbQuery *tsdb.TsdbQuery) error {
 	pointsBySeries := make(map[string]*tsdb.TimeSeries)
 	seriesByQueryOrder := list.New()
-	columnNames, err := rows.Columns()
 
+	columnNames, err := rows.Columns()
 	if err != nil {
 		return err
 	}
@@ -176,6 +189,18 @@ func (e MysqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.
 	rowLimit := 1000000
 	rowCount := 0
 
+	fillMissing := query.Model.Get("fill").MustBool(false)
+	var fillInterval float64
+	fillValue := null.Float{}
+	if fillMissing {
+		fillInterval = query.Model.Get("fillInterval").MustFloat64() * 1000
+		if query.Model.Get("fillNull").MustBool(false) == false {
+			fillValue.Float64 = query.Model.Get("fillValue").MustFloat64()
+			fillValue.Valid = true
+		}
+
+	}
+
 	for ; rows.Next(); rowCount++ {
 		if rowCount > rowLimit {
 			return fmt.Errorf("MySQL query row limit exceeded, limit %d", rowLimit)
@@ -195,19 +220,50 @@ func (e MysqlQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.
 			return fmt.Errorf("Found row with no time value")
 		}
 
-		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})
+		series, exist := pointsBySeries[rowData.metric]
+		if exist == false {
+			series = &tsdb.TimeSeries{Name: rowData.metric}
 			pointsBySeries[rowData.metric] = series
 			seriesByQueryOrder.PushBack(rowData.metric)
 		}
+
+		if fillMissing {
+			var intervalStart float64
+			if exist == false {
+				intervalStart = float64(tsdbQuery.TimeRange.MustGetFrom().UnixNano() / 1e6)
+			} else {
+				intervalStart = series.Points[len(series.Points)-1][1].Float64 + fillInterval
+			}
+
+			// align interval start
+			intervalStart = math.Floor(intervalStart/fillInterval) * fillInterval
+
+			for i := intervalStart; i < rowData.time.Float64; i += fillInterval {
+				series.Points = append(series.Points, tsdb.TimePoint{fillValue, null.FloatFrom(i)})
+				rowCount++
+			}
+		}
+
+		series.Points = append(series.Points, tsdb.TimePoint{rowData.value, rowData.time})
 	}
 
 	for elem := seriesByQueryOrder.Front(); elem != nil; elem = elem.Next() {
 		key := elem.Value.(string)
 		result.Series = append(result.Series, pointsBySeries[key])
+
+		if fillMissing {
+			series := pointsBySeries[key]
+			// fill in values from last fetched value till interval end
+			intervalStart := series.Points[len(series.Points)-1][1].Float64
+			intervalEnd := float64(tsdbQuery.TimeRange.MustGetTo().UnixNano() / 1e6)
+
+			// align interval start
+			intervalStart = math.Floor(intervalStart/fillInterval) * fillInterval
+			for i := intervalStart + fillInterval; i < intervalEnd; i += fillInterval {
+				series.Points = append(series.Points, tsdb.TimePoint{fillValue, null.FloatFrom(i)})
+				rowCount++
+			}
+		}
 	}
 
 	result.Meta.Set("rowCount", rowCount)

pkg/tsdb/mysql/mysql_test.go

@@ -30,19 +30,19 @@ func TestMySQL(t *testing.T) {
 		defer sess.Close()
 
 		sql := "CREATE TABLE `mysql_types` ("
-		sql += "`atinyint` tinyint(1),"
-		sql += "`avarchar` varchar(3),"
+		sql += "`atinyint` tinyint(1) NOT NULL,"
+		sql += "`avarchar` varchar(3) NOT NULL,"
 		sql += "`achar` char(3),"
-		sql += "`amediumint` mediumint,"
-		sql += "`asmallint` smallint,"
-		sql += "`abigint` bigint,"
-		sql += "`aint` int(11),"
+		sql += "`amediumint` mediumint NOT NULL,"
+		sql += "`asmallint` smallint NOT NULL,"
+		sql += "`abigint` bigint NOT NULL,"
+		sql += "`aint` int(11) NOT NULL,"
 		sql += "`adouble` double(10,2),"
 		sql += "`anewdecimal` decimal(10,2),"
-		sql += "`afloat` float(10,2),"
+		sql += "`afloat` float(10,2) NOT NULL,"
 		sql += "`atimestamp` timestamp NOT NULL,"
-		sql += "`adatetime` datetime,"
-		sql += "`atime` time,"
+		sql += "`adatetime` datetime NOT NULL,"
+		sql += "`atime` time NOT NULL,"
 		// sql += "`ayear` year," // Crashes xorm when running cleandb
 		sql += "`abit` bit(1),"
 		sql += "`atinytext` tinytext,"
@@ -55,7 +55,12 @@ func TestMySQL(t *testing.T) {
 		sql += "`alongblob` longblob,"
 		sql += "`aenum` enum('val1', 'val2'),"
 		sql += "`aset` set('a', 'b', 'c', 'd'),"
-		sql += "`adate` date"
+		sql += "`adate` date,"
+		sql += "`time_sec` datetime(6),"
+		sql += "`aintnull` int(11),"
+		sql += "`afloatnull` float(10,2),"
+		sql += "`avarcharnull` varchar(3),"
+		sql += "`adecimalnull` decimal(10,2)"
 		sql += ") ENGINE=InnoDB DEFAULT CHARSET=latin1;"
 		_, err := sess.Exec(sql)
 		So(err, ShouldBeNil)
@@ -64,11 +69,11 @@ func TestMySQL(t *testing.T) {
 		sql += "(`atinyint`, `avarchar`, `achar`, `amediumint`, `asmallint`, `abigint`, `aint`, `adouble`, "
 		sql += "`anewdecimal`, `afloat`, `adatetime`, `atimestamp`, `atime`, `abit`, `atinytext`, "
 		sql += "`atinyblob`, `atext`, `ablob`, `amediumtext`, `amediumblob`, `alongtext`, `alongblob`, "
-		sql += "`aenum`, `aset`, `adate`) "
+		sql += "`aenum`, `aset`, `adate`, `time_sec`) "
 		sql += "VALUES(1, 'abc', 'def', 1, 10, 100, 1420070400, 1.11, "
 		sql += "2.22, 3.33, now(), current_timestamp(), '11:11:11', 1, 'tinytext', "
 		sql += "'tinyblob', 'text', 'blob', 'mediumtext', 'mediumblob', 'longtext', 'longblob', "
-		sql += "'val2', 'a,b', curdate());"
+		sql += "'val2', 'a,b', curdate(), '2018-01-01 00:01:01.123456');"
 		_, err = sess.Exec(sql)
 		So(err, ShouldBeNil)
 
@@ -90,32 +95,38 @@ func TestMySQL(t *testing.T) {
 			So(err, ShouldBeNil)
 
 			column := queryResult.Tables[0].Rows[0]
+
 			So(*column[0].(*int8), ShouldEqual, 1)
-			So(*column[1].(*string), ShouldEqual, "abc")
-			So(*column[2].(*string), ShouldEqual, "def")
+			So(column[1].(string), ShouldEqual, "abc")
+			So(column[2].(string), ShouldEqual, "def")
 			So(*column[3].(*int32), ShouldEqual, 1)
 			So(*column[4].(*int16), ShouldEqual, 10)
 			So(*column[5].(*int64), ShouldEqual, 100)
-			So(*column[6].(*int), ShouldEqual, 1420070400)
-			So(*column[7].(*float64), ShouldEqual, 1.11)
-			So(*column[8].(*float64), ShouldEqual, 2.22)
-			So(*column[9].(*float64), ShouldEqual, 3.33)
+			So(*column[6].(*int32), ShouldEqual, 1420070400)
+			So(column[7].(float64), ShouldEqual, 1.11)
+			So(column[8].(float64), ShouldEqual, 2.22)
+			So(*column[9].(*float32), ShouldEqual, 3.33)
 			_, offset := time.Now().Zone()
-			So((*column[10].(*time.Time)), ShouldHappenWithin, time.Duration(10*time.Second), time.Now().Add(time.Duration(offset)*time.Second))
-			So(*column[11].(*time.Time), ShouldHappenWithin, time.Duration(10*time.Second), time.Now().Add(time.Duration(offset)*time.Second))
-			So(*column[12].(*string), ShouldEqual, "11:11:11")
+			So(column[10].(time.Time), ShouldHappenWithin, time.Duration(10*time.Second), time.Now().Add(time.Duration(offset)*time.Second))
+			So(column[11].(time.Time), ShouldHappenWithin, time.Duration(10*time.Second), time.Now().Add(time.Duration(offset)*time.Second))
+			So(column[12].(string), ShouldEqual, "11:11:11")
 			So(*column[13].(*[]byte), ShouldHaveSameTypeAs, []byte{1})
-			So(*column[14].(*string), ShouldEqual, "tinytext")
-			So(*column[15].(*string), ShouldEqual, "tinyblob")
-			So(*column[16].(*string), ShouldEqual, "text")
-			So(*column[17].(*string), ShouldEqual, "blob")
-			So(*column[18].(*string), ShouldEqual, "mediumtext")
-			So(*column[19].(*string), ShouldEqual, "mediumblob")
-			So(*column[20].(*string), ShouldEqual, "longtext")
-			So(*column[21].(*string), ShouldEqual, "longblob")
-			So(*column[22].(*string), ShouldEqual, "val2")
-			So(*column[23].(*string), ShouldEqual, "a,b")
-			So(*column[24].(*string), ShouldEqual, time.Now().Format("2006-01-02T00:00:00Z"))
+			So(column[14].(string), ShouldEqual, "tinytext")
+			So(column[15].(string), ShouldEqual, "tinyblob")
+			So(column[16].(string), ShouldEqual, "text")
+			So(column[17].(string), ShouldEqual, "blob")
+			So(column[18].(string), ShouldEqual, "mediumtext")
+			So(column[19].(string), ShouldEqual, "mediumblob")
+			So(column[20].(string), ShouldEqual, "longtext")
+			So(column[21].(string), ShouldEqual, "longblob")
+			So(column[22].(string), ShouldEqual, "val2")
+			So(column[23].(string), ShouldEqual, "a,b")
+			So(column[24].(time.Time).Format("2006-01-02T00:00:00Z"), ShouldEqual, time.Now().Format("2006-01-02T00:00:00Z"))
+			So(column[25].(float64), ShouldEqual, 1514764861)
+			So(column[26], ShouldEqual, nil)
+			So(column[27], ShouldEqual, nil)
+			So(column[28], ShouldEqual, "")
+			So(column[29], ShouldEqual, nil)
 		})
 	})
 }

pkg/tsdb/opentsdb/opentsdb.go

@@ -22,23 +22,10 @@ import (
 )
 
 type OpenTsdbExecutor struct {
-	//*models.DataSource
-	//httpClient *http.Client
 }
 
 func NewOpenTsdbExecutor(datasource *models.DataSource) (tsdb.TsdbQueryEndpoint, error) {
-	/*
-		httpClient, err := datasource.GetHttpClient()
-
-		if err != nil {
-			return nil, err
-		}
-	*/
-
-	return &OpenTsdbExecutor{
-	//DataSource: datasource,
-	//httpClient: httpClient,
-	}, nil
+	return &OpenTsdbExecutor{}, nil
 }
 
 var (

pkg/tsdb/postgres/macros.go

@@ -3,6 +3,7 @@ package postgres
 import (
 	"fmt"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -15,19 +16,25 @@ const sExpr = `\$` + rsIdentifier + `\(([^\)]*)\)`
 
 type PostgresMacroEngine struct {
 	TimeRange *tsdb.TimeRange
+	Query     *tsdb.Query
 }
 
 func NewPostgresMacroEngine() tsdb.SqlMacroEngine {
 	return &PostgresMacroEngine{}
 }
 
-func (m *PostgresMacroEngine) Interpolate(timeRange *tsdb.TimeRange, sql string) (string, error) {
+func (m *PostgresMacroEngine) Interpolate(query *tsdb.Query, timeRange *tsdb.TimeRange, sql string) (string, error) {
 	m.TimeRange = timeRange
+	m.Query = query
 	rExp, _ := regexp.Compile(sExpr)
 	var macroError error
 
 	sql = replaceAllStringSubmatchFunc(rExp, sql, func(groups []string) string {
-		res, err := m.evaluateMacro(groups[1], strings.Split(groups[2], ","))
+		args := strings.Split(groups[2], ",")
+		for i, arg := range args {
+			args[i] = strings.Trim(arg, " ")
+		}
+		res, err := m.evaluateMacro(groups[1], args)
 		if err != nil && macroError == nil {
 			macroError = err
 			return "macro_error()"
@@ -82,13 +89,26 @@ func (m *PostgresMacroEngine) evaluateMacro(name string, args []string) (string,
 	case "__timeTo":
 		return fmt.Sprintf("to_timestamp(%d)", uint64(m.TimeRange.GetToAsMsEpoch()/1000)), nil
 	case "__timeGroup":
-		if len(args) != 2 {
-			return "", fmt.Errorf("macro %v needs time column and interval", name)
+		if len(args) < 2 {
+			return "", fmt.Errorf("macro %v needs time column and interval and optional fill value", name)
 		}
-		interval, err := time.ParseDuration(strings.Trim(args[1], `' `))
+		interval, err := time.ParseDuration(strings.Trim(args[1], `'`))
 		if err != nil {
 			return "", fmt.Errorf("error parsing interval %v", args[1])
 		}
+		if len(args) == 3 {
+			m.Query.Model.Set("fill", true)
+			m.Query.Model.Set("fillInterval", interval.Seconds())
+			if args[2] == "NULL" {
+				m.Query.Model.Set("fillNull", true)
+			} else {
+				floatVal, err := strconv.ParseFloat(args[2], 64)
+				if err != nil {
+					return "", fmt.Errorf("error parsing fill value %v", args[2])
+				}
+				m.Query.Model.Set("fillValue", floatVal)
+			}
+		}
 		return fmt.Sprintf("(extract(epoch from %s)/%v)::bigint*%v AS time", args[0], interval.Seconds(), interval.Seconds()), nil
 	case "__unixEpochFilter":
 		if len(args) == 0 {

pkg/tsdb/postgres/macros_test.go

@@ -10,31 +10,32 @@ import (
 func TestMacroEngine(t *testing.T) {
 	Convey("MacroEngine", t, func() {
 		engine := &PostgresMacroEngine{}
+		query := &tsdb.Query{}
 		timeRange := &tsdb.TimeRange{From: "5m", To: "now"}
 
 		Convey("interpolate __time function", func() {
-			sql, err := engine.Interpolate(nil, "select $__time(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__time(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select time_column AS \"time\"")
 		})
 
 		Convey("interpolate __time function wrapped in aggregation", func() {
-			sql, err := engine.Interpolate(nil, "select min($__time(time_column))")
+			sql, err := engine.Interpolate(query, timeRange, "select min($__time(time_column))")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select min(time_column AS \"time\")")
 		})
 
 		Convey("interpolate __timeFilter function", func() {
-			sql, err := engine.Interpolate(timeRange, "WHERE $__timeFilter(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "WHERE $__timeFilter(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "WHERE extract(epoch from time_column) BETWEEN 18446744066914186738 AND 18446744066914187038")
 		})
 
 		Convey("interpolate __timeFrom function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__timeFrom(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeFrom(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select to_timestamp(18446744066914186738)")
@@ -42,35 +43,43 @@ func TestMacroEngine(t *testing.T) {
 
 		Convey("interpolate __timeGroup function", func() {
 
-			sql, err := engine.Interpolate(timeRange, "GROUP BY $__timeGroup(time_column,'5m')")
+			sql, err := engine.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column,'5m')")
+			So(err, ShouldBeNil)
+
+			So(sql, ShouldEqual, "GROUP BY (extract(epoch from time_column)/300)::bigint*300 AS time")
+		})
+
+		Convey("interpolate __timeGroup function with spaces between args", func() {
+
+			sql, err := engine.Interpolate(query, timeRange, "GROUP BY $__timeGroup(time_column , '5m')")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "GROUP BY (extract(epoch from time_column)/300)::bigint*300 AS time")
 		})
 
 		Convey("interpolate __timeTo function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__timeTo(time_column)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__timeTo(time_column)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select to_timestamp(18446744066914187038)")
 		})
 
 		Convey("interpolate __unixEpochFilter function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFilter(18446744066914186738)")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFilter(18446744066914186738)")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914186738 >= 18446744066914186738 AND 18446744066914186738 <= 18446744066914187038")
 		})
 
 		Convey("interpolate __unixEpochFrom function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochFrom()")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochFrom()")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914186738")
 		})
 
 		Convey("interpolate __unixEpochTo function", func() {
-			sql, err := engine.Interpolate(timeRange, "select $__unixEpochTo()")
+			sql, err := engine.Interpolate(query, timeRange, "select $__unixEpochTo()")
 			So(err, ShouldBeNil)
 
 			So(sql, ShouldEqual, "select 18446744066914187038")

pkg/tsdb/postgres/postgres.go

@@ -4,6 +4,7 @@ import (
 	"container/list"
 	"context"
 	"fmt"
+	"math"
 	"net/url"
 	"strconv"
 	"time"
@@ -53,14 +54,15 @@ func generateConnectionString(datasource *models.DataSource) string {
 	}
 
 	sslmode := datasource.JsonData.Get("sslmode").MustString("verify-full")
-	return fmt.Sprintf("postgres://%s:%s@%s/%s?sslmode=%s", url.PathEscape(datasource.User), url.PathEscape(password), url.PathEscape(datasource.Url), url.PathEscape(datasource.Database), url.QueryEscape(sslmode))
+	u := &url.URL{Scheme: "postgres", User: url.UserPassword(datasource.User, password), Host: datasource.Url, Path: datasource.Database, RawQuery: "sslmode=" + sslmode}
+	return u.String()
 }
 
 func (e *PostgresQueryEndpoint) Query(ctx context.Context, dsInfo *models.DataSource, tsdbQuery *tsdb.TsdbQuery) (*tsdb.Response, error) {
 	return e.sqlEngine.Query(ctx, dsInfo, tsdbQuery, e.transformToTimeSeries, e.transformToTable)
 }
 
-func (e PostgresQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+func (e PostgresQueryEndpoint) transformToTable(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult, tsdbQuery *tsdb.TsdbQuery) error {
 
 	columnNames, err := rows.Columns()
 	if err != nil {
@@ -157,7 +159,7 @@ func (e PostgresQueryEndpoint) getTypedRowData(rows *core.Rows) (tsdb.RowValues,
 	return values, nil
 }
 
-func (e PostgresQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult) error {
+func (e PostgresQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *core.Rows, result *tsdb.QueryResult, tsdbQuery *tsdb.TsdbQuery) error {
 	pointsBySeries := make(map[string]*tsdb.TimeSeries)
 	seriesByQueryOrder := list.New()
 
@@ -198,6 +200,18 @@ func (e PostgresQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *co
 		return fmt.Errorf("Found no column named time")
 	}
 
+	fillMissing := query.Model.Get("fill").MustBool(false)
+	var fillInterval float64
+	fillValue := null.Float{}
+	if fillMissing {
+		fillInterval = query.Model.Get("fillInterval").MustFloat64() * 1000
+		if query.Model.Get("fillNull").MustBool(false) == false {
+			fillValue.Float64 = query.Model.Get("fillValue").MustFloat64()
+			fillValue.Valid = true
+		}
+
+	}
+
 	for rows.Next() {
 		var timestamp float64
 		var value null.Float
@@ -220,14 +234,14 @@ func (e PostgresQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *co
 		case time.Time:
 			timestamp = float64(columnValue.UnixNano() / 1e6)
 		default:
-			return fmt.Errorf("Invalid type for column time, must be of type timestamp or unix timestamp")
+			return fmt.Errorf("Invalid type for column time, must be of type timestamp or unix timestamp, got: %T %v", columnValue, columnValue)
 		}
 
 		if metricIndex >= 0 {
 			if columnValue, ok := values[metricIndex].(string); ok == true {
 				metric = columnValue
 			} else {
-				return fmt.Errorf("Column metric must be of type char,varchar or text")
+				return fmt.Errorf("Column metric must be of type char,varchar or text, got: %T %v", values[metricIndex], values[metricIndex])
 			}
 		}
 
@@ -249,7 +263,34 @@ func (e PostgresQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *co
 			if metricIndex == -1 {
 				metric = col
 			}
-			e.appendTimePoint(pointsBySeries, seriesByQueryOrder, metric, timestamp, value)
+
+			series, exist := pointsBySeries[metric]
+			if exist == false {
+				series = &tsdb.TimeSeries{Name: metric}
+				pointsBySeries[metric] = series
+				seriesByQueryOrder.PushBack(metric)
+			}
+
+			if fillMissing {
+				var intervalStart float64
+				if exist == false {
+					intervalStart = float64(tsdbQuery.TimeRange.MustGetFrom().UnixNano() / 1e6)
+				} else {
+					intervalStart = series.Points[len(series.Points)-1][1].Float64 + fillInterval
+				}
+
+				// align interval start
+				intervalStart = math.Floor(intervalStart/fillInterval) * fillInterval
+
+				for i := intervalStart; i < timestamp; i += fillInterval {
+					series.Points = append(series.Points, tsdb.TimePoint{fillValue, null.FloatFrom(i)})
+					rowCount++
+				}
+			}
+
+			series.Points = append(series.Points, tsdb.TimePoint{value, null.FloatFrom(timestamp)})
+
+			e.log.Debug("Rows", "metric", metric, "time", timestamp, "value", value)
 			rowCount++
 
 		}
@@ -258,20 +299,22 @@ func (e PostgresQueryEndpoint) transformToTimeSeries(query *tsdb.Query, rows *co
 	for elem := seriesByQueryOrder.Front(); elem != nil; elem = elem.Next() {
 		key := elem.Value.(string)
 		result.Series = append(result.Series, pointsBySeries[key])
+
+		if fillMissing {
+			series := pointsBySeries[key]
+			// fill in values from last fetched value till interval end
+			intervalStart := series.Points[len(series.Points)-1][1].Float64
+			intervalEnd := float64(tsdbQuery.TimeRange.MustGetTo().UnixNano() / 1e6)
+
+			// align interval start
+			intervalStart = math.Floor(intervalStart/fillInterval) * fillInterval
+			for i := intervalStart + fillInterval; i < intervalEnd; i += fillInterval {
+				series.Points = append(series.Points, tsdb.TimePoint{fillValue, null.FloatFrom(i)})
+				rowCount++
+			}
+		}
 	}
 
 	result.Meta.Set("rowCount", rowCount)
 	return nil
 }
-
-func (e PostgresQueryEndpoint) 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)
-	}
-	e.log.Debug("Rows", "metric", metric, "time", timestamp, "value", value)
-}

pkg/tsdb/sql_engine.go

@@ -17,15 +17,15 @@ type SqlEngine interface {
 		ctx context.Context,
 		ds *models.DataSource,
 		query *TsdbQuery,
-		transformToTimeSeries func(query *Query, rows *core.Rows, result *QueryResult) error,
-		transformToTable func(query *Query, rows *core.Rows, result *QueryResult) error,
+		transformToTimeSeries func(query *Query, rows *core.Rows, result *QueryResult, tsdbQuery *TsdbQuery) error,
+		transformToTable func(query *Query, rows *core.Rows, result *QueryResult, tsdbQuery *TsdbQuery) error,
 	) (*Response, error)
 }
 
-// SqlMacroEngine interpolates macros into sql. It takes in the timeRange to be able to
-// generate queries that use from and to.
+// SqlMacroEngine interpolates macros into sql. It takes in the Query to have access to query context and
+// timeRange to be able to generate queries that use from and to.
 type SqlMacroEngine interface {
-	Interpolate(timeRange *TimeRange, sql string) (string, error)
+	Interpolate(query *Query, timeRange *TimeRange, sql string) (string, error)
 }
 
 type DefaultSqlEngine struct {
@@ -77,8 +77,8 @@ func (e *DefaultSqlEngine) Query(
 	ctx context.Context,
 	dsInfo *models.DataSource,
 	tsdbQuery *TsdbQuery,
-	transformToTimeSeries func(query *Query, rows *core.Rows, result *QueryResult) error,
-	transformToTable func(query *Query, rows *core.Rows, result *QueryResult) error,
+	transformToTimeSeries func(query *Query, rows *core.Rows, result *QueryResult, tsdbQuery *TsdbQuery) error,
+	transformToTable func(query *Query, rows *core.Rows, result *QueryResult, tsdbQuery *TsdbQuery) error,
 ) (*Response, error) {
 	result := &Response{
 		Results: make(map[string]*QueryResult),
@@ -97,7 +97,7 @@ func (e *DefaultSqlEngine) Query(
 		queryResult := &QueryResult{Meta: simplejson.New(), RefId: query.RefId}
 		result.Results[query.RefId] = queryResult
 
-		rawSql, err := e.MacroEngine.Interpolate(tsdbQuery.TimeRange, rawSql)
+		rawSql, err := e.MacroEngine.Interpolate(query, tsdbQuery.TimeRange, rawSql)
 		if err != nil {
 			queryResult.Error = err
 			continue
@@ -117,13 +117,13 @@ func (e *DefaultSqlEngine) Query(
 
 		switch format {
 		case "time_series":
-			err := transformToTimeSeries(query, rows, queryResult)
+			err := transformToTimeSeries(query, rows, queryResult, tsdbQuery)
 			if err != nil {
 				queryResult.Error = err
 				continue
 			}
 		case "table":
-			err := transformToTable(query, rows, queryResult)
+			err := transformToTable(query, rows, queryResult, tsdbQuery)
 			if err != nil {
 				queryResult.Error = err
 				continue

pkg/tsdb/time_range.go

@@ -11,14 +11,14 @@ func NewTimeRange(from, to string) *TimeRange {
 	return &TimeRange{
 		From: from,
 		To:   to,
-		Now:  time.Now(),
+		now:  time.Now(),
 	}
 }
 
 type TimeRange struct {
 	From string
 	To   string
-	Now  time.Time
+	now  time.Time
 }
 
 func (tr *TimeRange) GetFromAsMsEpoch() int64 {
@@ -65,12 +65,12 @@ func (tr *TimeRange) ParseFrom() (time.Time, error) {
 		return time.Time{}, err
 	}
 
-	return tr.Now.Add(diff), nil
+	return tr.now.Add(diff), nil
 }
 
 func (tr *TimeRange) ParseTo() (time.Time, error) {
 	if tr.To == "now" {
-		return tr.Now, nil
+		return tr.now, nil
 	} else if strings.HasPrefix(tr.To, "now-") {
 		withoutNow := strings.Replace(tr.To, "now-", "", 1)
 
@@ -79,7 +79,7 @@ func (tr *TimeRange) ParseTo() (time.Time, error) {
 			return time.Time{}, nil
 		}
 
-		return tr.Now.Add(diff), nil
+		return tr.now.Add(diff), nil
 	}
 
 	if res, ok := tryParseUnixMsEpoch(tr.To); ok {

pkg/tsdb/time_range_test.go

@@ -16,7 +16,7 @@ func TestTimeRange(t *testing.T) {
 			tr := TimeRange{
 				From: "5m",
 				To:   "now",
-				Now:  now,
+				now:  now,
 			}
 
 			Convey("5m ago ", func() {
@@ -39,7 +39,7 @@ func TestTimeRange(t *testing.T) {
 			tr := TimeRange{
 				From: "5h",
 				To:   "now-10m",
-				Now:  now,
+				now:  now,
 			}
 
 			Convey("5h ago ", func() {
@@ -65,7 +65,7 @@ func TestTimeRange(t *testing.T) {
 			tr := TimeRange{
 				From: "1474973725473",
 				To:   "1474975757930",
-				Now:  now,
+				now:  now,
 			}
 
 			res, err := tr.ParseFrom()
@@ -82,7 +82,7 @@ func TestTimeRange(t *testing.T) {
 			tr := TimeRange{
 				From: "asdf",
 				To:   "asdf",
-				Now:  now,
+				now:  now,
 			}
 
 			_, err = tr.ParseFrom()