Use new container update function

Signed-off-by: Lantao Liu <lantaol@google.com>
This commit is contained in:
Lantao Liu 2017-10-06 01:41:03 +00:00
parent f7eaa29c67
commit e78c85f76b
15 changed files with 64 additions and 117 deletions

View File

@ -60,7 +60,7 @@ func TestUpdateContainerResources(t *testing.T) {
t.Logf("Check memory limit in container OCI spec") t.Logf("Check memory limit in container OCI spec")
container, err := containerdClient.LoadContainer(context.Background(), cn) container, err := containerdClient.LoadContainer(context.Background(), cn)
require.NoError(t, err) require.NoError(t, err)
spec, err := container.Spec() spec, err := container.Spec(context.Background())
require.NoError(t, err) require.NoError(t, err)
checkMemoryLimit(t, spec, 2*1024*1024) checkMemoryLimit(t, spec, 2*1024*1024)
@ -71,9 +71,7 @@ func TestUpdateContainerResources(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
t.Logf("Check memory limit in container OCI spec") t.Logf("Check memory limit in container OCI spec")
container, err = containerdClient.LoadContainer(context.Background(), cn) spec, err = container.Spec(context.Background())
require.NoError(t, err)
spec, err = container.Spec()
require.NoError(t, err) require.NoError(t, err)
checkMemoryLimit(t, spec, 4*1024*1024) checkMemoryLimit(t, spec, 4*1024*1024)
@ -96,9 +94,7 @@ func TestUpdateContainerResources(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
t.Logf("Check memory limit in container OCI spec") t.Logf("Check memory limit in container OCI spec")
container, err = containerdClient.LoadContainer(context.Background(), cn) spec, err = container.Spec(context.Background())
require.NoError(t, err)
spec, err = container.Spec()
require.NoError(t, err) require.NoError(t, err)
checkMemoryLimit(t, spec, 8*1024*1024) checkMemoryLimit(t, spec, 8*1024*1024)

View File

@ -61,7 +61,7 @@ func (c *criContainerdService) attachContainer(ctx context.Context, id string, s
return fmt.Errorf("container is in %s state", criContainerStateToString(state)) return fmt.Errorf("container is in %s state", criContainerStateToString(state))
} }
task, err := cntr.Container.Get().Task(ctx, nil) task, err := cntr.Container.Task(ctx, nil)
if err != nil { if err != nil {
return fmt.Errorf("failed to load task: %v", err) return fmt.Errorf("failed to load task: %v", err)
} }

View File

@ -87,8 +87,8 @@ func (c *criContainerdService) execInContainer(ctx context.Context, id string, o
return nil, fmt.Errorf("container is in %s state", criContainerStateToString(state)) return nil, fmt.Errorf("container is in %s state", criContainerStateToString(state))
} }
container := cntr.Container.Get() container := cntr.Container
spec, err := container.Spec() spec, err := container.Spec(ctx)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to get container spec: %v", err) return nil, fmt.Errorf("failed to get container spec: %v", err)
} }

View File

