forked from containerd/containerd
-
Notifications
You must be signed in to change notification settings - Fork 0
/
process.go
130 lines (118 loc) · 2.81 KB
/
process.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
package containerd
import (
"context"
"syscall"
eventsapi "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/api/services/tasks/v1"
"github.com/containerd/containerd/typeurl"
specs "github.com/opencontainers/runtime-spec/specs-go"
)
type process struct {
id string
task *task
pid uint32
io *IO
spec *specs.Process
}
func (p *process) ID() string {
return p.id
}
// Pid returns the pid of the process
// The pid is not set until start is called and returns
func (p *process) Pid() uint32 {
return p.pid
}
// Start starts the exec process
func (p *process) Start(ctx context.Context) error {
any, err := typeurl.MarshalAny(p.spec)
if err != nil {
return err
}
request := &tasks.ExecProcessRequest{
ContainerID: p.task.id,
ExecID: p.id,
Terminal: p.io.Terminal,
Stdin: p.io.Stdin,
Stdout: p.io.Stdout,
Stderr: p.io.Stderr,
Spec: any,
}
response, err := p.task.client.TaskService().Exec(ctx, request)
if err != nil {
p.io.Cancel()
p.io.Wait()
p.io.Close()
return err
}
p.pid = response.Pid
return nil
}
func (p *process) Kill(ctx context.Context, s syscall.Signal) error {
_, err := p.task.client.TaskService().Kill(ctx, &tasks.KillRequest{
Signal: uint32(s),
ContainerID: p.task.id,
ExecID: p.id,
})
return err
}
func (p *process) Wait(ctx context.Context) (uint32, error) {
eventstream, err := p.task.client.EventService().Subscribe(ctx, &eventsapi.SubscribeRequest{})
if err != nil {
return UnknownExitStatus, err
}
for {
evt, err := eventstream.Recv()
if err != nil {
return UnknownExitStatus, err
}
if typeurl.Is(evt.Event, &eventsapi.TaskExit{}) {
v, err := typeurl.UnmarshalAny(evt.Event)
if err != nil {
return UnknownExitStatus, err
}
e := v.(*eventsapi.TaskExit)
if e.ID == p.id && e.ContainerID == p.task.id {
return e.ExitStatus, nil
}
}
}
}
func (p *process) CloseIO(ctx context.Context, opts ...IOCloserOpts) error {
r := &tasks.CloseIORequest{
ContainerID: p.task.id,
ExecID: p.id,
}
var i IOCloseInfo
for _, o := range opts {
o(&i)
}
r.Stdin = i.Stdin
_, err := p.task.client.TaskService().CloseIO(ctx, r)
return err
}
func (p *process) IO() *IO {
return p.io
}
func (p *process) Resize(ctx context.Context, w, h uint32) error {
_, err := p.task.client.TaskService().ResizePty(ctx, &tasks.ResizePtyRequest{
ContainerID: p.task.id,
Width: w,
Height: h,
ExecID: p.id,
})
return err
}
func (p *process) Delete(ctx context.Context) (uint32, error) {
if p.io != nil {
p.io.Wait()
p.io.Close()
}
r, err := p.task.client.TaskService().DeleteProcess(ctx, &tasks.DeleteProcessRequest{
ContainerID: p.task.id,
ExecID: p.id,
})
if err != nil {
return UnknownExitStatus, err
}
return r.ExitStatus, nil
}