work on scheduled queries for lossy counting

config/morgoth_conf.go

@@ -6,7 +6,7 @@ import (
 )
 
 type MorgothConf struct {
-	MetaDir string      `yaml:"meta_dir" validate:"nonzero" default:"meta"`
+	MetaDir string `yaml:"meta_dir" validate:"nonzero" default:"meta"`
 }
 
 func (self MorgothConf) Validate() error {

detector/mgof/config.go

@@ -7,13 +7,13 @@ import (
 )
 
 type MGOFConf struct {
-	Min             float64 `yaml:"min"`
-	Max             float64 `yaml:"max"`
+	Min float64 `yaml:"min"`
+	Max float64 `yaml:"max"`
 	// A smaller number means that the algorithm is more distinguishing
-	NullConfidence  uint    `yaml:"null_confidence" validate:"min=1,max=10" default:"10"`
-	NBins           uint    `yaml:"nbins" validate:"nonzero" default:"15"`
-	NormalCount     uint    `yaml:"normal_count" validate:"nonzero" default:"2"`
-	MaxFingerprints uint    `yaml:"max_fingerprints" validate:"nonzero" default:"20"`
+	NullConfidence  uint `yaml:"null_confidence" validate:"min=1,max=10" default:"10"`
+	NBins           uint `yaml:"nbins" validate:"nonzero" default:"15"`
+	NormalCount     uint `yaml:"normal_count" validate:"nonzero" default:"2"`
+	MaxFingerprints uint `yaml:"max_fingerprints" validate:"nonzero" default:"20"`
 }
 
 func (self *MGOFConf) Default() {

engine/influxdb/query.go

@@ -0,0 +1,73 @@
+package influxdb
+
+import (
+	"fmt"
+	"errors"
+	"github.com/influxdb/influxdb/influxql"
+	"time"
+)
+
+type Query struct {
+	statement *influxql.SelectStatement
+	startTL *influxql.TimeLiteral
+	stopTL *influxql.TimeLiteral
+}
+
+func NewQuery(quertStr string) (*Query, error) {
+
+	s, err := influxql.ParseStatement(quertStr)
+	if err != nil {
+		return nil, err
+	}
+	stmt, ok := s.(*influxql.SelectStatement)
+	if !ok {
+		return nil, errors.New(fmt.Sprintf("Query must be a select statement '%s'", quertStr))
+	}
+
+
+	//Add New BinaryExpr for time clause
+	startTL := &influxql.TimeLiteral{}
+	startExpr := &influxql.BinaryExpr{
+		Op:  influxql.GT,
+		LHS: &influxql.VarRef{Val: "time"},
+		RHS: startTL,
+	}
+
+	stopTL := &influxql.TimeLiteral{}
+	stopExpr := &influxql.BinaryExpr{
+		Op:  influxql.LT,
+		LHS: &influxql.VarRef{Val: "time"},
+		RHS: stopTL,
+	}
+	if stmt.Condition != nil {
+		stmt.Condition = &influxql.BinaryExpr{
+			Op:  influxql.AND,
+			LHS: stmt.Condition,
+			RHS: &influxql.BinaryExpr{
+				Op:  influxql.AND,
+				LHS: startExpr,
+				RHS: stopExpr,
+			},
+		}
+	} else {
+		stmt.Condition = &influxql.BinaryExpr{
+			Op:  influxql.AND,
+			LHS: startExpr,
+			RHS: stopExpr,
+		}
+	}
+
+	return &Query{
+		statement: stmt,
+		startTL: startTL,
+		stopTL:  stopTL,
+	}, nil
+}
+
+func (self *Query) QueryForTimeRange(start, stop time.Time) string {
+
+	self.startTL.Val = start
+	self.stopTL.Val = stop
+
+	return self.statement.String()
+}

engine/influxdb/query_test.go

@@ -0,0 +1,39 @@
+package influxdb_test
+
+import (
+	"fmt"
+	"github.com/nvcook42/morgoth/Godeps/_workspace/src/github.com/stretchr/testify/assert"
+	"github.com/nvcook42/morgoth/engine/influxdb"
+	"testing"
+	"time"
+)
+
+func TestScheduleQueryShouldAddTimeWhereCondition(t *testing.T) {
+	assert := assert.New(t)
+
+	type testCase struct {
+		quertStr string
+		start time.Time
+		stop time.Time
+		expectedStr string
+	}
+
+	testCases := []testCase{
+		testCase{
+			quertStr: "SELECT value FROM kbps",
+			start: time.Unix(1433378783, 0),
+			stop: time.Unix(1433379783, 0),
+			expectedStr: "SELECT value FROM kbps WHERE time > \"2015-06-04 00:46:23\" AND time < \"2015-06-04 01:03:03\"",
+		},
+	}
+
+	test := func(queryStr, start, stop, expectedStr) {
+		q, err := influxdb.NewQuery(quertStr)
+		assert.Nil(err)
+		assert.Equal(
+			expectedStr,
+			q.QueryForTimeRange(start, stop),
+		)
+	}
+
+}

engine/query.go

@@ -0,0 +1,10 @@
+package engine
+
+import (
+	"time"
+)
+
+type Query interface {
+	QueryForTimeRange(start, stop time.Time)
+}
+

fitting/grafana/config_test.go

@@ -4,8 +4,8 @@ import (
 	"github.com/nvcook42/morgoth/Godeps/_workspace/src/github.com/golang/glog"
 	"github.com/nvcook42/morgoth/Godeps/_workspace/src/github.com/stretchr/testify/assert"
 	"github.com/nvcook42/morgoth/Godeps/_workspace/src/gopkg.in/yaml.v2"
-	"github.com/nvcook42/morgoth/fitting/grafana"
 	_ "github.com/nvcook42/morgoth/engine/list"
+	"github.com/nvcook42/morgoth/fitting/grafana"
 	"testing"
 )
 
@@ -28,9 +28,9 @@ func TestGrafanaConfShouldValidate(t *testing.T) {
 	assert := assert.New(t)
 
 	rc := grafana.GrafanaConf{
-		URL: "http://localhost/grafana.tar.gz",
-		Port: 42,
-		Dir: "/tmp/grafana",
+		URL:       "http://localhost/grafana.tar.gz",
+		Port:      42,
+		Dir:       "/tmp/grafana",
 		GrafanaDB: "grafanadb",
 	}
 

fitting/grafana/grafana.go

@@ -4,12 +4,12 @@ import (
 	"fmt"
 	"github.com/nvcook42/morgoth/Godeps/_workspace/src/github.com/golang/glog"
 	app "github.com/nvcook42/morgoth/app/types"
+	"io"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
 	"path"
-	"io"
-	"io/ioutil"
 )
 
 type GrafanaFitting struct {

fitting/list/list.go

@@ -3,7 +3,7 @@
 package list
 
 import (
-	_ "github.com/nvcook42/morgoth/fitting/graphite"
 	_ "github.com/nvcook42/morgoth/fitting/grafana"
+	_ "github.com/nvcook42/morgoth/fitting/graphite"
 	_ "github.com/nvcook42/morgoth/fitting/rest"
 )