@ -75,7 +75,7 @@ func (c *criContainerdService) RemoveContainer(ctx context.Context, r *runtime.R
} }
// Delete containerd container. // Delete containerd container.
if err := container.Container.Get().Delete(ctx, containerd.WithSnapshotCleanup); err != nil { if err := container.Container.Delete(ctx, containerd.WithSnapshotCleanup); err != nil {
if !errdefs.IsNotFound(err) { if !errdefs.IsNotFound(err) {
return nil, fmt.Errorf("failed to delete containerd container %q: %v", id, err) return nil, fmt.Errorf("failed to delete containerd container %q: %v", id, err)
} }

View File

@ -59,7 +59,7 @@ func (c *criContainerdService) startContainer(ctx context.Context,
status *containerstore.Status) (retErr error) { status *containerstore.Status) (retErr error) {
id := cntr.ID id := cntr.ID
meta := cntr.Metadata meta := cntr.Metadata
container := cntr.Container.Get() container := cntr.Container
config := meta.Config config := meta.Config
// Return error if container is not in created state. // Return error if container is not in created state.

View File

@ -88,7 +88,7 @@ func (c *criContainerdService) stopContainer(ctx context.Context, container cont
} }
} }
glog.V(2).Infof("Stop container %q with signal %v", id, stopSignal) glog.V(2).Infof("Stop container %q with signal %v", id, stopSignal)
task, err := container.Container.Get().Task(ctx, nil) task, err := container.Container.Task(ctx, nil)
if err != nil { if err != nil {
if !errdefs.IsNotFound(err) { if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to stop container, task not found for container %q: %v", id, err) return fmt.Errorf("failed to stop container, task not found for container %q: %v", id, err)
@ -111,7 +111,7 @@ func (c *criContainerdService) stopContainer(ctx context.Context, container cont
glog.Errorf("Stop container %q timed out: %v", id, err) glog.Errorf("Stop container %q timed out: %v", id, err)
} }
task, err := container.Container.Get().Task(ctx, nil) task, err := container.Container.Task(ctx, nil)
if err != nil { if err != nil {
if !errdefs.IsNotFound(err) { if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to stop container, task not found for container %q: %v", id, err) return fmt.Errorf("failed to stop container, task not found for container %q: %v", id, err)

View File

@ -17,9 +17,11 @@ limitations under the License.
package server package server
import ( import (
gocontext "context"
"fmt" "fmt"
"github.com/containerd/containerd" "github.com/containerd/containerd"
"github.com/containerd/containerd/containers"
"github.com/containerd/containerd/errdefs" "github.com/containerd/containerd/errdefs"
"github.com/containerd/typeurl" "github.com/containerd/typeurl"
"github.com/golang/glog" "github.com/golang/glog"
@ -63,7 +65,7 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
// spec makes sure that the resource limits are correct when start; // spec makes sure that the resource limits are correct when start;
// if the container is already started, updating spec is still required, // if the container is already started, updating spec is still required,
// the spec will become our source of truth for resource limits. // the spec will become our source of truth for resource limits.
oldSpec, err := cntr.Container.Get().Spec() oldSpec, err := cntr.Container.Spec(ctx)
if err != nil { if err != nil {
return fmt.Errorf("failed to get container spec: %v", err) return fmt.Errorf("failed to get container spec: %v", err)
} }
@ -72,42 +74,15 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
return fmt.Errorf("failed to update resource in spec: %v", err) return fmt.Errorf("failed to update resource in spec: %v", err)
} }
info := cntr.Container.Get().Info() if err := updateContainerSpec(ctx, cntr.Container, newSpec); err != nil {
any, err := typeurl.MarshalAny(newSpec) return err
if err != nil {
return fmt.Errorf("failed to marshal spec %+v: %v", newSpec, err)
}
info.Spec = any
// TODO(random-liu): Add helper function in containerd to do the update.
if _, err := c.client.ContainerService().Update(ctx, info, "spec"); err != nil {
return fmt.Errorf("failed to update container spec: %v", err)
} }
defer func() { defer func() {
if retErr != nil { if retErr != nil {
// Reset spec on error. // Reset spec on error.
any, err := typeurl.MarshalAny(oldSpec) if err := updateContainerSpec(ctx, cntr.Container, oldSpec); err != nil {
if err != nil { glog.Errorf("Failed to update spec %+v for container %q: %v", oldSpec, id, err)
glog.Errorf("Failed to marshal spec %+v for container %q: %v", oldSpec, id, err)
return
} }
info.Spec = any
if _, err := c.client.ContainerService().Update(ctx, info, "spec"); err != nil {
glog.Errorf("Failed to recover spec %+v for container %q: %v", oldSpec, id, err)
}
}
}()
container, err := c.client.LoadContainer(ctx, id)
if err != nil {
return fmt.Errorf("failed to load container: %v", err)
}
defer func() {
if retErr == nil {
// Update container client if no error is returned.
// NOTE(random-liu): By updating container client, we'll be able
// to get latest OCI spec from it, which includes the up-to-date
// container resource limits. This will be useful after the debug
// api is introduced.
cntr.Container.Set(container)
} }
}() }()
@ -117,7 +92,7 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
return nil return nil
} }
task, err := container.Task(ctx, nil) task, err := cntr.Container.Task(ctx, nil)
if err != nil { if err != nil {
if errdefs.IsNotFound(err) { if errdefs.IsNotFound(err) {
// Task exited already. // Task exited already.
@ -136,6 +111,21 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
return nil return nil
} }
// updateContainerSpec updates container spec.
func updateContainerSpec(ctx context.Context, cntr containerd.Container, spec *runtimespec.Spec) error {
any, err := typeurl.MarshalAny(spec)
if err != nil {
return fmt.Errorf("failed to marshal spec %+v: %v", spec, err)
}
if err := cntr.Update(ctx, func(ctx gocontext.Context, client *containerd.Client, c *containers.Container) error {
c.Spec = any
return nil
}); err != nil {
return fmt.Errorf("failed to update container spec: %v", err)
}
return nil
}
// updateOCILinuxResource updates container resource limit. // updateOCILinuxResource updates container resource limit.
func updateOCILinuxResource(spec *runtimespec.Spec, new *runtime.LinuxContainerResources) (*runtimespec.Spec, error) { func updateOCILinuxResource(spec *runtimespec.Spec, new *runtime.LinuxContainerResources) (*runtimespec.Spec, error) {
// Copy to make sure old spec is not changed. // Copy to make sure old spec is not changed.

View File

@ -104,7 +104,7 @@ func (em *eventMonitor) handleEvent(evt *events.Envelope) {
return return
} }
// Attach container IO so that `Delete` could cleanup the stream properly. // Attach container IO so that `Delete` could cleanup the stream properly.
task, err := cntr.Container.Get().Task(context.Background(), task, err := cntr.Container.Task(context.Background(),
func(*containerd.FIFOSet) (containerd.IO, error) { func(*containerd.FIFOSet) (containerd.IO, error) {
return cntr.IO, nil return cntr.IO, nil
}, },

View File

@ -127,7 +127,11 @@ func loadContainer(ctx context.Context, cntr containerd.Container, containerDir
id := cntr.ID() id := cntr.ID()
var container containerstore.Container var container containerstore.Container
// Load container metadata. // Load container metadata.
ext, ok := cntr.Extensions()[containerMetadataExtension] exts, err := cntr.Extensions(ctx)
if err != nil {
return container, fmt.Errorf("failed to get container extensions: %v", err)
}
ext, ok := exts[containerMetadataExtension]
if !ok { if !ok {
return container, fmt.Errorf("metadata extension %q not found", containerMetadataExtension) return container, fmt.Errorf("metadata extension %q not found", containerMetadataExtension)
} }
@ -278,7 +282,11 @@ func unknownContainerStatus() containerstore.Status {
func loadSandbox(ctx context.Context, cntr containerd.Container) (sandboxstore.Sandbox, error) { func loadSandbox(ctx context.Context, cntr containerd.Container) (sandboxstore.Sandbox, error) {
var sandbox sandboxstore.Sandbox var sandbox sandboxstore.Sandbox
// Load sandbox metadata. // Load sandbox metadata.
ext, ok := cntr.Extensions()[sandboxMetadataExtension] exts, err := cntr.Extensions(ctx)
if err != nil {
return sandbox, fmt.Errorf("failed to get sandbox container extensions: %v", err)
}
ext, ok := exts[sandboxMetadataExtension]
if !ok { if !ok {
return sandbox, fmt.Errorf("metadata extension %q not found", sandboxMetadataExtension) return sandbox, fmt.Errorf("metadata extension %q not found", sandboxMetadataExtension)
} }

View File

@ -22,6 +22,7 @@ import (
tasks "github.com/containerd/containerd/api/services/tasks/v1" tasks "github.com/containerd/containerd/api/services/tasks/v1"
"github.com/containerd/containerd/api/types/task" "github.com/containerd/containerd/api/types/task"
"github.com/containerd/containerd/errdefs"
"golang.org/x/net/context" "golang.org/x/net/context"
"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime" "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
@ -55,7 +56,15 @@ func (c *criContainerdService) ListPodSandbox(ctx context.Context, r *runtime.Li
state = runtime.PodSandboxState_SANDBOX_READY state = runtime.PodSandboxState_SANDBOX_READY
} }
createdAt := sandboxInStore.Container.Info().CreatedAt info, err := sandboxInStore.Container.Info(ctx)
if err != nil {
// It's possible that container gets deleted during list.
if errdefs.IsNotFound(err) {
continue
}
return nil, fmt.Errorf("failed to get sandbox container %q info: %v", sandboxInStore.ID, err)
}
createdAt := info.CreatedAt
sandboxes = append(sandboxes, toCRISandbox(sandboxInStore.Metadata, state, createdAt)) sandboxes = append(sandboxes, toCRISandbox(sandboxInStore.Metadata, state, createdAt))
} }

View File

@ -34,15 +34,12 @@ import (
func (c *criContainerdService) PodSandboxStatus(ctx context.Context, r *runtime.PodSandboxStatusRequest) (*runtime.PodSandboxStatusResponse, error) { func (c *criContainerdService) PodSandboxStatus(ctx context.Context, r *runtime.PodSandboxStatusRequest) (*runtime.PodSandboxStatusResponse, error) {
sandbox, err := c.sandboxStore.Get(r.GetPodSandboxId()) sandbox, err := c.sandboxStore.Get(r.GetPodSandboxId())
if err != nil { if err != nil {
return nil, fmt.Errorf("an error occurred when try to find sandbox %q: %v", return nil, fmt.Errorf("an error occurred when try to find sandbox: %v", err)
r.GetPodSandboxId(), err)
} }
// Use the full sandbox id.
id := sandbox.ID
task, err := sandbox.Container.Task(ctx, nil) task, err := sandbox.Container.Task(ctx, nil)
if err != nil && !errdefs.IsNotFound(err) { if err != nil && !errdefs.IsNotFound(err) {
return nil, fmt.Errorf("failed to get sandbox container info for %q: %v", id, err) return nil, fmt.Errorf("failed to get sandbox container task: %v", err)
} }
// Set sandbox state to NOTREADY by default. // Set sandbox state to NOTREADY by default.
@ -51,7 +48,7 @@ func (c *criContainerdService) PodSandboxStatus(ctx context.Context, r *runtime.
if task != nil { if task != nil {
taskStatus, err := task.Status(ctx) taskStatus, err := task.Status(ctx)
if err != nil { if err != nil {
return nil, fmt.Errorf("failed to get task status for sandbox container %q: %v", id, err) return nil, fmt.Errorf("failed to get task status: %v", err)
} }
if taskStatus.Status == containerd.Running { if taskStatus.Status == containerd.Running {
@ -64,7 +61,11 @@ func (c *criContainerdService) PodSandboxStatus(ctx context.Context, r *runtime.
return nil, fmt.Errorf("failed to get sandbox ip: %v", err) return nil, fmt.Errorf("failed to get sandbox ip: %v", err)
} }
createdAt := sandbox.Container.Info().CreatedAt info, err := sandbox.Container.Info(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get sandbox container info: %v", err)
}
createdAt := info.CreatedAt
status := toCRISandboxStatus(sandbox.Metadata, state, createdAt, ip) status := toCRISandboxStatus(sandbox.Metadata, state, createdAt, ip)
return &runtime.PodSandboxStatusResponse{Status: status}, nil return &runtime.PodSandboxStatusResponse{Status: status}, nil
} }

View File

@ -1,45 +0,0 @@
/*
Copyright 2017 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package container
import (
"sync"
"github.com/containerd/containerd"
)
// Client holds the containerd container client.
// containerd.Container is a pointer underlying. New assignment won't affect
// the previous pointer, so simply lock around is enough.
type Client struct {
lock sync.RWMutex
container containerd.Container
}
// Get containerd container client.
func (c *Client) Get() containerd.Container {
c.lock.RLock()
defer c.lock.RUnlock()
return c.container
}
// Set containerd container client.
func (c *Client) Set(container containerd.Container) {
c.lock.Lock()
defer c.lock.Unlock()
c.container = container
}

View File

@ -33,7 +33,7 @@ type Container struct {
// Status stores the status of the container. // Status stores the status of the container.
Status StatusStorage Status StatusStorage
// Container is the containerd container client. // Container is the containerd container client.
Container *Client Container containerd.Container
// Container IO // Container IO
IO *cio.ContainerIO IO *cio.ContainerIO
// TODO(random-liu): Add stop channel to get rid of stop poll waiting. // TODO(random-liu): Add stop channel to get rid of stop poll waiting.
@ -45,7 +45,7 @@ type Opts func(*Container) error
// WithContainer adds the containerd Container to the internal data store. // WithContainer adds the containerd Container to the internal data store.
func WithContainer(cntr containerd.Container) Opts { func WithContainer(cntr containerd.Container) Opts {
return func(c *Container) error { return func(c *Container) error {
c.Container = &Client{container: cntr} c.Container = cntr
return nil return nil
} }
} }
@ -95,10 +95,6 @@ type Store struct {
// TODO(random-liu): Add trunc index. // TODO(random-liu): Add trunc index.
} }
// LoadStore loads containers from runtime.
// TODO(random-liu): Implement LoadStore.
func LoadStore() *Store { return nil }
// NewStore creates a container store. // NewStore creates a container store.
func NewStore() *Store { func NewStore() *Store {
return &Store{containers: make(map[string]Container)} return &Store{containers: make(map[string]Container)}

View File

@ -51,10 +51,6 @@ type Store struct {
// TODO(random-liu): Add trunc index. // TODO(random-liu): Add trunc index.
} }
// LoadStore loads images from runtime.
// TODO(random-liu): Implement LoadStore.
func LoadStore() *Store { return nil }
// NewStore creates an image store. // NewStore creates an image store.
func NewStore() *Store { func NewStore() *Store {
return &Store{images: make(map[string]Image)} return &Store{images: make(map[string]Image)}

View File

@ -42,10 +42,6 @@ type Store struct {
// TODO(random-liu): Add trunc index. // TODO(random-liu): Add trunc index.
} }
// LoadStore loads sandboxes from runtime.
// TODO(random-liu): Implement LoadStore.
func LoadStore() *Store { return nil }
// NewStore creates a sandbox store. // NewStore creates a sandbox store.
func NewStore() *Store { func NewStore() *Store {
return &Store{sandboxes: make(map[string]Sandbox)} return &Store{sandboxes: make(map[string]Sandbox)}