Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
storagenode/pieces/lazyfilewalker: fix gc-filewalker
This change fixes the GC filewalker subprocess output not being piped to the trashHandler. Previously, the `stdout` of the subprocess was set to an `io.ReadWriter`. This caused issues because the `trashHandler` (which implements `io.ReadWriter`) wasn't actively reading from the pipe, leading to data being buffered and not processed. This commit changes the `stdout` type to `io.Writer`. This clarifies the intent and ensures the subprocess can write its output without relying on the `trashHandler` to read from the pipe. Updates storj/storj-private#666 Change-Id: I6139018e0eaae5c0c7bee0dfdd5bbba69b948a99
- Loading branch information
Showing
4 changed files
with
139 additions
and
90 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| // Copyright (C) 2024 Storj Labs, Inc. | ||
| // See LICENSE for copying information. | ||
|
|
||
| package lazyfilewalker | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
|
|
||
| "go.uber.org/zap" | ||
|
|
||
| "storj.io/common/storj" | ||
| ) | ||
|
|
||
| type writer interface { | ||
| Write(p []byte) (n int, err error) | ||
| Decode(v interface{}) error | ||
| } | ||
|
|
||
| // check that genericWriter and trashHandler implement the writer interface. | ||
| var _ writer = (*genericWriter)(nil) | ||
| var _ writer = (*trashHandler)(nil) | ||
|
|
||
| // genericWriter is a writer that processes the output of the lazyfilewalker subprocess. | ||
| type genericWriter struct { | ||
| buf bytes.Buffer | ||
| log *zap.Logger | ||
| } | ||
|
|
||
| func newGenericWriter(log *zap.Logger) *genericWriter { | ||
| return &genericWriter{ | ||
| log: log, | ||
| } | ||
| } | ||
|
|
||
| // Decode decodes the data from the buffer into the provided value. | ||
| func (w *genericWriter) Decode(v interface{}) error { | ||
| if err := json.NewDecoder(&w.buf).Decode(&v); err != nil { | ||
| w.log.Error("failed to decode response from subprocess", zap.Error(err)) | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Write writes the provided bytes to the buffer. | ||
| func (w *genericWriter) Write(b []byte) (n int, err error) { | ||
| return w.buf.Write(b) | ||
| } | ||
|
|
||
| // trashHandler is a writer that processes the output of the gc-filewalker subprocess. | ||
| type trashHandler struct { | ||
| buf *genericWriter | ||
| log *zap.Logger | ||
| lineBuffer []byte | ||
|
|
||
| trashFunc func(pieceID storj.PieceID) error | ||
| } | ||
|
|
||
| func newTrashHandler(log *zap.Logger, trashFunc func(pieceID storj.PieceID) error) *trashHandler { | ||
| return &trashHandler{ | ||
| log: log.Named("trash-handler"), | ||
| trashFunc: trashFunc, | ||
| buf: newGenericWriter(log), | ||
| } | ||
| } | ||
|
|
||
| // Decode decodes the data from the buffer into the provided value. | ||
| func (t *trashHandler) Decode(v interface{}) error { | ||
| return t.buf.Decode(v) | ||
| } | ||
|
|
||
| // Write writes the provided bytes to the buffer. | ||
| func (t *trashHandler) Write(b []byte) (n int, err error) { | ||
| n = len(b) | ||
| t.lineBuffer = append(t.lineBuffer, b...) | ||
| for { | ||
| if b, err = t.writeLine(t.lineBuffer); err != nil { | ||
| return n, err | ||
| } | ||
| if len(b) == len(t.lineBuffer) { | ||
| break | ||
| } | ||
|
|
||
| t.lineBuffer = b | ||
| } | ||
|
|
||
| return n, nil | ||
| } | ||
|
|
||
| func (t *trashHandler) writeLine(b []byte) (remaining []byte, err error) { | ||
| idx := bytes.IndexByte(b, '\n') | ||
| if idx < 0 { | ||
| return b, nil | ||
| } | ||
|
|
||
| b, remaining = b[:idx], b[idx+1:] | ||
|
|
||
| return remaining, t.processTrashPiece(b) | ||
| } | ||
|
|
||
| func (t *trashHandler) processTrashPiece(b []byte) error { | ||
| var resp GCFilewalkerResponse | ||
| if err := json.Unmarshal(b, &resp); err != nil { | ||
| t.log.Error("failed to unmarshal data from subprocess", zap.Error(err)) | ||
| return err | ||
| } | ||
|
|
||
| if !resp.Completed { | ||
| for _, pieceID := range resp.PieceIDs { | ||
| t.log.Debug("trashing piece", zap.String("pieceID", pieceID.String())) | ||
| return t.trashFunc(pieceID) | ||
| } | ||
| } | ||
|
|
||
| _, err := t.buf.Write(b) | ||
| return err | ||
| } |