-
Notifications
You must be signed in to change notification settings - Fork 93
/
meta.go
157 lines (145 loc) · 3.95 KB
/
meta.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
package miner
import (
"context"
"fmt"
"sync"
"time"
"github.com/filecoin-project/go-address"
cbor "github.com/ipfs/go-ipld-cbor"
"github.com/textileio/filecoin/iplocation"
"go.opencensus.io/stats"
"go.opencensus.io/tag"
)
var (
metadataRefreshInterval = time.Second * 45
pingTimeout = time.Second * 3
pingRateLim = 100
)
var (
dsKeyMetaIndex = dsBase.ChildString("meta")
)
// metaWorker makes a pass on refreshing metadata information about known miners
func (mi *MinerIndex) metaWorker() {
defer func() { mi.finished <- struct{}{} }()
mi.chMeta <- struct{}{}
for {
select {
case <-mi.ctx.Done():
log.Info("graceful shutdown of meta updater")
return
case _, ok := <-mi.chMeta:
if !ok {
log.Info("meta worker channel closed")
return
}
log.Info("updating meta index...")
// ToDo: coud have smarter ways of electing which addrs to refresh, and then
// doing a merge. Will depend if this too slow, but might not be the case
mi.lock.Lock()
addrs := make([]string, 0, len(mi.index.Chain.Power))
for addr := range mi.index.Chain.Power {
addrs = append(addrs, addr)
}
mi.lock.Unlock()
newIndex, err := updateMetaIndex(mi.ctx, mi.api, mi.h, mi.lr, addrs)
if err != nil {
log.Errorf("error when updating meta index: %s", err)
break
}
if err := mi.persistMetaIndex(newIndex); err != nil {
log.Errorf("error when persisting meta index: %s", err)
}
mi.lock.Lock()
mi.index.Meta = newIndex
mi.lock.Unlock()
mi.signaler.Signal() // ToDo: consider a finer-grained signaling
}
}
}
// updateMetaIndex generates a new index that contains fresh metadata information
// of addrs miners.
func updateMetaIndex(ctx context.Context, api API, h P2PHost, lr iplocation.LocationResolver, addrs []string) (MetaIndex, error) {
index := MetaIndex{
Info: make(map[string]Meta),
}
rl := make(chan struct{}, pingRateLim)
var lock sync.Mutex
for i, a := range addrs {
rl <- struct{}{}
go func(a string) {
defer func() { <-rl }()
si, err := getMeta(ctx, api, h, lr, a)
if err != nil {
log.Debugf("error getting static info: %s", err)
return
}
lock.Lock()
index.Info[a] = si
lock.Unlock()
}(a)
if i%100 == 0 {
stats.Record(context.Background(), mMetaRefreshProgress.M(float64(i)/float64(len(addrs))))
}
}
for i := 0; i < pingRateLim; i++ {
rl <- struct{}{}
}
for _, v := range index.Info {
if v.Online {
index.Online++
}
}
index.Offline = uint32(len(addrs)) - index.Online
stats.Record(context.Background(), mMetaRefreshProgress.M(1))
ctx, _ = tag.New(context.Background(), tag.Insert(metricOnline, "online"))
stats.Record(ctx, mMetaPingCount.M(int64(index.Online)))
ctx, _ = tag.New(context.Background(), tag.Insert(metricOnline, "offline"))
stats.Record(ctx, mMetaPingCount.M(int64(index.Offline)))
return index, nil
}
// getMeta returns fresh metadata information about a miner
func getMeta(ctx context.Context, c API, h P2PHost, lr iplocation.LocationResolver, straddr string) (Meta, error) {
si := Meta{
LastUpdated: time.Now(),
}
addr, err := address.NewFromString(straddr)
if err != nil {
return si, err
}
pid, err := c.StateMinerPeerID(ctx, addr, nil)
if err != nil {
return si, err
}
ctx, cancel := context.WithTimeout(ctx, pingTimeout)
defer cancel()
if alive := h.Ping(ctx, pid); !alive {
return si, fmt.Errorf("peer didn't pong")
}
si.Online = true
if av := h.GetAgentVersion(pid); av != "" {
si.UserAgent = av
}
addrs := h.Addrs(pid)
if len(addrs) == 0 {
return si, nil
}
if l, err := lr.Resolve(addrs); err == nil {
si.Location = Location{
Country: l.Country,
Latitude: l.Latitude,
Longitude: l.Longitude,
}
}
return si, nil
}
// persisteMetaIndex saves to datastore a new MetaIndex
func (mi *MinerIndex) persistMetaIndex(index MetaIndex) error {
buf, err := cbor.DumpObject(index)
if err != nil {
return err
}
if err := mi.ds.Put(dsKeyMetaIndex, buf); err != nil {
return err
}
return nil
}