Merge pull request #58 from insolar/develop

Develop: fix WLT-1091
insolar · Sep 18, 2019 · 93f937d · 93f937d
2 parents bc2b271 + c449999
commit 93f937d
Show file tree

Hide file tree

Showing 20 changed files with 126 additions and 106 deletions.
diff --git a/internal/beauty/beautifier.go b/internal/beauty/beautifier.go
@@ -147,21 +147,21 @@ func (b *Beautifier) process(rec *record.Material) {
 	b.depositKeeper.Process(rec)
 }
 
-func (b *Beautifier) dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (b *Beautifier) dump(tx orm.DB, pub replicator.OnDumpSuccess, errors prometheus.Counter) error {
 	log.Infof("dump beautifier")
-	if err := b.memberComposer.Dump(tx, pub); err != nil {
+	if err := b.memberComposer.Dump(tx, pub, errors); err != nil {
 		return err
 	}
-	if err := b.memberBalanceUpdater.Dump(tx, pub); err != nil {
+	if err := b.memberBalanceUpdater.Dump(tx, pub, errors); err != nil {
 		return err
 	}
-	if err := b.transferComposer.Dump(tx, pub); err != nil {
+	if err := b.transferComposer.Dump(tx, pub, errors); err != nil {
 		return err
 	}
-	if err := b.depositComposer.Dump(tx, pub); err != nil {
+	if err := b.depositComposer.Dump(tx, pub, errors); err != nil {
 		return err
 	}
-	if err := b.migrationAddressComposer.Dump(tx, pub); err != nil {
+	if err := b.migrationAddressComposer.Dump(tx, pub, errors); err != nil {
 		return err
 	}
 	if err := b.migrationAddressKeeper.Dump(tx, pub); err != nil {

diff --git a/internal/beauty/member/member.go b/internal/beauty/member/member.go
@@ -196,12 +196,12 @@ func (c *Composer) compose(b *memberBuilder) {
 	delete(c.builders, origin)
 }
 
-func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess, errorCounter prometheus.Counter) error {
 	log.Infof("dump members")
 	c.updateStat()
 
 	for _, member := range c.cache {
-		if err := member.Dump(tx); err != nil {
+		if err := member.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump members")
 		}
 	}

diff --git a/internal/beauty/member/update_balance.go b/internal/beauty/member/update_balance.go
@@ -22,6 +22,7 @@ import (
 	"github.com/insolar/insolar/insolar/gen"
 	"github.com/insolar/insolar/insolar/record"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 
 	log "github.com/sirupsen/logrus"
 
@@ -73,11 +74,11 @@ func (u *BalanceUpdater) processAccountAmend(id insolar.ID, rec *record.Material
 	})
 }
 
-func (u *BalanceUpdater) Dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (u *BalanceUpdater) Dump(tx orm.DB, pub replicator.OnDumpSuccess, errorCounter prometheus.Counter) error {
 	log.Infof("dump member balances")
 
 	for _, acc := range u.technicalAccounts {
-		if err := acc.Dump(tx); err != nil {
+		if err := acc.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump internal member")
 		}
 	}

diff --git a/internal/beauty/migration/address.go b/internal/beauty/migration/address.go
@@ -119,14 +119,14 @@ func (c *Composer) Process(rec *record.Material) {
 	}
 }
 
-func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess, errorCounter prometheus.Counter) error {
 	log.Infof("dump migration addresses")
 
 	c.updateStat()
 
 	log.Infof("dump %d addresses", len(c.cache))
 	for _, addr := range c.cache {
-		if err := addr.Dump(tx); err != nil {
+		if err := addr.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump migration addresses addr=%s", addr.Addr)
 		}
 	}

diff --git a/internal/beauty/migration/deposit/deposit.go b/internal/beauty/migration/deposit/deposit.go
@@ -282,11 +282,11 @@ func (c *Composer) compose(b *depositBuilder) {
 	delete(c.builders, origin)
 }
 
-func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess, errorCounter prometheus.Counter) error {
 	log.Infof("dump deposits")
 
 	for _, dep := range c.cache {
-		if err := dep.Dump(tx); err != nil {
+		if err := dep.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump deposits")
 		}
 	}

diff --git a/internal/beauty/pulse/keeper.go b/internal/beauty/pulse/keeper.go
@@ -23,6 +23,7 @@ import (
 	"github.com/go-pg/pg/orm"
 	"github.com/insolar/insolar/insolar"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/insolar/observer/internal/configuration"
@@ -83,10 +84,10 @@ func (k *Keeper) process(pn insolar.PulseNumber, entropy insolar.Entropy, timest
 	})
 }
 
-func (k *Keeper) dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (k *Keeper) dump(tx orm.DB, pub replicator.OnDumpSuccess, errorCounter prometheus.Counter) error {
 	log.Infof("dump pulse keeper")
 	for _, p := range k.cache {
-		if err := p.Dump(tx); err != nil {
+		if err := p.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump pulse")
 		}
 	}

diff --git a/internal/beauty/transfer/transfer.go b/internal/beauty/transfer/transfer.go
@@ -188,13 +188,13 @@ func isTransferCall(request *dto.Request) bool {
 	return false
 }
 
-func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (c *Composer) Dump(tx orm.DB, pub replicator.OnDumpSuccess, errorCounter prometheus.Counter) error {
 	log.Infof("dump member transfers")
 
 	c.updateStat()
 
 	for _, transfer := range c.cache {
-		if err := transfer.Dump(tx); err != nil {
+		if err := transfer.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump transfers")
 		}
 	}

diff --git a/internal/dto/dumper.go b/internal/dto/dumper.go
@@ -22,6 +22,7 @@ import (
 	"github.com/go-pg/pg/orm"
 	"github.com/insolar/insolar/insolar/record"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/insolar/observer/internal/configuration"
@@ -89,25 +90,25 @@ func (d *Dumper) process(rn uint32, rec *record.Material) {
 	d.buildUnpacked(rec)
 }
 
-func (d *Dumper) dump(tx orm.DB, pub replicator.OnDumpSuccess) error {
+func (d *Dumper) dump(tx orm.DB, pub replicator.OnDumpSuccess, errorCounter prometheus.Counter) error {
 	log.Infof("dump raw records")
 	for _, rec := range d.records {
-		if err := rec.Dump(tx); err != nil {
+		if err := rec.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump raw records")
 		}
 	}
 	for _, req := range d.requests {
-		if err := req.Dump(tx); err != nil {
+		if err := req.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump raw requests")
 		}
 	}
 	for _, res := range d.results {
-		if err := res.Dump(tx); err != nil {
+		if err := res.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump raw results")
 		}
 	}
 	for _, obj := range d.objects {
-		if err := obj.Dump(tx); err != nil {
+		if err := obj.Dump(tx, errorCounter); err != nil {
 			return errors.Wrapf(err, "failed to dump raw objects")
 		}
 	}

diff --git a/internal/model/beauty/deposit.go b/internal/model/beauty/deposit.go
@@ -19,6 +19,8 @@ package beauty
 import (
 	"github.com/go-pg/pg/orm"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
 )
 
 type Deposit struct {
@@ -34,9 +36,15 @@ type Deposit struct {
 	DepositState    string `sql:",notnull"`
 }
 
-func (d *Deposit) Dump(tx orm.DB) error {
-	if err := tx.Insert(d); err != nil {
+func (d *Deposit) Dump(tx orm.DB, errorCounter prometheus.Counter) error {
+	res, err := tx.Model(d).OnConflict("DO NOTHING").Insert(d)
+	if err != nil {
 		return errors.Wrapf(err, "failed to insert deposit")
 	}
+
+	if res.RowsAffected() == 0 {
+		errorCounter.Inc()
+		logrus.Errorf("Failed to insert deposit: %v", d)
+	}
 	return nil
 }
diff --git a/internal/model/beauty/fee.go b/internal/model/beauty/fee.go
diff --git a/internal/model/beauty/member.go b/internal/model/beauty/member.go
@@ -19,6 +19,8 @@ package beauty
 import (
 	"github.com/go-pg/pg/orm"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
 )
 
 type Member struct {
@@ -32,9 +34,15 @@ type Member struct {
 	Status           string
 }
 
-func (m *Member) Dump(tx orm.DB) error {
-	if err := tx.Insert(m); err != nil {
+func (m *Member) Dump(tx orm.DB, errorCounter prometheus.Counter) error {
+	res, err := tx.Model(m).OnConflict("DO NOTHING").Insert(m)
+	if err != nil {
 		return errors.Wrapf(err, "failed to insert member")
 	}
+
+	if res.RowsAffected() == 0 {
+		errorCounter.Inc()
+		logrus.Errorf("Failed to insert member: %v", m)
+	}
 	return nil
 }
diff --git a/internal/model/beauty/migration_address.go b/internal/model/beauty/migration_address.go
@@ -19,7 +19,8 @@ package beauty
 import (
 	"github.com/go-pg/pg/orm"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
 )
 
 type MigrationAddress struct {
@@ -30,9 +31,15 @@ type MigrationAddress struct {
 	Wasted    bool
 }
 
-func (a *MigrationAddress) Dump(tx orm.DB) error {
-	if err := tx.Insert(a); err != nil {
-		log.Error(errors.Wrapf(err, "failed to insert migration address"))
+func (a *MigrationAddress) Dump(tx orm.DB, errorCounter prometheus.Counter) error {
+	res, err := tx.Model(a).OnConflict("DO NOTHING").Insert(a)
+	if err != nil {
+		return errors.Wrapf(err, "failed to insert migration address")
+	}
+
+	if res.RowsAffected() == 0 {
+		errorCounter.Inc()
+		logrus.Errorf("Failed to insert migration address: %v", a)
 	}
 	return nil
 }
diff --git a/internal/model/beauty/pulse.go b/internal/model/beauty/pulse.go
@@ -20,6 +20,8 @@ import (
 	"github.com/go-pg/pg/orm"
 	"github.com/insolar/insolar/insolar"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
 )
 
 type Pulse struct {
@@ -31,10 +33,15 @@ type Pulse struct {
 	RequestsCount uint32
 }
 
-func (p *Pulse) Dump(tx orm.DB) error {
-	err := tx.Insert(p)
+func (p *Pulse) Dump(tx orm.DB, errorCounter prometheus.Counter) error {
+	res, err := tx.Model(p).OnConflict("DO NOTHING").Insert(p)
 	if err != nil {
 		return errors.Wrapf(err, "failed to insert pulse %v", p)
 	}
+
+	if res.RowsAffected() == 0 {
+		errorCounter.Inc()
+		logrus.Errorf("Failed to insert pulse: %v", p)
+	}
 	return nil
 }
diff --git a/internal/model/beauty/transfer.go b/internal/model/beauty/transfer.go
@@ -18,8 +18,11 @@ package beauty
 
 import (
 	"github.com/go-pg/pg/orm"
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/insolar/insolar/insolar"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
 type Transfer struct {
@@ -39,9 +42,15 @@ type Transfer struct {
 	EthHash       string              `sql:",notnull"`
 }
 
-func (m *Transfer) Dump(tx orm.DB) error {
-	if err := tx.Insert(m); err != nil {
+func (t *Transfer) Dump(tx orm.DB, errorCounter prometheus.Counter) error {
+	res, err := tx.Model(t).OnConflict("DO NOTHING").Insert(t)
+	if err != nil {
 		return errors.Wrapf(err, "failed to insert transfer")
 	}
+
+	if res.RowsAffected() == 0 {
+		errorCounter.Inc()
+		logrus.Errorf("Failed to insert transfer: %v", t)
+	}
 	return nil
 }
diff --git a/internal/model/dumper.go b/internal/model/dumper.go