Use request context instead of background

This prevents goroutine leak:
If background context were used then push operation would continue even
if client aborted request by closing connection.

[NO TESTS NEEDED]

Signed-off-by: Matej Vasek <mvasek@redhat.com>
This commit is contained in:
Matej Vasek
2021-06-01 15:05:13 +02:00
parent 7dd463bad1
commit ad182976b6

View File

@ -1,7 +1,6 @@
package compat package compat
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io/ioutil" "io/ioutil"
@ -12,7 +11,6 @@ import (
"github.com/containers/podman/v3/libpod" "github.com/containers/podman/v3/libpod"
"github.com/containers/podman/v3/pkg/api/handlers/utils" "github.com/containers/podman/v3/pkg/api/handlers/utils"
"github.com/containers/podman/v3/pkg/auth" "github.com/containers/podman/v3/pkg/auth"
"github.com/containers/podman/v3/pkg/channel"
"github.com/containers/podman/v3/pkg/domain/entities" "github.com/containers/podman/v3/pkg/domain/entities"
"github.com/containers/podman/v3/pkg/domain/infra/abi" "github.com/containers/podman/v3/pkg/domain/infra/abi"
"github.com/containers/storage" "github.com/containers/storage"
@ -101,46 +99,33 @@ func PushImage(w http.ResponseWriter, r *http.Request) {
destination = imageName destination = imageName
} }
errorWriter := channel.NewWriter(make(chan []byte)) flush := func() {}
defer errorWriter.Close()
statusWriter := channel.NewWriter(make(chan []byte))
defer statusWriter.Close()
runCtx, cancel := context.WithCancel(context.Background())
var failed bool
go func() {
defer cancel()
statusWriter.Write([]byte(fmt.Sprintf("The push refers to repository [%s]", imageName)))
err := imageEngine.Push(runCtx, imageName, destination, options)
if err != nil {
if errors.Cause(err) != storage.ErrImageUnknown {
errorWriter.Write([]byte("An image does not exist locally with the tag: " + imageName))
} else {
errorWriter.Write([]byte(err.Error()))
}
}
}()
flush := func() {
if flusher, ok := w.(http.Flusher); ok { if flusher, ok := w.(http.Flusher); ok {
flusher.Flush() flush = flusher.Flush
}
} }
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
w.Header().Add("Content-Type", "application/json") w.Header().Add("Content-Type", "application/json")
flush() flush()
var report jsonmessage.JSONMessage
enc := json.NewEncoder(w) enc := json.NewEncoder(w)
enc.SetEscapeHTML(true) enc.SetEscapeHTML(true)
report.Status = fmt.Sprintf("The push refers to repository [%s]", imageName)
if err := enc.Encode(report); err != nil {
logrus.Warnf("Failed to json encode error %q", err.Error())
}
flush()
pushErrChan := make(chan error)
go func() {
pushErrChan <- imageEngine.Push(r.Context(), imageName, destination, options)
}()
loop: // break out of for/select infinite loop loop: // break out of for/select infinite loop
for { for {
var report jsonmessage.JSONMessage report = jsonmessage.JSONMessage{}
select { select {
case e := <-options.Progress: case e := <-options.Progress:
@ -160,29 +145,40 @@ loop: // break out of for/select infinite loop
} }
report.ID = e.Artifact.Digest.Encoded()[0:12] report.ID = e.Artifact.Digest.Encoded()[0:12]
if err := enc.Encode(report); err != nil { if err := enc.Encode(report); err != nil {
errorWriter.Write([]byte(err.Error())) logrus.Warnf("Failed to json encode error %q", err.Error())
} }
flush() flush()
case e := <-statusWriter.Chan(): case err := <-pushErrChan:
report.Status = string(e) if err != nil {
if err := enc.Encode(report); err != nil { var msg string
errorWriter.Write([]byte(err.Error())) if errors.Cause(err) != storage.ErrImageUnknown {
msg = "An image does not exist locally with the tag: " + imageName
} else {
msg = err.Error()
} }
flush()
case e := <-errorWriter.Chan():
failed = true
report.Error = &jsonmessage.JSONError{ report.Error = &jsonmessage.JSONError{
Message: string(e), Message: msg,
} }
report.ErrorMessage = string(e) report.ErrorMessage = msg
if err := enc.Encode(report); err != nil { if err := enc.Encode(report); err != nil {
logrus.Warnf("Failed to json encode error %q", err.Error()) logrus.Warnf("Failed to json encode error %q", err.Error())
} }
flush() flush()
case <-runCtx.Done(): break loop
if !failed { }
digestBytes, err := ioutil.ReadAll(digestFile) digestBytes, err := ioutil.ReadAll(digestFile)
if err == nil { if err != nil {
report.Error = &jsonmessage.JSONError{
Message: err.Error(),
}
report.ErrorMessage = err.Error()
if err := enc.Encode(report); err != nil {
logrus.Warnf("Failed to json encode error %q", err.Error())
}
flush()
break loop
}
tag := query.Tag tag := query.Tag
if tag == "" { if tag == "" {
tag = "latest" tag = "latest"
@ -191,12 +187,8 @@ loop: // break out of for/select infinite loop
if err := enc.Encode(report); err != nil { if err := enc.Encode(report); err != nil {
logrus.Warnf("Failed to json encode error %q", err.Error()) logrus.Warnf("Failed to json encode error %q", err.Error())
} }
flush() flush()
}
}
break loop // break out of for/select infinite loop
case <-r.Context().Done():
// Client has closed connection
break loop // break out of for/select infinite loop break loop // break out of for/select infinite loop
} }
} }