From fa2a8d72094297b99c988af687c859b325173a98 Mon Sep 17 00:00:00 2001 From: jiuker <2818723467@qq.com> Date: Mon, 21 Aug 2023 16:09:07 +0800 Subject: [PATCH] fix: drain the req.body into io.Discard correctly (#17881) --- internal/event/target/elasticsearch.go | 22 +++++++--------------- internal/event/target/webhook.go | 24 ++++++++++++++++++++++-- 2 files changed, 29 insertions(+), 17 deletions(-) diff --git a/internal/event/target/elasticsearch.go b/internal/event/target/elasticsearch.go index 282a81deb..ea567b61f 100644 --- a/internal/event/target/elasticsearch.go +++ b/internal/event/target/elasticsearch.go @@ -23,7 +23,6 @@ import ( "encoding/base64" "encoding/json" "fmt" - "io" "net/http" "net/url" "os" @@ -475,12 +474,10 @@ func (c *esClientV7) createIndex(args ElasticsearchArgs) error { if err != nil { return err } - defer resp.Body.Close() + defer DrainBody(resp.Body) if resp.IsError() { - err := fmt.Errorf("Create index err: %s", res.String()) - return err + return fmt.Errorf("Create index err: %v", res) } - io.Copy(io.Discard, resp.Body) return nil } return nil @@ -493,8 +490,7 @@ func (c *esClientV7) ping(ctx context.Context, _ ElasticsearchArgs) (bool, error if err != nil { return false, store.ErrNotConnected } - io.Copy(io.Discard, resp.Body) - resp.Body.Close() + DrainBody(resp.Body) return !resp.IsError(), nil } @@ -507,8 +503,7 @@ func (c *esClientV7) entryExists(ctx context.Context, index string, key string) if err != nil { return false, err } - io.Copy(io.Discard, res.Body) - res.Body.Close() + DrainBody(res.Body) return !res.IsError(), nil } @@ -523,8 +518,7 @@ func (c *esClientV7) removeEntry(ctx context.Context, index string, key string) if err != nil { return err } - defer res.Body.Close() - defer io.Copy(io.Discard, res.Body) + defer DrainBody(res.Body) if res.IsError() { return fmt.Errorf("Delete err: %s", res.String()) } @@ -552,8 +546,7 @@ func (c *esClientV7) updateEntry(ctx context.Context, index string, key string, if err != nil { return err } - defer res.Body.Close() - defer io.Copy(io.Discard, res.Body) + defer DrainBody(res.Body) if res.IsError() { return fmt.Errorf("Update err: %s", res.String()) } @@ -579,8 +572,7 @@ func (c *esClientV7) addEntry(ctx context.Context, index string, eventData event if err != nil { return err } - defer res.Body.Close() - defer io.Copy(io.Discard, res.Body) + defer DrainBody(res.Body) if res.IsError() { return fmt.Errorf("Add err: %s", res.String()) } diff --git a/internal/event/target/webhook.go b/internal/event/target/webhook.go index d40a3114d..1d048b6f1 100644 --- a/internal/event/target/webhook.go +++ b/internal/event/target/webhook.go @@ -34,6 +34,7 @@ import ( "syscall" "time" + "github.com/dustin/go-humanize" "github.com/minio/minio/internal/event" "github.com/minio/minio/internal/logger" "github.com/minio/minio/internal/once" @@ -194,8 +195,7 @@ func (target *WebhookTarget) send(eventData event.Event) error { if err != nil { return err } - defer resp.Body.Close() - io.Copy(io.Discard, resp.Body) + defer DrainBody(resp.Body) if resp.StatusCode < 200 || resp.StatusCode > 299 { return fmt.Errorf("sending event failed with %v", resp.Status) @@ -297,3 +297,23 @@ func NewWebhookTarget(ctx context.Context, id string, args WebhookArgs, loggerOn return target, nil } + +// DrainBody close non nil response with any response Body. +// convenient wrapper to drain any remaining data on response body. +// +// Subsequently this allows golang http RoundTripper +// to re-use the same connection for future requests. +func DrainBody(respBody io.ReadCloser) { + // Callers should close resp.Body when done reading from it. + // If resp.Body is not closed, the Client's underlying RoundTripper + // (typically Transport) may not be able to re-use a persistent TCP + // connection to the server for a subsequent "keep-alive" request. + if respBody != nil { + // Drain any remaining Body and then close the connection. + // Without this closing connection would disallow re-using + // the same connection for future uses. + // - http://stackoverflow.com/a/17961593/4465767 + defer respBody.Close() + io.CopyN(io.Discard, respBody, 1*humanize.MiByte) + } +}