Merge pull request #4502 from akshat-kmr/master

Add logging binary support when terminal is true
2020-10-08 12:14:39 +09:00
parent 515133671b 61da6986c0
commit 915263f269
9 changed files with 362 additions and 61 deletions
--- a/runtime/v1/shim/service_linux.go
+++ b/runtime/v1/shim/service_linux.go
@@ -19,10 +19,14 @@ package shim
 import (
 	"context"
 	"io"
+	"net/url"
+	"os"
 	"sync"
 	"syscall"

 	"github.com/containerd/console"
+	"github.com/containerd/containerd/namespaces"
+	"github.com/containerd/containerd/runtime"
 	"github.com/containerd/fifo"
 	"github.com/pkg/errors"
 )
@@ -31,7 +35,7 @@ type linuxPlatform struct {
 	epoller *console.Epoller
 }

-func (p *linuxPlatform) CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg *sync.WaitGroup) (console.Console, error) {
+func (p *linuxPlatform) CopyConsole(ctx context.Context, console console.Console, id, stdin, stdout, stderr string, wg *sync.WaitGroup) (cons console.Console, retErr error) {
 	if p.epoller == nil {
 		return nil, errors.New("uninitialized epoller")
 	}
@@ -59,26 +63,98 @@ func (p *linuxPlatform) CopyConsole(ctx context.Context, console console.Console
 		}()
 	}

-	outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
+	uri, err := url.Parse(stdout)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "unable to parse stdout uri")
 	}
-	outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
-	if err != nil {
-		return nil, err
+
+	switch uri.Scheme {
+	case "binary":
+		ns, err := namespaces.NamespaceRequired(ctx)
+		if err != nil {
+			return nil, err
+		}
+
+		cmd := runtime.NewBinaryCmd(uri, id, ns)
+
+		// In case of unexpected errors during logging binary start, close open pipes
+		var filesToClose []*os.File
+
+		defer func() {
+			if retErr != nil {
+				runtime.CloseFiles(filesToClose...)
+			}
+		}()
+
+		// Create pipe to be used by logging binary for Stdout
+		outR, outW, err := os.Pipe()
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to create stdout pipes")
+		}
+		filesToClose = append(filesToClose, outR)
+
+		// Stderr is created for logging binary but unused when terminal is true
+		serrR, _, err := os.Pipe()
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to create stderr pipes")
+		}
+		filesToClose = append(filesToClose, serrR)
+
+		r, w, err := os.Pipe()
+		if err != nil {
+			return nil, err
+		}
+		filesToClose = append(filesToClose, r)
+
+		cmd.ExtraFiles = append(cmd.ExtraFiles, outR, serrR, w)
+
+		wg.Add(1)
+		cwg.Add(1)
+		go func() {
+			cwg.Done()
+			io.Copy(outW, epollConsole)
+			outW.Close()
+			wg.Done()
+		}()
+
+		if err := cmd.Start(); err != nil {
+			return nil, errors.Wrap(err, "failed to start logging binary process")
+		}
+
+		// Close our side of the pipe after start
+		if err := w.Close(); err != nil {
+			return nil, errors.Wrap(err, "failed to close write pipe after start")
+		}
+
+		// Wait for the logging binary to be ready
+		b := make([]byte, 1)
+		if _, err := r.Read(b); err != nil && err != io.EOF {
+			return nil, errors.Wrap(err, "failed to read from logging binary")
+		}
+		cwg.Wait()
+
+	default:
+		outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
+		if err != nil {
+			return nil, err
+		}
+		outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
+		if err != nil {
+			return nil, err
+		}
+		wg.Add(1)
+		cwg.Add(1)
+		go func() {
+			cwg.Done()
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+			io.CopyBuffer(outw, epollConsole, *p)
+			outw.Close()
+			outr.Close()
+			wg.Done()
+		}()
+		cwg.Wait()
 	}
-	wg.Add(1)
-	cwg.Add(1)
-	go func() {
-		cwg.Done()
-		p := bufPool.Get().(*[]byte)
-		defer bufPool.Put(p)
-		io.CopyBuffer(outw, epollConsole, *p)
-		outw.Close()
-		outr.Close()
-		wg.Done()
-	}()
-	cwg.Wait()
 	return epollConsole, nil
 }

