diff --git a/Makefile b/Makefile index 49a8459..e2224a8 100644 --- a/Makefile +++ b/Makefile @@ -18,9 +18,20 @@ lint: @echo "lint" @golangci-lint --exclude=SA1019 run ./... --deadline=5m -test: +test: postgres go test -v -failfast ./... generate: go generate ./telemetry/sql - \ No newline at end of file + +run: build postgres + ./build/server --data-source-name=postgres://telemetry:newPassword@127.0.0.1:5432/telemetrydb?sslmode=disable + +postgres: + docker inspect telemetry-postgres > /dev/null ||\ + docker run --name telemetry-postgres -e POSTGRES_USER=telemetry -e POSTGRES_PASSWORD=newPassword -e POSTGRES_DB=telemetrydb -p 5432:5432 -d postgres &&\ + sleep 3 + +postgres-clean: + docker stop telemetry-postgres &&\ + docker rm telemetry-postgres diff --git a/cmd/server/main.go b/cmd/server/main.go index ff2ad9e..29aaec4 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -6,32 +6,41 @@ import ( "time" "github.com/status-im/dev-telemetry/telemetry" + "go.uber.org/zap" "github.com/robfig/cron/v3" ) func main() { + logger, err := zap.NewDevelopment() + if err != nil { + log.Fatal(err) + } port := flag.Int("port", 8080, "Port number") dataSourceName := flag.String("data-source-name", "", "DB URL") flag.Parse() - db := telemetry.OpenDb(*dataSourceName) + db := telemetry.OpenDb(*dataSourceName, logger) defer db.Close() - aggregator := telemetry.NewAggregator(db) + aggregator, err := telemetry.NewAggregator(db, logger) + if err != nil { + logger.Fatal("Error creating aggregator", zap.Error(err)) + } + c := cron.New() - _, err := c.AddFunc("0 * * * *", func() { + _, err = c.AddFunc("0 * * * *", func() { aggregator.Run(time.Hour) }) if err != nil { - log.Fatalf("Error adding cron job: %v", err) + logger.Fatal("Error adding cron job", zap.Error(err)) } c.Start() defer c.Stop() - server := telemetry.NewServer(db) + server := telemetry.NewServer(db, logger) server.Start(*port) } diff --git a/go.mod b/go.mod index 111a8fa..1666c1d 100644 --- a/go.mod +++ b/go.mod @@ -3,10 +3,11 @@ module github.com/status-im/dev-telemetry go 1.15 require ( + github.com/go-bindata/go-bindata v3.1.2+incompatible // indirect github.com/golang-migrate/migrate/v4 v4.15.2 github.com/gorilla/mux v1.8.0 github.com/lib/pq v1.10.3 - github.com/mattn/go-sqlite3 v1.14.10 github.com/robfig/cron/v3 v3.0.1 - github.com/stretchr/testify v1.7.0 + github.com/stretchr/testify v1.8.1 + go.uber.org/zap v1.27.0 ) diff --git a/go.sum b/go.sum index 3a988e4..cc4b06e 100644 --- a/go.sum +++ b/go.sum @@ -408,6 +408,8 @@ github.com/garyburd/redigo v0.0.0-20150301180006-535138d7bcd7/go.mod h1:NR3MbYis github.com/getsentry/raven-go v0.2.0/go.mod h1:KungGk8q33+aIAZUIVWZDr2OfAEBsO49PX4NzFV5kcQ= github.com/ghodss/yaml v0.0.0-20150909031657-73d445a93680/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= +github.com/go-bindata/go-bindata v3.1.2+incompatible h1:5vjJMVhowQdPzjE1LdxyFF7YFTXg5IgGVW4gBr5IbvE= +github.com/go-bindata/go-bindata v3.1.2+incompatible/go.mod h1:xK8Dsgwmeed+BBsSy2XTopBn/8uK2HWuGSnA11C3Joo= github.com/go-fonts/dejavu v0.1.0/go.mod h1:4Wt4I4OU2Nq9asgDCteaAaWZOV24E+0/Pwo0gppep4g= github.com/go-fonts/latin-modern v0.2.0/go.mod h1:rQVLdDMK+mK1xscDwsqM5J8U2jrRa3T0ecnM9pNujks= github.com/go-fonts/liberation v0.1.1/go.mod h1:K6qoJYypsmfVjWg8KOVDQhLc8UDgIK2HYqyqAO9z7GY= @@ -792,7 +794,6 @@ github.com/mattn/go-shellwords v1.0.6/go.mod h1:3xCvwCdWdlDJUrvuMn7Wuy9eWs4pE8vq github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lLtQsUlTZDWQ8Y= github.com/mattn/go-sqlite3 v1.9.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc= github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU= -github.com/mattn/go-sqlite3 v1.14.10 h1:MLn+5bFRlWMGoSRmJour3CL1w/qL96mvipqpwQW/Sfk= github.com/mattn/go-sqlite3 v1.14.10/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4= @@ -1018,14 +1019,19 @@ github.com/stretchr/objx v0.0.0-20180129172003-8a3f7159479f/go.mod h1:HFkY916IF+ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/testify v0.0.0-20180303142811-b89eecf5ca5d/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw= github.com/syndtr/gocapability v0.0.0-20170704070218-db04d3cc01c8/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww= github.com/syndtr/gocapability v0.0.0-20180916011248-d98352740cb2/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww= @@ -1120,13 +1126,19 @@ go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw= go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A= go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU= go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= +go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA= go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo= +go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= +go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= golang.org/x/crypto v0.0.0-20171113213409-9f005a07e0d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20181009213950-7c1a557ab941/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= @@ -1728,8 +1740,9 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo= gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gorm.io/driver/postgres v1.0.8/go.mod h1:4eOzrI1MUfm6ObJU/UcmbXyiHSs8jSwH95G5P5dxcAg= gorm.io/gorm v1.20.12/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw= gorm.io/gorm v1.21.4/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw= diff --git a/telemetry/aggregator.go b/telemetry/aggregator.go index ca39963..7c3e2ab 100644 --- a/telemetry/aggregator.go +++ b/telemetry/aggregator.go @@ -2,23 +2,25 @@ package telemetry import ( "database/sql" - "fmt" - "log" "time" + + "go.uber.org/zap" ) type Aggregator struct { - DB *sql.DB + DB *sql.DB + logger zap.Logger } -func NewAggregator(db *sql.DB) *Aggregator { +func NewAggregator(db *sql.DB, logger *zap.Logger) (*Aggregator, error) { return &Aggregator{ - DB: db, - } + DB: db, + logger: *logger, + }, nil } func (a *Aggregator) Run(d time.Duration) { - log.Printf("started aggregator for %s\n", d) + a.logger.Info("started aggregator", zap.Duration("duration", d)) // Define the duration starts and end. // Allow a buffer of the duration to define the start and end. // This is to ensure we wait for people not being connected or if they received messages with delay @@ -29,7 +31,7 @@ func (a *Aggregator) Run(d time.Duration) { // Query all received message for a specific duration receivedMessages, err := queryReceivedMessagesBetween(a.DB, startsAt, endsAt) if err != nil { - log.Fatalf("could not query received message: %s", err) + a.logger.Fatal("could not query received message", zap.Error(err)) } // Group the received messages by chat id and key uid @@ -43,7 +45,7 @@ func (a *Aggregator) Run(d time.Duration) { receivedMessage.ChatID, ) if err != nil { - log.Fatalf("could not check message id: %s, because of %s", fmt.Sprint(receivedMessage.ID), err) + a.logger.Fatal("could not check message", zap.Int("messageID", receivedMessage.ID), zap.Error(err)) } if !ok { continue @@ -56,7 +58,7 @@ func (a *Aggregator) Run(d time.Duration) { } if len(groupedMessages) == 0 { - log.Println("no record found, finishing early") + a.logger.Info("no record found, finishing early") return } @@ -94,10 +96,10 @@ func (a *Aggregator) Run(d time.Duration) { } err := rma.put(a.DB) if err != nil { - log.Fatalf("could not store received message aggregated: %s", err) + a.logger.Fatal("could not store received message aggregated", zap.Error(err)) } } - log.Printf("stored %d chat id records", len(rChatID)) + a.logger.Sugar().Infof("stored %d chat id records", len(rChatID)) // Calculate the global reliability R = (R(0) + R(1)+ .... + R(n)) / len(Rch) rChatIDTotal := 0.0 @@ -114,7 +116,7 @@ func (a *Aggregator) Run(d time.Duration) { } err = rma.put(a.DB) if err != nil { - log.Fatalf("could not store received message aggregated: %s", err) + a.logger.Fatal("could not store received message aggregateds", zap.Error(err)) } - log.Printf("finished aggregator for %s\n", d) + a.logger.Info("finished aggregator", zap.Duration("duration", d)) } diff --git a/telemetry/aggregator_test.go b/telemetry/aggregator_test.go index 16ec1e3..b399cec 100644 --- a/telemetry/aggregator_test.go +++ b/telemetry/aggregator_test.go @@ -8,10 +8,11 @@ import ( "time" "github.com/stretchr/testify/require" + "go.uber.org/zap" ) func NewMock() *sql.DB { - db, err := sql.Open("postgres", "postgres://telemetry:newPassword@127.0.0.1:5432/telemetry_test") + db, err := sql.Open("postgres", "postgres://telemetry:newPassword@127.0.0.1:5432/telemetrydb?sslmode=disable") if err != nil { log.Fatalf("an error '%s' was not expected when opening a stub database connection", err) } @@ -36,14 +37,44 @@ func dropTables(db *sql.DB) { log.Fatalf("an error '%s' was not expected when dropping the table", err) } + _, err = db.Exec("DROP TABLE IF EXISTS receivedEnvelopes") + if err != nil { + log.Fatalf("an error '%s' was not expected when dropping the table", err) + } + + _, err = db.Exec("DROP TABLE IF EXISTS protocolStatsRate") + if err != nil { + log.Fatalf("an error '%s' was not expected when dropping the table", err) + } + + _, err = db.Exec("DROP TABLE IF EXISTS protocolStatsTotals") + if err != nil { + log.Fatalf("an error '%s' was not expected when dropping the table", err) + } + _, err = db.Exec("DROP TABLE IF EXISTS schema_migrations") if err != nil { log.Fatalf("an error '%s' was not expected when dropping the table", err) } - _, err = db.Exec("DROP TABLE IF EXISTS receivedEnvelopes") + _, err = db.Exec("DROP INDEX IF EXISTS receivedEnvelopes") if err != nil { - log.Fatalf("an error '%s' was not expected when dropping the table", err) + log.Fatalf("an error '%s' was not expected when dropping the index", err) + } + + _, err = db.Exec("DROP INDEX IF EXISTS receivedMessageAggregated_runAt") + if err != nil { + log.Fatalf("an error '%s' was not expected when dropping the index", err) + } + + _, err = db.Exec("DROP INDEX IF EXISTS protocolStatsRate_idx1") + if err != nil { + log.Fatalf("an error '%s' was not expected when dropping the index", err) + } + + _, err = db.Exec("DROP INDEX IF EXISTS protocolStatsTotals_idx1") + if err != nil { + log.Fatalf("an error '%s' was not expected when dropping the index", err) } db.Close() @@ -118,7 +149,10 @@ func TestRunAggregatorSimple(t *testing.T) { err = updateCreatedAt(db, m) require.NoError(t, err) - agg := NewAggregator(db) + logger, err := zap.NewDevelopment() + require.NoError(t, err) + agg, err := NewAggregator(db, logger) + require.NoError(t, err) agg.Run(time.Hour) @@ -213,7 +247,10 @@ func TestRunAggregatorSimpleWithMessageMissing(t *testing.T) { err = updateCreatedAt(db, m) require.NoError(t, err) - agg := NewAggregator(db) + logger, err := zap.NewDevelopment() + require.NoError(t, err) + agg, err := NewAggregator(db, logger) + require.NoError(t, err) agg.Run(time.Hour) diff --git a/telemetry/bindata.go b/telemetry/bindata.go index 0992487..4052746 100644 --- a/telemetry/bindata.go +++ b/telemetry/bindata.go @@ -1,19 +1,18 @@ -// Code generated by go-bindata. DO NOT EDIT. +// Code generated for package telemetry by go-bindata DO NOT EDIT. (@generated) // sources: -// 000001_message_type.up.sql (66B) -// 000002_bandwidth_protocol.up.sql (719B) -// 000003_index_truncate.up.sql (598B) -// 000004_envelope.table.up.sql (531B) -// doc.go (73B) - +// 000001_message_type.up.sql +// 000002_bandwidth_protocol.up.sql +// 000003_index_truncate.up.sql +// 000004_envelope.table.up.sql +// doc.go package telemetry import ( "bytes" "compress/gzip" - "crypto/sha256" "fmt" "io" + "io/ioutil" "os" "path/filepath" "strings" @@ -23,7 +22,7 @@ import ( func bindataRead(data []byte, name string) ([]byte, error) { gz, err := gzip.NewReader(bytes.NewBuffer(data)) if err != nil { - return nil, fmt.Errorf("read %q: %w", name, err) + return nil, fmt.Errorf("Read %q: %v", name, err) } var buf bytes.Buffer @@ -31,7 +30,7 @@ func bindataRead(data []byte, name string) ([]byte, error) { clErr := gz.Close() if err != nil { - return nil, fmt.Errorf("read %q: %w", name, err) + return nil, fmt.Errorf("Read %q: %v", name, err) } if clErr != nil { return nil, err @@ -41,9 +40,8 @@ func bindataRead(data []byte, name string) ([]byte, error) { } type asset struct { - bytes []byte - info os.FileInfo - digest [sha256.Size]byte + bytes []byte + info os.FileInfo } type bindataFileInfo struct { @@ -53,21 +51,32 @@ type bindataFileInfo struct { modTime time.Time } +// Name return file name func (fi bindataFileInfo) Name() string { return fi.name } + +// Size return file size func (fi bindataFileInfo) Size() int64 { return fi.size } + +// Mode return file mode func (fi bindataFileInfo) Mode() os.FileMode { return fi.mode } + +// Mode return file modify time func (fi bindataFileInfo) ModTime() time.Time { return fi.modTime } + +// IsDir return file whether a directory func (fi bindataFileInfo) IsDir() bool { - return false + return fi.mode&os.ModeDir != 0 } + +// Sys return file is sys mode func (fi bindataFileInfo) Sys() interface{} { return nil } @@ -87,8 +96,8 @@ func _000001_message_typeUpSql() (*asset, error) { return nil, err } - info := bindataFileInfo{name: "000001_message_type.up.sql", size: 66, mode: os.FileMode(0644), modTime: time.Unix(1697552711, 0)} - a := &asset{bytes: bytes, info: info, digest: [32]uint8{0xe2, 0x43, 0xcc, 0xef, 0xad, 0x5f, 0x44, 0x58, 0x8d, 0x47, 0x70, 0x5d, 0x23, 0x30, 0xe2, 0x1f, 0xdb, 0x4d, 0xad, 0x6e, 0xd9, 0xe7, 0x50, 0x19, 0x43, 0x1c, 0x37, 0x57, 0xea, 0xc6, 0x57, 0xab}} + info := bindataFileInfo{name: "000001_message_type.up.sql", size: 66, mode: os.FileMode(436), modTime: time.Unix(1715855770, 0)} + a := &asset{bytes: bytes, info: info} return a, nil } @@ -107,8 +116,8 @@ func _000002_bandwidth_protocolUpSql() (*asset, error) { return nil, err } - info := bindataFileInfo{name: "000002_bandwidth_protocol.up.sql", size: 719, mode: os.FileMode(0644), modTime: time.Unix(1697552711, 0)} - a := &asset{bytes: bytes, info: info, digest: [32]uint8{0xfe, 0x83, 0x69, 0xab, 0x3e, 0xf5, 0x8d, 0x44, 0xb2, 0x6e, 0x52, 0x8d, 0x27, 0xe8, 0x95, 0x28, 0x3c, 0xea, 0x29, 0x93, 0x6d, 0xa3, 0x10, 0xde, 0x9b, 0xc8, 0xa6, 0xb9, 0x80, 0xa1, 0x3, 0x6f}} + info := bindataFileInfo{name: "000002_bandwidth_protocol.up.sql", size: 719, mode: os.FileMode(436), modTime: time.Unix(1715855770, 0)} + a := &asset{bytes: bytes, info: info} return a, nil } @@ -127,8 +136,8 @@ func _000003_index_truncateUpSql() (*asset, error) { return nil, err } - info := bindataFileInfo{name: "000003_index_truncate.up.sql", size: 598, mode: os.FileMode(0644), modTime: time.Unix(1698325445, 0)} - a := &asset{bytes: bytes, info: info, digest: [32]uint8{0xcf, 0x8, 0x4, 0x47, 0xc8, 0x65, 0x38, 0x79, 0x3e, 0x37, 0xec, 0x4e, 0x1a, 0x24, 0x50, 0x3c, 0x1c, 0x75, 0xe8, 0x3b, 0x2, 0x62, 0x2, 0x52, 0x50, 0xff, 0x4a, 0x8f, 0x9d, 0x71, 0x79, 0xf6}} + info := bindataFileInfo{name: "000003_index_truncate.up.sql", size: 598, mode: os.FileMode(436), modTime: time.Unix(1716989343, 0)} + a := &asset{bytes: bytes, info: info} return a, nil } @@ -147,8 +156,8 @@ func _000004_envelopeTableUpSql() (*asset, error) { return nil, err } - info := bindataFileInfo{name: "000004_envelope.table.up.sql", size: 531, mode: os.FileMode(0644), modTime: time.Unix(1698655313, 0)} - a := &asset{bytes: bytes, info: info, digest: [32]uint8{0x32, 0xee, 0x49, 0xa0, 0x48, 0x2b, 0x8b, 0xe8, 0xd3, 0x6a, 0xae, 0x7f, 0x62, 0x65, 0x8a, 0x45, 0xbb, 0x8a, 0xee, 0xcd, 0x13, 0xde, 0xd6, 0x33, 0xe2, 0x3f, 0x32, 0xff, 0xfe, 0xf4, 0xda, 0xe7}} + info := bindataFileInfo{name: "000004_envelope.table.up.sql", size: 531, mode: os.FileMode(436), modTime: time.Unix(1715855770, 0)} + a := &asset{bytes: bytes, info: info} return a, nil } @@ -167,8 +176,8 @@ func docGo() (*asset, error) { return nil, err } - info := bindataFileInfo{name: "doc.go", size: 73, mode: os.FileMode(0644), modTime: time.Unix(1697552711, 0)} - a := &asset{bytes: bytes, info: info, digest: [32]uint8{0xae, 0x4f, 0xb8, 0x11, 0x84, 0x79, 0xbb, 0x6c, 0xf, 0xed, 0xc, 0xfc, 0x18, 0x32, 0x9d, 0xf1, 0x7, 0x2c, 0x20, 0xde, 0xe9, 0x97, 0x0, 0x62, 0x9f, 0x5e, 0x24, 0xfc, 0x8e, 0xc2, 0xd9, 0x2d}} + info := bindataFileInfo{name: "doc.go", size: 73, mode: os.FileMode(436), modTime: time.Unix(1715855770, 0)} + a := &asset{bytes: bytes, info: info} return a, nil } @@ -176,8 +185,8 @@ func docGo() (*asset, error) { // It returns an error if the asset could not be found or // could not be loaded. func Asset(name string) ([]byte, error) { - canonicalName := strings.Replace(name, "\\", "/", -1) - if f, ok := _bindata[canonicalName]; ok { + cannonicalName := strings.Replace(name, "\\", "/", -1) + if f, ok := _bindata[cannonicalName]; ok { a, err := f() if err != nil { return nil, fmt.Errorf("Asset %s can't read by error: %v", name, err) @@ -187,12 +196,6 @@ func Asset(name string) ([]byte, error) { return nil, fmt.Errorf("Asset %s not found", name) } -// AssetString returns the asset contents as a string (instead of a []byte). -func AssetString(name string) (string, error) { - data, err := Asset(name) - return string(data), err -} - // MustAsset is like Asset but panics when Asset would return an error. // It simplifies safe initialization of global variables. func MustAsset(name string) []byte { @@ -204,18 +207,12 @@ func MustAsset(name string) []byte { return a } -// MustAssetString is like AssetString but panics when Asset would return an -// error. It simplifies safe initialization of global variables. -func MustAssetString(name string) string { - return string(MustAsset(name)) -} - // AssetInfo loads and returns the asset info for the given name. // It returns an error if the asset could not be found or // could not be loaded. func AssetInfo(name string) (os.FileInfo, error) { - canonicalName := strings.Replace(name, "\\", "/", -1) - if f, ok := _bindata[canonicalName]; ok { + cannonicalName := strings.Replace(name, "\\", "/", -1) + if f, ok := _bindata[cannonicalName]; ok { a, err := f() if err != nil { return nil, fmt.Errorf("AssetInfo %s can't read by error: %v", name, err) @@ -225,33 +222,6 @@ func AssetInfo(name string) (os.FileInfo, error) { return nil, fmt.Errorf("AssetInfo %s not found", name) } -// AssetDigest returns the digest of the file with the given name. It returns an -// error if the asset could not be found or the digest could not be loaded. -func AssetDigest(name string) ([sha256.Size]byte, error) { - canonicalName := strings.Replace(name, "\\", "/", -1) - if f, ok := _bindata[canonicalName]; ok { - a, err := f() - if err != nil { - return [sha256.Size]byte{}, fmt.Errorf("AssetDigest %s can't read by error: %v", name, err) - } - return a.digest, nil - } - return [sha256.Size]byte{}, fmt.Errorf("AssetDigest %s not found", name) -} - -// Digests returns a map of all known files and their checksums. -func Digests() (map[string][sha256.Size]byte, error) { - mp := make(map[string][sha256.Size]byte, len(_bindata)) - for name := range _bindata { - a, err := _bindata[name]() - if err != nil { - return nil, err - } - mp[name] = a.digest - } - return mp, nil -} - // AssetNames returns the names of the assets. func AssetNames() []string { names := make([]string, 0, len(_bindata)) @@ -270,29 +240,24 @@ var _bindata = map[string]func() (*asset, error){ "doc.go": docGo, } -// AssetDebug is true if the assets were built with the debug flag enabled. -const AssetDebug = false - // AssetDir returns the file names below a certain // directory embedded in the file by go-bindata. // For example if you run go-bindata on data/... and data contains the // following hierarchy: -// -// data/ -// foo.txt -// img/ -// a.png -// b.png -// -// then AssetDir("data") would return []string{"foo.txt", "img"}, -// AssetDir("data/img") would return []string{"a.png", "b.png"}, -// AssetDir("foo.txt") and AssetDir("notexist") would return an error, and +// data/ +// foo.txt +// img/ +// a.png +// b.png +// then AssetDir("data") would return []string{"foo.txt", "img"} +// AssetDir("data/img") would return []string{"a.png", "b.png"} +// AssetDir("foo.txt") and AssetDir("notexist") would return an error // AssetDir("") will return []string{"data"}. func AssetDir(name string) ([]string, error) { node := _bintree if len(name) != 0 { - canonicalName := strings.Replace(name, "\\", "/", -1) - pathList := strings.Split(canonicalName, "/") + cannonicalName := strings.Replace(name, "\\", "/", -1) + pathList := strings.Split(cannonicalName, "/") for _, p := range pathList { node = node.Children[p] if node == nil { @@ -316,14 +281,14 @@ type bintree struct { } var _bintree = &bintree{nil, map[string]*bintree{ - "000001_message_type.up.sql": {_000001_message_typeUpSql, map[string]*bintree{}}, - "000002_bandwidth_protocol.up.sql": {_000002_bandwidth_protocolUpSql, map[string]*bintree{}}, - "000003_index_truncate.up.sql": {_000003_index_truncateUpSql, map[string]*bintree{}}, - "000004_envelope.table.up.sql": {_000004_envelopeTableUpSql, map[string]*bintree{}}, - "doc.go": {docGo, map[string]*bintree{}}, + "000001_message_type.up.sql": &bintree{_000001_message_typeUpSql, map[string]*bintree{}}, + "000002_bandwidth_protocol.up.sql": &bintree{_000002_bandwidth_protocolUpSql, map[string]*bintree{}}, + "000003_index_truncate.up.sql": &bintree{_000003_index_truncateUpSql, map[string]*bintree{}}, + "000004_envelope.table.up.sql": &bintree{_000004_envelopeTableUpSql, map[string]*bintree{}}, + "doc.go": &bintree{docGo, map[string]*bintree{}}, }} -// RestoreAsset restores an asset under the given directory. +// RestoreAsset restores an asset under the given directory func RestoreAsset(dir, name string) error { data, err := Asset(name) if err != nil { @@ -337,14 +302,18 @@ func RestoreAsset(dir, name string) error { if err != nil { return err } - err = os.WriteFile(_filePath(dir, name), data, info.Mode()) + err = ioutil.WriteFile(_filePath(dir, name), data, info.Mode()) if err != nil { return err } - return os.Chtimes(_filePath(dir, name), info.ModTime(), info.ModTime()) + err = os.Chtimes(_filePath(dir, name), info.ModTime(), info.ModTime()) + if err != nil { + return err + } + return nil } -// RestoreAssets restores an asset under the given directory recursively. +// RestoreAssets restores an asset under the given directory recursively func RestoreAssets(dir, name string) error { children, err := AssetDir(name) // File @@ -362,6 +331,6 @@ func RestoreAssets(dir, name string) error { } func _filePath(dir, name string) string { - canonicalName := strings.Replace(name, "\\", "/", -1) - return filepath.Join(append([]string{dir}, strings.Split(canonicalName, "/")...)...) + cannonicalName := strings.Replace(name, "\\", "/", -1) + return filepath.Join(append([]string{dir}, strings.Split(cannonicalName, "/")...)...) } diff --git a/telemetry/db.go b/telemetry/db.go index 5281d87..7829d5b 100644 --- a/telemetry/db.go +++ b/telemetry/db.go @@ -9,6 +9,7 @@ import ( "github.com/golang-migrate/migrate/v4/database/postgres" bindata "github.com/golang-migrate/migrate/v4/source/go_bindata" _ "github.com/lib/pq" + "go.uber.org/zap" ) // Migrate applies migrations. @@ -41,7 +42,7 @@ func migrateDB(db *sql.DB, resources *bindata.AssetSource, driver database.Drive return nil } -func OpenDb(dataSourceName string) *sql.DB { +func OpenDb(dataSourceName string, logger *zap.Logger) *sql.DB { db, err := sql.Open("postgres", dataSourceName) if err != nil { log.Fatalf("could not connect to database: %v", err) @@ -50,12 +51,12 @@ func OpenDb(dataSourceName string) *sql.DB { if err := db.Ping(); err != nil { log.Fatalf("unable to reach database: %v", err) } - log.Println("Connected to database") + logger.Info("Connected to database") if err := createTables(db); err != nil { log.Fatalf("unable to create the table: %v", err) } - log.Println("DB initialized") + logger.Info("DB initialized") return db } diff --git a/telemetry/server.go b/telemetry/server.go index 6bb7e09..c6146ed 100644 --- a/telemetry/server.go +++ b/telemetry/server.go @@ -11,17 +11,20 @@ import ( "time" "github.com/gorilla/mux" + "go.uber.org/zap" ) type Server struct { Router *mux.Router DB *sql.DB + logger *zap.Logger } -func NewServer(db *sql.DB) *Server { +func NewServer(db *sql.DB, logger *zap.Logger) *Server { server := &Server{ Router: mux.NewRouter().StrictSlash(true), DB: db, + logger: logger, } server.Router.HandleFunc("/protocol-stats", server.createProtocolStats).Methods("POST") @@ -43,11 +46,11 @@ func (s *Server) createReceivedMessages(w http.ResponseWriter, r *http.Request) var receivedMessages []ReceivedMessage decoder := json.NewDecoder(r.Body) if err := decoder.Decode(&receivedMessages); err != nil { - log.Println(err) + s.logger.Error("failed to decode messages", zap.Error(err)) err := respondWithError(w, http.StatusBadRequest, "Invalid request payload") if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) } return } @@ -56,7 +59,7 @@ func (s *Server) createReceivedMessages(w http.ResponseWriter, r *http.Request) var ids []int for _, receivedMessage := range receivedMessages { if err := receivedMessage.put(s.DB); err != nil { - log.Println("could not save message", err, receivedMessage) + s.logger.Error("could not save message", zap.Error(err), zap.Any("receivedMessage", receivedMessage)) continue } ids = append(ids, receivedMessage.ID) @@ -65,21 +68,22 @@ func (s *Server) createReceivedMessages(w http.ResponseWriter, r *http.Request) if len(ids) != len(receivedMessages) { err := respondWithError(w, http.StatusInternalServerError, "Could not save all record") if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) } return } err := respondWithJSON(w, http.StatusCreated, receivedMessages) if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) + return } - log.Printf( - "%s\t%s\t%s", - r.Method, - r.RequestURI, - time.Since(start), + s.logger.Info( + "handled received message", + zap.String("method", r.Method), + zap.String("requestURI", r.RequestURI), + zap.Duration("duration", time.Since(start)), ) } @@ -88,11 +92,12 @@ func (s *Server) createReceivedEnvelope(w http.ResponseWriter, r *http.Request) var receivedEnvelope ReceivedEnvelope decoder := json.NewDecoder(r.Body) if err := decoder.Decode(&receivedEnvelope); err != nil { - log.Println(err) + s.logger.Error("failed to decode envelope", zap.Error(err)) err := respondWithError(w, http.StatusBadRequest, "Invalid request payload") if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) + return } return } @@ -100,19 +105,24 @@ func (s *Server) createReceivedEnvelope(w http.ResponseWriter, r *http.Request) err := receivedEnvelope.put(s.DB) if err != nil { - log.Println("could not save envelope", err, receivedEnvelope) + s.logger.Error("could not save envelope", zap.Error(err), zap.Any("envelope", receivedEnvelope)) + err := respondWithError(w, http.StatusBadRequest, "Could not save the envelope") + if err != nil { + s.logger.Error("failed to respond", zap.Error(err)) + } } err = respondWithJSON(w, http.StatusCreated, receivedEnvelope) if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) + return } - log.Printf( - "%s\t%s\t%s", - r.Method, - r.RequestURI, - time.Since(start), + s.logger.Info( + "handled received envelope", + zap.String("method", r.Method), + zap.String("requestURI", r.RequestURI), + zap.Duration("duration", time.Since(start)), ) } @@ -120,13 +130,13 @@ func (s *Server) updateEnvelope(w http.ResponseWriter, r *http.Request) { start := time.Now() var receivedEnvelope ReceivedEnvelope decoder := json.NewDecoder(r.Body) - log.Println("Update envelope") + s.logger.Info("update envelope") if err := decoder.Decode(&receivedEnvelope); err != nil { - log.Println(err) + s.logger.Error("failed to decode envelope", zap.Error(err)) err := respondWithError(w, http.StatusBadRequest, "Invalid request payload") if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) } return } @@ -134,19 +144,25 @@ func (s *Server) updateEnvelope(w http.ResponseWriter, r *http.Request) { err := receivedEnvelope.updateProcessingError(s.DB) if err != nil { - log.Println("could not update envelope", err, receivedEnvelope) + s.logger.Error("could not update envelope", zap.Error(err), zap.Any("envelope", receivedEnvelope)) + err := respondWithError(w, http.StatusBadRequest, "Could not update the envelope") + if err != nil { + s.logger.Error("failed to respond", zap.Error(err)) + } + return } err = respondWithJSON(w, http.StatusCreated, receivedEnvelope) if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) + return } - log.Printf( - "%s\t%s\t%s", - r.Method, - r.RequestURI, - time.Since(start), + s.logger.Info( + "handled update message", + zap.String("method", r.Method), + zap.String("requestURI", r.RequestURI), + zap.Duration("duration", time.Since(start)), ) } @@ -155,11 +171,11 @@ func (s *Server) createProtocolStats(w http.ResponseWriter, r *http.Request) { var protocolStats ProtocolStats decoder := json.NewDecoder(r.Body) if err := decoder.Decode(&protocolStats); err != nil { - log.Println(err) + s.logger.Error("failed to decode protocol stats", zap.Error(err)) err := respondWithError(w, http.StatusBadRequest, "Invalid request payload") if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) } return } @@ -169,27 +185,29 @@ func (s *Server) createProtocolStats(w http.ResponseWriter, r *http.Request) { protocolStats.PeerID = hex.EncodeToString(peerIDHash[:]) if err := protocolStats.put(s.DB); err != nil { + s.logger.Error("failed to save protocol stats", zap.Error(err)) err := respondWithError(w, http.StatusInternalServerError, "Could not save protocol stats") if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) } return } err := respondWithJSON(w, http.StatusCreated, map[string]string{"error": ""}) if err != nil { - log.Println(err) + s.logger.Error("failed to respond", zap.Error(err)) + return } - log.Printf( - "%s\t%s\t%s", - r.Method, - r.RequestURI, - time.Since(start), + s.logger.Info( + "handled protocol stats", + zap.String("method", r.Method), + zap.String("requestURI", r.RequestURI), + zap.Duration("duration", time.Since(start)), ) } func (s *Server) Start(port int) { - log.Printf("Starting server on port %d", port) + s.logger.Info("Starting server", zap.Int("port", port)) log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", port), s.Router)) }