From 9f4262e93994b3d20f649eb4aa244757c2264c5d Mon Sep 17 00:00:00 2001 From: Sebastien Tardif Date: Fri, 23 Oct 2026 08:05:40 -0700 Subject: [PATCH] Close the MinIO object body after download minio-go v7.2.1 copies the object in FGetObject and never closes the body. Go 1.27 does not finalize response bodies, so every successful Bucket download keeps that connection until exit. Download with GetObject and close the reader. Keep the etag match. Write to a hashed part file, the same scheme minio-go uses, and rename it into place only after the full copy and a successful file close. A failed copy removes the part file. Assisted-by: Grok/grok-4.7 Signed-off-by: Sebastien Tardif --- internal/bucket/minio/minio.go | 73 ++++++++++++++++- internal/bucket/minio/minio_test.go | 122 ++++++++++++++++++++++++++++ 2 files changed, 194 insertions(+), 1 deletion(-) diff --git a/internal/bucket/minio/minio.go b/internal/bucket/minio/minio.go index 026200a83..38965f123 100644 --- a/internal/bucket/minio/minio.go +++ b/internal/bucket/minio/minio.go @@ -18,17 +18,23 @@ package minio import ( "context" + "crypto/sha256" "crypto/tls" + "encoding/hex" "errors" "fmt" + "io" "net/http" "net/url" + "os" + "path/filepath" "strings" "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" "github.com/minio/minio-go/v7/pkg/s3utils" corev1 "k8s.io/api/core/v1" + ctrl "sigs.k8s.io/controller-runtime" "github.com/fluxcd/pkg/auth" awsauth "github.com/fluxcd/pkg/auth/aws" @@ -330,12 +336,77 @@ func (c *MinioClient) FGetObject(ctx context.Context, bucketName, objectName, lo if err = opts.SetMatchETag(stat.ETag); err != nil { return "", err } - if err = c.Client.FGetObject(ctx, bucketName, objectName, localPath, opts); err != nil { + + st, err := os.Stat(localPath) + if err == nil { + if st.IsDir() { + return "", errors.New("fileName is a directory.") + } + } else if !os.IsNotExist(err) { + return "", err + } + objectDir, _ := filepath.Split(localPath) + if objectDir != "" { + if err = os.MkdirAll(objectDir, 0o700); err != nil { + return "", err + } + } + + // GetObject, not Client.FGetObject: minio-go v7.2.1 leaves the GET body open. + object, err := c.Client.GetObject(ctx, bucketName, objectName, opts) + if err != nil { + return "", err + } + defer func() { + if cerr := object.Close(); cerr != nil { + ctrl.LoggerFrom(ctx).Error(cerr, "failed to close object reader") + } + }() + + partPath := partFilePath(localPath, stat.ETag) + f, err := os.OpenFile(partPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) + if err != nil { + return "", err + } + discardPart := func(copyErr error) (string, error) { + if cerr := f.Close(); cerr != nil { + ctrl.LoggerFrom(ctx).Error(cerr, "failed to close file") + } + if rerr := os.Remove(partPath); rerr != nil && !os.IsNotExist(rerr) { + ctrl.LoggerFrom(ctx).Error(rerr, "failed to remove partial download") + } + return "", copyErr + } + + n, err := io.Copy(f, object) + if err != nil { + return discardPart(err) + } + if n != stat.Size { + return discardPart(fmt.Errorf("downloaded %d bytes from object %q, stat size is %d", n, objectName, stat.Size)) + } + if err = f.Close(); err != nil { + if rerr := os.Remove(partPath); rerr != nil && !os.IsNotExist(rerr) { + ctrl.LoggerFrom(ctx).Error(rerr, "failed to remove partial download") + } + return "", err + } + if err = os.Rename(partPath, localPath); err != nil { + if rerr := os.Remove(partPath); rerr != nil && !os.IsNotExist(rerr) { + ctrl.LoggerFrom(ctx).Error(rerr, "failed to remove partial download") + } return "", err } return stat.ETag, nil } +// Hashed like minio-go so ".part.minio" can be a different object key. +func partFilePath(localPath, etag string) string { + sum := sha256.Sum256([]byte(filepath.Base(localPath) + etag)) + name := hex.EncodeToString(sum[:]) + ".part.minio" + return filepath.Join(filepath.Dir(localPath), name) +} + // VisitObjects iterates over the items in the provided object storage // bucket, calling visit for every item. // If the underlying client or the visit callback returns an error, diff --git a/internal/bucket/minio/minio_test.go b/internal/bucket/minio/minio_test.go index 8d5878f49..24d377c6a 100644 --- a/internal/bucket/minio/minio_test.go +++ b/internal/bucket/minio/minio_test.go @@ -17,12 +17,14 @@ limitations under the License. package minio import ( + "bytes" "context" "crypto/tls" "crypto/x509" "encoding/xml" "errors" "fmt" + "io" "log" "net/http" "net/http/httptest" @@ -30,11 +32,15 @@ import ( "os" "path/filepath" "regexp" + "strconv" + "strings" "sync" + "sync/atomic" "testing" "time" "github.com/google/uuid" + miniogo "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" . "github.com/onsi/gomega" corev1 "k8s.io/api/core/v1" @@ -313,6 +319,122 @@ func TestFGetObject(t *testing.T) { g.Expect(err).NotTo(HaveOccurred()) } +type closeTrackTransport struct { + closed *atomic.Bool +} + +func (t *closeTrackTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := http.DefaultTransport.RoundTrip(req) + if err != nil || resp == nil || resp.Body == nil || req.Method != http.MethodGet { + return resp, err + } + resp.Body = &closeTrackBody{ReadCloser: resp.Body, closed: t.closed} + return resp, nil +} + +type closeTrackBody struct { + io.ReadCloser + closed *atomic.Bool +} + +func (b *closeTrackBody) Close() error { + b.closed.Store(true) + return b.ReadCloser.Close() +} + +func newTrackingMinioClient(t *testing.T, handler http.Handler, closed *atomic.Bool) *MinioClient { + t.Helper() + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + + mc, err := miniogo.New(strings.TrimPrefix(srv.URL, "http://"), &miniogo.Options{ + Creds: credentials.NewStaticV4("fluxcd", "passw0rd!", ""), + Secure: false, + Region: "us-east-1", + Transport: &closeTrackTransport{closed: closed}, + }) + NewWithT(t).Expect(err).NotTo(HaveOccurred()) + return &MinioClient{Client: mc} +} + +func s3ObjectHandler(payload []byte) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT") + w.Header().Set("ETag", `"abc123"`) + w.Header().Set("Content-Length", strconv.Itoa(len(payload))) + if r.Method == http.MethodGet { + _, _ = w.Write(payload) + } + } +} + +func TestMinioClient_FGetObject_ClosesResponseBody(t *testing.T) { + payload := bytes.Repeat([]byte("flux"), 30000) + var closed atomic.Bool + var ifMatch atomic.Value + dir := t.TempDir() + partDir := filepath.Join(dir, "nested") + var partMu sync.Mutex + var partNames []string + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + ifMatch.Store(r.Header.Get("If-Match")) + matches, _ := filepath.Glob(filepath.Join(partDir, "*.part.minio")) + partMu.Lock() + partNames = append(partNames, matches...) + partMu.Unlock() + } + s3ObjectHandler(payload)(w, r) + }) + client := newTrackingMinioClient(t, handler, &closed) + + path := filepath.Join(partDir, "obj") + etag, err := client.FGetObject(context.Background(), "test-bucket", "obj", path) + g := NewWithT(t) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(etag).To(Equal("abc123")) + g.Expect(ifMatch.Load()).To(Equal(`"abc123"`)) + got, err := os.ReadFile(path) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(got).To(Equal(payload)) + partMu.Lock() + seenParts := append([]string(nil), partNames...) + partMu.Unlock() + g.Expect(seenParts).NotTo(BeEmpty()) + for _, name := range seenParts { + g.Expect(filepath.Base(name)).NotTo(Equal("obj.part.minio")) + } + matches, globErr := filepath.Glob(filepath.Join(partDir, "*.part.minio")) + g.Expect(globErr).NotTo(HaveOccurred()) + g.Expect(matches).To(BeEmpty()) + g.Eventually(closed.Load, 2*time.Second, 10*time.Millisecond).Should(BeTrue()) +} + +func TestMinioClient_FGetObject_CopyError(t *testing.T) { + var closed atomic.Bool + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT") + w.Header().Set("ETag", `"abc123"`) + w.Header().Set("Content-Length", "64") + if r.Method == http.MethodGet { + _, _ = w.Write([]byte("short")) + } + }) + client := newTrackingMinioClient(t, handler, &closed) + + dir := t.TempDir() + path := filepath.Join(dir, "obj") + _, err := client.FGetObject(context.Background(), "test-bucket", "obj", path) + g := NewWithT(t) + g.Expect(err).To(HaveOccurred()) + _, statErr := os.Stat(path) + g.Expect(os.IsNotExist(statErr)).To(BeTrue()) + matches, globErr := filepath.Glob(filepath.Join(dir, "*.part.minio")) + g.Expect(globErr).NotTo(HaveOccurred()) + g.Expect(matches).To(BeEmpty()) + g.Eventually(closed.Load, 2*time.Second, 10*time.Millisecond).Should(BeTrue()) +} + func TestFGetObjectInvalidCredentials(t *testing.T) { invalidSecret := secret.DeepCopy() invalidSecret.Data["accesskey"] = []byte("invalid")