438 lines
12 KiB
Go
438 lines
12 KiB
Go
package server
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"syscall"
|
|
"time"
|
|
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
|
|
"github.com/docker/containerd"
|
|
"github.com/docker/containerd/api/grpc/types"
|
|
"github.com/docker/containerd/osutils"
|
|
"github.com/docker/containerd/runtime"
|
|
"github.com/docker/containerd/supervisor"
|
|
"github.com/golang/protobuf/ptypes"
|
|
"golang.org/x/net/context"
|
|
)
|
|
|
|
type apiServer struct {
|
|
sv *supervisor.Supervisor
|
|
}
|
|
|
|
// NewServer returns grpc server instance
|
|
func NewServer(sv *supervisor.Supervisor) types.APIServer {
|
|
return &apiServer{
|
|
sv: sv,
|
|
}
|
|
}
|
|
|
|
func (s *apiServer) GetServerVersion(ctx context.Context, c *types.GetServerVersionRequest) (*types.GetServerVersionResponse, error) {
|
|
return &types.GetServerVersionResponse{
|
|
Major: containerd.VersionMajor,
|
|
Minor: containerd.VersionMinor,
|
|
Patch: containerd.VersionPatch,
|
|
Revision: containerd.GitCommit,
|
|
}, nil
|
|
}
|
|
|
|
func (s *apiServer) CreateContainer(ctx context.Context, c *types.CreateContainerRequest) (*types.CreateContainerResponse, error) {
|
|
if c.BundlePath == "" {
|
|
return nil, errors.New("empty bundle path")
|
|
}
|
|
e := &supervisor.StartTask{}
|
|
e.ID = c.Id
|
|
e.BundlePath = c.BundlePath
|
|
e.Stdin = c.Stdin
|
|
e.Stdout = c.Stdout
|
|
e.Stderr = c.Stderr
|
|
e.Labels = c.Labels
|
|
e.NoPivotRoot = c.NoPivotRoot
|
|
e.Runtime = c.Runtime
|
|
e.RuntimeArgs = c.RuntimeArgs
|
|
e.StartResponse = make(chan supervisor.StartResponse, 1)
|
|
if c.Checkpoint != "" {
|
|
e.CheckpointDir = c.CheckpointDir
|
|
e.Checkpoint = &runtime.Checkpoint{
|
|
Name: c.Checkpoint,
|
|
}
|
|
}
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
r := <-e.StartResponse
|
|
apiC, err := createAPIContainer(r.Container, false)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &types.CreateContainerResponse{
|
|
Container: apiC,
|
|
}, nil
|
|
}
|
|
|
|
func (s *apiServer) CreateCheckpoint(ctx context.Context, r *types.CreateCheckpointRequest) (*types.CreateCheckpointResponse, error) {
|
|
e := &supervisor.CreateCheckpointTask{}
|
|
e.ID = r.Id
|
|
e.CheckpointDir = r.CheckpointDir
|
|
e.Checkpoint = &runtime.Checkpoint{
|
|
Name: r.Checkpoint.Name,
|
|
Exit: r.Checkpoint.Exit,
|
|
TCP: r.Checkpoint.Tcp,
|
|
UnixSockets: r.Checkpoint.UnixSockets,
|
|
Shell: r.Checkpoint.Shell,
|
|
EmptyNS: r.Checkpoint.EmptyNS,
|
|
}
|
|
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
return &types.CreateCheckpointResponse{}, nil
|
|
}
|
|
|
|
func (s *apiServer) DeleteCheckpoint(ctx context.Context, r *types.DeleteCheckpointRequest) (*types.DeleteCheckpointResponse, error) {
|
|
if r.Name == "" {
|
|
return nil, errors.New("checkpoint name cannot be empty")
|
|
}
|
|
e := &supervisor.DeleteCheckpointTask{}
|
|
e.ID = r.Id
|
|
e.CheckpointDir = r.CheckpointDir
|
|
e.Checkpoint = &runtime.Checkpoint{
|
|
Name: r.Name,
|
|
}
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
return &types.DeleteCheckpointResponse{}, nil
|
|
}
|
|
|
|
func (s *apiServer) ListCheckpoint(ctx context.Context, r *types.ListCheckpointRequest) (*types.ListCheckpointResponse, error) {
|
|
e := &supervisor.GetContainersTask{}
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
var container runtime.Container
|
|
for _, c := range e.Containers {
|
|
if c.ID() == r.Id {
|
|
container = c
|
|
break
|
|
}
|
|
}
|
|
if container == nil {
|
|
return nil, grpc.Errorf(codes.NotFound, "no such containers")
|
|
}
|
|
var out []*types.Checkpoint
|
|
checkpoints, err := container.Checkpoints(r.CheckpointDir)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, c := range checkpoints {
|
|
out = append(out, &types.Checkpoint{
|
|
Name: c.Name,
|
|
Tcp: c.TCP,
|
|
Shell: c.Shell,
|
|
UnixSockets: c.UnixSockets,
|
|
// TODO: figure out timestamp
|
|
//Timestamp: c.Timestamp,
|
|
})
|
|
}
|
|
return &types.ListCheckpointResponse{Checkpoints: out}, nil
|
|
}
|
|
|
|
func (s *apiServer) Signal(ctx context.Context, r *types.SignalRequest) (*types.SignalResponse, error) {
|
|
e := &supervisor.SignalTask{}
|
|
e.ID = r.Id
|
|
e.PID = r.Pid
|
|
e.Signal = syscall.Signal(int(r.Signal))
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
return &types.SignalResponse{}, nil
|
|
}
|
|
|
|
func (s *apiServer) State(ctx context.Context, r *types.StateRequest) (*types.StateResponse, error) {
|
|
|
|
getState := func(c runtime.Container) (interface{}, error) {
|
|
return createAPIContainer(c, true)
|
|
}
|
|
|
|
e := &supervisor.GetContainersTask{}
|
|
e.ID = r.Id
|
|
e.GetState = getState
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
m := s.sv.Machine()
|
|
state := &types.StateResponse{
|
|
Machine: &types.Machine{
|
|
Cpus: uint32(m.Cpus),
|
|
Memory: uint64(m.Memory),
|
|
},
|
|
}
|
|
for idx := range e.Containers {
|
|
state.Containers = append(state.Containers, e.States[idx].(*types.Container))
|
|
}
|
|
return state, nil
|
|
}
|
|
|
|
func createAPIContainer(c runtime.Container, getPids bool) (*types.Container, error) {
|
|
processes, err := c.Processes()
|
|
if err != nil {
|
|
return nil, grpc.Errorf(codes.Internal, "get processes for container: "+err.Error())
|
|
}
|
|
var procs []*types.Process
|
|
for _, p := range processes {
|
|
oldProc := p.Spec()
|
|
stdio := p.Stdio()
|
|
proc := &types.Process{
|
|
Pid: p.ID(),
|
|
SystemPid: uint32(p.SystemPid()),
|
|
Terminal: oldProc.Terminal,
|
|
Args: oldProc.Args,
|
|
Env: oldProc.Env,
|
|
Cwd: oldProc.Cwd,
|
|
Stdin: stdio.Stdin,
|
|
Stdout: stdio.Stdout,
|
|
Stderr: stdio.Stderr,
|
|
}
|
|
proc.User = &types.User{
|
|
Uid: oldProc.User.UID,
|
|
Gid: oldProc.User.GID,
|
|
AdditionalGids: oldProc.User.AdditionalGids,
|
|
}
|
|
proc.Capabilities = oldProc.Capabilities
|
|
proc.ApparmorProfile = oldProc.ApparmorProfile
|
|
proc.SelinuxLabel = oldProc.SelinuxLabel
|
|
proc.NoNewPrivileges = oldProc.NoNewPrivileges
|
|
for _, rl := range oldProc.Rlimits {
|
|
proc.Rlimits = append(proc.Rlimits, &types.Rlimit{
|
|
Type: rl.Type,
|
|
Soft: rl.Soft,
|
|
Hard: rl.Hard,
|
|
})
|
|
}
|
|
procs = append(procs, proc)
|
|
}
|
|
var pids []int
|
|
state := c.State()
|
|
if getPids && (state == runtime.Running || state == runtime.Paused) {
|
|
if pids, err = c.Pids(); err != nil {
|
|
return nil, grpc.Errorf(codes.Internal, "get all pids for container: "+err.Error())
|
|
}
|
|
}
|
|
return &types.Container{
|
|
Id: c.ID(),
|
|
BundlePath: c.Path(),
|
|
Processes: procs,
|
|
Labels: c.Labels(),
|
|
Status: string(state),
|
|
Pids: toUint32(pids),
|
|
Runtime: c.Runtime(),
|
|
}, nil
|
|
}
|
|
|
|
func toUint32(its []int) []uint32 {
|
|
o := []uint32{}
|
|
for _, i := range its {
|
|
o = append(o, uint32(i))
|
|
}
|
|
return o
|
|
}
|
|
|
|
func (s *apiServer) UpdateContainer(ctx context.Context, r *types.UpdateContainerRequest) (*types.UpdateContainerResponse, error) {
|
|
e := &supervisor.UpdateTask{}
|
|
e.ID = r.Id
|
|
e.State = runtime.State(r.Status)
|
|
if r.Resources != nil {
|
|
rs := r.Resources
|
|
e.Resources = &runtime.Resource{}
|
|
if rs.CpuShares != 0 {
|
|
e.Resources.CPUShares = int64(rs.CpuShares)
|
|
}
|
|
if rs.BlkioWeight != 0 {
|
|
e.Resources.BlkioWeight = uint16(rs.BlkioWeight)
|
|
}
|
|
if rs.CpuPeriod != 0 {
|
|
e.Resources.CPUPeriod = int64(rs.CpuPeriod)
|
|
}
|
|
if rs.CpuQuota != 0 {
|
|
e.Resources.CPUQuota = int64(rs.CpuQuota)
|
|
}
|
|
if rs.CpusetCpus != "" {
|
|
e.Resources.CpusetCpus = rs.CpusetCpus
|
|
}
|
|
if rs.CpusetMems != "" {
|
|
e.Resources.CpusetMems = rs.CpusetMems
|
|
}
|
|
if rs.KernelMemoryLimit != 0 {
|
|
e.Resources.KernelMemory = int64(rs.KernelMemoryLimit)
|
|
}
|
|
if rs.KernelTCPMemoryLimit != 0 {
|
|
e.Resources.KernelTCPMemory = int64(rs.KernelTCPMemoryLimit)
|
|
}
|
|
if rs.MemoryLimit != 0 {
|
|
e.Resources.Memory = int64(rs.MemoryLimit)
|
|
}
|
|
if rs.MemoryReservation != 0 {
|
|
e.Resources.MemoryReservation = int64(rs.MemoryReservation)
|
|
}
|
|
if rs.MemorySwap != 0 {
|
|
e.Resources.MemorySwap = int64(rs.MemorySwap)
|
|
}
|
|
}
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
return &types.UpdateContainerResponse{}, nil
|
|
}
|
|
|
|
func (s *apiServer) UpdateProcess(ctx context.Context, r *types.UpdateProcessRequest) (*types.UpdateProcessResponse, error) {
|
|
e := &supervisor.UpdateProcessTask{}
|
|
e.ID = r.Id
|
|
e.PID = r.Pid
|
|
e.Height = int(r.Height)
|
|
e.Width = int(r.Width)
|
|
e.CloseStdin = r.CloseStdin
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
return &types.UpdateProcessResponse{}, nil
|
|
}
|
|
|
|
func (s *apiServer) Events(r *types.EventsRequest, stream types.API_EventsServer) error {
|
|
t := time.Time{}
|
|
if r.Timestamp != nil {
|
|
from, err := ptypes.Timestamp(r.Timestamp)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
t = from
|
|
}
|
|
if r.StoredOnly && t.IsZero() {
|
|
return fmt.Errorf("invalid parameter: StoredOnly cannot be specified without setting a valid Timestamp")
|
|
}
|
|
events := s.sv.Events(t, r.StoredOnly, r.Id)
|
|
defer s.sv.Unsubscribe(events)
|
|
for e := range events {
|
|
tsp, err := ptypes.TimestampProto(e.Timestamp)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if r.Id == "" || e.ID == r.Id {
|
|
if err := stream.Send(&types.Event{
|
|
Id: e.ID,
|
|
Type: e.Type,
|
|
Timestamp: tsp,
|
|
Pid: e.PID,
|
|
Status: uint32(e.Status),
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func convertToPb(st *runtime.Stat) *types.StatsResponse {
|
|
tsp, _ := ptypes.TimestampProto(st.Timestamp)
|
|
pbSt := &types.StatsResponse{
|
|
Timestamp: tsp,
|
|
CgroupStats: &types.CgroupStats{},
|
|
}
|
|
systemUsage, _ := osutils.GetSystemCPUUsage()
|
|
pbSt.CgroupStats.CpuStats = &types.CpuStats{
|
|
CpuUsage: &types.CpuUsage{
|
|
TotalUsage: st.CPU.Usage.Total,
|
|
PercpuUsage: st.CPU.Usage.Percpu,
|
|
UsageInKernelmode: st.CPU.Usage.Kernel,
|
|
UsageInUsermode: st.CPU.Usage.User,
|
|
},
|
|
ThrottlingData: &types.ThrottlingData{
|
|
Periods: st.CPU.Throttling.Periods,
|
|
ThrottledPeriods: st.CPU.Throttling.ThrottledPeriods,
|
|
ThrottledTime: st.CPU.Throttling.ThrottledTime,
|
|
},
|
|
SystemUsage: systemUsage,
|
|
}
|
|
pbSt.CgroupStats.MemoryStats = &types.MemoryStats{
|
|
Cache: st.Memory.Cache,
|
|
Usage: &types.MemoryData{
|
|
Usage: st.Memory.Usage.Usage,
|
|
MaxUsage: st.Memory.Usage.Max,
|
|
Failcnt: st.Memory.Usage.Failcnt,
|
|
Limit: st.Memory.Usage.Limit,
|
|
},
|
|
SwapUsage: &types.MemoryData{
|
|
Usage: st.Memory.Swap.Usage,
|
|
MaxUsage: st.Memory.Swap.Max,
|
|
Failcnt: st.Memory.Swap.Failcnt,
|
|
Limit: st.Memory.Swap.Limit,
|
|
},
|
|
KernelUsage: &types.MemoryData{
|
|
Usage: st.Memory.Kernel.Usage,
|
|
MaxUsage: st.Memory.Kernel.Max,
|
|
Failcnt: st.Memory.Kernel.Failcnt,
|
|
Limit: st.Memory.Kernel.Limit,
|
|
},
|
|
Stats: st.Memory.Raw,
|
|
}
|
|
pbSt.CgroupStats.BlkioStats = &types.BlkioStats{
|
|
IoServiceBytesRecursive: convertBlkioEntryToPb(st.Blkio.IoServiceBytesRecursive),
|
|
IoServicedRecursive: convertBlkioEntryToPb(st.Blkio.IoServicedRecursive),
|
|
IoQueuedRecursive: convertBlkioEntryToPb(st.Blkio.IoQueuedRecursive),
|
|
IoServiceTimeRecursive: convertBlkioEntryToPb(st.Blkio.IoServiceTimeRecursive),
|
|
IoWaitTimeRecursive: convertBlkioEntryToPb(st.Blkio.IoWaitTimeRecursive),
|
|
IoMergedRecursive: convertBlkioEntryToPb(st.Blkio.IoMergedRecursive),
|
|
IoTimeRecursive: convertBlkioEntryToPb(st.Blkio.IoTimeRecursive),
|
|
SectorsRecursive: convertBlkioEntryToPb(st.Blkio.SectorsRecursive),
|
|
}
|
|
pbSt.CgroupStats.HugetlbStats = make(map[string]*types.HugetlbStats)
|
|
for k, st := range st.Hugetlb {
|
|
pbSt.CgroupStats.HugetlbStats[k] = &types.HugetlbStats{
|
|
Usage: st.Usage,
|
|
MaxUsage: st.Max,
|
|
Failcnt: st.Failcnt,
|
|
}
|
|
}
|
|
pbSt.CgroupStats.PidsStats = &types.PidsStats{
|
|
Current: st.Pids.Current,
|
|
Limit: st.Pids.Limit,
|
|
}
|
|
return pbSt
|
|
}
|
|
|
|
func convertBlkioEntryToPb(b []runtime.BlkioEntry) []*types.BlkioStatsEntry {
|
|
var pbEs []*types.BlkioStatsEntry
|
|
for _, e := range b {
|
|
pbEs = append(pbEs, &types.BlkioStatsEntry{
|
|
Major: e.Major,
|
|
Minor: e.Minor,
|
|
Op: e.Op,
|
|
Value: e.Value,
|
|
})
|
|
}
|
|
return pbEs
|
|
}
|
|
|
|
func (s *apiServer) Stats(ctx context.Context, r *types.StatsRequest) (*types.StatsResponse, error) {
|
|
e := &supervisor.StatsTask{}
|
|
e.ID = r.Id
|
|
e.Stat = make(chan *runtime.Stat, 1)
|
|
s.sv.SendTask(e)
|
|
if err := <-e.ErrorCh(); err != nil {
|
|
return nil, err
|
|
}
|
|
stats := <-e.Stat
|
|
t := convertToPb(stats)
|
|
return t, nil
|
|
}
|