From 0379eb6c0fc041933dfc7b8a3fdad406ad956559 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Du=C5=A1an=20Borov=C4=8Danin?= Date: Sun, 18 Nov 2018 16:42:39 +0100 Subject: [PATCH] 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 * 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 * 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 * Use channel and publisher as tag keys Move all the other Message fields to the field keys. Signed-off-by: Dusan Borovcanin --- readers/cassandra/messages_test.go | 3 +++ readers/influxdb/messages.go | 30 +++++++++++++++++-------- readers/influxdb/messages_test.go | 6 +++-- readers/mongodb/messages_test.go | 4 +++- writers/cassandra/messages_test.go | 3 +++ writers/influxdb/messages.go | 36 ++++++++++++++++-------------- writers/influxdb/messages_test.go | 3 ++- writers/mongodb/messages_test.go | 4 ++++ 8 files changed, 59 insertions(+), 30 deletions(-) diff --git a/readers/cassandra/messages_test.go b/readers/cassandra/messages_test.go index 7a97e4d8..da9f2fe5 100644 --- a/readers/cassandra/messages_test.go +++ b/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)) diff --git a/readers/influxdb/messages.go b/readers/influxdb/messages.go index 0bd9bdf4..d446a3fe 100644 --- a/readers/influxdb/messages.go +++ b/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) } diff --git a/readers/influxdb/messages_test.go b/readers/influxdb/messages_test.go index c569d032..8b0afd74 100644 --- a/readers/influxdb/messages_test.go +++ b/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)) } } diff --git a/readers/mongodb/messages_test.go b/readers/mongodb/messages_test.go index 6ed60dd9..6d2427da 100644 --- a/readers/mongodb/messages_test.go +++ b/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)) diff --git a/writers/cassandra/messages_test.go b/writers/cassandra/messages_test.go index 083be4e4..c6e3a409 100644 --- a/writers/cassandra/messages_test.go +++ b/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)) diff --git a/writers/influxdb/messages.go b/writers/influxdb/messages.go index bf63deb0..44b75b62 100644 --- a/writers/influxdb/messages.go +++ b/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, - "Publisher": publisher, - "Protocol": msg.Protocol, - "Name": msg.Name, - "Unit": msg.Unit, - "Link": msg.Link, - "Time": time, - "UpdateTime": update, + "channel": channel, + "publisher": publisher, } } 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 diff --git a/writers/influxdb/messages_test.go b/writers/influxdb/messages_test.go index 568d152a..7c793e0b 100644 --- a/writers/influxdb/messages_test.go +++ b/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)) diff --git a/writers/mongodb/messages_test.go b/writers/mongodb/messages_test.go index 70a73268..416ccdf5 100644 --- a/writers/mongodb/messages_test.go +++ b/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) }