github/containerd
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Use event service post for shim events

Browse Source 
Signed-off-by: Michael Crosby <crosbymichael@gmail.com>
This commit is contained in:
Michael Crosby
2017-07-07 16:12:33 -07:00
parent f39693eabe
commit 6578565216
12 changed files with 149 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
linux/bundle.go
View File

@@ -56,8 +56,8 @@ type bundle struct {
}
// NewShim connects to the shim managing the bundle and tasks
func (b *bundle) NewShim(ctx context.Context, binary string, remote bool) (*client.Client, error) {
opt := client.WithStart(binary)
func (b *bundle) NewShim(ctx context.Context, binary, grpcAddress string, remote bool) (*client.Client, error) {
opt := client.WithStart(binary, grpcAddress)
if !remote {
opt = client.WithLocal
}

86
linux/runtime.go
View File

@@ -8,7 +8,6 @@ import (
"io/ioutil"
"os"
"path/filepath"
"strings"
"google.golang.org/grpc"
@@ -84,22 +83,17 @@ func New(ic *plugin.InitContext) (interface{}, error) {
return nil, err
}
cfg := ic.Config.(*Config)
c, cancel := context.WithCancel(ic.Context)
r := &Runtime{
root: ic.Root,
remote: !cfg.NoShim,
shim: cfg.Shim,
runtime: cfg.Runtime,
events: make(chan *eventsapi.RuntimeEvent, 2048),
eventsContext: c,
eventsCancel: cancel,
monitor: monitor.(runtime.TaskMonitor),
tasks: newTaskList(),
emitter: events.GetPoster(ic.Context),
db: m.(*bolt.DB),
root: ic.Root,
remote: !cfg.NoShim,
shim: cfg.Shim,
runtime: cfg.Runtime,
monitor: monitor.(runtime.TaskMonitor),
tasks: newTaskList(),
emitter: events.GetPoster(ic.Context),
db: m.(*bolt.DB),
address: ic.Address,
}
// set the events output for a monitor if it generates events
r.monitor.Events(r.events)
tasks, err := r.restoreTasks(ic.Context)
if err != nil {
return nil, err
@@ -108,9 +102,6 @@ func New(ic *plugin.InitContext) (interface{}, error) {
if err := r.tasks.addWithNamespace(t.namespace, t); err != nil {
return nil, err
}
if err := r.handleEvents(ic.Context, t.shim); err != nil {
return nil, err
}
}
return r, nil
}
@@ -120,14 +111,12 @@ type Runtime struct {
shim string
runtime string
remote bool
address string
events chan *eventsapi.RuntimeEvent
eventsContext context.Context
eventsCancel func()
monitor runtime.TaskMonitor
tasks *taskList
emitter events.Poster
db *bolt.DB
monitor runtime.TaskMonitor
tasks *taskList
emitter events.Poster
db *bolt.DB
}
func (r *Runtime) ID() string {
@@ -148,7 +137,7 @@ func (r *Runtime) Create(ctx context.Context, id string, opts runtime.CreateOpts
bundle.Delete()
}
}()
s, err := bundle.NewShim(ctx, r.shim, r.remote)
s, err := bundle.NewShim(ctx, r.shim, r.address, r.remote)
if err != nil {
return nil, err
}
@@ -159,9 +148,6 @@ func (r *Runtime) Create(ctx context.Context, id string, opts runtime.CreateOpts
}
}
}()
if err = r.handleEvents(ctx, s); err != nil {
return nil, err
}
sopts := &shim.CreateTaskRequest{
ID: id,
Bundle: bundle.path,
@@ -332,48 +318,6 @@ func (r *Runtime) loadTasks(ctx context.Context, ns string) ([]*Task, error) {
return o, nil
}
func (r *Runtime) handleEvents(ctx context.Context, s *client.Client) error {
events, err := s.Stream(r.eventsContext, &shim.StreamEventsRequest{})
if err != nil {
return err
}
go r.forward(ctx, events)
return nil
}
// forward forwards events from a shim to the events service and monitors
func (r *Runtime) forward(ctx context.Context, events shim.Shim_StreamClient) {
for {
e, err := events.Recv()
if err != nil {
if !strings.HasSuffix(err.Error(), "transport is closing") {
log.G(r.eventsContext).WithError(err).Error("get event from shim")
}
return
}
r.events <- e
if err := r.emit(ctx, "/runtime/"+getTopic(e), e); err != nil {
return
}
}
}
func getTopic(e *eventsapi.RuntimeEvent) string {
switch e.Type {
case eventsapi.RuntimeEvent_CREATE:
return "task-create"
case eventsapi.RuntimeEvent_START:
return "task-start"
case eventsapi.RuntimeEvent_EXEC_ADDED:
return "task-execadded"
case eventsapi.RuntimeEvent_OOM:
return "task-oom"
case eventsapi.RuntimeEvent_EXIT:
return "task-exit"
}
return ""
}
func (r *Runtime) terminate(ctx context.Context, bundle *bundle, ns, id string) error {
ctx = namespaces.WithNamespace(ctx, ns)
rt, err := r.getRuntime(ctx, ns, id)

32
linux/shim/client.go
View File

@@ -1,4 +1,4 @@
// +build linux
// +build !windows
package shim
@@ -10,7 +10,6 @@ import (
"os"
"os/exec"
"strings"
"syscall"
"time"
"golang.org/x/sys/unix"
@@ -28,7 +27,7 @@ import (
type ClientOpt func(context.Context, Config) (shim.ShimClient, io.Closer, error)
// WithStart executes a new shim process
func WithStart(binary string) ClientOpt {
func WithStart(binary, address string) ClientOpt {
return func(ctx context.Context, config Config) (shim.ShimClient, io.Closer, error) {
socket, err := newSocket(config)
if err != nil {
@@ -41,7 +40,7 @@ func WithStart(binary string) ClientOpt {
}
defer f.Close()
cmd := newCommand(binary, config, f)
cmd := newCommand(binary, address, config, f)
if err := reaper.Default.Start(cmd); err != nil {
return nil, nil, errors.Wrapf(err, "failed to start shim")
}
@@ -56,9 +55,10 @@ func WithStart(binary string) ClientOpt {
}
}
func newCommand(binary string, config Config, socket *os.File) *exec.Cmd {
func newCommand(binary, address string, config Config, socket *os.File) *exec.Cmd {
args := []string{
"--namespace", config.Namespace,
"--address", address,
}
if config.Debug {
args = append(args, "--debug")
@@ -68,11 +68,12 @@ func newCommand(binary string, config Config, socket *os.File) *exec.Cmd {
// make sure the shim can be re-parented to system init
// and is cloned in a new mount namespace because the overlay/filesystems
// will be mounted by the shim
cmd.SysProcAttr = &syscall.SysProcAttr{
Cloneflags: syscall.CLONE_NEWNS,
Setpgid: true,
}
cmd.SysProcAttr = &atter
cmd.ExtraFiles = append(cmd.ExtraFiles, socket)
if config.Debug {
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
}
return cmd
}
@@ -88,12 +89,12 @@ func newSocket(config Config) (*net.UnixListener, error) {
return l.(*net.UnixListener), nil
}
func connect(address string) (*grpc.ClientConn, error) {
func connect(address string, d func(string, time.Duration) (net.Conn, error)) (*grpc.ClientConn, error) {
gopts := []grpc.DialOption{
grpc.WithBlock(),
grpc.WithInsecure(),
grpc.WithTimeout(100 * time.Second),
grpc.WithDialer(dialer),
grpc.WithDialer(d),
grpc.FailOnNonTempDialError(true),
}
conn, err := grpc.Dial(dialAddress(address), gopts...)
@@ -104,6 +105,11 @@ func connect(address string) (*grpc.ClientConn, error) {
}
func dialer(address string, timeout time.Duration) (net.Conn, error) {
address = strings.TrimPrefix(address, "unix://")
return net.DialTimeout("unix", address, timeout)
}
func annonDialer(address string, timeout time.Duration) (net.Conn, error) {
address = strings.TrimPrefix(address, "unix://")
return net.DialTimeout("unix", "\x00"+address, timeout)
}
@@ -114,7 +120,7 @@ func dialAddress(address string) string {
// WithConnect connects to an existing shim
func WithConnect(ctx context.Context, config Config) (shim.ShimClient, io.Closer, error) {
conn, err := connect(config.Address)
conn, err := connect(config.Address, annonDialer)
if err != nil {
return nil, nil, err
}
@@ -123,7 +129,7 @@ func WithConnect(ctx context.Context, config Config) (shim.ShimClient, io.Closer
// WithLocal uses an in process shim
func WithLocal(ctx context.Context, config Config) (shim.ShimClient, io.Closer, error) {
service, err := NewService(config.Path, config.Namespace)
service, err := NewService(config.Path, config.Namespace, "")
if err != nil {
return nil, nil, err
}

10
linux/shim/client_linux.go Normal file
View File

@@ -0,0 +1,10 @@
// +build linux
package shim
import "syscall"
var atter = syscall.SysProcAttr{
Cloneflags: syscall.CLONE_NEWNS,
Setpgid: true,
}

9
linux/shim/client_unix.go Normal file
View File

@@ -0,0 +1,9 @@
// +build !linux,!windows
package shim
import "syscall"
var atter = syscall.SysProcAttr{
Setpgid: true,
}

17
linux/shim/local.go
View File

@@ -6,6 +6,7 @@ import (
"path/filepath"
events "github.com/containerd/containerd/api/services/events/v1"
evt "github.com/containerd/containerd/events"
shimapi "github.com/containerd/containerd/linux/shim/v1"
google_protobuf "github.com/golang/protobuf/ptypes/empty"
"golang.org/x/net/context"
@@ -129,3 +130,19 @@ func (e *streamEvents) SendMsg(m interface{}) error {
func (e *streamEvents) RecvMsg(m interface{}) error {
return nil
}
type poster interface {
Post(ctx context.Context, in *events.PostEventRequest, opts ...grpc.CallOption) (*google_protobuf.Empty, error)
}
type localEventsClient struct {
emitter evt.Poster
}
func (l *localEventsClient) Post(ctx context.Context, r *events.PostEventRequest, opts ...grpc.CallOption) (*google_protobuf.Empty, error) {
ctx = evt.WithTopic(ctx, r.Envelope.Topic)
if err := l.emitter.Post(ctx, r.Envelope); err != nil {
return nil, err
}
return empty, nil
}

63
linux/shim/service.go
View File

@@ -6,6 +6,7 @@ import (
"fmt"
"os"
"sync"
"time"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
@@ -13,8 +14,12 @@ import (
"github.com/containerd/console"
events "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/api/types/task"
evt "github.com/containerd/containerd/events"
shimapi "github.com/containerd/containerd/linux/shim/v1"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/namespaces"
"github.com/containerd/containerd/reaper"
"github.com/containerd/containerd/typeurl"
google_protobuf "github.com/golang/protobuf/ptypes/empty"
"github.com/pkg/errors"
"golang.org/x/net/context"
@@ -29,16 +34,32 @@ var empty = &google_protobuf.Empty{}
const RuncRoot = "/run/containerd/runc"
// NewService returns a new shim service that can be used via GRPC
func NewService(path, namespace string) (*Service, error) {
func NewService(path, namespace, address string) (*Service, error) {
if namespace == "" {
return nil, fmt.Errorf("shim namespace cannot be empty")
}
return &Service{
context := namespaces.WithNamespace(context.Background(), namespace)
var client poster
if address != "" {
conn, err := connect(address, dialer)
if err != nil {
return nil, errors.Wrapf(err, "failed to dial %q", address)
}
client = events.NewEventsClient(conn)
} else {
client = &localEventsClient{
emitter: evt.GetPoster(context),
}
}
s := &Service{
path: path,
processes: make(map[string]process),
events: make(chan *events.RuntimeEvent, 4096),
namespace: namespace,
}, nil
context: context,
}
go s.forward(client)
return s, nil
}
type Service struct {
@@ -52,6 +73,7 @@ type Service struct {
eventsMu sync.Mutex
deferredEvent *events.RuntimeEvent
namespace string
context context.Context
}
func (s *Service) Create(ctx context.Context, r *shimapi.CreateTaskRequest) (*shimapi.CreateTaskResponse, error) {
@@ -367,3 +389,38 @@ func (s *Service) getContainerPids(ctx context.Context, id string) ([]uint32, er
}
return pids, nil
}
func (s *Service) forward(client poster) {
for e := range s.events {
a, err := typeurl.MarshalAny(e)
if err != nil {
log.G(s.context).WithError(err).Error("marshal event")
continue
}
if _, err := client.Post(s.context, &events.PostEventRequest{
Envelope: &events.Envelope{
Timestamp: time.Now(),
Topic: "/runtime/" + getTopic(e),
Event: a,
},
}); err != nil {
log.G(s.context).WithError(err).Error("post event")
}
}
}
func getTopic(e *events.RuntimeEvent) string {
switch e.Type {
case events.RuntimeEvent_CREATE:
return "task-create"
case events.RuntimeEvent_START:
return "task-start"
case events.RuntimeEvent_EXEC_ADDED:
return "task-execadded"
case events.RuntimeEvent_OOM:
return "task-oom"
case events.RuntimeEvent_EXIT:
return "task-exit"
}
return "?"
}