| 1 | package dagcmd |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "io" |
| 6 | "os" |
| 7 | |
| 8 | "github.com/dustin/go-humanize" |
| 9 | mdag "github.com/ipfs/boxo/ipld/merkledag" |
| 10 | "github.com/ipfs/boxo/ipld/merkledag/traverse" |
| 11 | cid "github.com/ipfs/go-cid" |
| 12 | cmds "github.com/ipfs/go-ipfs-cmds" |
| 13 | "github.com/ipfs/kubo/core/commands/cmdenv" |
| 14 | "github.com/ipfs/kubo/core/commands/cmdutils" |
| 15 | "github.com/ipfs/kubo/core/commands/e" |
| 16 | ) |
| 17 | |
| 18 | // TODO cache every cid traversal in a dp cache |
| 19 | // if the cid exists in the cache, don't traverse it, and use the cached result |
| 20 | // to compute the new state |
| 21 | |
| 22 | func dagStat(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error { |
| 23 | // Default to true (emit intermediate states) for HTTP/RPC clients that want progress |
| 24 | progressive := true |
| 25 | if val, specified := req.Options[progressOptionName].(bool); specified { |
| 26 | progressive = val |
| 27 | } |
| 28 | api, err := cmdenv.GetApi(env, req) |
| 29 | if err != nil { |
| 30 | return err |
| 31 | } |
| 32 | |
| 33 | enc, err := cmdenv.GetCidEncoder(req) |
| 34 | if err != nil { |
| 35 | return err |
| 36 | } |
| 37 | |
| 38 | nodeGetter := mdag.NewSession(req.Context, api.Dag()) |
| 39 | |
| 40 | cidSet := cid.NewSet() |
| 41 | dagStatSummary := &DagStatSummary{DagStatsArray: []*DagStat{}} |
| 42 | for _, a := range req.Arguments { |
| 43 | p, err := cmdutils.PathOrCidPath(a) |
| 44 | if err != nil { |
| 45 | return err |
| 46 | } |
| 47 | rp, remainder, err := api.ResolvePath(req.Context, p) |
| 48 | if err != nil { |
| 49 | return err |
| 50 | } |
| 51 | if len(remainder) > 0 { |
| 52 | return fmt.Errorf("cannot return size for anything other than a DAG with a root CID") |
| 53 | } |
| 54 | |
| 55 | obj, err := nodeGetter.Get(req.Context, rp.RootCid()) |
| 56 | if err != nil { |
| 57 | return err |
| 58 | } |
| 59 | dagstats := &DagStat{Cid: enc.Encode(rp.RootCid())} |
| 60 | dagStatSummary.appendStats(dagstats) |
| 61 | err = traverse.Traverse(obj, traverse.Options{ |
| 62 | DAG: nodeGetter, |
| 63 | Order: traverse.DFSPre, |
| 64 | Func: func(current traverse.State) error { |
| 65 | currentNodeSize := uint64(len(current.Node.RawData())) |
| 66 | dagstats.Size += currentNodeSize |
| 67 | dagstats.NumBlocks++ |
| 68 | if !cidSet.Has(current.Node.Cid()) { |
| 69 | dagStatSummary.incrementTotalSize(currentNodeSize) |
| 70 | } |
| 71 | dagStatSummary.incrementRedundantSize(currentNodeSize) |
| 72 | cidSet.Add(current.Node.Cid()) |
| 73 | if progressive { |
| 74 | if err := res.Emit(dagStatSummary); err != nil { |
| 75 | return err |
| 76 | } |
| 77 | } |
| 78 | return nil |
| 79 | }, |
| 80 | ErrFunc: nil, |
| 81 | SkipDuplicates: true, |
| 82 | }) |
| 83 | if err != nil { |
| 84 | return fmt.Errorf("error traversing DAG: %w", err) |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | dagStatSummary.UniqueBlocks = cidSet.Len() |
| 89 | dagStatSummary.calculateSummary() |
| 90 | |
| 91 | if err := res.Emit(dagStatSummary); err != nil { |
| 92 | return err |
| 93 | } |
| 94 | return nil |
| 95 | } |
| 96 | |
| 97 | func finishCLIStat(res cmds.Response, re cmds.ResponseEmitter) error { |
| 98 | showProgress := cmdenv.ShouldShowProgress(res.Request(), progressOptionName) |
| 99 | |
| 100 | var dagStats *DagStatSummary |
| 101 | for { |
| 102 | v, err := res.Next() |
| 103 | if err != nil { |
| 104 | if err == io.EOF { |
| 105 | break |
| 106 | } |
| 107 | return err |
| 108 | } |
| 109 | switch out := v.(type) { |
| 110 | case *DagStatSummary: |
| 111 | dagStats = out |
| 112 | // Ratio == 0 means this is a progress update (not final result) |
| 113 | if showProgress && dagStats.Ratio == 0 { |
| 114 | // Sum up total progress across all DAGs being scanned |
| 115 | var totalBlocks int64 |
| 116 | var totalSize uint64 |
| 117 | for _, stat := range dagStats.DagStatsArray { |
| 118 | totalBlocks += stat.NumBlocks |
| 119 | totalSize += stat.Size |
| 120 | } |
| 121 | fmt.Fprintf(os.Stderr, "Fetched/Processed %d blocks, %d bytes (%s)\r", totalBlocks, totalSize, humanize.Bytes(totalSize)) |
| 122 | } |
| 123 | default: |
| 124 | return e.TypeErr(out, v) |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | // Clear the progress line before final output |
| 129 | if showProgress { |
| 130 | fmt.Fprint(os.Stderr, "\033[2K\r") |
| 131 | } |
| 132 | |
| 133 | return re.Emit(dagStats) |
| 134 | } |