cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge branch 'master' into next

Browse Source
This commit is contained in:
Aayush Rajasekaran
2021-12-13 13:24:28 -05:00
parent ac31651a42 dc26f3bd74
commit 80d5e52923
138 changed files with 7416 additions and 2140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/lotus-miner/main.go
+5 -3
View File
@@ -7,7 +7,6 @@ import (
"github.com/fatih/color"
logging "github.com/ipfs/go-log/v2"
"github.com/urfave/cli/v2"
"go.opencensus.io/trace"
"golang.org/x/xerrors"
cliutil "github.com/filecoin-project/lotus/cli/util"
@@ -55,10 +54,11 @@ func main() {
lcli.WithCategory("storage", sealingCmd),
lcli.WithCategory("retrieval", piecesCmd),
}
jaeger := tracing.SetupJaegerTracing("lotus")
defer func() {
if jaeger != nil {
jaeger.Flush()
_ = jaeger.ForceFlush(context.Background())
}
}()
@@ -66,7 +66,9 @@ func main() {
cmd := cmd
originBefore := cmd.Before
cmd.Before = func(cctx *cli.Context) error {
trace.UnregisterExporter(jaeger)
if jaeger != nil {
_ = jaeger.Shutdown(cctx.Context)
}
jaeger = tracing.SetupJaegerTracing("lotus/" + cmd.Name)
if cctx.IsSet("color") {
cmd/lotus-miner/pieces.go
+53 -1
View File
@@ -6,6 +6,7 @@ import (
"text/tabwriter"
lcli "github.com/filecoin-project/lotus/cli"
"github.com/filecoin-project/lotus/lib/tablewriter"
"github.com/ipfs/go-cid"
"github.com/urfave/cli/v2"
)
@@ -48,6 +49,12 @@ var piecesListPiecesCmd = &cli.Command{
var piecesListCidInfosCmd = &cli.Command{
Name: "list-cids",
Usage: "list registered payload CIDs",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "verbose",
Aliases: []string{"v"},
},
},
Action: func(cctx *cli.Context) error {
nodeApi, closer, err := lcli.GetMarketsAPI(cctx)
if err != nil {
@@ -61,9 +68,54 @@ var piecesListCidInfosCmd = &cli.Command{
return err
}
w := tablewriter.New(tablewriter.Col("CID"),
tablewriter.Col("Piece"),
tablewriter.Col("BlockOffset"),
tablewriter.Col("BlockLen"),
tablewriter.Col("Deal"),
tablewriter.Col("Sector"),
tablewriter.Col("DealOffset"),
tablewriter.Col("DealLen"),
)
for _, c := range cids {
fmt.Println(c)
if !cctx.Bool("verbose") {
fmt.Println(c)
continue
}
ci, err := nodeApi.PiecesGetCIDInfo(ctx, c)
if err != nil {
fmt.Printf("Error getting CID info: %s\n", err)
continue
}
for _, location := range ci.PieceBlockLocations {
pi, err := nodeApi.PiecesGetPieceInfo(ctx, location.PieceCID)
if err != nil {
fmt.Printf("Error getting piece info: %s\n", err)
continue
}
for _, deal := range pi.Deals {
w.Write(map[string]interface{}{
"CID": c,
"Piece": location.PieceCID,
"BlockOffset": location.RelOffset,
"BlockLen": location.BlockSize,
"Deal": deal.DealID,
"Sector": deal.SectorID,
"DealOffset": deal.Offset,
"DealLen": deal.Length,
})
}
}
}
if cctx.Bool("verbose") {
return w.Flush(os.Stdout)
}
return nil
},
}
cmd/lotus-miner/retrieval-deals.go
+5
View File
@@ -3,6 +3,7 @@ package main
import (
"fmt"
"os"
"sort"
"text/tabwriter"
"github.com/docker/go-units"
@@ -137,6 +138,10 @@ var retrievalDealsListCmd = &cli.Command{
return err
}
sort.Slice(deals, func(i, j int) bool {
return deals[i].ID < deals[j].ID
})
w := tabwriter.NewWriter(os.Stdout, 2, 4, 2, ' ', 0)
_, _ = fmt.Fprintf(w, "Receiver\tDealID\tPayload\tState\tPricePerByte\tBytesSent\tMessage\n")
cmd/lotus-miner/sealing.go
+42 -43
View File
@@ -4,6 +4,7 @@ import (
"encoding/hex"
"encoding/json"
"fmt"
"math"
"os"
"sort"
"strings"
@@ -32,6 +33,17 @@ var sealingCmd = &cli.Command{
},
}
var barCols = float64(64)
func barString(total, y, g float64) string {
yBars := int(math.Round(y / total * barCols))
gBars := int(math.Round(g / total * barCols))
eBars := int(barCols) - yBars - gBars
return color.YellowString(strings.Repeat("|", yBars)) +
color.GreenString(strings.Repeat("|", gBars)) +
strings.Repeat(" ", eBars)
}
var sealingWorkersCmd = &cli.Command{
Name: "workers",
Usage: "list workers",
@@ -77,7 +89,7 @@ var sealingWorkersCmd = &cli.Command{
for _, stat := range st {
gpuUse := "not "
gpuCol := color.FgBlue
if stat.GpuUsed {
if stat.GpuUsed > 0 {
gpuCol = color.FgGreen
gpuUse = ""
}
@@ -89,56 +101,43 @@ var sealingWorkersCmd = &cli.Command{
fmt.Printf("Worker %s, host %s%s\n", stat.id, color.MagentaString(stat.Info.Hostname), disabled)
var barCols = uint64(64)
cpuBars := int(stat.CpuUse * barCols / stat.Info.Resources.CPUs)
cpuBar := strings.Repeat("|", cpuBars)
if int(barCols)-cpuBars >= 0 {
cpuBar += strings.Repeat(" ", int(barCols)-cpuBars)
}
fmt.Printf("\tCPU: [%s] %d/%d core(s) in use\n",
color.GreenString(cpuBar), stat.CpuUse, stat.Info.Resources.CPUs)
barString(float64(stat.Info.Resources.CPUs), 0, float64(stat.CpuUse)), stat.CpuUse, stat.Info.Resources.CPUs)
ramBarsRes := int(stat.Info.Resources.MemReserved * barCols / stat.Info.Resources.MemPhysical)
ramBarsUsed := int(stat.MemUsedMin * barCols / stat.Info.Resources.MemPhysical)
ramRepeatSpace := int(barCols) - (ramBarsUsed + ramBarsRes)
colorFunc := color.YellowString
if ramRepeatSpace < 0 {
ramRepeatSpace = 0
colorFunc = color.RedString
ramTotal := stat.Info.Resources.MemPhysical
ramTasks := stat.MemUsedMin
ramUsed := stat.Info.Resources.MemUsed
var ramReserved uint64 = 0
if ramUsed > ramTasks {
ramReserved = ramUsed - ramTasks
}
ramBar := colorFunc(strings.Repeat("|", ramBarsRes)) +
color.GreenString(strings.Repeat("|", ramBarsUsed)) +
strings.Repeat(" ", ramRepeatSpace)
vmem := stat.Info.Resources.MemPhysical + stat.Info.Resources.MemSwap
vmemBarsRes := int(stat.Info.Resources.MemReserved * barCols / vmem)
vmemBarsUsed := int(stat.MemUsedMax * barCols / vmem)
vmemRepeatSpace := int(barCols) - (vmemBarsUsed + vmemBarsRes)
colorFunc = color.YellowString
if vmemRepeatSpace < 0 {
vmemRepeatSpace = 0
colorFunc = color.RedString
}
vmemBar := colorFunc(strings.Repeat("|", vmemBarsRes)) +
color.GreenString(strings.Repeat("|", vmemBarsUsed)) +
strings.Repeat(" ", vmemRepeatSpace)
ramBar := barString(float64(ramTotal), float64(ramReserved), float64(ramTasks))
fmt.Printf("\tRAM: [%s] %d%% %s/%s\n", ramBar,
(stat.Info.Resources.MemReserved+stat.MemUsedMin)*100/stat.Info.Resources.MemPhysical,
types.SizeStr(types.NewInt(stat.Info.Resources.MemReserved+stat.MemUsedMin)),
(ramTasks+ramReserved)*100/stat.Info.Resources.MemPhysical,
types.SizeStr(types.NewInt(ramTasks+ramUsed)),
types.SizeStr(types.NewInt(stat.Info.Resources.MemPhysical)))
fmt.Printf("\tVMEM: [%s] %d%% %s/%s\n", vmemBar,
(stat.Info.Resources.MemReserved+stat.MemUsedMax)*100/vmem,
types.SizeStr(types.NewInt(stat.Info.Resources.MemReserved+stat.MemUsedMax)),
types.SizeStr(types.NewInt(vmem)))
vmemTotal := stat.Info.Resources.MemPhysical + stat.Info.Resources.MemSwap
vmemTasks := stat.MemUsedMax
vmemUsed := stat.Info.Resources.MemUsed + stat.Info.Resources.MemSwapUsed
var vmemReserved uint64 = 0
if vmemUsed > vmemTasks {
vmemReserved = vmemUsed - vmemTasks
}
vmemBar := barString(float64(vmemTotal), float64(vmemReserved), float64(vmemTasks))
fmt.Printf("\tVMEM: [%s] %d%% %s/%s\n", vmemBar,
(vmemTasks+vmemReserved)*100/vmemTotal,
types.SizeStr(types.NewInt(vmemTasks+vmemReserved)),
types.SizeStr(types.NewInt(vmemTotal)))
if len(stat.Info.Resources.GPUs) > 0 {
gpuBar := barString(float64(len(stat.Info.Resources.GPUs)), 0, stat.GpuUsed)
fmt.Printf("\tGPU: [%s] %.f%% %.2f/%d gpu(s) in use\n", color.GreenString(gpuBar),
stat.GpuUsed*100/float64(len(stat.Info.Resources.GPUs)),
stat.GpuUsed, len(stat.Info.Resources.GPUs))
}
for _, gpu := range stat.Info.Resources.GPUs {
fmt.Printf("\tGPU: %s\n", color.New(gpuCol).Sprintf("%s, %sused", gpu, gpuUse))
}
cmd/lotus-miner/storage.go
+59 -1
View File
@@ -48,6 +48,7 @@ stored while moving through the sealing pipeline (references as 'seal').`,
storageListCmd,
storageFindCmd,
storageCleanupCmd,
storageLocks,
},
}
@@ -95,6 +96,14 @@ over time
Name: "max-storage",
Usage: "(for init) limit storage space for sectors (expensive for very large paths!)",
},
&cli.StringSliceFlag{
Name: "groups",
Usage: "path group names",
},
&cli.StringSliceFlag{
Name: "allow-to",
Usage: "path groups allowed to pull data from this path (allow all if not specified)",
},
},
Action: func(cctx *cli.Context) error {
nodeApi, closer, err := lcli.GetStorageMinerAPI(cctx)
@@ -142,6 +151,8 @@ over time
CanSeal: cctx.Bool("seal"),
CanStore: cctx.Bool("store"),
MaxStorage: uint64(maxStor),
Groups: cctx.StringSlice("groups"),
AllowTo: cctx.StringSlice("allow-to"),
}
if !(cfg.CanStore || cfg.CanSeal) {
@@ -322,10 +333,17 @@ var storageListCmd = &cli.Command{
if si.CanStore {
fmt.Print(color.CyanString("Store"))
}
fmt.Println("")
} else {
fmt.Print(color.HiYellowString("Use: ReadOnly"))
}
fmt.Println()
if len(si.Groups) > 0 {
fmt.Printf("\tGroups: %s\n", strings.Join(si.Groups, ", "))
}
if len(si.AllowTo) > 0 {
fmt.Printf("\tAllowTo: %s\n", strings.Join(si.AllowTo, ", "))
}
if localPath, ok := local[s.ID]; ok {
fmt.Printf("\tLocal: %s\n", color.GreenString(localPath))
@@ -741,3 +759,43 @@ func cleanupRemovedSectorData(ctx context.Context, api api.StorageMiner, napi v0
return nil
}
var storageLocks = &cli.Command{
Name: "locks",
Usage: "show active sector locks",
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetStorageMinerAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
locks, err := api.StorageGetLocks(ctx)
if err != nil {
return err
}
for _, lock := range locks.Locks {
st, err := api.SectorsStatus(ctx, lock.Sector.Number, false)
if err != nil {
return xerrors.Errorf("getting sector status(%d): %w", lock.Sector.Number, err)
}
lockstr := fmt.Sprintf("%d\t%s\t", lock.Sector.Number, color.New(stateOrder[sealing.SectorState(st.State)].col).Sprint(st.State))
for i := 0; i < storiface.FileTypes; i++ {
if lock.Write[i] > 0 {
lockstr += fmt.Sprintf("%s(%s) ", storiface.SectorFileType(1<<i).String(), color.RedString("W"))
}
if lock.Read[i] > 0 {
lockstr += fmt.Sprintf("%s(%s:%d) ", storiface.SectorFileType(1<<i).String(), color.GreenString("R"), lock.Read[i])
}
}
fmt.Println(lockstr)
}
return nil
},
}
cmd/lotus-pcr/main.go
+13 -12
View File
@@ -17,6 +17,7 @@ import (
"time"
"github.com/filecoin-project/lotus/chain/actors/builtin"
lcli "github.com/filecoin-project/lotus/cli"
miner2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/miner"
@@ -41,7 +42,7 @@ import (
"github.com/filecoin-project/lotus/chain/actors/builtin/market"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/tools/stats"
"github.com/filecoin-project/lotus/tools/stats/sync"
)
var log = logging.Logger("main")
@@ -160,15 +161,15 @@ var findMinersCmd = &cli.Command{
},
Action: func(cctx *cli.Context) error {
ctx := context.Background()
api, closer, err := stats.GetFullNodeAPI(cctx.Context, cctx.String("lotus-path"))
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
log.Fatal(err)
return err
}
defer closer()
if !cctx.Bool("no-sync") {
if err := stats.WaitForSyncComplete(ctx, api); err != nil {
log.Fatal(err)
if err := sync.SyncWait(ctx, api); err != nil {
return err
}
}
@@ -245,7 +246,7 @@ var recoverMinersCmd = &cli.Command{
},
Action: func(cctx *cli.Context) error {
ctx := context.Background()
api, closer, err := stats.GetFullNodeAPI(cctx.Context, cctx.String("lotus-path"))
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
log.Fatal(err)
}
@@ -266,8 +267,8 @@ var recoverMinersCmd = &cli.Command{
}
if !cctx.Bool("no-sync") {
if err := stats.WaitForSyncComplete(ctx, api); err != nil {
log.Fatal(err)
if err := sync.SyncWait(ctx, api); err != nil {
return err
}
}
@@ -427,7 +428,7 @@ var runCmd = &cli.Command{
}()
ctx := context.Background()
api, closer, err := stats.GetFullNodeAPI(cctx.Context, cctx.String("lotus-path"))
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
log.Fatal(err)
}
@@ -448,12 +449,12 @@ var runCmd = &cli.Command{
}
if !cctx.Bool("no-sync") {
if err := stats.WaitForSyncComplete(ctx, api); err != nil {
log.Fatal(err)
if err := sync.SyncWait(ctx, api); err != nil {
return err
}
}
tipsetsCh, err := stats.GetTips(ctx, api, r.Height(), cctx.Int("head-delay"))
tipsetsCh, err := sync.BufferedTipsetChannel(ctx, api, r.Height(), cctx.Int("head-delay"))
if err != nil {
log.Fatal(err)
}
cmd/lotus-seal-worker/info.go
+5 -2
View File
@@ -58,8 +58,11 @@ var infoCmd = &cli.Command{
fmt.Printf("Hostname: %s\n", info.Hostname)
fmt.Printf("CPUs: %d; GPUs: %v\n", info.Resources.CPUs, info.Resources.GPUs)
fmt.Printf("RAM: %s; Swap: %s\n", types.SizeStr(types.NewInt(info.Resources.MemPhysical)), types.SizeStr(types.NewInt(info.Resources.MemSwap)))
fmt.Printf("Reserved memory: %s\n", types.SizeStr(types.NewInt(info.Resources.MemReserved)))
fmt.Printf("RAM: %s/%s; Swap: %s/%s\n",
types.SizeStr(types.NewInt(info.Resources.MemUsed)),
types.SizeStr(types.NewInt(info.Resources.MemPhysical)),
types.SizeStr(types.NewInt(info.Resources.MemSwapUsed)),
types.SizeStr(types.NewInt(info.Resources.MemSwap)))
fmt.Printf("Task types: ")
for _, t := range ttList(tt) {
cmd/lotus-seal-worker/main.go
+1
View File
@@ -60,6 +60,7 @@ func main() {
storageCmd,
setCmd,
waitQuietCmd,
resourcesCmd,
tasksCmd,
}
cmd/lotus-seal-worker/resources.go
+72
View File
@@ -0,0 +1,72 @@
package main
import (
"fmt"
"os"
"sort"
"github.com/urfave/cli/v2"
"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
)
var resourcesCmd = &cli.Command{
Name: "resources",
Usage: "Manage resource table overrides",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "all",
Usage: "print all resource envvars",
},
&cli.BoolFlag{
Name: "default",
Usage: "print default resource envvars",
},
},
Action: func(cctx *cli.Context) error {
def := map[string]string{}
set := map[string]string{}
all := map[string]string{}
_, err := storiface.ParseResourceEnv(func(key, d string) (string, bool) {
if d != "" {
all[key] = d
def[key] = d
}
s, ok := os.LookupEnv(key)
if ok {
all[key] = s
set[key] = s
}
return s, ok
})
if err != nil {
return err
}
printMap := func(m map[string]string) {
var arr []string
for k, v := range m {
arr = append(arr, fmt.Sprintf("%s=%s", k, v))
}
sort.Strings(arr)
for _, s := range arr {
fmt.Println(s)
}
}
if cctx.Bool("default") {
printMap(def)
} else {
if cctx.Bool("all") {
printMap(all)
} else {
printMap(set)
}
}
return nil
},
}
cmd/lotus-seal-worker/storage.go
+10
View File
@@ -51,6 +51,14 @@ var storageAttachCmd = &cli.Command{
Name: "max-storage",
Usage: "(for init) limit storage space for sectors (expensive for very large paths!)",
},
&cli.StringSliceFlag{
Name: "groups",
Usage: "path group names",
},
&cli.StringSliceFlag{
Name: "allow-to",
Usage: "path groups allowed to pull data from this path (allow all if not specified)",
},
},
Action: func(cctx *cli.Context) error {
nodeApi, closer, err := lcli.GetWorkerAPI(cctx)
@@ -98,6 +106,8 @@ var storageAttachCmd = &cli.Command{
CanSeal: cctx.Bool("seal"),
CanStore: cctx.Bool("store"),
MaxStorage: uint64(maxStor),
Groups: cctx.StringSlice("groups"),
AllowTo: cctx.StringSlice("allow-to"),
}
if !(cfg.CanStore || cfg.CanSeal) {
cmd/lotus-shed/balancer.go
+222
View File
@@ -0,0 +1,222 @@
package main
import (
"fmt"
"strings"
"time"
"github.com/ipfs/go-cid"
"github.com/urfave/cli/v2"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/go-state-types/exitcode"
lapi "github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
lcli "github.com/filecoin-project/lotus/cli"
)
var balancerCmd = &cli.Command{
Name: "balancer",
Usage: "Utility for balancing tokens between multiple wallets",
Description: `Tokens are balanced based on the specification provided in arguments
Each argument specifies an address, role, and role parameters separated by ';'
Supported roles:
- request;[addr];[low];[high] - request tokens when balance drops to [low], topping up to [high]
- provide;[addr];[min] - provide tokens to other addresses as long as the balance is above [min]
`,
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetFullNodeAPIV1(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
type request struct {
addr address.Address
low, high abi.TokenAmount
}
type provide struct {
addr address.Address
min abi.TokenAmount
}
var requests []request
var provides []provide
for i, s := range cctx.Args().Slice() {
ss := strings.Split(s, ";")
switch ss[0] {
case "request":
if len(ss) != 4 {
return xerrors.Errorf("request role needs 4 parameters (arg %d)", i)
}
addr, err := address.NewFromString(ss[1])
if err != nil {
return xerrors.Errorf("parsing address in arg %d: %w", i, err)
}
low, err := types.ParseFIL(ss[2])
if err != nil {
return xerrors.Errorf("parsing low in arg %d: %w", i, err)
}
high, err := types.ParseFIL(ss[3])
if err != nil {
return xerrors.Errorf("parsing high in arg %d: %w", i, err)
}
if abi.TokenAmount(low).GreaterThanEqual(abi.TokenAmount(high)) {
return xerrors.Errorf("low must be less than high in arg %d", i)
}
requests = append(requests, request{
addr: addr,
low: abi.TokenAmount(low),
high: abi.TokenAmount(high),
})
case "provide":
if len(ss) != 3 {
return xerrors.Errorf("provide role needs 3 parameters (arg %d)", i)
}
addr, err := address.NewFromString(ss[1])
if err != nil {
return xerrors.Errorf("parsing address in arg %d: %w", i, err)
}
min, err := types.ParseFIL(ss[2])
if err != nil {
return xerrors.Errorf("parsing min in arg %d: %w", i, err)
}
provides = append(provides, provide{
addr: addr,
min: abi.TokenAmount(min),
})
default:
return xerrors.Errorf("unknown role '%s' in arg %d", ss[0], i)
}
}
if len(provides) == 0 {
return xerrors.Errorf("no provides specified")
}
if len(requests) == 0 {
return xerrors.Errorf("no requests specified")
}
const confidence = 16
var notifs <-chan []*lapi.HeadChange
for {
if notifs == nil {
notifs, err = api.ChainNotify(ctx)
if err != nil {
return xerrors.Errorf("chain notify error: %w", err)
}
}
var ts *types.TipSet
loop:
for {
time.Sleep(150 * time.Millisecond)
select {
case n := <-notifs:
for _, change := range n {
if change.Type != store.HCApply {
continue
}
ts = change.Val
}
case <-ctx.Done():
return nil
default:
break loop
}
}
type send struct {
to address.Address
amt abi.TokenAmount
filled bool
}
var toSend []*send
for _, req := range requests {
bal, err := api.StateGetActor(ctx, req.addr, ts.Key())
if err != nil {
return err
}
if bal.Balance.LessThan(req.low) {
toSend = append(toSend, &send{
to: req.addr,
amt: big.Sub(req.high, bal.Balance),
})
}
}
for _, s := range toSend {
fmt.Printf("REQUEST %s for %s\n", types.FIL(s.amt), s.to)
}
var msgs []cid.Cid
for _, prov := range provides {
bal, err := api.StateGetActor(ctx, prov.addr, ts.Key())
if err != nil {
return err
}
avail := big.Sub(bal.Balance, prov.min)
for _, s := range toSend {
if s.filled {
continue
}
if avail.LessThan(s.amt) {
continue
}
m, err := api.MpoolPushMessage(ctx, &types.Message{
From: prov.addr,
To: s.to,
Value: s.amt,
}, nil)
if err != nil {
fmt.Printf("SEND ERROR %s\n", err.Error())
}
fmt.Printf("SEND %s; %s from %s TO %s\n", m.Cid(), types.FIL(s.amt), s.to, prov.addr)
msgs = append(msgs, m.Cid())
s.filled = true
avail = big.Sub(avail, s.amt)
}
}
if len(msgs) > 0 {
fmt.Printf("WAITING FOR %d MESSAGES\n", len(msgs))
}
for _, msg := range msgs {
ml, err := api.StateWaitMsg(ctx, msg, confidence, lapi.LookbackNoLimit, true)
if err != nil {
return err
}
if ml.Receipt.ExitCode != exitcode.Ok {
fmt.Printf("MSG %s NON-ZERO EXITCODE: %s\n", msg, ml.Receipt.ExitCode)
}
}
}
},
}
cmd/lotus-shed/main.go
+1
View File
@@ -64,6 +64,7 @@ func main() {
splitstoreCmd,
fr32Cmd,
chainCmd,
balancerCmd,
}
app := &cli.App{
cmd/lotus-shed/msg.go
+9
View File
@@ -148,6 +148,15 @@ func printMessage(cctx *cli.Context, msg *types.Message) error {
fmt.Println("Params:", p)
if msg, err := messageFromBytes(cctx, msg.Params); err == nil {
fmt.Println("---")
color.Red("Params message:")
if err := printMessage(cctx, msg.VMMessage()); err != nil {
return err
}
}
return nil
}
cmd/lotus-shed/sectors.go
+114 -24
View File
@@ -2,11 +2,14 @@ package main
import (
"bytes"
"context"
"encoding/base64"
"encoding/binary"
"fmt"
"image"
"image/color"
"image/png"
"io"
"os"
"sort"
"strconv"
@@ -23,6 +26,7 @@ import (
miner2 "github.com/filecoin-project/specs-actors/v2/actors/builtin/miner"
"github.com/filecoin-project/lotus/api/v0api"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
"github.com/filecoin-project/lotus/chain/types"
@@ -38,6 +42,7 @@ var sectorsCmd = &cli.Command{
terminateSectorCmd,
terminateSectorPenaltyEstimationCmd,
visAllocatedSectorsCmd,
dumpRLESectorCmd,
},
}
@@ -275,6 +280,113 @@ var terminateSectorPenaltyEstimationCmd = &cli.Command{
},
}
func activeMiners(ctx context.Context, api v0api.FullNode) ([]address.Address, error) {
miners, err := api.StateListMiners(ctx, types.EmptyTSK)
if err != nil {
return nil, err
}
powCache := make(map[address.Address]types.BigInt)
var lk sync.Mutex
parmap.Par(32, miners, func(a address.Address) {
pow, err := api.StateMinerPower(ctx, a, types.EmptyTSK)
lk.Lock()
if err == nil {
powCache[a] = pow.MinerPower.QualityAdjPower
} else {
powCache[a] = types.NewInt(0)
}
lk.Unlock()
})
sort.Slice(miners, func(i, j int) bool {
return powCache[miners[i]].GreaterThan(powCache[miners[j]])
})
n := sort.Search(len(miners), func(i int) bool {
pow := powCache[miners[i]]
return pow.IsZero()
})
return append(miners[0:0:0], miners[:n]...), nil
}
var dumpRLESectorCmd = &cli.Command{
Name: "dump-rles",
Usage: "Dump AllocatedSectors RLEs from miners passed as arguments as run lengths in uint64 LE format.\nIf no arguments are passed, dumps all active miners in the state tree.",
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
var miners []address.Address
if cctx.NArg() == 0 {
miners, err = activeMiners(ctx, api)
if err != nil {
return xerrors.Errorf("getting active miners: %w", err)
}
} else {
for _, mS := range cctx.Args().Slice() {
mA, err := address.NewFromString(mS)
if err != nil {
return xerrors.Errorf("parsing address '%s': %w", mS, err)
}
miners = append(miners, mA)
}
}
wbuf := make([]byte, 8)
buf := &bytes.Buffer{}
for i := 0; i < len(miners); i++ {
buf.Reset()
err := func() error {
state, err := api.StateReadState(ctx, miners[i], types.EmptyTSK)
if err != nil {
return xerrors.Errorf("getting state: %+v", err)
}
allocSString := state.State.(map[string]interface{})["AllocatedSectors"].(map[string]interface{})["/"].(string)
allocCid, err := cid.Decode(allocSString)
if err != nil {
return xerrors.Errorf("decoding cid: %+v", err)
}
rle, err := api.ChainReadObj(ctx, allocCid)
if err != nil {
return xerrors.Errorf("reading AllocatedSectors: %+v", err)
}
var bf bitfield.BitField
err = bf.UnmarshalCBOR(bytes.NewReader(rle))
if err != nil {
return xerrors.Errorf("decoding bitfield: %w", err)
}
ri, err := bf.RunIterator()
if err != nil {
return xerrors.Errorf("creating iterator: %w", err)
}
for ri.HasNext() {
run, err := ri.NextRun()
if err != nil {
return xerrors.Errorf("getting run: %w", err)
}
binary.LittleEndian.PutUint64(wbuf, run.Len)
buf.Write(wbuf)
}
_, err = io.Copy(os.Stdout, buf)
if err != nil {
return xerrors.Errorf("copy: %w", err)
}
return nil
}()
if err != nil {
log.Errorf("miner %d: %s: %+v", i, miners[i], err)
}
}
return nil
},
}
var visAllocatedSectorsCmd = &cli.Command{
Name: "vis-allocated",
Usage: "Produces a html with visualisation of allocated sectors",
@@ -287,32 +399,10 @@ var visAllocatedSectorsCmd = &cli.Command{
ctx := lcli.ReqContext(cctx)
var miners []address.Address
if cctx.NArg() == 0 {
miners, err = api.StateListMiners(ctx, types.EmptyTSK)
miners, err = activeMiners(ctx, api)
if err != nil {
return err
return xerrors.Errorf("getting active miners: %w", err)
}
powCache := make(map[address.Address]types.BigInt)
var lk sync.Mutex
parmap.Par(32, miners, func(a address.Address) {
pow, err := api.StateMinerPower(ctx, a, types.EmptyTSK)
lk.Lock()
if err == nil {
powCache[a] = pow.MinerPower.QualityAdjPower
} else {
powCache[a] = types.NewInt(0)
}
lk.Unlock()
})
sort.Slice(miners, func(i, j int) bool {
return powCache[miners[i]].GreaterThan(powCache[miners[j]])
})
n := sort.Search(len(miners), func(i int) bool {
pow := powCache[miners[i]]
log.Infof("pow @%d = %s", i, pow)
return pow.IsZero()
})
miners = miners[:n]
} else {
for _, mS := range cctx.Args().Slice() {
mA, err := address.NewFromString(mS)
cmd/lotus-stats/docker-compose.yml
+1 -1
View File
@@ -2,7 +2,7 @@ version: '3'
services:
influxdb:
image: influxdb:latest
image: influxdb:1.8
container_name: influxdb
ports:
- "18086:8086"
cmd/lotus-stats/main.go
+126 -20
View File
@@ -2,18 +2,36 @@ package main
import (
"context"
"net/http"
_ "net/http/pprof"
"os"
"time"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/build"
lcli "github.com/filecoin-project/lotus/cli"
"github.com/filecoin-project/lotus/tools/stats"
"github.com/filecoin-project/lotus/tools/stats/influx"
"github.com/filecoin-project/lotus/tools/stats/ipldstore"
"github.com/filecoin-project/lotus/tools/stats/metrics"
"github.com/filecoin-project/lotus/tools/stats/points"
"github.com/filecoin-project/lotus/tools/stats/sync"
logging "github.com/ipfs/go-log/v2"
"github.com/urfave/cli/v2"
"contrib.go.opencensus.io/exporter/prometheus"
stats "go.opencensus.io/stats"
"go.opencensus.io/stats/view"
)
var log = logging.Logger("stats")
func init() {
if err := view.Register(metrics.DefaultViews...); err != nil {
log.Fatal(err)
}
}
func main() {
local := []*cli.Command{
runCmd,
@@ -37,7 +55,7 @@ func main() {
},
},
Before: func(cctx *cli.Context) error {
return logging.SetLogLevel("stats", cctx.String("log-level"))
return logging.SetLogLevelRegex("stats/*", cctx.String("log-level"))
},
Commands: local,
}
@@ -104,6 +122,12 @@ var runCmd = &cli.Command{
Usage: "do not wait for chain sync to complete",
Value: false,
},
&cli.IntFlag{
Name: "ipld-store-cache-size",
Usage: "size of lru cache for ChainReadObj",
EnvVars: []string{"LOTUS_STATS_IPLD_STORE_CACHE_SIZE"},
Value: 2 << 15,
},
},
Action: func(cctx *cli.Context) error {
ctx := context.Background()
@@ -118,30 +142,35 @@ var runCmd = &cli.Command{
influxPasswordFlag := cctx.String("influx-password")
influxDatabaseFlag := cctx.String("influx-database")
ipldStoreCacheSizeFlag := cctx.Int("ipld-store-cache-size")
log.Infow("opening influx client", "hostname", influxHostnameFlag, "username", influxUsernameFlag, "database", influxDatabaseFlag)
influx, err := stats.InfluxClient(influxHostnameFlag, influxUsernameFlag, influxPasswordFlag)
influxClient, err := influx.NewClient(influxHostnameFlag, influxUsernameFlag, influxPasswordFlag)
if err != nil {
log.Fatal(err)
return err
}
exporter, err := prometheus.NewExporter(prometheus.Options{
Namespace: "lotus_stats",
})
if err != nil {
return err
}
go func() {
http.Handle("/metrics", exporter)
if err := http.ListenAndServe(":6688", nil); err != nil {
log.Errorw("failed to start http server", "err", err)
}
}()
if resetFlag {
if err := stats.ResetDatabase(influx, influxDatabaseFlag); err != nil {
log.Fatal(err)
if err := influx.ResetDatabase(influxClient, influxDatabaseFlag); err != nil {
return err
}
}
height := int64(heightFlag)
if !resetFlag && height == 0 {
h, err := stats.GetLastRecordedHeight(influx, influxDatabaseFlag)
if err != nil {
log.Info(err)
}
height = h
}
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
return err
@@ -149,12 +178,89 @@ var runCmd = &cli.Command{
defer closer()
if !noSyncFlag {
if err := stats.WaitForSyncComplete(ctx, api); err != nil {
log.Fatal(err)
if err := sync.SyncWait(ctx, api); err != nil {
return err
}
}
stats.Collect(ctx, api, influx, influxDatabaseFlag, height, headLagFlag)
gtp, err := api.ChainGetGenesis(ctx)
if err != nil {
return err
}
genesisTime := time.Unix(int64(gtp.MinTimestamp()), 0)
// When height is set to `0` we will resume from the best height we can.
// The goal is to ensure we have data in the last 60 tipsets
height := int64(heightFlag)
if !resetFlag && height == 0 {
lastHeight, err := influx.GetLastRecordedHeight(influxClient, influxDatabaseFlag)
if err != nil {
return err
}
sinceGenesis := build.Clock.Now().Sub(genesisTime)
expectedHeight := int64(sinceGenesis.Seconds()) / int64(build.BlockDelaySecs)
startOfWindowHeight := expectedHeight - 60
if lastHeight > startOfWindowHeight {
height = lastHeight
} else {
height = startOfWindowHeight
}
ts, err := api.ChainHead(ctx)
if err != nil {
return err
}
headHeight := int64(ts.Height())
if headHeight < height {
height = headHeight
}
}
go func() {
t := time.NewTicker(time.Second)
for {
select {
case <-t.C:
sinceGenesis := build.Clock.Now().Sub(genesisTime)
expectedHeight := int64(sinceGenesis.Seconds()) / int64(build.BlockDelaySecs)
stats.Record(ctx, metrics.TipsetCollectionHeightExpected.M(expectedHeight))
}
}
}()
store, err := ipldstore.NewApiIpldStore(ctx, api, ipldStoreCacheSizeFlag)
if err != nil {
return err
}
collector, err := points.NewChainPointCollector(ctx, store, api)
if err != nil {
return err
}
tipsets, err := sync.BufferedTipsetChannel(ctx, api, abi.ChainEpoch(height), headLagFlag)
if err != nil {
return err
}
wq := influx.NewWriteQueue(ctx, influxClient)
defer wq.Close()
for tipset := range tipsets {
if nb, err := collector.Collect(ctx, tipset); err != nil {
log.Warnw("failed to collect points", "err", err)
} else {
nb.SetDatabase(influxDatabaseFlag)
wq.AddBatch(nb)
}
}
return nil
},
cmd/lotus/main.go
+4 -2
View File
@@ -39,7 +39,7 @@ func main() {
jaeger := tracing.SetupJaegerTracing("lotus")
defer func() {
if jaeger != nil {
jaeger.Flush()
_ = jaeger.ForceFlush(context.Background())
}
}()
@@ -47,7 +47,9 @@ func main() {
cmd := cmd
originBefore := cmd.Before
cmd.Before = func(cctx *cli.Context) error {
trace.UnregisterExporter(jaeger)
if jaeger != nil {
_ = jaeger.Shutdown(cctx.Context)
}
jaeger = tracing.SetupJaegerTracing("lotus/" + cmd.Name)
if originBefore != nil {