Skip to content

Commit 915263f

Browse files
AkihiroSudaAkihiroSuda
authored
Merge pull request #4502 from akshat-kmr/master
Add logging binary support when terminal is true
2 parents 5151336 + 61da698 commit 915263f

9 files changed

Lines changed: 362 additions & 61 deletions

File tree

‎cio/io.go‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,26 @@ func BinaryIO(binary string, args map[string]string) Creator {
260260
}
261261
}
262262

263+
// TerminalBinaryIO forwards container STDOUT|STDERR directly to a logging binary
264+
// It also sets the terminal option to true
265+
func TerminalBinaryIO(binary string, args map[string]string) Creator {
266+
return func(_ string) (IO, error) {
267+
uri, err := LogURIGenerator("binary", binary, args)
268+
if err != nil {
269+
return nil, err
270+
}
271+
272+
res := uri.String()
273+
return &logURI{
274+
config: Config{
275+
Stdout: res,
276+
Stderr: res,
277+
Terminal: true,
278+
},
279+
}, nil
280+
}
281+
}
282+
263283
// LogFile creates a file on disk that logs the task's STDOUT,STDERR.
264284
// If the log file already exists, the logs will be appended to the file.
265285
func LogFile(path string) Creator {

‎pkg/process/exec.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -221,7 +221,7 @@ func (e *execProcess) start(ctx context.Context) (err error) {
221221
if err != nil {
222222
return errors.Wrap(err, "failed to retrieve console master")
223223
}
224-
if e.console, err = e.parent.Platform.CopyConsole(ctx, console, e.stdio.Stdin, e.stdio.Stdout, e.stdio.Stderr, &e.wg); err != nil {
224+
if e.console, err = e.parent.Platform.CopyConsole(ctx, console, e.id, e.stdio.Stdin, e.stdio.Stdout, e.stdio.Stderr, &e.wg); err != nil {
225225
return errors.Wrap(err, "failed to start console copy")
226226
}
227227
} else {

‎pkg/process/init.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ func (p *Init) Create(ctx context.Context, r *CreateConfig) error {
157157
if err != nil {
158158
return errors.Wrap(err, "failed to retrieve console master")
159159
}
160-
console, err = p.Platform.CopyConsole(ctx, console, r.Stdin, r.Stdout, r.Stderr, &p.wg)
160+
console, err = p.Platform.CopyConsole(ctx, console, p.id, r.Stdin, r.Stdout, r.Stderr, &p.wg)
161161
if err != nil {
162162
return errors.Wrap(err, "failed to start console copy")
163163
}

‎pkg/process/init_state.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -172,7 +172,7 @@ func (s *createdCheckpointState) Start(ctx context.Context) error {
172172
if err != nil {
173173
return errors.Wrap(err, "failed to retrieve console master")
174174
}
175-
console, err = p.Platform.CopyConsole(ctx, console, sio.Stdin, sio.Stdout, sio.Stderr, &p.wg)
175+
console, err = p.Platform.CopyConsole(ctx, console, p.id, sio.Stdin, sio.Stdout, sio.Stderr, &p.wg)
176176
if err != nil {
177177
return errors.Wrap(err, "failed to start console copy")
178178
}

‎pkg/stdio/platform.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ import (
2626
// Platform handles platform-specific behavior that may differs across
2727
// platform implementations
2828
type Platform interface {
29-
CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string,
29+
CopyConsole(ctx context.Context, console console.Console, id, stdin, stdout, stderr string,
3030
wg *sync.WaitGroup) (console.Console, error)
3131
ShutdownConsole(ctx context.Context, console console.Console) error
3232
Close() error

‎runtime/io.go‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
/*
2+
Copyright The containerd Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package runtime
18+
19+
import (
20+
"net/url"
21+
"os"
22+
"os/exec"
23+
)
24+
25+
// NewBinaryCmd returns a Cmd to be used to start a logging binary.
26+
// The Cmd is generated from the provided uri, and the container ID and
27+
// namespace are appended to the Cmd environment.
28+
func NewBinaryCmd(binaryURI *url.URL, id, ns string) *exec.Cmd {
29+
var args []string
30+
for k, vs := range binaryURI.Query() {
31+
args = append(args, k)
32+
if len(vs) > 0 {
33+
args = append(args, vs[0])
34+
}
35+
}
36+
37+
cmd := exec.Command(binaryURI.Path, args...)
38+
39+
cmd.Env = append(cmd.Env,
40+
"CONTAINER_ID="+id,
41+
"CONTAINER_NAMESPACE="+ns,
42+
)
43+
44+
return cmd
45+
}
46+
47+
// CloseFiles closes any files passed in.
48+
// It it used for cleanup in the event of unexpected errors.
49+
func CloseFiles(files ...*os.File) {
50+
for _, file := range files {
51+
file.Close()
52+
}
53+
}

‎runtime/v1/shim/service_linux.go‎

Lines changed: 94 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,14 @@ package shim
1919
import (
2020
"context"
2121
"io"
22+
"net/url"
23+
"os"
2224
"sync"
2325
"syscall"
2426

2527
"github.com/containerd/console"
28+
"github.com/containerd/containerd/namespaces"
29+
"github.com/containerd/containerd/runtime"
2630
"github.com/containerd/fifo"
2731
"github.com/pkg/errors"
2832
)
@@ -31,7 +35,7 @@ type linuxPlatform struct {
3135
epoller *console.Epoller
3236
}
3337

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

62-
outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
66+
uri, err := url.Parse(stdout)
6367
if err != nil {
64-
return nil, err
68+
return nil, errors.Wrap(err, "unable to parse stdout uri")
6569
}
66-
outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
67-
if err != nil {
68-
return nil, err
70+
71+
switch uri.Scheme {
72+
case "binary":
73+
ns, err := namespaces.NamespaceRequired(ctx)
74+
if err != nil {
75+
return nil, err
76+
}
77+
78+
cmd := runtime.NewBinaryCmd(uri, id, ns)
79+
80+
// In case of unexpected errors during logging binary start, close open pipes
81+
var filesToClose []*os.File
82+
83+
defer func() {
84+
if retErr != nil {
85+
runtime.CloseFiles(filesToClose...)
86+
}
87+
}()
88+
89+
// Create pipe to be used by logging binary for Stdout
90+
outR, outW, err := os.Pipe()
91+
if err != nil {
92+
return nil, errors.Wrap(err, "failed to create stdout pipes")
93+
}
94+
filesToClose = append(filesToClose, outR)
95+
96+
// Stderr is created for logging binary but unused when terminal is true
97+
serrR, _, err := os.Pipe()
98+
if err != nil {
99+
return nil, errors.Wrap(err, "failed to create stderr pipes")
100+
}
101+
filesToClose = append(filesToClose, serrR)
102+
103+
r, w, err := os.Pipe()
104+
if err != nil {
105+
return nil, err
106+
}
107+
filesToClose = append(filesToClose, r)
108+
109+
cmd.ExtraFiles = append(cmd.ExtraFiles, outR, serrR, w)
110+
111+
wg.Add(1)
112+
cwg.Add(1)
113+
go func() {
114+
cwg.Done()
115+
io.Copy(outW, epollConsole)
116+
outW.Close()
117+
wg.Done()
118+
}()
119+
120+
if err := cmd.Start(); err != nil {
121+
return nil, errors.Wrap(err, "failed to start logging binary process")
122+
}
123+
124+
// Close our side of the pipe after start
125+
if err := w.Close(); err != nil {
126+
return nil, errors.Wrap(err, "failed to close write pipe after start")
127+
}
128+
129+
// Wait for the logging binary to be ready
130+
b := make([]byte, 1)
131+
if _, err := r.Read(b); err != nil && err != io.EOF {
132+
return nil, errors.Wrap(err, "failed to read from logging binary")
133+
}
134+
cwg.Wait()
135+
136+
default:
137+
outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
138+
if err != nil {
139+
return nil, err
140+
}
141+
outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
142+
if err != nil {
143+
return nil, err
144+
}
145+
wg.Add(1)
146+
cwg.Add(1)
147+
go func() {
148+
cwg.Done()
149+
p := bufPool.Get().(*[]byte)
150+
defer bufPool.Put(p)
151+
io.CopyBuffer(outw, epollConsole, *p)
152+
outw.Close()
153+
outr.Close()
154+
wg.Done()
155+
}()
156+
cwg.Wait()
69157
}
70-
wg.Add(1)
71-
cwg.Add(1)
72-
go func() {
73-
cwg.Done()
74-
p := bufPool.Get().(*[]byte)
75-
defer bufPool.Put(p)
76-
io.CopyBuffer(outw, epollConsole, *p)
77-
outw.Close()
78-
outr.Close()
79-
wg.Done()
80-
}()
81-
cwg.Wait()
82158
return epollConsole, nil
83159
}
84160

‎runtime/v1/shim/service_unix.go‎

Lines changed: 95 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -21,17 +21,22 @@ package shim
2121
import (
2222
"context"
2323
"io"
24+
"net/url"
25+
"os"
2426
"sync"
2527
"syscall"
2628

2729
"github.com/containerd/console"
30+
"github.com/containerd/containerd/namespaces"
31+
"github.com/containerd/containerd/runtime"
2832
"github.com/containerd/fifo"
33+
"github.com/pkg/errors"
2934
)
3035

3136
type unixPlatform struct {
3237
}
3338

34-
func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console, stdin, stdout, stderr string, wg *sync.WaitGroup) (console.Console, error) {
39+
func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console, id, stdin, stdout, stderr string, wg *sync.WaitGroup) (cons console.Console, retErr error) {
3540
var cwg sync.WaitGroup
3641
if stdin != "" {
3742
in, err := fifo.OpenFifo(ctx, stdin, syscall.O_RDONLY, 0)
@@ -47,28 +52,98 @@ func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console,
4752
io.CopyBuffer(console, in, *p)
4853
}()
4954
}
50-
outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
55+
uri, err := url.Parse(stdout)
5156
if err != nil {
52-
return nil, err
57+
return nil, errors.Wrap(err, "unable to parse stdout uri")
5358
}
54-
outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
55-
if err != nil {
56-
return nil, err
59+
60+
switch uri.Scheme {
61+
case "binary":
62+
ns, err := namespaces.NamespaceRequired(ctx)
63+
if err != nil {
64+
return nil, err
65+
}
66+
67+
cmd := runtime.NewBinaryCmd(uri, id, ns)
68+
69+
// In case of unexpected errors during logging binary start, close open pipes
70+
var filesToClose []*os.File
71+
72+
defer func() {
73+
if retErr != nil {
74+
runtime.CloseFiles(filesToClose...)
75+
}
76+
}()
77+
78+
// Create pipe to be used by logging binary for Stdout
79+
outR, outW, err := os.Pipe()
80+
if err != nil {
81+
return nil, errors.Wrap(err, "failed to create stdout pipes")
82+
}
83+
filesToClose = append(filesToClose, outR)
84+
85+
// Stderr is created for logging binary but unused when terminal is true
86+
serrR, _, err := os.Pipe()
87+
if err != nil {
88+
return nil, errors.Wrap(err, "failed to create stderr pipes")
89+
}
90+
filesToClose = append(filesToClose, serrR)
91+
92+
r, w, err := os.Pipe()
93+
if err != nil {
94+
return nil, err
95+
}
96+
filesToClose = append(filesToClose, r)
97+
98+
cmd.ExtraFiles = append(cmd.ExtraFiles, outR, serrR, w)
99+
100+
wg.Add(1)
101+
cwg.Add(1)
102+
go func() {
103+
cwg.Done()
104+
io.Copy(outW, console)
105+
outW.Close()
106+
wg.Done()
107+
}()
108+
109+
if err := cmd.Start(); err != nil {
110+
return nil, errors.Wrap(err, "failed to start logging binary process")
111+
}
112+
113+
// Close our side of the pipe after start
114+
if err := w.Close(); err != nil {
115+
return nil, errors.Wrap(err, "failed to close write pipe after start")
116+
}
117+
118+
// Wait for the logging binary to be ready
119+
b := make([]byte, 1)
120+
if _, err := r.Read(b); err != nil && err != io.EOF {
121+
return nil, errors.Wrap(err, "failed to read from logging binary")
122+
}
123+
cwg.Wait()
124+
125+
default:
126+
outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
127+
if err != nil {
128+
return nil, err
129+
}
130+
outr, err := fifo.OpenFifo(ctx, stdout, syscall.O_RDONLY, 0)
131+
if err != nil {
132+
return nil, err
133+
}
134+
wg.Add(1)
135+
cwg.Add(1)
136+
go func() {
137+
cwg.Done()
138+
p := bufPool.Get().(*[]byte)
139+
defer bufPool.Put(p)
140+
io.CopyBuffer(outw, console, *p)
141+
outw.Close()
142+
outr.Close()
143+
wg.Done()
144+
}()
145+
cwg.Wait()
57146
}
58-
wg.Add(1)
59-
cwg.Add(1)
60-
go func() {
61-
cwg.Done()
62-
p := bufPool.Get().(*[]byte)
63-
defer bufPool.Put(p)
64-
65-
io.CopyBuffer(outw, console, *p)
66-
console.Close()
67-
outr.Close()
68-
outw.Close()
69-
wg.Done()
70-
}()
71-
cwg.Wait()
72147
return console, nil
73148
}
74149

0 commit comments

Comments
 (0)