MF-454 - Use message Time field as a time for InfluxDB points (#455)

* Use message time as Point time in InfluxDB

Signed-off-by: Dusan Borovcanin <dusan.borovcanin@mainflux.com>

* Use actual message time

Update all Reader and Writer services to use time from the message
instead of time given from the corrseponding Writer service.

Signed-off-by: Dusan Borovcanin <dusan.borovcanin@mainflux.com>

* Remove message time check

Messages time less than 2**28 represent time relative to the current time so Writer used to convert this to the correct value, i.e. msg.Time += time.Now(). However, this step is optional and should really be a part of the app on top of Mainflux or could be introduced with minor changes in Normalizer, Reader or Writer services, so there is no need for this to be supported out of box.

Signed-off-by: Dusan Borovcanin <dusan.borovcanin@mainflux.com>

* Use channel and publisher as tag keys

Move all the other Message fields to the field keys.

Signed-off-by: Dusan Borovcanin <dusan.borovcanin@mainflux.com>

readers/cassandra/messages_test.go

@@ -10,6 +10,7 @@ package cassandra_test
 import (
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/mainflux/mainflux"
 	readers "github.com/mainflux/mainflux/readers/cassandra"
@@ -41,6 +42,7 @@ func TestReadAll(t *testing.T) {
 	writer := writers.New(session)
 
 	messages := []mainflux.Message{}
+	now := time.Now().Unix()
 	for i := 0; i < msgsNum; i++ {
 		// Mix possible values as well as value sum.
 		count := i % valueFields
@@ -58,6 +60,7 @@ func TestReadAll(t *testing.T) {
 		case 5:
 			msg.ValueSum = &mainflux.SumValue{Value: 45}
 		}
+		msg.Time = float64(now + int64(i))
 
 		err := writer.Save(msg)
 		require.Nil(t, err, fmt.Sprintf("failed to store message to Cassandra: %s", err))

readers/influxdb/messages.go

@@ -6,6 +6,7 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/mainflux/mainflux/readers"
 
@@ -22,9 +23,6 @@ type influxRepository struct {
 	client   influxdata.Client
 }
 
-type fields map[string]interface{}
-type tags map[string]string
-
 // New returns new InfluxDB reader.
 func New(client influxdata.Client, database string) (readers.MessageRepository, error) {
 	return &influxRepository{database, client}, nil
@@ -35,7 +33,7 @@ func (repo *influxRepository) ReadAll(chanID, offset, limit uint64) []mainflux.M
 		limit = maxLimit
 	}
 
-	cmd := fmt.Sprintf(`SELECT * from messages WHERE Channel='%d' LIMIT %d OFFSET %d`, chanID, limit, offset)
+	cmd := fmt.Sprintf(`SELECT * from messages WHERE channel='%d' LIMIT %d OFFSET %d`, chanID, limit, offset)
 	q := influxdata.Query{
 		Command:  cmd,
 		Database: repo.database,
@@ -51,6 +49,7 @@ func (repo *influxRepository) ReadAll(chanID, offset, limit uint64) []mainflux.M
 	if len(resp.Results) < 1 || len(resp.Results[0].Series) < 1 {
 		return ret
 	}
+
 	result := resp.Results[0].Series[0]
 	for _, v := range result.Values {
 		ret = append(ret, parseMessage(result.Columns, v))
@@ -63,17 +62,19 @@ func (repo *influxRepository) ReadAll(chanID, offset, limit uint64) []mainflux.M
 // results in form of rows and columns, this obscure message conversion is needed
 // to return actual []mainflux.Message from the query result.
 func parseValues(value interface{}, name string, msg *mainflux.Message) {
-	if name == "ValueSum" && value != nil {
+	if name == "valueSum" && value != nil {
 		if sum, ok := value.(json.Number); ok {
 			valSum, err := sum.Float64()
 			if err != nil {
 				return
 			}
+
 			msg.ValueSum = &mainflux.SumValue{Value: valSum}
 		}
 		return
 	}
-	if strings.HasSuffix(name, "Value") {
+
+	if strings.HasSuffix(strings.ToLower(name), "value") {
 		switch value.(type) {
 		case bool:
 			msg.Value = &mainflux.Message_BoolValue{value.(bool)}
@@ -85,12 +86,12 @@ func parseValues(value interface{}, name string, msg *mainflux.Message) {
 
 			msg.Value = &mainflux.Message_FloatValue{num}
 		case string:
-			if strings.HasPrefix(name, "String") {
+			if strings.HasPrefix(name, "string") {
 				msg.Value = &mainflux.Message_StringValue{value.(string)}
 				return
 			}
 
-			if strings.HasPrefix(name, "Data") {
+			if strings.HasPrefix(name, "data") {
 				msg.Value = &mainflux.Message_DataValue{value.(string)}
 			}
 		}
@@ -102,7 +103,7 @@ func parseMessage(names []string, fields []interface{}) mainflux.Message {
 	v := reflect.ValueOf(&m).Elem()
 	for i, name := range names {
 		parseValues(fields[i], name, &m)
-		msgField := v.FieldByName(name)
+		msgField := v.FieldByName(strings.Title(name))
 		if !msgField.IsValid() {
 			continue
 		}
@@ -117,6 +118,17 @@ func parseMessage(names []string, fields []interface{}) mainflux.Message {
 			u, _ := strconv.ParseUint(fields[i].(string), 10, 64)
 			msgField.SetUint(u)
 		case float64:
+			if name == "time" {
+				t, err := time.Parse(time.RFC3339, fields[i].(string))
+				if err != nil {
+					continue
+				}
+
+				v := float64(t.Unix())
+				msgField.SetFloat(v)
+				continue
+			}
+
 			val, _ := strconv.ParseFloat(fields[i].(string), 64)
 			msgField.SetFloat(val)
 		}

readers/influxdb/messages_test.go

@@ -42,7 +42,7 @@ var (
 		Value:      &mainflux.Message_FloatValue{5},
 		ValueSum:   &mainflux.SumValue{Value: 45},
 		Time:       123456,
-		UpdateTime: 1234567,
+		UpdateTime: 1234,
 		Link:       "link",
 	}
 )
@@ -55,6 +55,7 @@ func TestReadAll(t *testing.T) {
 	require.Nil(t, err, fmt.Sprintf("Creating new InfluxDB writer expected to succeed: %s.\n", err))
 
 	messages := []mainflux.Message{}
+	now := time.Now().Unix()
 	for i := 0; i < msgsNum; i++ {
 		// Mix possible values as well as value sum.
 		count := i % valueFields
@@ -72,6 +73,7 @@ func TestReadAll(t *testing.T) {
 		case 5:
 			msg.ValueSum = &mainflux.SumValue{Value: 45}
 		}
+		msg.Time = float64(now + int64(i))
 
 		err := writer.Save(msg)
 		require.Nil(t, err, fmt.Sprintf("failed to store message to InfluxDB: %s", err))
@@ -115,6 +117,6 @@ func TestReadAll(t *testing.T) {
 
 	for desc, tc := range cases {
 		result := reader.ReadAll(tc.chanID, tc.offset, tc.limit)
-		assert.ElementsMatch(t, tc.messages, result, fmt.Sprintf("%s: expected %v got %v", desc, tc.messages, result))
+		assert.ElementsMatch(t, tc.messages, result, fmt.Sprintf("%s: expected: %v \n-------------\n got: %v", desc, tc.messages, result))
 	}
 }

readers/mongodb/messages_test.go

@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"os"
 	"testing"
+	"time"
 
 	readers "github.com/mainflux/mainflux/readers/mongodb"
 	writers "github.com/mainflux/mainflux/writers/mongodb"
@@ -48,10 +49,10 @@ func TestReadAll(t *testing.T) {
 	require.Nil(t, err, fmt.Sprintf("Creating new MongoDB client expected to succeed: %s.\n", err))
 
 	db := client.Database(testDB)
-
 	writer := writers.New(db)
 
 	messages := []mainflux.Message{}
+	now := time.Now().Unix()
 	for i := 0; i < msgsNum; i++ {
 		// Mix possible values as well as value sum.
 		count := i % valueFields
@@ -69,6 +70,7 @@ func TestReadAll(t *testing.T) {
 		case 5:
 			msg.ValueSum = &mainflux.SumValue{Value: 45}
 		}
+		msg.Time = float64(now + int64(i))
 
 		err := writer.Save(msg)
 		require.Nil(t, err, fmt.Sprintf("failed to store message to Cassandra: %s", err))

writers/cassandra/messages_test.go

@@ -10,6 +10,7 @@ package cassandra_test
 import (
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/mainflux/mainflux"
 	"github.com/mainflux/mainflux/writers/cassandra"
@@ -35,6 +36,7 @@ func TestSave(t *testing.T) {
 	require.Nil(t, err, fmt.Sprintf("failed to connect to Cassandra: %s", err))
 
 	repo := cassandra.New(session)
+	now := time.Now().Unix()
 	for i := 0; i < msgsNum; i++ {
 		// Mix possible values as well as value sum.
 		count := i % valueFields
@@ -52,6 +54,7 @@ func TestSave(t *testing.T) {
 		case 5:
 			msg.ValueSum = &mainflux.SumValue{Value: 45}
 		}
+		msg.Time = float64(now + int64(i))
 
 		err = repo.Save(msg)
 		assert.Nil(t, err, fmt.Sprintf("expected no error, got %s", err))

writers/influxdb/messages.go

@@ -112,7 +112,9 @@ func (repo *influxRepo) savePoint(point *influxdata.Point) error {
 
 func (repo *influxRepo) Save(msg mainflux.Message) error {
 	tags, fields := repo.tagsOf(&msg), repo.fieldsOf(&msg)
-	pt, err := influxdata.NewPoint(pointName, tags, fields, time.Now())
+	t := time.Unix(int64(msg.Time), 0)
+
+	pt, err := influxdata.NewPoint(pointName, tags, fields, t)
 	if err != nil {
 		return err
 	}
@@ -121,38 +123,38 @@ func (repo *influxRepo) Save(msg mainflux.Message) error {
 }
 
 func (repo *influxRepo) tagsOf(msg *mainflux.Message) tags {
-	time := strconv.FormatFloat(msg.Time, 'f', -1, 64)
-	update := strconv.FormatFloat(msg.UpdateTime, 'f', -1, 64)
 	channel := strconv.FormatUint(msg.Channel, 10)
 	publisher := strconv.FormatUint(msg.Publisher, 10)
 
 	return tags{
-		"Channel":    channel,
+		"channel":   channel,
-		"Publisher":  publisher,
+		"publisher": publisher,
-		"Protocol":   msg.Protocol,
-		"Name":       msg.Name,
-		"Unit":       msg.Unit,
-		"Link":       msg.Link,
-		"Time":       time,
-		"UpdateTime": update,
 	}
 }
 
 func (repo *influxRepo) fieldsOf(msg *mainflux.Message) fields {
-	ret := fields{}
+	updateTime := strconv.FormatFloat(msg.UpdateTime, 'f', -1, 64)
+	ret := fields{
+		"protocol":   msg.Protocol,
+		"name":       msg.Name,
+		"unit":       msg.Unit,
+		"link":       msg.Link,
+		"updateTime": updateTime,
+	}
+
 	switch msg.Value.(type) {
 	case *mainflux.Message_FloatValue:
-		ret["Value"] = msg.GetFloatValue()
+		ret["value"] = msg.GetFloatValue()
 	case *mainflux.Message_StringValue:
-		ret["StringValue"] = msg.GetStringValue()
+		ret["stringValue"] = msg.GetStringValue()
 	case *mainflux.Message_DataValue:
-		ret["DataValue"] = msg.GetDataValue()
+		ret["dataValue"] = msg.GetDataValue()
 	case *mainflux.Message_BoolValue:
-		ret["BoolValue"] = msg.GetBoolValue()
+		ret["boolValue"] = msg.GetBoolValue()
 	}
 
 	if msg.ValueSum != nil {
-		ret["ValueSum"] = msg.GetValueSum().GetValue()
+		ret["valueSum"] = msg.GetValueSum().GetValue()
 	}
 
 	return ret

writers/influxdb/messages_test.go

@@ -47,7 +47,6 @@ var (
 		Unit:       "km",
 		Value:      &mainflux.Message_FloatValue{24},
 		ValueSum:   &mainflux.SumValue{Value: 22},
-		Time:       13451312,
 		UpdateTime: 5456565466,
 		Link:       "link",
 	}
@@ -145,6 +144,7 @@ func TestSave(t *testing.T) {
 		row, err := queryDB(dropMsgs)
 		require.Nil(t, err, fmt.Sprintf("Cleaning data from InfluxDB expected to succeed: %s.\n", err))
 
+		now := time.Now().Unix()
 		for i := 0; i < tc.msgsNum; i++ {
 			// Mix possible values as well as value sum.
 			count := i % valueFields
@@ -162,6 +162,7 @@ func TestSave(t *testing.T) {
 			case 5:
 				msg.ValueSum = &mainflux.SumValue{Value: 45}
 			}
+			msg.Time = float64(now + int64(i))
 
 			err := tc.repo.Save(msg)
 			assert.Nil(t, err, fmt.Sprintf("Save operation expected to succeed: %s.\n", err))

writers/mongodb/messages_test.go

@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -53,6 +54,8 @@ func TestSave(t *testing.T) {
 
 	db := client.Database(testDB)
 	repo := mongodb.New(db)
+
+	now := time.Now().Unix()
 	for i := 0; i < msgsNum; i++ {
 		// Mix possible values as well as value sum.
 		count := i % valueFields
@@ -70,6 +73,7 @@ func TestSave(t *testing.T) {
 		case 5:
 			msg.ValueSum = &mainflux.SumValue{Value: 45}
 		}
+		msg.Time = float64(now + int64(i))
 
 		err = repo.Save(msg)
 	}