kvm2: stop handing go-libvirt a reader that can error (H2)
uploadImage streamed the HTTP body straight into StorageVolUpload. When that body failed part way through, go-libvirt deadlocked: requestStream starts its stream sender on two unbuffered channels and, on the error path, returns without ever reading outErr, so SendStream emits a StatusError packet and parks on `outErr <-` forever while the main goroutine parks on `abort <- true` forever. StorageVolUpload never returned. The 20-minute transfer budget rescued the caller only, and the socket kill did nothing, because both ends are blocked on channel sends rather than on I/O -- which is what the comment at backend.go:152 got wrong. Each failed upload leaked kvm2's call goroutine, go-libvirt's sender goroutine and a 4 MiB buffer, permanently, against a 128Mi pod limit and a reconciler that retries on backoff. The defect is upstream and stays there: no fork, no replace. Instead eofReader, at the go-libvirt seam so it covers the seed and marker uploads too, converts any source error to io.EOF -- the one outcome that ends a stream the normal way -- and keeps the real error for the caller. countingReader sits underneath untouched, so the existing length and digest checks report exactly what they did before. The download also gets its own deadline a minute inside the transfer budget. fetchImage already bounded the whole fetch at 20 minutes, so the body's death and the budget's socket kill were racing and the socket usually won; with the body a minute ahead, a stalled mirror ends through its reader and unwinds cleanly instead of leaking. discardVolume moves from uploadImage into fetchImage's upload-error branch, so the dedicated connection is closed and libvirtd has let go of the stream's fd before the file is unlinked. Moving it rather than adding a second Close keeps image_test's closeCount() == 1 true. That discard is still unfenced against a second provider entry sharing the volume name; H4 closes that by making the name per-entry. What remains: a failure originating in libvirtd mid-upload still leaks, since SendStream has already returned by then. Confirmed unfixable from here. Known gaps quotes the two-character upstream fix (make(chan bool, 1) / make(chan error, 1)) and the replace escape hatch. Unverified at ~85%: that libvirtd answers a short-but-clean stream finish with OK rather than an error. virFDStreamWrite enforces length only on writes past it and nothing compares offset to length on the close path, which is also why virsh vol-upload of a small file into a large volume works. If it errors we are back to a deadlock by a different door, which is exactly today's behaviour, so this cannot be worse. The execution file carries a probe recipe to settle it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -2,6 +2,7 @@ package kvm2
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
@@ -90,6 +91,12 @@ const smallUploadBytes = 16 << 20
|
||||
// closeBudget bounds the courtesy ConnectClose RPC in Close.
|
||||
const closeBudget = 5 * time.Second
|
||||
|
||||
// transferGrace keeps the image download's own deadline inside the transfer
|
||||
// budget. A stalled mirror then ends by its body failing, which the upload
|
||||
// seam turns into a clean end of stream, instead of by the budget killing
|
||||
// the socket, which go-libvirt's stream sender does not survive.
|
||||
const transferGrace = time.Minute
|
||||
|
||||
func (b budgets) upload(length uint64) time.Duration {
|
||||
if length <= smallUploadBytes {
|
||||
return b.seedUpload
|
||||
@@ -325,9 +332,50 @@ func (b *libvirtBackend) StorageVolCreateXML(ctx context.Context, pool libvirt.S
|
||||
}
|
||||
|
||||
func (b *libvirtBackend) StorageVolUpload(ctx context.Context, vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error {
|
||||
return b.link.run(ctx, "StorageVolUpload", b.budgets.upload(length), func() error {
|
||||
return b.lv.StorageVolUpload(vol, r, offset, length, flags)
|
||||
src := &eofReader{r: r}
|
||||
err := b.link.run(ctx, "StorageVolUpload", b.budgets.upload(length), func() error {
|
||||
return b.lv.StorageVolUpload(vol, src, offset, length, flags)
|
||||
})
|
||||
if err == nil {
|
||||
if srcErr := src.sourceErr(); srcErr != nil {
|
||||
return fmt.Errorf("reading the upload source for %s: %w", vol.Name, srcErr)
|
||||
}
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// eofReader hides a source error from go-libvirt's stream sender and keeps
|
||||
// it for the caller. SendStream answers a non-EOF error with a StatusError
|
||||
// packet and then blocks forever handing its result to an unbuffered
|
||||
// channel that requestStream no longer reads (go-libvirt rpc.go), which
|
||||
// wedges the call and leaks the goroutine along with its 4 MiB buffer;
|
||||
// closing the socket does not help, because both ends are parked on
|
||||
// channel sends rather than on I/O. io.EOF is the one outcome that ends
|
||||
// the stream the normal way. The field is guarded because an abandoned
|
||||
// call returns while go-libvirt is still reading.
|
||||
type eofReader struct {
|
||||
r io.Reader
|
||||
mu sync.Mutex
|
||||
err error
|
||||
}
|
||||
|
||||
func (e *eofReader) Read(b []byte) (int, error) {
|
||||
n, err := e.r.Read(b)
|
||||
if err == nil || errors.Is(err, io.EOF) {
|
||||
return n, err
|
||||
}
|
||||
e.mu.Lock()
|
||||
if e.err == nil {
|
||||
e.err = err
|
||||
}
|
||||
e.mu.Unlock()
|
||||
return n, io.EOF
|
||||
}
|
||||
|
||||
func (e *eofReader) sourceErr() error {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
return e.err
|
||||
}
|
||||
|
||||
func (b *libvirtBackend) StorageVolDownload(ctx context.Context, vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error {
|
||||
|
||||
@@ -3,6 +3,7 @@ package kvm2
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
@@ -213,6 +214,57 @@ func TestMarkDead_slowKillDoesNotBlockCallers(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestEOFReader_sourceErrorNeverReachesTheStream pins the one property the
|
||||
// upload seam needs: go-libvirt's stream sender must never see an error
|
||||
// other than io.EOF, because it answers one by parking forever on a channel
|
||||
// nothing reads. The real error has to survive for the caller.
|
||||
func TestEOFReader_sourceErrorNeverReachesTheStream(t *testing.T) {
|
||||
t.Parallel()
|
||||
boom := errors.New("connection reset by peer")
|
||||
tests := []struct {
|
||||
name string
|
||||
reader io.Reader
|
||||
wantN int
|
||||
wantErr error
|
||||
}{
|
||||
{name: "error after data", reader: io.MultiReader(strings.NewReader("abc"), errReader{boom}), wantN: 3, wantErr: boom},
|
||||
{name: "error with no data", reader: errReader{boom}, wantErr: boom},
|
||||
{name: "clean eof passes through", reader: strings.NewReader("abc"), wantN: 3},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
e := &eofReader{r: tc.reader}
|
||||
n, err := io.Copy(io.Discard, e)
|
||||
if err != nil {
|
||||
t.Fatalf("io.Copy through eofReader = %v, want nil: the stream sender must only ever see io.EOF", err)
|
||||
}
|
||||
if n != int64(tc.wantN) {
|
||||
t.Errorf("copied %d bytes, want %d", n, tc.wantN)
|
||||
}
|
||||
if !errors.Is(e.sourceErr(), tc.wantErr) {
|
||||
t.Errorf("sourceErr() = %v, want %v", e.sourceErr(), tc.wantErr)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEOFReader_keepsTheFirstError(t *testing.T) {
|
||||
t.Parallel()
|
||||
first, second := errors.New("first"), errors.New("second")
|
||||
e := &eofReader{r: io.MultiReader(errReader{first}, errReader{second})}
|
||||
if _, err := io.Copy(io.Discard, e); err != nil {
|
||||
t.Fatalf("io.Copy through eofReader = %v, want nil", err)
|
||||
}
|
||||
if !errors.Is(e.sourceErr(), first) {
|
||||
t.Errorf("sourceErr() = %v, want the first error", e.sourceErr())
|
||||
}
|
||||
}
|
||||
|
||||
type errReader struct{ err error }
|
||||
|
||||
func (r errReader) Read([]byte) (int, error) { return 0, r.err }
|
||||
|
||||
func TestBudgets_upload(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
|
||||
@@ -214,7 +214,12 @@ func (p *Provider) fetchImage(ctx context.Context, host provider.KVMHost, pool l
|
||||
log := p.logger(ctx).WithValues("host", host.Name, "volume", name)
|
||||
start := time.Now()
|
||||
|
||||
resp, err := p.do(ctx, http.MethodGet, src.URL)
|
||||
// The body runs out of time before the surrounding budget does, so a
|
||||
// mirror that stalls mid-image ends the upload through its reader
|
||||
// rather than through the killed socket the budget would leave.
|
||||
httpCtx, cancelHTTP := context.WithTimeout(ctx, defaultBudgets.transfer-transferGrace)
|
||||
defer cancelHTTP()
|
||||
resp, err := p.do(httpCtx, http.MethodGet, src.URL)
|
||||
if err == nil {
|
||||
err = checkResponse(src.URL, resp)
|
||||
}
|
||||
@@ -227,8 +232,11 @@ func (p *Provider) fetchImage(ctx context.Context, host provider.KVMHost, pool l
|
||||
span.SetAttributes(attribute.Int64("kvm2.bytes", size))
|
||||
log.V(1).Info("fetching base image into host pool", "url", src.URL, "bytes", size)
|
||||
|
||||
vol, digest, err := p.uploadImage(ctx, host, pool, be, src, resp.Body, size)
|
||||
vol, digest, err := p.uploadImage(ctx, host, pool, src, resp.Body, size)
|
||||
if err != nil {
|
||||
// uploadImage has returned, so its dedicated connection is closed
|
||||
// and libvirtd has let go of the stream before the file goes.
|
||||
p.discardVolume(ctx, log, be, pool, name)
|
||||
span.SetStatus(codes.Error, err.Error())
|
||||
return libvirt.StorageVol{}, err
|
||||
}
|
||||
@@ -249,10 +257,10 @@ func (p *Provider) fetchImage(ctx context.Context, host provider.KVMHost, pool l
|
||||
|
||||
// uploadImage creates the base volume and streams body into it over a
|
||||
// dedicated connection with a running sha256, then checks length and
|
||||
// digest. On any failure the volume is discarded and the error says why:
|
||||
// a body that ended early is Transient, a digest that differs from the
|
||||
// configured one is Permanent.
|
||||
func (p *Provider) uploadImage(ctx context.Context, host provider.KVMHost, pool libvirt.StoragePool, be backend, src imageSource, body io.Reader, size int64) (libvirt.StorageVol, string, error) {
|
||||
// digest. The error says why: a body that ended early is Transient, a
|
||||
// digest that differs from the configured one is Permanent. Discarding
|
||||
// the volume is the caller's, once this connection is closed.
|
||||
func (p *Provider) uploadImage(ctx context.Context, host provider.KVMHost, pool libvirt.StoragePool, src imageSource, body io.Reader, size int64) (libvirt.StorageVol, string, error) {
|
||||
name := src.volumeName()
|
||||
xmlDoc, err := buildVolumeXML(volumeSpec{Name: name, CapacityBytes: size, Format: src.Format})
|
||||
if err != nil {
|
||||
@@ -289,7 +297,6 @@ func (p *Provider) uploadImage(ctx context.Context, host provider.KVMHost, pool
|
||||
err = permanentf("image %s: sha256 %s does not match the configured %s", src.URL, digest, strings.ToLower(src.SHA256))
|
||||
}
|
||||
if err != nil {
|
||||
p.discardVolume(ctx, p.logger(ctx).WithValues("host", host.Name, "volume", name), be, pool, name)
|
||||
return libvirt.StorageVol{}, "", err
|
||||
}
|
||||
return vol, digest, nil
|
||||
|
||||
Reference in New Issue
Block a user