--- a/runtime/v1/shim/service_unix.go
+++ b/runtime/v1/shim/service_unix.go
@@ -21,17 +21,22 @@ package shim
 import (
 	"context"
 	"io"
+	"net/url"
+	"os"
 	"sync"
 	"syscall"

 	"github.com/containerd/console"
+	"github.com/containerd/containerd/namespaces"
+	"github.com/containerd/containerd/runtime"
 	"github.com/containerd/fifo"
+	"github.com/pkg/errors"
 )

 type unixPlatform struct {
 }

-func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg *sync.WaitGroup) (console.Console, error) {
+func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console, id, stdin, stdout, stderr string, wg *sync.WaitGroup) (cons console.Console, retErr error) {
 	var cwg sync.WaitGroup
 	if stdin != "" {
 		in, err := fifo.OpenFifo(ctx, stdin, syscall.O_RDONLY, 0)
@@ -47,28 +52,98 @@ func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console,
 			io.CopyBuffer(console, in, *p)
 		}()
 	}
-	outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
+	uri, err := url.Parse(stdout)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "unable to parse stdout uri")
 	}
-	outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
-	if err != nil {
-		return nil, err
-	}
-	wg.Add(1)
-	cwg.Add(1)
-	go func() {
-		cwg.Done()
-		p := bufPool.Get().(*[]byte)
-		defer bufPool.Put(p)

-		io.CopyBuffer(outw, console, *p)
-		console.Close()
-		outr.Close()
-		outw.Close()
-		wg.Done()
-	}()
-	cwg.Wait()
+	switch uri.Scheme {
+	case "binary":
+		ns, err := namespaces.NamespaceRequired(ctx)
+		if err != nil {
+			return nil, err
+		}
+
+		cmd := runtime.NewBinaryCmd(uri, id, ns)
+
+		// In case of unexpected errors during logging binary start, close open pipes
+		var filesToClose []*os.File
+
+		defer func() {
+			if retErr != nil {
+				runtime.CloseFiles(filesToClose...)
+			}
+		}()
+
+		// Create pipe to be used by logging binary for Stdout
+		outR, outW, err := os.Pipe()
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to create stdout pipes")
+		}
+		filesToClose = append(filesToClose, outR)
+
+		// Stderr is created for logging binary but unused when terminal is true
+		serrR, _, err := os.Pipe()
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to create stderr pipes")
+		}
+		filesToClose = append(filesToClose, serrR)
+
+		r, w, err := os.Pipe()
+		if err != nil {
+			return nil, err
+		}
+		filesToClose = append(filesToClose, r)
+
+		cmd.ExtraFiles = append(cmd.ExtraFiles, outR, serrR, w)
+
+		wg.Add(1)
+		cwg.Add(1)
+		go func() {
+			cwg.Done()
+			io.Copy(outW, console)
+			outW.Close()
+			wg.Done()
+		}()
+
+		if err := cmd.Start(); err != nil {
+			return nil, errors.Wrap(err, "failed to start logging binary process")
+		}
+
+		// Close our side of the pipe after start
+		if err := w.Close(); err != nil {
+			return nil, errors.Wrap(err, "failed to close write pipe after start")
+		}
+
+		// Wait for the logging binary to be ready
+		b := make([]byte, 1)
+		if _, err := r.Read(b); err != nil && err != io.EOF {
+			return nil, errors.Wrap(err, "failed to read from logging binary")
+		}
+		cwg.Wait()
+
+	default:
+		outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
+		if err != nil {
+			return nil, err
+		}
+		outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
+		if err != nil {
+			return nil, err
+		}
+		wg.Add(1)
+		cwg.Add(1)
+		go func() {
+			cwg.Done()
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+			io.CopyBuffer(outw, console, *p)
+			outw.Close()
+			outr.Close()
+			wg.Done()
+		}()
+		cwg.Wait()
+	}
 	return console, nil
 }