Handle stderr in stream processors

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby 2019-08-06 21:06:18 +00:00
parent 3fded74bc7
commit 552a0b1be5
3 changed files with 86 additions and 18 deletions

View File

@ -76,11 +76,14 @@ func (s *fsApplier) Apply(ctx context.Context, desc ocispec.Descriptor, mounts [
}
defer ra.Close()
var processors []diff.StreamProcessor
processor := diff.NewProcessorChain(desc.MediaType, content.NewReader(ra))
processors = append(processors, processor)
for {
if processor, err = diff.GetProcessor(ctx, processor, config.ProcessorPayloads); err != nil {
return emptyDesc, errors.Wrapf(err, "failed to get stream processor for %s", desc.MediaType)
}
processors = append(processors, processor)
if processor.MediaType() == ocispec.MediaTypeImageLayer {
break
}
@ -102,6 +105,16 @@ func (s *fsApplier) Apply(ctx context.Context, desc ocispec.Descriptor, mounts [
}); err != nil {
return emptyDesc, err
}
for _, p := range processors {
if ep, ok := p.(interface {
Err() error
}); ok {
if err := ep.Err(); err != nil {
return emptyDesc, err
}
}
}
return ocispec.Descriptor{
MediaType: ocispec.MediaTypeImageLayer,
Size: rc.c,

View File

@ -25,14 +25,17 @@ import (
"io"
"os"
"os/exec"
"sync"
"github.com/gogo/protobuf/proto"
"github.com/gogo/protobuf/types"
"github.com/pkg/errors"
)
// NewBinaryProcessor returns a binary processor for use with processing content streams
func NewBinaryProcessor(ctx context.Context, imt, rmt string, stream StreamProcessor, name string, args []string, payload *types.Any) (StreamProcessor, error) {
cmd := exec.CommandContext(ctx, name, args...)
cmd.Env = os.Environ()
var payloadC io.Closer
if payload != nil {
@ -71,10 +74,19 @@ func NewBinaryProcessor(ctx context.Context, imt, rmt string, stream StreamProce
}
cmd.Stdout = w
stderr := bytes.NewBuffer(nil)
cmd.Stderr = stderr
if err := cmd.Start(); err != nil {
return nil, err
}
go cmd.Wait()
p := &binaryProcessor{
cmd: cmd,
r: r,
mt: rmt,
stderr: stderr,
}
go p.wait()
// close after start and dup
w.Close()
@ -84,17 +96,33 @@ func NewBinaryProcessor(ctx context.Context, imt, rmt string, stream StreamProce
if payloadC != nil {
payloadC.Close()
}
return &binaryProcessor{
cmd: cmd,
r: r,
mt: rmt,
}, nil
return p, nil
}
type binaryProcessor struct {
cmd *exec.Cmd
r *os.File
mt string
cmd *exec.Cmd
r *os.File
mt string
stderr *bytes.Buffer
mu sync.Mutex
err error
}
func (c *binaryProcessor) Err() error {
c.mu.Lock()
defer c.mu.Unlock()
return c.err
}
func (c *binaryProcessor) wait() {
if err := c.cmd.Wait(); err != nil {
if _, ok := err.(*exec.ExitError); ok {
c.mu.Lock()
c.err = errors.New(c.stderr.String())
c.mu.Unlock()
}
}
}
func (c *binaryProcessor) File() *os.File {

View File

@ -27,10 +27,12 @@ import (
"os"
"os/exec"
"path/filepath"
"sync"
winio "github.com/Microsoft/go-winio"
"github.com/gogo/protobuf/proto"
"github.com/gogo/protobuf/types"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
@ -39,6 +41,7 @@ const processorPipe = "STREAM_PROCESSOR_PIPE"
// NewBinaryProcessor returns a binary processor for use with processing content streams
func NewBinaryProcessor(ctx context.Context, imt, rmt string, stream StreamProcessor, name string, args []string, payload *types.Any) (StreamProcessor, error) {
cmd := exec.CommandContext(ctx, name, args...)
cmd.Env = os.Environ()
if payload != nil {
data, err := proto.Marshal(payload)
@ -84,28 +87,52 @@ func NewBinaryProcessor(ctx context.Context, imt, rmt string, stream StreamProce
return nil, err
}
cmd.Stdout = w
stderr := bytes.NewBuffer(nil)
cmd.Stderr = stderr
if err := cmd.Start(); err != nil {
return nil, err
}
go cmd.Wait()
p := &binaryProcessor{
cmd: cmd,
r: r,
mt: rmt,
stderr: stderr,
}
go p.wait()
// close after start and dup
w.Close()
if closer != nil {
closer()
}
return &binaryProcessor{
cmd: cmd,
r: r,
mt: rmt,
}, nil
return p, nil
}
type binaryProcessor struct {
cmd *exec.Cmd
r *os.File
mt string
cmd *exec.Cmd
r *os.File
mt string
stderr *bytes.Buffer
mu sync.Mutex
err error
}
func (c *binaryProcessor) Err() error {
c.mu.Lock()
defer c.mu.Unlock()
return c.err
}
func (c *binaryProcessor) wait() {
if err := c.cmd.Wait(); err != nil {
if _, ok := err.(*exec.ExitError); ok {
c.mu.Lock()
c.err = errors.New(c.stderr.String())
c.mu.Unlock()
}
}
}
func (c *binaryProcessor) File() *os.File {