| 1 | package corerepo |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/ipfs/kubo/core" |
| 10 | "github.com/ipfs/kubo/gc" |
| 11 | "github.com/ipfs/kubo/repo" |
| 12 | |
| 13 | "github.com/dustin/go-humanize" |
| 14 | "github.com/ipfs/boxo/mfs" |
| 15 | "github.com/ipfs/go-cid" |
| 16 | logging "github.com/ipfs/go-log/v2" |
| 17 | ) |
| 18 | |
| 19 | var log = logging.Logger("corerepo") |
| 20 | |
| 21 | var ErrMaxStorageExceeded = errors.New("maximum storage limit exceeded. Try to unpin some files") |
| 22 | |
| 23 | type GC struct { |
| 24 | Node *core.IpfsNode |
| 25 | Repo repo.Repo |
| 26 | StorageMax uint64 |
| 27 | StorageGC uint64 |
| 28 | SlackGB uint64 |
| 29 | Storage uint64 |
| 30 | } |
| 31 | |
| 32 | func NewGC(n *core.IpfsNode) (*GC, error) { |
| 33 | r := n.Repo |
| 34 | cfg, err := r.Config() |
| 35 | if err != nil { |
| 36 | return nil, err |
| 37 | } |
| 38 | |
| 39 | // check if cfg has these fields initialized |
| 40 | // TODO: there should be a general check for all of the cfg fields |
| 41 | // maybe distinguish between user config file and default struct? |
| 42 | if cfg.Datastore.StorageMax == "" { |
| 43 | if err := r.SetConfigKey("Datastore.StorageMax", "10GB"); err != nil { |
| 44 | return nil, err |
| 45 | } |
| 46 | cfg.Datastore.StorageMax = "10GB" |
| 47 | } |
| 48 | if cfg.Datastore.StorageGCWatermark == 0 { |
| 49 | if err := r.SetConfigKey("Datastore.StorageGCWatermark", 90); err != nil { |
| 50 | return nil, err |
| 51 | } |
| 52 | cfg.Datastore.StorageGCWatermark = 90 |
| 53 | } |
| 54 | |
| 55 | storageMax, err := humanize.ParseBytes(cfg.Datastore.StorageMax) |
| 56 | if err != nil { |
| 57 | return nil, err |
| 58 | } |
| 59 | storageGC := storageMax * uint64(cfg.Datastore.StorageGCWatermark) / 100 |
| 60 | |
| 61 | // calculate the slack space between StorageMax and StorageGCWatermark |
| 62 | // used to limit GC duration |
| 63 | slackGB := max((storageMax-storageGC)/10e9, 1) |
| 64 | |
| 65 | return &GC{ |
| 66 | Node: n, |
| 67 | Repo: r, |
| 68 | StorageMax: storageMax, |
| 69 | StorageGC: storageGC, |
| 70 | SlackGB: slackGB, |
| 71 | }, nil |
| 72 | } |
| 73 | |
| 74 | func BestEffortRoots(filesRoot *mfs.Root) ([]cid.Cid, error) { |
| 75 | rootDag, err := filesRoot.GetDirectory().GetNode() |
| 76 | if err != nil { |
| 77 | return nil, err |
| 78 | } |
| 79 | |
| 80 | return []cid.Cid{rootDag.Cid()}, nil |
| 81 | } |
| 82 | |
| 83 | func GarbageCollect(n *core.IpfsNode, ctx context.Context) error { |
| 84 | roots, err := BestEffortRoots(n.FilesRoot) |
| 85 | if err != nil { |
| 86 | return err |
| 87 | } |
| 88 | rmed := gc.GC(ctx, n.Blockstore, n.Repo.Datastore(), n.Pinning, roots) |
| 89 | |
| 90 | return CollectResult(ctx, rmed, nil) |
| 91 | } |
| 92 | |
| 93 | // CollectResult collects the output of a garbage collection run and calls the |
| 94 | // given callback for each object removed. It also collects all errors into a |
| 95 | // MultiError which is returned after the gc is completed. |
| 96 | func CollectResult(ctx context.Context, gcOut <-chan gc.Result, cb func(cid.Cid)) error { |
| 97 | var errors []error |
| 98 | loop: |
| 99 | for { |
| 100 | select { |
| 101 | case res, ok := <-gcOut: |
| 102 | if !ok { |
| 103 | break loop |
| 104 | } |
| 105 | if res.Error != nil { |
| 106 | errors = append(errors, res.Error) |
| 107 | } else if res.KeyRemoved.Defined() && cb != nil { |
| 108 | cb(res.KeyRemoved) |
| 109 | } |
| 110 | case <-ctx.Done(): |
| 111 | errors = append(errors, ctx.Err()) |
| 112 | break loop |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | switch len(errors) { |
| 117 | case 0: |
| 118 | return nil |
| 119 | case 1: |
| 120 | return errors[0] |
| 121 | default: |
| 122 | return NewMultiError(errors...) |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | // NewMultiError creates a new MultiError object from a given slice of errors. |
| 127 | func NewMultiError(errs ...error) *MultiError { |
| 128 | return &MultiError{errs[:len(errs)-1], errs[len(errs)-1]} |
| 129 | } |
| 130 | |
| 131 | // MultiError contains the results of multiple errors. |
| 132 | type MultiError struct { |
| 133 | Errors []error |
| 134 | Summary error |
| 135 | } |
| 136 | |
| 137 | func (e *MultiError) Error() string { |
| 138 | var buf bytes.Buffer |
| 139 | for _, err := range e.Errors { |
| 140 | buf.WriteString(err.Error()) |
| 141 | buf.WriteString("; ") |
| 142 | } |
| 143 | buf.WriteString(e.Summary.Error()) |
| 144 | return buf.String() |
| 145 | } |
| 146 | |
| 147 | func GarbageCollectAsync(n *core.IpfsNode, ctx context.Context) <-chan gc.Result { |
| 148 | roots, err := BestEffortRoots(n.FilesRoot) |
| 149 | if err != nil { |
| 150 | out := make(chan gc.Result) |
| 151 | out <- gc.Result{Error: err} |
| 152 | close(out) |
| 153 | return out |
| 154 | } |
| 155 | |
| 156 | return gc.GC(ctx, n.Blockstore, n.Repo.Datastore(), n.Pinning, roots) |
| 157 | } |
| 158 | |
| 159 | func PeriodicGC(ctx context.Context, node *core.IpfsNode) error { |
| 160 | cfg, err := node.Repo.Config() |
| 161 | if err != nil { |
| 162 | return err |
| 163 | } |
| 164 | |
| 165 | if cfg.Datastore.GCPeriod == "" { |
| 166 | cfg.Datastore.GCPeriod = "1h" |
| 167 | } |
| 168 | |
| 169 | period, err := time.ParseDuration(cfg.Datastore.GCPeriod) |
| 170 | if err != nil { |
| 171 | return err |
| 172 | } |
| 173 | if int64(period) == 0 { |
| 174 | // if duration is 0, it means GC is disabled. |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | gc, err := NewGC(node) |
| 179 | if err != nil { |
| 180 | return err |
| 181 | } |
| 182 | |
| 183 | for { |
| 184 | select { |
| 185 | case <-ctx.Done(): |
| 186 | return nil |
| 187 | case <-time.After(period): |
| 188 | // the private func maybeGC doesn't compute storageMax, storageGC, slackGC so that they are not re-computed for every cycle |
| 189 | if err := gc.maybeGC(ctx, 0); err != nil { |
| 190 | log.Error(err) |
| 191 | } |
| 192 | } |
| 193 | } |
| 194 | } |
| 195 | |
| 196 | func ConditionalGC(ctx context.Context, node *core.IpfsNode, offset uint64) error { |
| 197 | gc, err := NewGC(node) |
| 198 | if err != nil { |
| 199 | return err |
| 200 | } |
| 201 | return gc.maybeGC(ctx, offset) |
| 202 | } |
| 203 | |
| 204 | func (gc *GC) maybeGC(ctx context.Context, offset uint64) error { |
| 205 | storage, err := gc.Repo.GetStorageUsage(ctx) |
| 206 | if err != nil { |
| 207 | return err |
| 208 | } |
| 209 | |
| 210 | if storage+offset > gc.StorageGC { |
| 211 | if storage+offset > gc.StorageMax { |
| 212 | log.Warnf("pre-GC: %s", ErrMaxStorageExceeded) |
| 213 | } |
| 214 | |
| 215 | // Do GC here |
| 216 | log.Info("Watermark exceeded. Starting repo GC...") |
| 217 | |
| 218 | if err := GarbageCollect(gc.Node, ctx); err != nil { |
| 219 | return err |
| 220 | } |
| 221 | log.Infof("Repo GC done. See `ipfs repo stat` to see how much space got freed.\n") |
| 222 | } |
| 223 | return nil |
| 224 | } |