Add now() function

Add now() built-in, tick stateless function that returns a time.Time

CHANGELOG.md

@@ -12,6 +12,7 @@
     The breaking change is that the Combine and Flatten nodes previously, but erroneously, operated across batch boundaries; this has been fixed.
 - [#1497](https://github.com/influxdata/kapacitor/pull/1497): Add support for Docker Swarm autoscaling services.
 - [#1485](https://github.com/influxdata/kapacitor/issues/1485): Add bools field types to UDFs.
+- [#1549](https://github.com/influxdata/kapacitor/issues/1549): Add stateless now() function to get the current local time.
 - [#1545](https://github.com/influxdata/kapacitor/pull/1545): Add support for timeout, tags and service template in the Alerta AlertNode
 - [#1568](https://github.com/influxdata/kapacitor/issues/1568): Add support for custom HTTP Post bodies via a template system.
 - [#1569](https://github.com/influxdata/kapacitor/issues/1569): Add support for add the HTTP status code as a field when using httpPost

integrations/data/TestStream_EvalNow.srpl

@@ -0,0 +1,3 @@
+dbname
+rpname
+account,owner=ownerA expiration=315533000000000000i 0000000001

integrations/data/TestStream_LambdaNow.srpl

@@ -0,0 +1,9 @@
+dbname
+rpname
+account,owner=ownerA expiration=315533000000000000i 0000000001
+dbname
+rpname
+account,owner=ownerB expiration=4102440000000000000i 0000000001
+dbname
+rpname
+account,owner=ownerC expiration=656419000000000000i 0000000001

integrations/streamer_test.go

@@ -9037,6 +9037,83 @@ stream
 	}
 }
 
+func TestStream_LambdaNow(t *testing.T) {
+	var script = `
+stream
+	|from()
+		.measurement('account')
+	|where(lambda: "expiration" < unixNano(now()))
+	|groupBy('owner')
+	|httpOut('TestStream_LambdaNow')
+`
+
+	expectedOutput := models.Result{
+		Series: models.Rows{
+			{
+				Name:    "account",
+				Tags:    map[string]string{"owner": "ownerA"},
+				Columns: []string{"time", "expiration"},
+				Values: [][]interface{}{
+					{
+						time.Date(1971, 1, 1, 0, 0, 0, 0, time.UTC),
+						float64(3.15533e+17), // 1980-01-01 00:00:00 (Unix ns timestamp)
+
+						// we expect "expiration" to be float64 and not int64 (even with input data consisting of ints)
+						// because httpOut uses JSON as serialization format for the results,
+						// resulting in the integers becoming floats
+					},
+				},
+			},
+
+			// the point with expiration = 4102440000000000000 should not be in the results
+			// as it represents a date past now (2100-01-01 00:00:00)
+
+			{
+				Name:    "account",
+				Tags:    map[string]string{"owner": "ownerC"},
+				Columns: []string{"time", "expiration"},
+				Values: [][]interface{}{
+					{
+						time.Date(1971, 1, 1, 0, 0, 0, 0, time.UTC),
+						float64(6.56419e+17), // 1990-10-20 10:42:42
+					},
+				},
+			},
+		},
+	}
+
+	testStreamerWithOutput(t, "TestStream_LambdaNow", script, time.Second, expectedOutput, false, nil)
+}
+
+func TestStream_EvalNow(t *testing.T) {
+	var script = `
+stream
+	|from()
+		.measurement('account')
+	|eval(lambda: year(now()))
+		.as('currentYear')
+	|httpOut('TestStream_EvalNow')
+`
+
+	expectedOutput := models.Result{
+		Series: models.Rows{
+			{
+				Name:    "account",
+				Tags:    map[string]string{"owner": "ownerA"},
+				Columns: []string{"time", "currentYear"},
+				Values: [][]interface{}{
+					{
+						time.Date(1971, 1, 1, 0, 0, 0, 0, time.UTC),
+						float64(time.Now().Year()),
+					},
+				},
+			},
+		},
+	}
+
+	testStreamerWithOutput(t, "TestStream_EvalNow", script, time.Second, expectedOutput, false, nil)
+}
+
 func TestStream_Autoscale(t *testing.T) {
 	testCases := map[string]struct {
 		script           string

tick/stateful/functions.go

@@ -225,6 +225,7 @@ func init() {
 	statelessFuncs["day"] = day{}
 	statelessFuncs["month"] = month{}
 	statelessFuncs["year"] = year{}
+	statelessFuncs["now"] = now{}
 
 	// Humanize functions
 	statelessFuncs["humanBytes"] = humanBytes{}
@@ -1354,6 +1355,34 @@ func (year) Signature() map[Domain]ast.ValueType {
 	return timeFuncSignature
 }
 
+var nowFuncSignature = map[Domain]ast.ValueType{}
+
+// Initialize Now function signature
+func init() {
+	d := Domain{}
+	nowFuncSignature[d] = ast.TTime
+}
+
+type now struct {
+}
+
+func (now) Reset() {
+}
+
+// Return the current local time.
+func (now) Call(args ...interface{}) (v interface{}, err error) {
+	if len(args) != 0 {
+		return 0, errors.New("now expects exactly zero argument")
+	}
+	v = time.Now()
+
+	return
+}
+
+func (now) Signature() map[Domain]ast.ValueType {
+	return nowFuncSignature
+}
+
 type humanBytes struct {
 }