Skip to content

Commit a037eaa

Browse files
antmatantmat
committed
feat: market shredder
1 parent 10630dd commit a037eaa

File tree

4 files changed

+88
-32
lines changed

4 files changed

+88
-32
lines changed

‎etc/node.yaml

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,3 @@
1-
migration_dwh:
2-
endpoint:
3-
4-
51
# Local Node settings
62
node:
73
# Node's port to listen for client connection

‎insonmnia/migration/migration.go

Lines changed: 35 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,42 +3,69 @@ package migration
33
import (
44
"context"
55
"crypto/ecdsa"
6+
"fmt"
67

78
"github.com/ethereum/go-ethereum/crypto"
89
"github.com/sonm-io/core/blockchain"
910
"github.com/sonm-io/core/insonmnia/dwh"
1011
"github.com/sonm-io/core/proto"
1112
"github.com/sonm-io/core/util/multierror"
1213
"github.com/sonm-io/core/util/xgrpc"
14+
"google.golang.org/grpc/credentials"
1315
)
1416

15-
type MarketShredder struct {
17+
type MarketShredder interface {
18+
WeDontNeedNoWaterLetTheMothefuckerBurn(ctx context.Context, pKey *ecdsa.PrivateKey) error
19+
}
20+
21+
type nilMarketShredder struct {
22+
}
23+
24+
func (m *nilMarketShredder) WeDontNeedNoWaterLetTheMothefuckerBurn(ctx context.Context, pKey *ecdsa.PrivateKey) error {
25+
return nil
26+
}
27+
28+
type marketShredder struct {
1629
api blockchain.API
1730
dwh sonm.DWHClient
1831
}
1932

20-
func NewV1MarketShredder(ctx context.Context, bcCfg *blockchain.Config, dwhCfg dwh.YAMLConfig) (*MarketShredder, error) {
21-
api, err := blockchain.NewAPI(ctx, blockchain.WithConfig(bcCfg), blockchain.WithVersion(1))
33+
type MigrationConfig struct {
34+
Blockchain *blockchain.Config
35+
MigrationDWH *dwh.YAMLConfig
36+
Enabled *bool `default:"true"`
37+
Version uint
38+
}
39+
40+
func NewV1MarketShredder(ctx context.Context, cfg *MigrationConfig, credentials credentials.TransportCredentials) (MarketShredder, error) {
41+
if !*cfg.Enabled {
42+
return &nilMarketShredder{}, nil
43+
}
44+
if cfg.MigrationDWH == nil {
45+
return nil, fmt.Errorf("dwh config is required for enabled migrartion")
46+
}
47+
api, err := blockchain.NewAPI(ctx, blockchain.WithConfig(cfg.Blockchain), blockchain.WithVersion(1))
2248
if err != nil {
2349
return nil, err
2450
}
2551

26-
cc, err := xgrpc.NewClient(ctx, dwhCfg.Endpoint, credentials)
52+
cc, err := xgrpc.NewClient(ctx, cfg.MigrationDWH.Endpoint, credentials)
2753
if err != nil {
28-
return err
54+
return nil, err
2955
}
3056

3157
dwh := sonm.NewDWHClient(cc)
58+
return NewMarketShredder(api, dwh), nil
3259
}
3360

34-
func NewMarketShredder(api blockchain.API, dwh sonm.DWHClient) *MarketShredder {
35-
return &MarketShredder{
61+
func NewMarketShredder(api blockchain.API, dwh sonm.DWHClient) *marketShredder {
62+
return &marketShredder{
3663
api: api,
3764
dwh: dwh,
3865
}
3966
}
4067

41-
func (m *MarketShredder) WeDontNeedNoWaterLetTheMothefuckerBurn(ctx context.Context, pKey *ecdsa.PrivateKey) error {
68+
func (m *marketShredder) WeDontNeedNoWaterLetTheMothefuckerBurn(ctx context.Context, pKey *ecdsa.PrivateKey) error {
4269
author := crypto.PubkeyToAddress(pKey.PublicKey)
4370
ordersRequest := &sonm.OrdersRequest{
4471
AuthorID: sonm.NewEthAddress(author),

‎insonmnia/worker/config.go

Lines changed: 28 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package worker
22

33
import (
4+
"fmt"
5+
46
"github.com/ethereum/go-ethereum/common"
57
"github.com/jinzhu/configor"
68
"github.com/opencontainers/runtime-spec/specs-go"
@@ -9,6 +11,7 @@ import (
911
"github.com/sonm-io/core/insonmnia/dwh"
1012
"github.com/sonm-io/core/insonmnia/logging"
1113
"github.com/sonm-io/core/insonmnia/matcher"
14+
"github.com/sonm-io/core/insonmnia/migration"
1215
"github.com/sonm-io/core/insonmnia/npp"
1316
"github.com/sonm-io/core/insonmnia/state"
1417
"github.com/sonm-io/core/insonmnia/worker/plugin"
@@ -37,26 +40,27 @@ type DevConfig struct {
3740
}
3841

3942
type Config struct {
40-
Endpoint string `yaml:"endpoint" required:"true"`
41-
Logging logging.Config `yaml:"logging"`
42-
Resources *ResourcesConfig `yaml:"resources" required:"false"`
43-
Blockchain *blockchain.Config `yaml:"blockchain"`
44-
NPP npp.Config `yaml:"npp"`
45-
SSH *SSHConfig `yaml:"ssh" required:"false" `
46-
PublicIPs []string `yaml:"public_ip_addrs" required:"false" `
47-
Plugins plugin.Config `yaml:"plugins"`
48-
Storage state.StorageConfig `yaml:"store"`
49-
Benchmarks benchmarks.Config `yaml:"benchmarks"`
50-
Whitelist WhitelistConfig `yaml:"whitelist"`
51-
MetricsListenAddr string `yaml:"metrics_listen_addr" default:"127.0.0.1:14000"`
52-
DWH dwh.YAMLConfig `yaml:"dwh"`
53-
Matcher *matcher.YAMLConfig `yaml:"matcher"`
54-
Salesman salesman.YAMLConfig `yaml:"salesman"`
55-
Master common.Address `yaml:"master" required:"true"`
56-
Development DevConfig `yaml:"development"`
57-
Admin *common.Address `yaml:"admin"`
58-
MetricsCollector *common.Address `yaml:"metrics_collector"`
59-
Debug *debug.Config `yaml:"debug"`
43+
Endpoint string `yaml:"endpoint" required:"true"`
44+
Logging logging.Config `yaml:"logging"`
45+
Resources *ResourcesConfig `yaml:"resources" required:"false"`
46+
Blockchain *blockchain.Config `yaml:"blockchain"`
47+
NPP npp.Config `yaml:"npp"`
48+
SSH *SSHConfig `yaml:"ssh" required:"false" `
49+
PublicIPs []string `yaml:"public_ip_addrs" required:"false" `
50+
Plugins plugin.Config `yaml:"plugins"`
51+
Storage state.StorageConfig `yaml:"store"`
52+
Benchmarks benchmarks.Config `yaml:"benchmarks"`
53+
Whitelist WhitelistConfig `yaml:"whitelist"`
54+
MetricsListenAddr string `yaml:"metrics_listen_addr" default:"127.0.0.1:14000"`
55+
DWH dwh.YAMLConfig `yaml:"dwh"`
56+
Matcher *matcher.YAMLConfig `yaml:"matcher"`
57+
Salesman salesman.YAMLConfig `yaml:"salesman"`
58+
Master common.Address `yaml:"master" required:"true"`
59+
Development DevConfig `yaml:"development"`
60+
Admin *common.Address `yaml:"admin"`
61+
MetricsCollector *common.Address `yaml:"metrics_collector"`
62+
Debug *debug.Config `yaml:"debug"`
63+
Migration *migration.MigrationConfig `yaml:"migration" required:"true"`
6064
}
6165

6266
// NewConfig creates a new Worker config from the specified YAML file.
@@ -73,5 +77,9 @@ func NewConfig(path string) (*Config, error) {
7377
cfg.Whitelist.PrivilegedIdentityLevel = sonm.IdentityLevel_ANONYMOUS
7478
}
7579

80+
if cfg.Migration.MigrationDWH == nil && *cfg.Migration.Enabled {
81+
return nil, fmt.Errorf("dwh config is required for enabled migrartion")
82+
}
83+
7684
return cfg, nil
7785
}

‎insonmnia/worker/server.go

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ import (
4343
"github.com/sonm-io/core/insonmnia/hardware"
4444
"github.com/sonm-io/core/insonmnia/hardware/disk"
4545
"github.com/sonm-io/core/insonmnia/matcher"
46+
"github.com/sonm-io/core/insonmnia/migration"
4647
"github.com/sonm-io/core/insonmnia/npp"
4748
"github.com/sonm-io/core/insonmnia/resource"
4849
"github.com/sonm-io/core/insonmnia/state"
@@ -172,6 +173,8 @@ type Worker struct {
172173
country *geoip2.Country
173174
// Hardware metrics for various hardware types.
174175
metrics *metrics.Handler
176+
177+
shredder migration.MarketShredder
175178
}
176179

177180
func NewWorker(cfg *Config, storage *state.Storage, options ...Option) (*Worker, error) {
@@ -245,6 +248,10 @@ func NewWorker(cfg *Config, storage *state.Storage, options ...Option) (*Worker,
245248
return nil, err
246249
}
247250

251+
if err := m.setupMarketShredder(); err != nil {
252+
return nil, err
253+
}
254+
248255
dg.CancelExec()
249256

250257
return m, nil
@@ -530,6 +537,15 @@ func (m *Worker) setupSSH(view OverseerView) error {
530537
return nil
531538
}
532539

540+
func (m *Worker) setupMarketShredder() error {
541+
shredder, err := migration.NewV1MarketShredder(m.ctx, m.cfg.Migration, m.credentials)
542+
if err != nil {
543+
return err
544+
}
545+
m.shredder = shredder
546+
return nil
547+
}
548+
533549
func (m *Worker) loadOrGenerateSSHSigner() (ssh.Signer, error) {
534550
var privateKeyData []byte
535551
ok, err := m.storage.Load(sshPrivateKeyKey, &privateKeyData)
@@ -594,6 +610,15 @@ func (m *Worker) Serve() error {
594610

595611
return m.externalGrpc.Serve(m.listener)
596612
})
613+
wg.Go(func() error {
614+
log.S(ctx).Infof("shredding old market")
615+
if err := m.shredder.WeDontNeedNoWaterLetTheMothefuckerBurn(ctx, m.key); err != nil {
616+
log.S(ctx).Warnf("failed to shred old market")
617+
return err
618+
}
619+
log.S(ctx).Infof("successfully shredded old market")
620+
return nil
621+
})
597622

598623
<-ctx.Done()
599624
m.Close()

0 commit comments

Comments
 (0)