@cryptotaxi247 / kubo / commits / 726eabead

feat: add deduplication ratio to 'ipfs dag stat' (#9787)

Arthur Gavazza committed Jun 6, 2023 at 10:17 UTC 726eabead42086d151a1d4aed1f4d92ec796d6a5
7 files changed +296 -87
core/commands/dag/dag.go
+103 -6
@@ -1,6 +1,8 @@
1 package dagcmd
2
3 import (
4 + "encoding/csv"
5 + "encoding/json"
6 "fmt"
7 "io"
8
@@ -276,12 +278,81 @@ CAR file follows the CARv1 format: https://ipld.io/specs/transport/car/carv1/
278
279 // DagStat is a dag stat command response
280 type DagStat struct {
279 - Size uint64
280 - NumBlocks int64
281 + Cid cid.Cid `json:",omitempty"`
282 + Size uint64 `json:",omitempty"`
283 + NumBlocks int64 `json:",omitempty"`
284 }
285
286 func (s *DagStat) String() string {
284 - return fmt.Sprintf("Size: %d, NumBlocks: %d", s.Size, s.NumBlocks)
287 + return fmt.Sprintf("%s %d %d", s.Cid.String()[:20], s.Size, s.NumBlocks)
288 +}
289 +
290 +func (s *DagStat) MarshalJSON() ([]byte, error) {
291 + type Alias DagStat
292 + /*
293 + We can't rely on cid.Cid.MarshalJSON since it uses the {"/": "..."}
294 + format. To make the output consistent and follow the Kubo API patterns
295 + we use the Cid.String method
296 + */
297 + return json.Marshal(struct {
298 + Cid string `json:"Cid"`
299 + *Alias
300 + }{
301 + Cid: s.Cid.String(),
302 + Alias: (*Alias)(s),
303 + })
304 +}
305 +
306 +func (s *DagStat) UnmarshalJSON(data []byte) error {
307 + /*
308 + We can't rely on cid.Cid.UnmarshalJSON since it uses the {"/": "..."}
309 + format. To make the output consistent and follow the Kubo API patterns
310 + we use the Cid.Parse method
311 + */
312 + type Alias DagStat
313 + aux := struct {
314 + Cid string `json:"Cid"`
315 + *Alias
316 + }{
317 + Alias: (*Alias)(s),
318 + }
319 + if err := json.Unmarshal(data, &aux); err != nil {
320 + return err
321 + }
322 + Cid, err := cid.Parse(aux.Cid)
323 + if err != nil {
324 + return err
325 + }
326 + s.Cid = Cid
327 + return nil
328 +}
329 +
330 +type DagStatSummary struct {
331 + redundantSize uint64 `json:"-"`
332 + UniqueBlocks int `json:",omitempty"`
333 + TotalSize uint64 `json:",omitempty"`
334 + SharedSize uint64 `json:",omitempty"`
335 + Ratio float32 `json:",omitempty"`
336 + DagStatsArray []*DagStat `json:"DagStats,omitempty"`
337 +}
338 +
339 +func (s *DagStatSummary) String() string {
340 + return fmt.Sprintf("Total Size: %d\nUnique Blocks: %d\nShared Size: %d\nRatio: %f", s.TotalSize, s.UniqueBlocks, s.SharedSize, s.Ratio)
341 +}
342 +
343 +func (s *DagStatSummary) incrementTotalSize(size uint64) {
344 + s.TotalSize += size
345 +}
346 +func (s *DagStatSummary) incrementRedundantSize(size uint64) {
347 + s.redundantSize += size
348 +}
349 +func (s *DagStatSummary) appendStats(stats *DagStat) {
350 + s.DagStatsArray = append(s.DagStatsArray, stats)
351 +}
352 +
353 +func (s *DagStatSummary) calculateSummary() {
354 + s.Ratio = float32(s.redundantSize) / float32(s.TotalSize)
355 + s.SharedSize = s.redundantSize - s.TotalSize
356 }
357
358 // DagStatCmd is a command for getting size information about an ipfs-stored dag
@@ -296,24 +367,50 @@ Note: This command skips duplicate blocks in reporting both size and the number
367 `,
368 },
369 Arguments: []cmds.Argument{
299 - cmds.StringArg("root", true, false, "CID of a DAG root to get statistics for").EnableStdin(),
370 + cmds.StringArg("root", true, true, "CID of a DAG root to get statistics for").EnableStdin(),
371 },
372 Options: []cmds.Option{
373 cmds.BoolOption(progressOptionName, "p", "Return progressive data while reading through the DAG").WithDefault(true),
374 },
375 Run: dagStat,
305 - Type: DagStat{},
376 + Type: DagStatSummary{},
377 PostRun: cmds.PostRunMap{
378 cmds.CLI: finishCLIStat,
379 },
380 Encoders: cmds.EncoderMap{
310 - cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, event *DagStat) error {
381 + cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, event *DagStatSummary) error {
382 + fmt.Fprintln(w)
383 + csvWriter := csv.NewWriter(w)
384 + csvWriter.Comma = '\t'
385 + cidSpacing := len(event.DagStatsArray[0].Cid.String())
386 + header := []string{fmt.Sprintf("%-*s", cidSpacing, "CID"), fmt.Sprintf("%-15s", "Blocks"), "Size"}
387 + if err := csvWriter.Write(header); err != nil {
388 + return err
389 + }
390 + for _, dagStat := range event.DagStatsArray {
391 + numBlocksStr := fmt.Sprint(dagStat.NumBlocks)
392 + err := csvWriter.Write([]string{
393 + dagStat.Cid.String(),
394 + fmt.Sprintf("%-15s", numBlocksStr),
395 + fmt.Sprint(dagStat.Size),
396 + })
397 + if err != nil {
398 + return err
399 + }
400 + }
401 + csvWriter.Flush()
402 + fmt.Fprint(w, "\nSummary\n")
403 _, err := fmt.Fprintf(
404 w,
405 "%v\n",
406 event,
407 )
408 + fmt.Fprint(w, "\n\n")
409 return err
410 }),
411 + cmds.JSON: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, event *DagStatSummary) error {
412 + return json.NewEncoder(w).Encode(event)
413 + },
414 + ),
415 },
416 }
core/commands/dag/stat.go
+66 -47
@@ -6,70 +6,82 @@ import (
6 "os"
7
8 "github.com/ipfs/boxo/coreiface/path"
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/e"
12 -
13 - mdag "github.com/ipfs/boxo/ipld/merkledag"
14 - cmds "github.com/ipfs/go-ipfs-cmds"
15 )
16
17 +// TODO cache every cid traversal in a dp cache
18 +// if the cid exists in the cache, don't traverse it, and use the cached result
19 +// to compute the new state
20 +
21 func dagStat(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
22 progressive := req.Options[progressOptionName].(bool)
19 -
23 api, err := cmdenv.GetApi(env, req)
24 if err != nil {
25 return err
26 }
24 -
25 - rp, err := api.ResolvePath(req.Context, path.New(req.Arguments[0]))
26 - if err != nil {
27 - return err
28 - }
29 -
30 - if len(rp.Remainder()) > 0 {
31 - return fmt.Errorf("cannot return size for anything other than a DAG with a root CID")
32 - }
33 -
27 nodeGetter := mdag.NewSession(req.Context, api.Dag())
35 - obj, err := nodeGetter.Get(req.Context, rp.Cid())
36 - if err != nil {
37 - return err
38 - }
28
40 - dagstats := &DagStat{}
41 - err = traverse.Traverse(obj, traverse.Options{
42 - DAG: nodeGetter,
43 - Order: traverse.DFSPre,
44 - Func: func(current traverse.State) error {
45 - dagstats.Size += uint64(len(current.Node.RawData()))
46 - dagstats.NumBlocks++
47 -
48 - if progressive {
49 - if err := res.Emit(dagstats); err != nil {
50 - return err
51 - }
52 - }
53 - return nil
54 - },
55 - ErrFunc: nil,
56 - SkipDuplicates: true,
57 - })
58 - if err != nil {
59 - return fmt.Errorf("error traversing DAG: %w", err)
60 - }
29 + cidSet := cid.NewSet()
30 + dagStatSummary := &DagStatSummary{DagStatsArray: []*DagStat{}}
31 + for _, a := range req.Arguments {
32 + rp, err := api.ResolvePath(req.Context, path.New(a))
33 + if err != nil {
34 + return err
35 + }
36 + if len(rp.Remainder()) > 0 {
37 + return fmt.Errorf("cannot return size for anything other than a DAG with a root CID")
38 + }
39
62 - if !progressive {
63 - if err := res.Emit(dagstats); err != nil {
40 + obj, err := nodeGetter.Get(req.Context, rp.Cid())
41 + if err != nil {
42 return err
43 }
44 + dagstats := &DagStat{Cid: rp.Cid()}
45 + dagStatSummary.appendStats(dagstats)
46 + err = traverse.Traverse(obj, traverse.Options{
47 + DAG: nodeGetter,
48 + Order: traverse.DFSPre,
49 + Func: func(current traverse.State) error {
50 + fmt.Println("previousDagStatSize:", dagstats.Size)
51 + currentNodeSize := uint64(len(current.Node.RawData()))
52 + dagstats.Size += currentNodeSize
53 + dagstats.NumBlocks++
54 + if !cidSet.Has(current.Node.Cid()) {
55 + dagStatSummary.incrementTotalSize(currentNodeSize)
56 + }
57 + dagStatSummary.incrementRedundantSize(currentNodeSize)
58 + cidSet.Add(current.Node.Cid())
59 + if progressive {
60 + if err := res.Emit(dagStatSummary); err != nil {
61 + return err
62 + }
63 + }
64 + return nil
65 + },
66 + ErrFunc: nil,
67 + SkipDuplicates: true,
68 + })
69 + if err != nil {
70 + return fmt.Errorf("error traversing DAG: %w", err)
71 + }
72 }
73
74 + dagStatSummary.UniqueBlocks = cidSet.Len()
75 + dagStatSummary.calculateSummary()
76 +
77 + if err := res.Emit(dagStatSummary); err != nil {
78 + return err
79 + }
80 return nil
81 }
82
83 func finishCLIStat(res cmds.Response, re cmds.ResponseEmitter) error {
72 - var dagStats *DagStat
84 + var dagStats *DagStatSummary
85 for {
86 v, err := res.Next()
87 if err != nil {
@@ -78,13 +90,20 @@ func finishCLIStat(res cmds.Response, re cmds.ResponseEmitter) error {
90 }
91 return err
92 }
81 -
82 - out, ok := v.(*DagStat)
83 - if !ok {
93 + switch out := v.(type) {
94 + case *DagStatSummary:
95 + dagStats = out
96 + if dagStats.Ratio == 0 {
97 + length := len(dagStats.DagStatsArray)
98 + if length > 0 {
99 + currentStat := dagStats.DagStatsArray[length-1]
100 + fmt.Fprintf(os.Stderr, "CID: %s, Size: %d, NumBlocks: %d\n", currentStat.Cid, currentStat.Size, currentStat.NumBlocks)
101 + }
102 + }
103 + default:
104 return e.TypeErr(out, v)
105 +
106 }
86 - dagStats = out
87 - fmt.Fprintf(os.Stderr, "%v\r", out)
107 }
108 return re.Emit(dagStats)
109 }
test/cli/dag_test.go new
+105
@@ -0,0 +1,105 @@
1 +package cli
2 +
3 +import (
4 + "encoding/json"
5 + "io"
6 + "os"
7 + "testing"
8 +
9 + "github.com/ipfs/kubo/test/cli/harness"
10 + "github.com/ipfs/kubo/test/cli/testutils"
11 + "github.com/stretchr/testify/assert"
12 +)
13 +
14 +const (
15 + fixtureFile = "./fixtures/TestDagStat.car"
16 + textOutputPath = "./fixtures/TestDagStatExpectedOutput.txt"
17 + node1Cid = "bafyreibmdfd7c5db4kls4ty57zljfhqv36gi43l6txl44pi423wwmeskwy"
18 + node2Cid = "bafyreie3njilzdi4ixumru4nzgecsnjtu7fzfcwhg7e6s4s5i7cnbslvn4"
19 + fixtureCid = "bafyreifrm6uf5o4dsaacuszf35zhibyojlqclabzrms7iak67pf62jygaq"
20 +)
21 +
22 +type DagStat struct {
23 + Cid string `json:"Cid"`
24 + Size int `json:"Size"`
25 + NumBlocks int `json:"NumBlocks"`
26 +}
27 +
28 +type Data struct {
29 + UniqueBlocks int `json:"UniqueBlocks"`
30 + TotalSize int `json:"TotalSize"`
31 + SharedSize int `json:"SharedSize"`
32 + Ratio float64 `json:"Ratio"`
33 + DagStats []DagStat `json:"DagStats"`
34 +}
35 +
36 +// The Fixture file represents a dag where 2 nodes of size = 46B each, have a common child of 7B
37 +// when traversing the DAG from the root's children (node1 and node2) we count (46 + 7)x2 bytes (counting redundant bytes) = 106
38 +// since both nodes share a common child of 7 bytes we actually had to read (46)x2 + 7 = 99 bytes
39 +// we should get a dedup ratio of 106/99 that results in approximatelly 1.0707071
40 +
41 +func TestDag(t *testing.T) {
42 + t.Parallel()
43 +
44 + t.Run("ipfs dag stat --enc=json", func(t *testing.T) {
45 + t.Parallel()
46 + node := harness.NewT(t).NewNode().Init().StartDaemon()
47 + // Import fixture
48 + r, err := os.Open(fixtureFile)
49 + assert.Nil(t, err)
50 + defer r.Close()
51 + err = node.IPFSDagImport(r, fixtureCid)
52 + assert.NoError(t, err)
53 + stat := node.RunIPFS("dag", "stat", "--progress=false", "--enc=json", node1Cid, node2Cid)
54 + var data Data
55 + err = json.Unmarshal(stat.Stdout.Bytes(), &data)
56 + assert.NoError(t, err)
57 +
58 + expectedUniqueBlocks := 3
59 + expectedSharedSize := 7
60 + expectedTotalSize := 99
61 + expectedRatio := float64(expectedSharedSize+expectedTotalSize) / float64(expectedTotalSize)
62 + expectedDagStatsLength := 2
63 + // Validate UniqueBlocks
64 + assert.Equal(t, expectedUniqueBlocks, data.UniqueBlocks)
65 + assert.Equal(t, expectedSharedSize, data.SharedSize)
66 + assert.Equal(t, expectedTotalSize, data.TotalSize)
67 + assert.Equal(t, testutils.FloatTruncate(expectedRatio, 4), testutils.FloatTruncate(data.Ratio, 4))
68 +
69 + // Validate DagStats
70 + assert.Equal(t, expectedDagStatsLength, len(data.DagStats))
71 + node1Output := data.DagStats[0]
72 + node2Output := data.DagStats[1]
73 +
74 + assert.Equal(t, node1Output.Cid, node1Cid)
75 + assert.Equal(t, node2Output.Cid, node2Cid)
76 +
77 + expectedNode1Size := (expectedTotalSize + expectedSharedSize) / 2
78 + expectedNode2Size := (expectedTotalSize + expectedSharedSize) / 2
79 + assert.Equal(t, expectedNode1Size, node1Output.Size)
80 + assert.Equal(t, expectedNode2Size, node2Output.Size)
81 +
82 + expectedNode1Blocks := 2
83 + expectedNode2Blocks := 2
84 + assert.Equal(t, expectedNode1Blocks, node1Output.NumBlocks)
85 + assert.Equal(t, expectedNode2Blocks, node2Output.NumBlocks)
86 + })
87 +
88 + t.Run("ipfs dag stat", func(t *testing.T) {
89 + t.Parallel()
90 + node := harness.NewT(t).NewNode().Init().StartDaemon()
91 + r, err := os.Open(fixtureFile)
92 + assert.NoError(t, err)
93 + defer r.Close()
94 + f, err := os.Open(textOutputPath)
95 + assert.NoError(t, err)
96 + defer f.Close()
97 + content, err := io.ReadAll(f)
98 + assert.NoError(t, err)
99 + err = node.IPFSDagImport(r, fixtureCid)
100 + assert.NoError(t, err)
101 + stat := node.RunIPFS("dag", "stat", "--progress=false", node1Cid, node2Cid)
102 + assert.Equal(t, content, stat.Stdout.Bytes())
103 + })
104 +
105 +}
test/cli/fixtures/TestDagStat.car
Binary files /dev/null and b/test/cli/fixtures/TestDagStat.car differ
test/cli/fixtures/TestDagStatExpectedOutput.txt new
+12
@@ -0,0 +1,12 @@
1 +
2 +CID Blocks Size
3 +bafyreibmdfd7c5db4kls4ty57zljfhqv36gi43l6txl44pi423wwmeskwy 2 53
4 +bafyreie3njilzdi4ixumru4nzgecsnjtu7fzfcwhg7e6s4s5i7cnbslvn4 2 53
5 +
6 +Summary
7 +Total Size: 99
8 +Unique Blocks: 3
9 +Shared Size: 7
10 +Ratio: 1.070707
11 +
12 +
test/cli/testutils/floats.go new
+9
@@ -0,0 +1,9 @@
1 +package testutils
2 +
3 +func FloatTruncate(value float64, decimalPlaces int) float64 {
4 + pow := 1.0
5 + for i := 0; i < decimalPlaces; i++ {
6 + pow *= 10.0
7 + }
8 + return float64(int(value*pow)) / pow
9 +}
test/sharness/t0053-dag.sh
+1 -34
@@ -428,40 +428,7 @@ test_expect_success "'ipfs dag put' check block size" '
428 test_cmp resolve_data_exp resolve_data
429 '
430
431 - test_expect_success "dag stat of simple IPLD object" '
432 - ipfs dag stat $NESTED_HASH > actual_stat_inner_ipld_obj &&
433 - echo "Size: 8, NumBlocks: 1" > exp_stat_inner_ipld_obj &&
434 - test_cmp exp_stat_inner_ipld_obj actual_stat_inner_ipld_obj &&
435 - ipfs dag stat $HASH > actual_stat_ipld_obj &&
436 - echo "Size: 54, NumBlocks: 2" > exp_stat_ipld_obj &&
437 - test_cmp exp_stat_ipld_obj actual_stat_ipld_obj
438 - '
439 -
440 - test_expect_success "dag stat of simple UnixFS object" '
441 - BASIC_UNIXFS=$(echo "1234" | ipfs add --pin=false -q) &&
442 - ipfs dag stat $BASIC_UNIXFS > actual_stat_basic_unixfs &&
443 - echo "Size: 13, NumBlocks: 1" > exp_stat_basic_unixfs &&
444 - test_cmp exp_stat_basic_unixfs actual_stat_basic_unixfs
445 - '
446 -
447 - # The multiblock file is just 10000000 copies of the number 1
448 - # As most of its data is replicated it should have a small number of blocks
449 - test_expect_success "dag stat of multiblock UnixFS object" '
450 - MULTIBLOCK_UNIXFS=$(printf "1%.0s" {1..10000000} | ipfs add --pin=false -q) &&
451 - ipfs dag stat $MULTIBLOCK_UNIXFS > actual_stat_multiblock_unixfs &&
452 - echo "Size: 302582, NumBlocks: 3" > exp_stat_multiblock_unixfs &&
453 - test_cmp exp_stat_multiblock_unixfs actual_stat_multiblock_unixfs
454 - '
455 -
456 - test_expect_success "dag stat of directory of UnixFS objects" '
457 - mkdir -p unixfsdir &&
458 - echo "1234" > unixfsdir/small.txt
459 - printf "1%.0s" {1..10000000} > unixfsdir/many1s.txt &&
460 - DIRECTORY_UNIXFS=$(ipfs add -r --pin=false -Q unixfsdir) &&
461 - ipfs dag stat $DIRECTORY_UNIXFS > actual_stat_directory_unixfs &&
462 - echo "Size: 302705, NumBlocks: 5" > exp_stat_directory_unixfs &&
463 - test_cmp exp_stat_directory_unixfs actual_stat_directory_unixfs
464 - '
431 +
432 }
433
434 # should work offline