status-go/protocol/anonmetrics/server.go

190 lines
4.0 KiB
Go
Raw Normal View History

Anon Metrics Broadcast (#2198) * Protobufs and adapters * Added basic anon metric service and config init * Added fibonacci interval incrementer * Added basic Client.Start func and integrated interval incrementer * Added new processed field to app metrics table * Added id column to app metrics table * Added migration clean up * Added appmetrics GetUnprocessed and SetToProcessedByIDs and tests There was a wierd bug where metrics in the db that did not explicitly insert a value would be NULL, so could not be found by . In addition I've added a new primary id field to the app_metrics table so that updates could be done against very specific metric rows. * Updated adaptors and db to handle proto_id I need a way to distinguish individual metric items from each other so that I can ignore the ones that have been seen before. * Moved incrementer into dedicated file * Resolve incrementer test fail * Finalised the main loop functionality * Implemented delete loop framework * Updated adaptors file name * Added delete loop delay and quit, and tweak on RawMessage gen * Completed delete loop logic * Added DBLock to prevent deletion during mainLoop * Added postgres DB connection, integrated into anonmetrics.Server * Removed proto_id from SQL migration and model * Integrated postgres with Server and updated adaptors * Function name update * Added sample config files for client and server * Fixes and testing for low level e2e * make generate * Fix lint * Fix for receiving an anonMetricBatch not in server mode * Postgres test fixes * Tidy up, make vendor and make generate * delinting * Fixing database tests * Attempted fix of does: cannot open `does' (No such file or directory) not: cannot open `not' (No such file or directory) exist: cannot open `exist' (No such file or directory) error on sql resource loas * Moved all anon metric postgres migration logic and sources into a the protocol/anonmetrics package or sub packages. I don't know if this will fix the does: cannot open `does' (No such file or directory) not: cannot open `not' (No such file or directory) exist: cannot open `exist' (No such file or directory) error that happens in Jenkins but this could work * Lint for the lint god * Why doesn't the linter list all its problems at once? * test tweaks * Fix for wakuV2 change * DB reset change * Fix for postgres db migrations fails * More robust implementation of postgres test setup and teardown * Added block for anon metrics functionality * Version Bump to 0.84.0 * Added test to check anon metrics broadcast is deactivated * Protobufs and adapters * Added basic anon metric service and config init * Added new processed field to app metrics table * Added id column to app metrics table * Added migration clean up * Added appmetrics GetUnprocessed and SetToProcessedByIDs and tests There was a wierd bug where metrics in the db that did not explicitly insert a value would be NULL, so could not be found by . In addition I've added a new primary id field to the app_metrics table so that updates could be done against very specific metric rows. * Updated adaptors and db to handle proto_id I need a way to distinguish individual metric items from each other so that I can ignore the ones that have been seen before. * Added postgres DB connection, integrated into anonmetrics.Server * Removed proto_id from SQL migration and model * Integrated postgres with Server and updated adaptors * Added sample config files for client and server * Fix lint * Fix for receiving an anonMetricBatch not in server mode * Postgres test fixes * Tidy up, make vendor and make generate * Moved all anon metric postgres migration logic and sources into a the protocol/anonmetrics package or sub packages. I don't know if this will fix the does: cannot open `does' (No such file or directory) not: cannot open `not' (No such file or directory) exist: cannot open `exist' (No such file or directory) error that happens in Jenkins but this could work
2021-09-01 12:02:18 +00:00
package anonmetrics
import (
"database/sql"
// Import postgres driver
_ "github.com/lib/pq"
"go.uber.org/zap"
"github.com/status-im/migrate/v4"
"github.com/status-im/migrate/v4/database/postgres"
bindata "github.com/status-im/migrate/v4/source/go_bindata"
"github.com/status-im/status-go/appmetrics"
"github.com/status-im/status-go/protocol/anonmetrics/migrations"
"github.com/status-im/status-go/protocol/protobuf"
)
const ActiveServerPhrase = "I was thinking that it would be a pretty nice idea if the server functionality was working now, I express gratitude in the anticipation"
type ServerConfig struct {
Enabled bool
PostgresURI string
Active string
}
type Server struct {
Config *ServerConfig
Logger *zap.Logger
PostgresDB *sql.DB
}
func NewServer(postgresURI string) (*Server, error) {
postgresMigration := bindata.Resource(migrations.AssetNames(), migrations.Asset)
db, err := NewMigratedDB(postgresURI, postgresMigration)
if err != nil {
return nil, err
}
return &Server{
PostgresDB: db,
}, nil
}
func (s *Server) Stop() error {
if s.PostgresDB != nil {
return s.PostgresDB.Close()
}
return nil
}
func (s *Server) StoreMetrics(appMetricsBatch protobuf.AnonymousMetricBatch) (appMetrics []*appmetrics.AppMetric, err error) {
if s.Config.Active != ActiveServerPhrase {
return nil, nil
}
s.Logger.Debug("StoreMetrics() triggered with payload",
zap.Reflect("appMetricsBatch", appMetricsBatch))
appMetrics, err = adaptProtoBatchToModels(appMetricsBatch)
if err != nil {
return
}
var (
tx *sql.Tx
insert *sql.Stmt
)
// start txn
tx, err = s.PostgresDB.Begin()
if err != nil {
return
}
defer func() {
if err == nil {
err = tx.Commit()
return
}
_ = tx.Rollback()
}()
//noinspection ALL
query := `INSERT INTO app_metrics (message_id, event, value, app_version, operating_system, session_id, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (message_id) DO NOTHING;`
insert, err = tx.Prepare(query)
if err != nil {
return
}
for _, metric := range appMetrics {
_, err = insert.Exec(
metric.MessageID,
metric.Event,
metric.Value,
metric.AppVersion,
metric.OS,
metric.SessionID,
metric.CreatedAt,
)
if err != nil {
return
}
}
return
}
func (s *Server) getFromRows(rows *sql.Rows) (appMetrics []appmetrics.AppMetric, err error) {
for rows.Next() {
metric := appmetrics.AppMetric{}
err = rows.Scan(
&metric.ID,
&metric.MessageID,
&metric.Event,
&metric.Value,
&metric.AppVersion,
&metric.OS,
&metric.SessionID,
&metric.CreatedAt,
&metric.Processed,
&metric.ReceivedAt,
)
if err != nil {
return nil, err
}
appMetrics = append(appMetrics, metric)
}
return appMetrics, nil
}
func (s *Server) GetAppMetrics(limit int, offset int) ([]appmetrics.AppMetric, error) {
if s.Config.Active != ActiveServerPhrase {
return nil, nil
}
rows, err := s.PostgresDB.Query("SELECT id, message_id, event, value, app_version, operating_system, session_id, created_at, processed, received_at FROM app_metrics LIMIT $1 OFFSET $2", limit, offset)
if err != nil {
return nil, err
}
defer rows.Close()
return s.getFromRows(rows)
}
func NewMigratedDB(uri string, migrationResource *bindata.AssetSource) (*sql.DB, error) {
db, err := sql.Open("postgres", uri)
if err != nil {
return nil, err
}
if err := setup(db, migrationResource); err != nil {
return nil, err
}
return db, nil
}
func setup(d *sql.DB, migrationResource *bindata.AssetSource) error {
m, err := MakeMigration(d, migrationResource)
if err != nil {
return err
}
if err = m.Up(); err != migrate.ErrNoChange {
return err
}
return nil
}
func MakeMigration(d *sql.DB, migrationResource *bindata.AssetSource) (*migrate.Migrate, error) {
source, err := bindata.WithInstance(migrationResource)
if err != nil {
return nil, err
}
driver, err := postgres.WithInstance(d, &postgres.Config{})
if err != nil {
return nil, err
}
return migrate.NewWithInstance(
"go-bindata",
source,
"postgres",
driver)
}