From 780b05610ecf9ea107ec69c913dd074d69c1dc88 Mon Sep 17 00:00:00 2001 From: baude Date: Tue, 18 Jun 2019 13:56:18 -0500 Subject: account for varlink calls that dont use more the commit and pull varlink endpoints were not working correctly when 'more' was not being specified. Fixes: #3317 Fixes: #3318 Fixes: #3526 Signed-off-by: baude --- pkg/channelwriter/channelwriter.go | 34 ++++++++++++ pkg/varlinkapi/images.go | 108 +++++++++++++------------------------ pkg/varlinkapi/util.go | 40 ++++++++++++++ 3 files changed, 112 insertions(+), 70 deletions(-) create mode 100644 pkg/channelwriter/channelwriter.go diff --git a/pkg/channelwriter/channelwriter.go b/pkg/channelwriter/channelwriter.go new file mode 100644 index 000000000..d51400eb3 --- /dev/null +++ b/pkg/channelwriter/channelwriter.go @@ -0,0 +1,34 @@ +package channelwriter + +import "github.com/pkg/errors" + +// Writer is an io.writer-like object that "writes" to a channel +// instead of a buffer or file, etc. It is handy for varlink endpoints when +// needing to handle endpoints that do logging "real-time" +type Writer struct { + ByteChannel chan []byte +} + +// NewChannelWriter creates a new channel writer and adds a +// byte slice channel into it. +func NewChannelWriter() *Writer { + byteChannel := make(chan []byte) + return &Writer{ + ByteChannel: byteChannel, + } +} + +// Write method for Writer +func (c *Writer) Write(w []byte) (int, error) { + if c.ByteChannel == nil { + return 0, errors.New("channel writer channel cannot be nil") + } + c.ByteChannel <- w + return len(w), nil +} + +// Close method for Writer +func (c *Writer) Close() error { + close(c.ByteChannel) + return nil +} diff --git a/pkg/varlinkapi/images.go b/pkg/varlinkapi/images.go index 0e2ad6bbf..2bebfd406 100644 --- a/pkg/varlinkapi/images.go +++ b/pkg/varlinkapi/images.go @@ -25,6 +25,7 @@ import ( "github.com/containers/libpod/libpod" "github.com/containers/libpod/libpod/define" "github.com/containers/libpod/libpod/image" + "github.com/containers/libpod/pkg/channelwriter" "github.com/containers/libpod/pkg/util" "github.com/containers/libpod/utils" "github.com/containers/storage/pkg/archive" @@ -495,9 +496,19 @@ func (i *LibpodAPI) DeleteUnusedImages(call iopodman.VarlinkCall) error { // Commit ... func (i *LibpodAPI) Commit(call iopodman.VarlinkCall, name, imageName string, changes []string, author, message string, pause bool, manifestType string) error { - var newImage *image.Image + var ( + newImage *image.Image + log []string + mimeType string + ) + output := channelwriter.NewChannelWriter() + channelClose := func() { + if err := output.Close(); err != nil { + logrus.Errorf("failed to close channel writer: %q", err) + } + } + defer channelClose() - output := bytes.NewBuffer([]byte{}) ctr, err := i.Runtime.LookupContainer(name) if err != nil { return call.ReplyContainerNotFound(name, err.Error()) @@ -507,7 +518,6 @@ func (i *LibpodAPI) Commit(call iopodman.VarlinkCall, name, imageName string, ch return call.ReplyErrorOccurred(err.Error()) } sc := image.GetSystemContext(rtc.SignaturePolicyPath, "", false) - var mimeType string switch manifestType { case "oci", "": //nolint mimeType = buildah.OCIv1ImageManifest @@ -535,6 +545,7 @@ func (i *LibpodAPI) Commit(call iopodman.VarlinkCall, name, imageName string, ch } c := make(chan error) + defer close(c) go func() { newImage, err = ctr.Commit(getContext(), imageName, options) @@ -542,48 +553,22 @@ func (i *LibpodAPI) Commit(call iopodman.VarlinkCall, name, imageName string, ch c <- err } c <- nil - close(c) }() - var log []string - done := false - for { - line, err := output.ReadString('\n') - if err == nil { - log = append(log, line) - continue - } else if err == io.EOF { - select { - case err := <-c: - if err != nil { - logrus.Errorf("reading of output during commit failed for %s", name) - return call.ReplyErrorOccurred(err.Error()) - } - done = true - default: - if !call.WantsMore() { - break - } - br := iopodman.MoreResponse{ - Logs: log, - } - call.ReplyCommit(br) - log = []string{} - } - } else { - return call.ReplyErrorOccurred(err.Error()) - } - if done { - break - } + // reply is the func being sent to the output forwarder. in this case it is replying + // with a more response struct + reply := func(br iopodman.MoreResponse) error { + return call.ReplyCommit(br) + } + log, err = forwardOutput(log, c, call.WantsMore(), output, reply) + if err != nil { + return call.ReplyErrorOccurred(err.Error()) } call.Continues = false - br := iopodman.MoreResponse{ Logs: log, Id: newImage.ID(), } - return call.ReplyCommit(br) } @@ -636,6 +621,7 @@ func (i *LibpodAPI) ExportImage(call iopodman.VarlinkCall, name, destination str func (i *LibpodAPI) PullImage(call iopodman.VarlinkCall, name string) error { var ( imageID string + err error ) dockerRegistryOptions := image.DockerRegistryOptions{} so := image.SigningOptions{} @@ -643,8 +629,16 @@ func (i *LibpodAPI) PullImage(call iopodman.VarlinkCall, name string) error { if call.WantsMore() { call.Continues = true } - output := bytes.NewBuffer([]byte{}) + output := channelwriter.NewChannelWriter() + channelClose := func() { + if err := output.Close(); err != nil { + logrus.Errorf("failed to close channel writer: %q", err) + } + } + defer channelClose() c := make(chan error) + defer close(c) + go func() { if strings.HasPrefix(name, dockerarchive.Transport.Name()+":") { srcRef, err := alltransports.ParseImageName(name) @@ -666,43 +660,17 @@ func (i *LibpodAPI) PullImage(call iopodman.VarlinkCall, name string) error { } } c <- nil - close(c) }() var log []string - done := false - for { - line, err := output.ReadString('\n') - if err == nil { - log = append(log, line) - continue - } else if err == io.EOF { - select { - case err := <-c: - if err != nil { - logrus.Errorf("reading of output during pull failed for %s", name) - return call.ReplyErrorOccurred(err.Error()) - } - done = true - default: - if !call.WantsMore() { - break - } - br := iopodman.MoreResponse{ - Logs: log, - } - call.ReplyPullImage(br) - log = []string{} - } - } else { - return call.ReplyErrorOccurred(err.Error()) - } - if done { - break - } + reply := func(br iopodman.MoreResponse) error { + return call.ReplyPullImage(br) + } + log, err = forwardOutput(log, c, call.WantsMore(), output, reply) + if err != nil { + return call.ReplyErrorOccurred(err.Error()) } call.Continues = false - br := iopodman.MoreResponse{ Logs: log, Id: imageID, diff --git a/pkg/varlinkapi/util.go b/pkg/varlinkapi/util.go index a74105795..e8f74e6aa 100644 --- a/pkg/varlinkapi/util.go +++ b/pkg/varlinkapi/util.go @@ -13,6 +13,7 @@ import ( "github.com/containers/libpod/cmd/podman/varlink" "github.com/containers/libpod/libpod" "github.com/containers/libpod/libpod/define" + "github.com/containers/libpod/pkg/channelwriter" "github.com/containers/storage/pkg/archive" ) @@ -196,3 +197,42 @@ func makePsOpts(inOpts iopodman.PsOpts) shared.PsOptions { Sync: derefBool(inOpts.Sync), } } + +// forwardOutput is a helper method for varlink endpoints that employ both more and without +// more. it is capable of sending updates as the output writer gets them or append them +// all to a log. the chan error is the error from the libpod call so we can honor +// and error event in that case. +func forwardOutput(log []string, c chan error, wantsMore bool, output *channelwriter.Writer, reply func(br iopodman.MoreResponse) error) ([]string, error) { + done := false + for { + select { + // We need to check if the libpod func being called has returned an + // error yet + case err := <-c: + if err != nil { + return nil, err + } + done = true + // if no error is found, we pull what we can from the log writer and + // append it to log string slice + case line := <-output.ByteChannel: + log = append(log, string(line)) + // If the end point is being used in more mode, send what we have + if wantsMore { + br := iopodman.MoreResponse{ + Logs: log, + } + if err := reply(br); err != nil { + return nil, err + } + // "reset" the log to empty because we are sending what we + // get as we get it + log = []string{} + } + } + if done { + break + } + } + return log, nil +} -- cgit v1.2.3-54-g00ecf