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

Use github.com/pkg/errors

Browse Source 
Signed-off-by: Lantao Liu <lantaol@google.com>
This commit is contained in:
Lantao Liu 2018-03-17 02:15:06 +00:00
parent 916e99d0ad
commit e1fe1abff0
40 changed files with 345 additions and 349 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cli/cli.go
View File

@ -51,7 +51,7 @@ var loadCommand = cli.Command{
)
cl, err := client.NewCRIContainerdClient(address, timeout)
if err != nil {
return fmt.Errorf("failed to create grpc client: %v", err)
return errors.Wrap(err, "failed to create grpc client")
}
if timeout > 0 {
ctx, cancel = gocontext.WithTimeout(gocontext.Background(), timeout)

9
integration/container_stats_test.go
View File

@ -21,6 +21,7 @@ import (
"testing"
"time"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -168,7 +169,7 @@ func TestContainerListStatsWithIdFilter(t *testing.T) {
return false, err
}
if len(stats) != 1 {
return false, fmt.Errorf("unexpected stats length")
return false, errors.New("unexpected stats length")
}
if stats[0].GetWritableLayer().GetUsedBytes().GetValue() != 0 &&
stats[0].GetWritableLayer().GetInodesUsed().GetValue() != 0 {
@ -227,7 +228,7 @@ func TestContainerListStatsWithSandboxIdFilter(t *testing.T) {
return false, err
}
if len(stats) != 3 {
return false, fmt.Errorf("unexpected stats length")
return false, errors.New("unexpected stats length")
}
if stats[0].GetWritableLayer().GetUsedBytes().GetValue() != 0 &&
stats[0].GetWritableLayer().GetInodesUsed().GetValue() != 0 {
@ -283,7 +284,7 @@ func TestContainerListStatsWithIdSandboxIdFilter(t *testing.T) {
return false, err
}
if len(stats) != 1 {
return false, fmt.Errorf("unexpected stats length")
return false, errors.New("unexpected stats length")
}
if stats[0].GetWritableLayer().GetUsedBytes().GetValue() != 0 &&
stats[0].GetWritableLayer().GetInodesUsed().GetValue() != 0 {
@ -306,7 +307,7 @@ func TestContainerListStatsWithIdSandboxIdFilter(t *testing.T) {
return false, err
}
if len(stats) != 1 {
return false, fmt.Errorf("unexpected stats length")
return false, errors.New("unexpected stats length")
}
if stats[0].GetWritableLayer().GetUsedBytes().GetValue() != 0 &&
stats[0].GetWritableLayer().GetInodesUsed().GetValue() != 0 {

4
integration/imagefs_info_test.go
View File

@ -17,11 +17,11 @@ limitations under the License.
package integration
import (
"fmt"
"os"
"testing"
"time"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -57,7 +57,7 @@ func TestImageFSInfo(t *testing.T) {
return false, nil
}
if len(stats) >= 2 {
return false, fmt.Errorf("unexpected stats length: %d", len(stats))
return false, errors.Errorf("unexpected stats length: %d", len(stats))
}
info = stats[0]
if info.GetTimestamp() != 0 &&

18
integration/test_utils.go
View File

@ -17,7 +17,6 @@ limitations under the License.
package integration
import (
"errors"
"flag"
"fmt"
"os/exec"
@ -26,6 +25,7 @@ import (
"time"
"github.com/containerd/containerd"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"k8s.io/kubernetes/pkg/kubelet/apis/cri"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -66,30 +66,30 @@ func ConnectDaemons() error {
var err error
runtimeService, err = remote.NewRemoteRuntimeService(*criContainerdEndpoint, timeout)
if err != nil {
return fmt.Errorf("failed to create runtime service: %v", err)
return errors.Wrap(err, "failed to create runtime service")
}
imageService, err = remote.NewRemoteImageService(*criContainerdEndpoint, timeout)
if err != nil {
return fmt.Errorf("failed to create image service: %v", err)
return errors.Wrap(err, "failed to create image service")
}
// Since CRI grpc client doesn't have `WithBlock` specified, we
// need to check whether it is actually connected.
// TODO(random-liu): Extend cri remote client to accept extra grpc options.
_, err = runtimeService.ListContainers(&runtime.ContainerFilter{})
if err != nil {
return fmt.Errorf("failed to list containers: %v", err)
return errors.Wrap(err, "failed to list containers")
}
_, err = imageService.ListImages(&runtime.ImageFilter{})
if err != nil {
return fmt.Errorf("failed to list images: %v", err)
return errors.Wrap(err, "failed to list images")
}
containerdClient, err = containerd.New(containerdEndpoint, containerd.WithDefaultNamespace(k8sNamespace))
if err != nil {
return fmt.Errorf("failed to connect containerd: %v", err)
return errors.Wrap(err, "failed to connect containerd")
}
criContainerdClient, err = client.NewCRIContainerdClient(*criContainerdEndpoint, timeout)
if err != nil {
return fmt.Errorf("failed to connect cri plugin: %v", err)
return errors.Wrap(err, "failed to connect cri plugin")
}
return nil
}
@ -227,7 +227,7 @@ func Randomize(str string) string {
func KillProcess(name string) error {
output, err := exec.Command("pkill", "-x", fmt.Sprintf("^%s$", name)).CombinedOutput()
if err != nil {
return fmt.Errorf("failed to kill %q - error: %v, output: %q", name, err, output)
return errors.Errorf("failed to kill %q - error: %v, output: %q", name, err, output)
}
return nil
}
@ -238,7 +238,7 @@ func PidOf(name string) (int, error) {
output := strings.TrimSpace(string(b))
if err != nil {
if len(output) != 0 {
return 0, fmt.Errorf("failed to run pidof %q - error: %v, output: %q", name, err, output)
return 0, errors.Errorf("failed to run pidof %q - error: %v, output: %q", name, err, output)
}
return 0, nil
}

6
pkg/client/client.go
View File

@ -17,9 +17,9 @@ limitations under the License.
package client
import (
"fmt"
"time"
"github.com/pkg/errors"
"google.golang.org/grpc"
"k8s.io/kubernetes/pkg/kubelet/util"
@ -31,7 +31,7 @@ import (
func NewCRIContainerdClient(endpoint string, timeout time.Duration) (api.CRIContainerdServiceClient, error) {
addr, dialer, err := util.GetAddressAndDialer(endpoint)
if err != nil {
return nil, fmt.Errorf("failed to get dialer: %v", err)
return nil, errors.Wrap(err, "failed to get dialer")
}
conn, err := grpc.Dial(addr,
grpc.WithBlock(),
@ -41,7 +41,7 @@ func NewCRIContainerdClient(endpoint string, timeout time.Duration) (api.CRICont
grpc.WithDialer(dialer),
)
if err != nil {
return nil, fmt.Errorf("failed to dial: %v", err)
return nil, errors.Wrap(err, "failed to dial")
}
return api.NewCRIContainerdServiceClient(conn), nil
}

12
pkg/containerd/resolver/resolver.go
View File

@ -271,7 +271,7 @@ func (r *containerdResolver) base(refspec reference.Spec) (*dockerBase, error) {
if urls, ok := r.registry[host]; ok {
urls, err := r.getV2Urls(urls, prefix)
if err != nil {
return nil, fmt.Errorf("failed to fetch v2 urls: %v", err)
return nil, errors.Wrap(err, "failed to fetch v2 urls")
}
base = append(base, urls...)
} else if host == "docker.io" {
@ -434,7 +434,7 @@ func (r *dockerBase) setTokenAuth(ctx context.Context, params map[string]string)
realmURL, err := url.Parse(realm)
if err != nil {
return fmt.Errorf("invalid token auth challenge realm: %s", err)
return errors.Wrap(err, "invalid token auth challenge realm")
}
to := tokenOptions{
@ -444,7 +444,7 @@ func (r *dockerBase) setTokenAuth(ctx context.Context, params map[string]string)
to.scopes = getTokenScopes(ctx, params)
if len(to.scopes) == 0 {
return errors.Errorf("no scope specified for token auth challenge")
return errors.New("no scope specified for token auth challenge")
}
if r.secret != "" {
// Credential information is provided, use oauth POST endpoint
@ -517,7 +517,7 @@ func (r *dockerBase) fetchTokenWithOAuth(ctx context.Context, to tokenOptions) (
var tr postTokenResponse
if err = decoder.Decode(&tr); err != nil {
return "", fmt.Errorf("unable to decode token response: %s", err)
return "", errors.Wrap(err, "unable to decode token response")
}
return tr.AccessToken, nil
@ -569,7 +569,7 @@ func (r *dockerBase) getToken(ctx context.Context, to tokenOptions) (string, err
var tr getTokenResponse
if err = decoder.Decode(&tr); err != nil {
return "", fmt.Errorf("unable to decode token response: %s", err)
return "", errors.Wrap(err, "unable to decode token response")
}
// `access_token` is equivalent to `token` and if both are specified
@ -591,7 +591,7 @@ func (r *containerdResolver) getV2Urls(urls []string, imagePath string) ([]url.U
for _, u := range urls {
v2Url, err := url.Parse(u)
if err != nil {
return nil, fmt.Errorf("Failed to parse url during getv2 urls: %+v, err:%s", u, err)
return nil, errors.Wrapf(err, "failed to parse url during getv2 urls: %+v", u)
}
v2Url.Path = path.Join("/v2", imagePath)
v2Urls = append(v2Urls, *v2Url)

9
pkg/registrar/registrar.go
View File

@ -17,8 +17,9 @@ limitations under the License.
package registrar
import (
"fmt"
"sync"
"github.com/pkg/errors"
)
// Registrar stores one-to-one name<->key mappings.
@ -49,19 +50,19 @@ func (r *Registrar) Reserve(name, key string) error {
defer r.lock.Unlock()
if name == "" || key == "" {
return fmt.Errorf("invalid name %q or key %q", name, key)
return errors.Errorf("invalid name %q or key %q", name, key)
}
if k, exists := r.nameToKey[name]; exists {
if k != key {
return fmt.Errorf("name %q is reserved for %q", name, k)
return errors.Errorf("name %q is reserved for %q", name, k)
}
return nil
}
if n, exists := r.keyToName[key]; exists {
if n != name {
return fmt.Errorf("key %q is reserved for %q", key, n)
return errors.Errorf("key %q is reserved for %q", key, n)
}
return nil
}

12
pkg/server/container_attach.go
View File

@ -17,10 +17,10 @@ limitations under the License.
package server
import (
"fmt"
"io"
"github.com/containerd/containerd"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
"k8s.io/client-go/tools/remotecommand"
@ -33,11 +33,11 @@ import (
func (c *criContainerdService) Attach(ctx context.Context, r *runtime.AttachRequest) (*runtime.AttachResponse, error) {
cntr, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
return nil, fmt.Errorf("failed to find container in store: %v", err)
return nil, errors.Wrap(err, "failed to find container in store")
}
state := cntr.Status.Get().State()
if state != runtime.ContainerState_CONTAINER_RUNNING {
return nil, fmt.Errorf("container is in %s state", criContainerStateToString(state))
return nil, errors.Errorf("container is in %s state", criContainerStateToString(state))
}
return c.streamServer.GetAttach(r)
}
@ -47,18 +47,18 @@ func (c *criContainerdService) attachContainer(ctx context.Context, id string, s
// Get container from our container store.
cntr, err := c.containerStore.Get(id)
if err != nil {
return fmt.Errorf("failed to find container %q in store: %v", id, err)
return errors.Wrapf(err, "failed to find container %q in store", id)
}
id = cntr.ID
state := cntr.Status.Get().State()
if state != runtime.ContainerState_CONTAINER_RUNNING {
return fmt.Errorf("container is in %s state", criContainerStateToString(state))
return errors.Errorf("container is in %s state", criContainerStateToString(state))
}
task, err := cntr.Container.Task(ctx, nil)
if err != nil {
return fmt.Errorf("failed to load task: %v", err)
return errors.Wrap(err, "failed to load task")
}
handleResizing(resize, func(size remotecommand.TerminalSize) {
if err := task.Resize(ctx, uint32(size.Width), uint32(size.Height)); err != nil {

69
pkg/server/container_create.go
View File

@ -17,7 +17,6 @@ limitations under the License.
package server
import (
"fmt"
"os"
"path/filepath"
"strings"
@ -38,6 +37,7 @@ import (
"github.com/opencontainers/runtime-tools/generate"
"github.com/opencontainers/runtime-tools/validate"
"github.com/opencontainers/selinux/go-selinux/label"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/syndtr/gocapability/capability"
"golang.org/x/net/context"
@ -78,12 +78,12 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
sandboxConfig := r.GetSandboxConfig()
sandbox, err := c.sandboxStore.Get(r.GetPodSandboxId())
if err != nil {
return nil, fmt.Errorf("failed to find sandbox id %q: %v", r.GetPodSandboxId(), err)
return nil, errors.Wrapf(err, "failed to find sandbox id %q", r.GetPodSandboxId())
}
sandboxID := sandbox.ID
s, err := sandbox.Container.Task(ctx, nil)
if err != nil {
return nil, fmt.Errorf("failed to get sandbox container task: %v", err)
return nil, errors.Wrap(err, "failed to get sandbox container task")
}
sandboxPid := s.Pid()
@ -94,7 +94,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
name := makeContainerName(config.GetMetadata(), sandboxConfig.GetMetadata())
logrus.Debugf("Generated id %q for container %q", id, name)
if err = c.containerNameIndex.Reserve(name, id); err != nil {
return nil, fmt.Errorf("failed to reserve container name %q: %v", name, err)
return nil, errors.Wrapf(err, "failed to reserve container name %q", name)
}
defer func() {
// Release the name if the function returns with an error.
@ -116,17 +116,17 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
imageRef := config.GetImage().GetImage()
image, err := c.localResolve(ctx, imageRef)
if err != nil {
return nil, fmt.Errorf("failed to resolve image %q: %v", imageRef, err)
return nil, errors.Wrapf(err, "failed to resolve image %q", imageRef)
}
if image == nil {
return nil, fmt.Errorf("image %q not found", imageRef)
return nil, errors.Errorf("image %q not found", imageRef)
}
// Create container root directory.
containerRootDir := getContainerRootDir(c.config.RootDir, id)
if err = c.os.MkdirAll(containerRootDir, 0755); err != nil {
return nil, fmt.Errorf("failed to create container root directory %q: %v",
containerRootDir, err)
return nil, errors.Wrapf(err, "failed to create container root directory %q",
containerRootDir)
}
defer func() {
if retErr != nil {
@ -146,7 +146,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
spec, err := c.generateContainerSpec(id, sandboxID, sandboxPid, config, sandboxConfig, &image.ImageSpec.Config, append(mounts, volumeMounts...))
if err != nil {
return nil, fmt.Errorf("failed to generate container %q spec: %v", id, err)
return nil, errors.Wrapf(err, "failed to generate container %q spec", id)
}
logrus.Debugf("Container %q spec: %#+v", id, spew.NewFormatter(spec))
@ -179,7 +179,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
containerIO, err := cio.NewContainerIO(id,
cio.WithNewFIFOs(containerRootDir, config.GetTty(), config.GetStdin()))
if err != nil {
return nil, fmt.Errorf("failed to create container io: %v", err)
return nil, errors.Wrap(err, "failed to create container io")
}
defer func() {
if retErr != nil {
@ -206,7 +206,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
securityContext.GetPrivileged(),
c.apparmorEnabled)
if err != nil {
return nil, fmt.Errorf("failed to generate apparmor spec opts: %v", err)
return nil, errors.Wrap(err, "failed to generate apparmor spec opts")
}
if apparmorSpecOpts != nil {
specOpts = append(specOpts, apparmorSpecOpts)
@ -217,7 +217,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
securityContext.GetPrivileged(),
c.seccompEnabled)
if err != nil {
return nil, fmt.Errorf("failed to generate seccomp spec opts: %v", err)
return nil, errors.Wrap(err, "failed to generate seccomp spec opts")
}
if seccompSpecOpts != nil {
specOpts = append(specOpts, seccompSpecOpts)
@ -236,7 +236,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
containerd.WithContainerExtension(containerMetadataExtension, &meta))
var cntr containerd.Container
if cntr, err = c.client.NewContainer(ctx, id, opts...); err != nil {
return nil, fmt.Errorf("failed to create containerd container: %v", err)
return nil, errors.Wrap(err, "failed to create containerd container")
}
defer func() {
if retErr != nil {
@ -255,8 +255,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
containerstore.WithContainerIO(containerIO),
)
if err != nil {
return nil, fmt.Errorf("failed to create internal container object for %q: %v",
id, err)
return nil, errors.Wrapf(err, "failed to create internal container object for %q", id)
}
defer func() {
if retErr != nil {
@ -269,7 +268,7 @@ func (c *criContainerdService) CreateContainer(ctx context.Context, r *runtime.C
// Add container into container store.
if err := c.containerStore.Add(container); err != nil {
return nil, fmt.Errorf("failed to add container %q into store: %v", id, err)
return nil, errors.Wrapf(err, "failed to add container %q into store", id)
}
return &runtime.CreateContainerResponse{ContainerId: id}, nil
@ -316,30 +315,30 @@ func (c *criContainerdService) generateContainerSpec(id string, sandboxID string
selinuxOpt := securityContext.GetSelinuxOptions()
processLabel, mountLabel, err := initSelinuxOpts(selinuxOpt)
if err != nil {
return nil, fmt.Errorf("failed to init selinux options %+v: %v", securityContext.GetSelinuxOptions(), err)
return nil, errors.Wrapf(err, "failed to init selinux options %+v", securityContext.GetSelinuxOptions())
}
// Add extra mounts first so that CRI specified mounts can override.
mounts := append(extraMounts, config.GetMounts()...)
if err := c.addOCIBindMounts(&g, mounts, mountLabel); err != nil {
return nil, fmt.Errorf("failed to set OCI bind mounts %+v: %v", mounts, err)
return nil, errors.Wrapf(err, "failed to set OCI bind mounts %+v", mounts)
}
if securityContext.GetPrivileged() {
if !sandboxConfig.GetLinux().GetSecurityContext().GetPrivileged() {
return nil, fmt.Errorf("no privileged container allowed in sandbox")
return nil, errors.New("no privileged container allowed in sandbox")
}
if err := setOCIPrivileged(&g, config); err != nil {
return nil, err
}
} else { // not privileged
if err := c.addOCIDevices(&g, config.GetDevices()); err != nil {
return nil, fmt.Errorf("failed to set devices mapping %+v: %v", config.GetDevices(), err)
return nil, errors.Wrapf(err, "failed to set devices mapping %+v", config.GetDevices())
}
if err := setOCICapabilities(&g, securityContext.GetCapabilities()); err != nil {
return nil, fmt.Errorf("failed to set capabilities %+v: %v",
securityContext.GetCapabilities(), err)
return nil, errors.Wrapf(err, "failed to set capabilities %+v",
securityContext.GetCapabilities())
}
}
@ -457,7 +456,7 @@ func setOCIProcessArgs(g *generate.Generator, config *runtime.ContainerConfig, i
}
}
if len(command) == 0 && len(args) == 0 {
return fmt.Errorf("no command specified")
return errors.New("no command specified")
}
g.SetProcessArgs(append(command, args...))
return nil
@ -469,7 +468,7 @@ func addImageEnvs(g *generate.Generator, imageEnvs []string) error {
for _, e := range imageEnvs {
kv := strings.SplitN(e, "=", 2)
if len(kv) != 2 {
return fmt.Errorf("invalid environment variable %q", e)
return errors.Errorf("invalid environment variable %q", e)
}
g.AddProcessEnv(kv[0], kv[1])
}
@ -481,7 +480,7 @@ func setOCIPrivileged(g *generate.Generator, config *runtime.ContainerConfig) er
g.SetupPrivileged(true)
setOCIBindMountsPrivileged(g)
if err := setOCIDevicesPrivileged(g); err != nil {
return fmt.Errorf("failed to set devices mapping %+v: %v", config.GetDevices(), err)
return errors.Wrapf(err, "failed to set devices mapping %+v", config.GetDevices())
}
return nil
}
@ -570,17 +569,17 @@ func (c *criContainerdService) addOCIBindMounts(g *generate.Generator, mounts []
// TODO(random-liu): Add CRI validation test for this case.
if _, err := c.os.Stat(src); err != nil {
if !os.IsNotExist(err) {
return fmt.Errorf("failed to stat %q: %v", src, err)
return errors.Wrapf(err, "failed to stat %q", src)
}
if err := c.os.MkdirAll(src, 0755); err != nil {
return fmt.Errorf("failed to mkdir %q: %v", src, err)
return errors.Wrapf(err, "failed to mkdir %q", src)
}
}
// TODO(random-liu): Add cri-containerd integration test or cri validation test
// for this.
src, err := c.os.ResolveSymbolicLink(src)
if err != nil {
return fmt.Errorf("failed to resolve symlink %q: %v", src, err)
return errors.Wrapf(err, "failed to resolve symlink %q", src)
}
options := []string{"rbind"}
@ -619,7 +618,7 @@ func (c *criContainerdService) addOCIBindMounts(g *generate.Generator, mounts []
if mount.GetSelinuxRelabel() {
if err := label.Relabel(src, mountLabel, true); err != nil && err != unix.ENOTSUP {
return fmt.Errorf("relabel %q with %q failed: %v", src, mountLabel, err)
return errors.Wrapf(err, "relabel %q with %q failed", src, mountLabel)
}
}
g.AddBindMount(src, dst, options)
@ -773,7 +772,7 @@ func generateSeccompSpecOpts(seccompProf string, privileged, seccompEnabled bool
}
if !seccompEnabled {
if seccompProf != "" && seccompProf != unconfinedProfile {
return nil, fmt.Errorf("seccomp is not supported")
return nil, errors.New("seccomp is not supported")
}
return nil, nil
}
@ -787,7 +786,7 @@ func generateSeccompSpecOpts(seccompProf string, privileged, seccompEnabled bool
default:
// Require and Trim default profile name prefix
if !strings.HasPrefix(seccompProf, profileNamePrefix) {
return nil, fmt.Errorf("invalid seccomp profile %q", seccompProf)
return nil, errors.Errorf("invalid seccomp profile %q", seccompProf)
}
return seccomp.WithProfile(strings.TrimPrefix(seccompProf, profileNamePrefix)), nil
}
@ -799,7 +798,7 @@ func generateApparmorSpecOpts(apparmorProf string, privileged, apparmorEnabled b
// Should fail loudly if user try to specify apparmor profile
// but we don't support it.
if apparmorProf != "" && apparmorProf != unconfinedProfile {
return nil, fmt.Errorf("apparmor is not supported")
return nil, errors.New("apparmor is not supported")
}
return nil, nil
}
@ -819,7 +818,7 @@ func generateApparmorSpecOpts(apparmorProf string, privileged, apparmorEnabled b
default:
// Require and Trim default profile name prefix
if !strings.HasPrefix(apparmorProf, profileNamePrefix) {
return nil, fmt.Errorf("invalid apparmor profile %q", apparmorProf)
return nil, errors.Errorf("invalid apparmor profile %q", apparmorProf)
}
return apparmor.WithProfile(strings.TrimPrefix(apparmorProf, profileNamePrefix)), nil
}
@ -840,7 +839,7 @@ func ensureShared(path string, lookupMount func(string) (mount.Info, error)) err
}
}
return fmt.Errorf("path %q is mounted on %q but it is not a shared mount", path, mountInfo.Mountpoint)
return errors.Errorf("path %q is mounted on %q but it is not a shared mount", path, mountInfo.Mountpoint)
}
// Ensure mount point on which path is mounted, is either shared or slave.
@ -858,5 +857,5 @@ func ensureSharedOrSlave(path string, lookupMount func(string) (mount.Info, erro
return nil
}
}
return fmt.Errorf("path %q is mounted on %q but it is not a shared or slave mount", path, mountInfo.Mountpoint)
return errors.Errorf("path %q is mounted on %q but it is not a shared or slave mount", path, mountInfo.Mountpoint)
}

7
pkg/server/container_exec.go
View File

@ -17,8 +17,7 @@ limitations under the License.
package server
import (
"fmt"
"github.com/pkg/errors"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
)
@ -27,11 +26,11 @@ import (
func (c *criContainerdService) Exec(ctx context.Context, r *runtime.ExecRequest) (*runtime.ExecResponse, error) {
cntr, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
return nil, fmt.Errorf("failed to find container %q in store: %v", r.GetContainerId(), err)
return nil, errors.Wrapf(err, "failed to find container %q in store", r.GetContainerId())
}
state := cntr.Status.Get().State()
if state != runtime.ContainerState_CONTAINER_RUNNING {
return nil, fmt.Errorf("container is in %s state", criContainerStateToString(state))
return nil, errors.Errorf("container is in %s state", criContainerStateToString(state))
}
return c.streamServer.GetExec(r)
}

24
pkg/server/container_execsync.go
View File

@ -18,13 +18,13 @@ package server
import (
"bytes"
"fmt"
"io"
"time"
"github.com/containerd/containerd"
containerdio "github.com/containerd/containerd/cio"
"github.com/containerd/containerd/errdefs"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
"golang.org/x/sys/unix"
@ -48,7 +48,7 @@ func (c *criContainerdService) ExecSync(ctx context.Context, r *runtime.ExecSync
timeout: time.Duration(r.GetTimeout()) * time.Second,
})
if err != nil {
return nil, fmt.Errorf("failed to exec in container: %v", err)
return nil, errors.Wrap(err, "failed to exec in container")
}
return &runtime.ExecSyncResponse{
@ -81,23 +81,23 @@ func (c *criContainerdService) execInContainer(ctx context.Context, id string, o
// Get container from our container store.
cntr, err := c.containerStore.Get(id)
if err != nil {
return nil, fmt.Errorf("failed to find container %q in store: %v", id, err)
return nil, errors.Wrapf(err, "failed to find container %q in store", id)
}
id = cntr.ID
state := cntr.Status.Get().State()
if state != runtime.ContainerState_CONTAINER_RUNNING {
return nil, fmt.Errorf("container is in %s state", criContainerStateToString(state))
return nil, errors.Errorf("container is in %s state", criContainerStateToString(state))
}
container := cntr.Container
spec, err := container.Spec(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get container spec: %v", err)
return nil, errors.Wrap(err, "failed to get container spec")
}
task, err := container.Task(ctx, nil)
if err != nil {
return nil, fmt.Errorf("failed to load task: %v", err)
return nil, errors.Wrap(err, "failed to load task")
}
if opts.tty {
g := newSpecGenerator(spec)
@ -126,7 +126,7 @@ func (c *criContainerdService) execInContainer(ctx context.Context, id string, o
},
)
if err != nil {
return nil, fmt.Errorf("failed to create exec %q: %v", execID, err)
return nil, errors.Wrapf(err, "failed to create exec %q", execID)
}
defer func() {
deferCtx, deferCancel := ctrdutil.DeferContext()
@ -138,10 +138,10 @@ func (c *criContainerdService) execInContainer(ctx context.Context, id string, o
exitCh, err := process.Wait(ctx)
if err != nil {
return nil, fmt.Errorf("failed to wait for process %q: %v", execID, err)
return nil, errors.Wrapf(err, "failed to wait for process %q", execID)
}
if err := process.Start(ctx); err != nil {
return nil, fmt.Errorf("failed to start exec %q: %v", execID, err)
return nil, errors.Wrapf(err, "failed to start exec %q", execID)
}
handleResizing(opts.resize, func(size remotecommand.TerminalSize) {
@ -173,7 +173,7 @@ func (c *criContainerdService) execInContainer(ctx context.Context, id string, o
//TODO(Abhi) Use context.WithDeadline instead of timeout.
// Ignore the not found error because the process may exit itself before killing.
if err := process.Kill(ctx, unix.SIGKILL); err != nil && !errdefs.IsNotFound(err) {
return nil, fmt.Errorf("failed to kill exec %q: %v", execID, err)
return nil, errors.Wrapf(err, "failed to kill exec %q", execID)
}
// Wait for the process to be killed.
exitRes := <-exitCh
@ -181,12 +181,12 @@ func (c *criContainerdService) execInContainer(ctx context.Context, id string, o
execID, exitRes.ExitCode(), exitRes.Error())
<-attachDone
logrus.Debugf("Stream pipe for exec process %q done", execID)
return nil, fmt.Errorf("timeout %v exceeded", opts.timeout)
return nil, errors.Errorf("timeout %v exceeded", opts.timeout)
case exitRes := <-exitCh:
code, _, err := exitRes.Result()
logrus.Infof("Exec process %q exits with exit code %d and error %v", execID, code, err)
if err != nil {
return nil, fmt.Errorf("failed while waiting for exec %q: %v", execID, err)
return nil, errors.Wrapf(err, "failed while waiting for exec %q", execID)
}
<-attachDone
logrus.Debugf("Stream pipe for exec process %q done", execID)

6
pkg/server/container_log_reopen.go
View File

@ -17,7 +17,7 @@ limitations under the License.
package server
import (
"fmt"
"github.com/pkg/errors"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -28,11 +28,11 @@ import (
func (c *criContainerdService) ReopenContainerLog(ctx context.Context, r *runtime.ReopenContainerLogRequest) (*runtime.ReopenContainerLogResponse, error) {
container, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
return nil, fmt.Errorf("an error occurred when try to find container %q: %v", r.GetContainerId(), err)
return nil, errors.Wrapf(err, "an error occurred when try to find container %q", r.GetContainerId())
}
if container.Status.Get().State() != runtime.ContainerState_CONTAINER_RUNNING {
return nil, fmt.Errorf("container is not running")
return nil, errors.New("container is not running")
}
// Create new container logger and replace the existing ones.

19
pkg/server/container_remove.go
View File

@ -17,11 +17,10 @@ limitations under the License.
package server
import (
"fmt"
"github.com/containerd/containerd"
"github.com/containerd/containerd/errdefs"
"github.com/docker/docker/pkg/system"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -37,7 +36,7 @@ func (c *criContainerdService) RemoveContainer(ctx context.Context, r *runtime.R
container, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
if err != store.ErrNotExist {
return nil, fmt.Errorf("an error occurred when try to find container %q: %v", r.GetContainerId(), err)
return nil, errors.Wrapf(err, "an error occurred when try to find container %q", r.GetContainerId())
}
// Do not return error if container metadata doesn't exist.
log.Tracef("RemoveContainer called for container %q that does not exist", r.GetContainerId())
@ -48,7 +47,7 @@ func (c *criContainerdService) RemoveContainer(ctx context.Context, r *runtime.R
// Set removing state to prevent other start/remove operations against this container
// while it's being removed.
if err := setContainerRemoving(container); err != nil {
return nil, fmt.Errorf("failed to set removing state for container %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to set removing state for container %q", id)
}
defer func() {
if retErr != nil {
@ -67,20 +66,20 @@ func (c *criContainerdService) RemoveContainer(ctx context.Context, r *runtime.R
// Delete containerd container.
if err := container.Container.Delete(ctx, containerd.WithSnapshotCleanup); err != nil {
if !errdefs.IsNotFound(err) {
return nil, fmt.Errorf("failed to delete containerd container %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to delete containerd container %q", id)
}
log.Tracef("Remove called for containerd container %q that does not exist", id)
}
// Delete container checkpoint.
if err := container.Delete(); err != nil {
return nil, fmt.Errorf("failed to delete container checkpoint for %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to delete container checkpoint for %q", id)
}
containerRootDir := getContainerRootDir(c.config.RootDir, id)
if err := system.EnsureRemoveAll(containerRootDir); err != nil {
return nil, fmt.Errorf("failed to remove container root directory %q: %v",
containerRootDir, err)
return nil, errors.Wrapf(err, "failed to remove container root directory %q",
containerRootDir)
}
c.containerStore.Delete(id)
@ -96,10 +95,10 @@ func setContainerRemoving(container containerstore.Container) error {
return container.Status.Update(func(status containerstore.Status) (containerstore.Status, error) {
// Do not remove container if it's still running.
if status.State() == runtime.ContainerState_CONTAINER_RUNNING {
return status, fmt.Errorf("container is still running")
return status, errors.New("container is still running")
}
if status.Removing {
return status, fmt.Errorf("container is already in removing state")
return status, errors.New("container is already in removing state")
}
status.Removing = true
return status, nil

24
pkg/server/container_start.go
View File

@ -17,13 +17,13 @@ limitations under the License.
package server
import (
"fmt"
"io"
"time"
"github.com/containerd/containerd"
containerdio "github.com/containerd/containerd/cio"
"github.com/containerd/containerd/errdefs"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -38,7 +38,7 @@ import (
func (c *criContainerdService) StartContainer(ctx context.Context, r *runtime.StartContainerRequest) (retRes *runtime.StartContainerResponse, retErr error) {
container, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
return nil, fmt.Errorf("an error occurred when try to find container %q: %v", r.GetContainerId(), err)
return nil, errors.Wrapf(err, "an error occurred when try to find container %q", r.GetContainerId())
}
var startErr error
@ -51,7 +51,7 @@ func (c *criContainerdService) StartContainer(ctx context.Context, r *runtime.St
}); startErr != nil {
return nil, startErr
} else if err != nil {
return nil, fmt.Errorf("failed to update container %q metadata: %v", container.ID, err)
return nil, errors.Wrapf(err, "failed to update container %q metadata", container.ID)
}
return &runtime.StartContainerResponse{}, nil
}
@ -68,11 +68,11 @@ func (c *criContainerdService) startContainer(ctx context.Context,
// Return error if container is not in created state.
if status.State() != runtime.ContainerState_CONTAINER_CREATED {
return fmt.Errorf("container %q is in %s state", id, criContainerStateToString(status.State()))
return errors.Errorf("container %q is in %s state", id, criContainerStateToString(status.State()))
}
// Do not start the container when there is a removal in progress.
if status.Removing {
return fmt.Errorf("container %q is in removing state", id)
return errors.Errorf("container %q is in removing state", id)
}
defer func() {
@ -89,17 +89,17 @@ func (c *criContainerdService) startContainer(ctx context.Context,
// Get sandbox config from sandbox store.
sandbox, err := c.sandboxStore.Get(meta.SandboxID)
if err != nil {
return fmt.Errorf("sandbox %q not found: %v", meta.SandboxID, err)
return errors.Wrapf(err, "sandbox %q not found", meta.SandboxID)
}
sandboxID := meta.SandboxID
if sandbox.Status.Get().State != sandboxstore.StateReady {
return fmt.Errorf("sandbox container %q is not running", sandboxID)
return errors.Errorf("sandbox container %q is not running", sandboxID)
}
ioCreation := func(id string) (_ containerdio.IO, err error) {
stdoutWC, stderrWC, err := createContainerLoggers(meta.LogPath, config.GetTty())
if err != nil {
return nil, fmt.Errorf("failed to create container loggers: %v", err)
return nil, errors.Wrap(err, "failed to create container loggers")
}
defer func() {
if err != nil {
@ -118,7 +118,7 @@ func (c *criContainerdService) startContainer(ctx context.Context,
task, err := container.NewTask(ctx, ioCreation)
if err != nil {
return fmt.Errorf("failed to create containerd task: %v", err)
return errors.Wrap(err, "failed to create containerd task")
}
defer func() {
if retErr != nil {
@ -133,7 +133,7 @@ func (c *criContainerdService) startContainer(ctx context.Context,
// Start containerd task.
if err := task.Start(ctx); err != nil {
return fmt.Errorf("failed to start containerd task %q: %v", id, err)
return errors.Wrapf(err, "failed to start containerd task %q", id)
}
// Update container start timestamp.
@ -147,7 +147,7 @@ func createContainerLoggers(logPath string, tty bool) (stdout io.WriteCloser, st
if logPath != "" {
// Only generate container log when log path is specified.
if stdout, err = cio.NewCRILogger(logPath, cio.Stdout); err != nil {
return nil, nil, fmt.Errorf("failed to start container stdout logger: %v", err)
return nil, nil, errors.Wrap(err, "failed to start container stdout logger")
}
defer func() {
if err != nil {
@ -157,7 +157,7 @@ func createContainerLoggers(logPath string, tty bool) (stdout io.WriteCloser, st
// Only redirect stderr when there is no tty.
if !tty {
if stderr, err = cio.NewCRILogger(logPath, cio.Stderr); err != nil {
return nil, nil, fmt.Errorf("failed to start container stderr logger: %v", err)
return nil, nil, errors.Wrap(err, "failed to start container stderr logger")
}
}
} else {

11
pkg/server/container_stats.go
View File

@ -17,9 +17,8 @@ limitations under the License.
package server
import (
"fmt"
tasks "github.com/containerd/containerd/api/services/tasks/v1"
"github.com/pkg/errors"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
)
@ -29,20 +28,20 @@ import (
func (c *criContainerdService) ContainerStats(ctx context.Context, in *runtime.ContainerStatsRequest) (*runtime.ContainerStatsResponse, error) {
cntr, err := c.containerStore.Get(in.GetContainerId())
if err != nil {
return nil, fmt.Errorf("failed to find container: %v", err)
return nil, errors.Wrap(err, "failed to find container")
}
request := &tasks.MetricsRequest{Filters: []string{"id==" + cntr.ID}}
resp, err := c.client.TaskService().Metrics(ctx, request)
if err != nil {
return nil, fmt.Errorf("failed to fetch metrics for task: %v", err)
return nil, errors.Wrap(err, "failed to fetch metrics for task")
}
if len(resp.Metrics) != 1 {
return nil, fmt.Errorf("unexpected metrics response: %+v", resp.Metrics)
return nil, errors.Errorf("unexpected metrics response: %+v", resp.Metrics)
}
cs, err := c.getContainerMetrics(cntr.Metadata, resp.Metrics[0])
if err != nil {
return nil, fmt.Errorf("failed to decode container metrics: %v", err)
return nil, errors.Wrap(err, "failed to decode container metrics")
}
return &runtime.ContainerStatsResponse{Stats: cs}, nil
}

13
pkg/server/container_stats_list.go
View File

@ -17,12 +17,11 @@ limitations under the License.
package server
import (
"fmt"
"github.com/containerd/cgroups"
tasks "github.com/containerd/containerd/api/services/tasks/v1"
"github.com/containerd/containerd/api/types"
"github.com/containerd/typeurl"
"github.com/pkg/errors"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -36,15 +35,15 @@ func (c *criContainerdService) ListContainerStats(
) (*runtime.ListContainerStatsResponse, error) {
request, containers, err := c.buildTaskMetricsRequest(in)
if err != nil {
return nil, fmt.Errorf("failed to build metrics request: %v", err)
return nil, errors.Wrap(err, "failed to build metrics request")
}
resp, err := c.client.TaskService().Metrics(ctx, &request)
if err != nil {
return nil, fmt.Errorf("failed to fetch metrics for tasks: %v", err)
return nil, errors.Wrap(err, "failed to fetch metrics for tasks")
}
criStats, err := c.toCRIContainerStats(resp.Metrics, containers)
if err != nil {
return nil, fmt.Errorf("failed to convert to cri containerd stats format: %v", err)
return nil, errors.Wrap(err, "failed to convert to cri containerd stats format")
}
return criStats, nil
}
@ -61,7 +60,7 @@ func (c *criContainerdService) toCRIContainerStats(
for _, cntr := range containers {
cs, err := c.getContainerMetrics(cntr.Metadata, statsMap[cntr.ID])
if err != nil {
return nil, fmt.Errorf("failed to decode container metrics for %q: %v", cntr.ID, err)
return nil, errors.Wrapf(err, "failed to decode container metrics for %q", cntr.ID)
}
containerStats.Stats = append(containerStats.Stats, cs)
}
@ -99,7 +98,7 @@ func (c *criContainerdService) getContainerMetrics(
if stats != nil {
s, err := typeurl.UnmarshalAny(stats.Data)
if err != nil {
return nil, fmt.Errorf("failed to extract container metrics: %v", err)
return nil, errors.Wrap(err, "failed to extract container metrics")
}
metrics := s.(*cgroups.Metrics)
if metrics.CPU != nil && metrics.CPU.Usage != nil {

10
pkg/server/container_status.go
View File

@ -18,9 +18,9 @@ package server
import (
"encoding/json"
"fmt"
runtimespec "github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -32,7 +32,7 @@ import (
func (c *criContainerdService) ContainerStatus(ctx context.Context, r *runtime.ContainerStatusRequest) (*runtime.ContainerStatusResponse, error) {
container, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
return nil, fmt.Errorf("an error occurred when try to find container %q: %v", r.GetContainerId(), err)
return nil, errors.Wrapf(err, "an error occurred when try to find container %q", r.GetContainerId())
}
// TODO(random-liu): Clean up the following logic in CRI.
@ -44,7 +44,7 @@ func (c *criContainerdService) ContainerStatus(ctx context.Context, r *runtime.C
imageRef := container.ImageRef
image, err := c.imageStore.Get(imageRef)
if err != nil {
return nil, fmt.Errorf("failed to get image %q: %v", imageRef, err)
return nil, errors.Wrapf(err, "failed to get image %q", imageRef)
}
if len(image.RepoTags) > 0 {
// Based on current behavior of dockershim, this field should be
@ -58,7 +58,7 @@ func (c *criContainerdService) ContainerStatus(ctx context.Context, r *runtime.C
status := toCRIContainerStatus(container, spec, imageRef)
info, err := toCRIContainerInfo(ctx, container, r.GetVerbose())
if err != nil {
return nil, fmt.Errorf("failed to get verbose container info: %v", err)
return nil, errors.Wrap(err, "failed to get verbose container info")
}
return &runtime.ContainerStatusResponse{
@ -145,7 +145,7 @@ func toCRIContainerInfo(ctx context.Context, container containerstore.Container,
infoBytes, err := json.Marshal(ci)
if err != nil {
return nil, fmt.Errorf("failed to marshal info %v: %v", ci, err)
return nil, errors.Wrapf(err, "failed to marshal info %v", ci)
}
return map[string]string{
"info": string(infoBytes),

24
pkg/server/container_stop.go
View File

@ -17,12 +17,12 @@ limitations under the License.
package server
import (
"fmt"
"time"
"github.com/containerd/containerd"
"github.com/containerd/containerd/errdefs"
"github.com/docker/docker/pkg/signal"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
"golang.org/x/sys/unix"
@ -40,7 +40,7 @@ func (c *criContainerdService) StopContainer(ctx context.Context, r *runtime.Sto
// Get container config from container store.
container, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
return nil, fmt.Errorf("an error occurred when try to find container %q: %v", r.GetContainerId(), err)
return nil, errors.Wrapf(err, "an error occurred when try to find container %q", r.GetContainerId())
}
if err := c.stopContainer(ctx, container, time.Duration(r.GetTimeout())*time.Second); err != nil {
@ -71,27 +71,27 @@ func (c *criContainerdService) stopContainer(ctx context.Context, container cont
// deleted and image is garbage collected before this point. However,
// the chance is really slim, even it happens, it's still fine to return
// an error here.
return fmt.Errorf("failed to get image metadata %q: %v", container.ImageRef, err)
return errors.Wrapf(err, "failed to get image metadata %q", container.ImageRef)
}
if image.ImageSpec.Config.StopSignal != "" {
stopSignal, err = signal.ParseSignal(image.ImageSpec.Config.StopSignal)
if err != nil {
return fmt.Errorf("failed to parse stop signal %q: %v",
image.ImageSpec.Config.StopSignal, err)
return errors.Wrapf(err, "failed to parse stop signal %q",
image.ImageSpec.Config.StopSignal)
}
}
logrus.Infof("Stop container %q with signal %v", id, stopSignal)
task, err := container.Container.Task(ctx, nil)
if err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to stop container, task not found for container %q: %v", id, err)
return errors.Wrapf(err, "failed to stop container, task not found for container %q", id)
}
return nil
}
if task != nil {
if err = task.Kill(ctx, stopSignal); err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to stop container %q: %v", id, err)
return errors.Wrapf(err, "failed to stop container %q", id)
}
// Move on to make sure container status is updated.
}
@ -107,7 +107,7 @@ func (c *criContainerdService) stopContainer(ctx context.Context, container cont
task, err := container.Container.Task(ctx, nil)
if err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to stop container, task not found for container %q: %v", id, err)
return errors.Wrapf(err, "failed to stop container, task not found for container %q", id)
}
return nil
}
@ -116,7 +116,7 @@ func (c *criContainerdService) stopContainer(ctx context.Context, container cont
if task != nil {
if err = task.Kill(ctx, unix.SIGKILL, containerd.WithKillAll); err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to kill container %q: %v", id, err)
return errors.Wrapf(err, "failed to kill container %q", id)
}
// Move on to make sure container status is updated.
}
@ -124,7 +124,7 @@ func (c *criContainerdService) stopContainer(ctx context.Context, container cont
// Wait for a fixed timeout until container stop is observed by event monitor.
if err := c.waitContainerStop(ctx, container, killContainerTimeout); err != nil {
return fmt.Errorf("an error occurs during waiting for container %q to stop: %v", id, err)
return errors.Wrapf(err, "an error occurs during waiting for container %q to stop", id)
}
return nil
}
@ -135,9 +135,9 @@ func (c *criContainerdService) waitContainerStop(ctx context.Context, container
defer timeoutTimer.Stop()
select {
case <-ctx.Done():
return fmt.Errorf("wait container %q is cancelled", container.ID)
return errors.Errorf("wait container %q is cancelled", container.ID)
case <-timeoutTimer.C:
return fmt.Errorf("wait container %q stop timeout", container.ID)
return errors.Errorf("wait container %q stop timeout", container.ID)
case <-container.Stopped():
return nil
}

22
pkg/server/container_update_resources.go
View File

@ -18,13 +18,13 @@ package server
import (
gocontext "context"
"fmt"
"github.com/containerd/containerd"
"github.com/containerd/containerd/containers"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/typeurl"
runtimespec "github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -38,7 +38,7 @@ import (
func (c *criContainerdService) UpdateContainerResources(ctx context.Context, r *runtime.UpdateContainerResourcesRequest) (retRes *runtime.UpdateContainerResourcesResponse, retErr error) {
container, err := c.containerStore.Get(r.GetContainerId())
if err != nil {
return nil, fmt.Errorf("failed to find container: %v", err)
return nil, errors.Wrap(err, "failed to find container")
}
// Update resources in status update transaction, so that:
// 1) There won't be race condition with container start.
@ -46,7 +46,7 @@ func (c *criContainerdService) UpdateContainerResources(ctx context.Context, r *
if err := container.Status.Update(func(status containerstore.Status) (containerstore.Status, error) {
return status, c.updateContainerResources(ctx, container, r.GetLinux(), status)
}); err != nil {
return nil, fmt.Errorf("failed to update resources: %v", err)
return nil, errors.Wrap(err, "failed to update resources")
}
return &runtime.UpdateContainerResourcesResponse{}, nil
}
@ -58,7 +58,7 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
id := cntr.ID
// Do not update the container when there is a removal in progress.
if status.Removing {
return fmt.Errorf("container %q is in removing state", id)
return errors.Errorf("container %q is in removing state", id)
}
// Update container spec. If the container is not started yet, updating
@ -67,11 +67,11 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
// the spec will become our source of truth for resource limits.
oldSpec, err := cntr.Container.Spec(ctx)
if err != nil {
return fmt.Errorf("failed to get container spec: %v", err)
return errors.Wrap(err, "failed to get container spec")
}
newSpec, err := updateOCILinuxResource(oldSpec, resources)
if err != nil {
return fmt.Errorf("failed to update resource in spec: %v", err)
return errors.Wrap(err, "failed to update resource in spec")
}
if err := updateContainerSpec(ctx, cntr.Container, newSpec); err != nil {
@ -100,7 +100,7 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
// Task exited already.
return nil
}
return fmt.Errorf("failed to get task: %v", err)
return errors.Wrap(err, "failed to get task")
}
// newSpec.Linux won't be nil
if err := task.Update(ctx, containerd.WithResources(newSpec.Linux.Resources)); err != nil {
@ -108,7 +108,7 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
// Task exited already.
return nil
}
return fmt.Errorf("failed to update resources: %v", err)
return errors.Wrap(err, "failed to update resources")
}
return nil
}
@ -117,13 +117,13 @@ func (c *criContainerdService) updateContainerResources(ctx context.Context,
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)
return errors.Wrapf(err, "failed to marshal spec %+v", spec)
}
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 errors.Wrap(err, "failed to update container spec")
}
return nil
}
@ -133,7 +133,7 @@ func updateOCILinuxResource(spec *runtimespec.Spec, new *runtime.LinuxContainerR
// Copy to make sure old spec is not changed.
var cloned runtimespec.Spec
if err := util.DeepCopy(&cloned, spec); err != nil {
return nil, fmt.Errorf("failed to deep copy: %v", err)
return nil, errors.Wrap(err, "failed to deep copy")
}
g := newSpecGenerator(&cloned)

33
pkg/server/events.go
View File

@ -17,8 +17,6 @@ limitations under the License.
package server
import (
"errors"
"fmt"
"time"
eventtypes "github.com/containerd/containerd/api/events"
@ -27,6 +25,7 @@ import (
"github.com/containerd/containerd/events"
"github.com/containerd/typeurl"
gogotypes "github.com/gogo/protobuf/types"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
"k8s.io/apimachinery/pkg/util/clock"
@ -99,7 +98,7 @@ func convertEvent(e *gogotypes.Any) (string, interface{}, error) {
containerID := ""
evt, err := typeurl.UnmarshalAny(e)
if err != nil {
return "", nil, fmt.Errorf("failed to unmarshalany %+v", err)
return "", nil, errors.Wrap(err, "failed to unmarshalany")
}
switch evt.(type) {
@ -108,7 +107,7 @@ func convertEvent(e *gogotypes.Any) (string, interface{}, error) {
case *eventtypes.TaskOOM:
containerID = evt.(*eventtypes.TaskOOM).ContainerID
default:
return "", nil, fmt.Errorf("unsupported event")
return "", nil, errors.New("unsupported event")
}
return containerID, evt, nil
}
@ -183,21 +182,21 @@ func (em *eventMonitor) handleEvent(any interface{}) error {
cntr, err := em.containerStore.Get(e.ContainerID)
if err == nil {
if err := handleContainerExit(ctx, e, cntr); err != nil {
return fmt.Errorf("failed to handle container TaskExit event:%+v", err)
return errors.Wrap(err, "failed to handle container TaskExit event")
}
return nil
} else if err != store.ErrNotExist {
return fmt.Errorf("can't find container for TaskExit event:%+v", err)
return errors.Wrap(err, "can't find container for TaskExit event")
}
// Use GetAll to include sandbox in unknown state.
sb, err := em.sandboxStore.GetAll(e.ContainerID)
if err == nil {
if err := handleSandboxExit(ctx, e, sb); err != nil {
return fmt.Errorf("failed to handle sandbox TaskExit event:%+v", err)
return errors.Wrap(err, "failed to handle sandbox TaskExit event")
}
return nil
} else if err != store.ErrNotExist {
return fmt.Errorf("can't find sandbox for TaskExit event:%+v", err)
return errors.Wrap(err, "can't find sandbox for TaskExit event")
}
return nil
case *eventtypes.TaskOOM:
@ -206,11 +205,11 @@ func (em *eventMonitor) handleEvent(any interface{}) error {
cntr, err := em.containerStore.Get(e.ContainerID)
if err != nil {
if err != store.ErrNotExist {
return fmt.Errorf("can't find container for TaskOOM event:%+v", err)
return errors.Wrap(err, "can't find container for TaskOOM event")
}
if _, err = em.sandboxStore.Get(e.ContainerID); err != nil {
if err != store.ErrNotExist {
return fmt.Errorf("can't find sandbox for TaskOOM event:%+v", err)
return errors.Wrap(err, "can't find sandbox for TaskOOM event")
}
return nil
}
@ -221,7 +220,7 @@ func (em *eventMonitor) handleEvent(any interface{}) error {
return status, nil
})
if err != nil {
return fmt.Errorf("failed to update container status for TaskOOM event:%+v", err)
return errors.Wrap(err, "failed to update container status for TaskOOM event")
}
}
@ -242,13 +241,13 @@ func handleContainerExit(ctx context.Context, e *eventtypes.TaskExit, cntr conta
)
if err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to load task for container: %v", err)
return errors.Wrapf(err, "failed to load task for container")
}
} else {
// TODO(random-liu): [P1] This may block the loop, we may want to spawn a worker
if _, err = task.Delete(ctx); err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to stop container: %v", err)
return errors.Wrap(err, "failed to stop container")
}
// Move on to make sure container status is updated.
}
@ -265,7 +264,7 @@ func handleContainerExit(ctx context.Context, e *eventtypes.TaskExit, cntr conta
return status, nil
})
if err != nil {
return fmt.Errorf("failed to update container state: %v", err)
return errors.Wrap(err, "failed to update container state")
}
// Using channel to propagate the information of container stop
cntr.Stop()
@ -282,13 +281,13 @@ func handleSandboxExit(ctx context.Context, e *eventtypes.TaskExit, sb sandboxst
task, err := sb.Container.Task(ctx, nil)
if err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to load task for sandbox: %v", err)
return errors.Wrap(err, "failed to load task for sandbox")
}
} else {
// TODO(random-liu): [P1] This may block the loop, we may want to spawn a worker
if _, err = task.Delete(ctx); err != nil {
if !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to stop sandbox: %v", err)
return errors.Wrap(err, "failed to stop sandbox")
}
// Move on to make sure container status is updated.
}
@ -307,7 +306,7 @@ func handleSandboxExit(ctx context.Context, e *eventtypes.TaskExit, sb sandboxst
return status, nil
})
if err != nil {
return fmt.Errorf("failed to update sandbox state: %v", err)
return errors.Wrap(err, "failed to update sandbox state")
}
// Using channel to propagate the information of sandbox stop
sb.Stop()

19
pkg/server/helpers.go
View File

@ -34,6 +34,7 @@ import (
"github.com/opencontainers/runtime-tools/generate"
"github.com/opencontainers/selinux/go-selinux"
"github.com/opencontainers/selinux/go-selinux/label"
"github.com/pkg/errors"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -251,7 +252,7 @@ func (c *criContainerdService) localResolve(ctx context.Context, refOrID string)
if err == store.ErrNotExist {
return nil, nil
}
return nil, fmt.Errorf("failed to get image %q : %v", imageID, err)
return nil, errors.Wrapf(err, "failed to get image %q", imageID)
}
return &image, nil
}
@ -280,7 +281,7 @@ func getUserFromImage(user string) (*int64, string) {
func (c *criContainerdService) ensureImageExists(ctx context.Context, ref string) (*imagestore.Image, error) {
image, err := c.localResolve(ctx, ref)
if err != nil {
return nil, fmt.Errorf("failed to resolve image %q: %v", ref, err)
return nil, errors.Wrapf(err, "failed to resolve image %q", ref)
}
if image != nil {
return image, nil
@ -288,13 +289,13 @@ func (c *criContainerdService) ensureImageExists(ctx context.Context, ref string
// Pull image to ensure the image exists
resp, err := c.PullImage(ctx, &runtime.PullImageRequest{Image: &runtime.ImageSpec{Image: ref}})
if err != nil {
return nil, fmt.Errorf("failed to pull image %q: %v", ref, err)
return nil, errors.Wrapf(err, "failed to pull image %q", ref)
}
imageID := resp.GetImageRef()
newImage, err := c.imageStore.Get(imageID)
if err != nil {
// It's still possible that someone removed the image right after it is pulled.
return nil, fmt.Errorf("failed to get image %q metadata after pulling: %v", imageID, err)
return nil, errors.Wrapf(err, "failed to get image %q metadata after pulling", imageID)
}
return &newImage, nil
}
@ -312,28 +313,28 @@ func getImageInfo(ctx context.Context, image containerd.Image) (*imageInfo, erro
// Get image information.
diffIDs, err := image.RootFS(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get image diffIDs: %v", err)
return nil, errors.Wrap(err, "failed to get image diffIDs")
}
chainID := identity.ChainID(diffIDs)
size, err := image.Size(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get image compressed resource size: %v", err)
return nil, errors.Wrap(err, "failed to get image compressed resource size")
}
desc, err := image.Config(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get image config descriptor: %v", err)
return nil, errors.Wrap(err, "failed to get image config descriptor")
}
id := desc.Digest.String()
rb, err := content.ReadBlob(ctx, image.ContentStore(), desc.Digest)
if err != nil {
return nil, fmt.Errorf("failed to read image config from content store: %v", err)
return nil, errors.Wrap(err, "failed to read image config from content store")
}
var ociimage imagespec.Image
if err := json.Unmarshal(rb, &ociimage); err != nil {
return nil, fmt.Errorf("failed to unmarshal image config %s: %v", rb, err)
return nil, errors.Wrapf(err, "failed to unmarshal image config %s", rb)
}
return &imageInfo{

18
pkg/server/image_load.go
View File

@ -17,12 +17,12 @@ limitations under the License.
package server
import (
"fmt"
"golang.org/x/net/context"
"os"
"path/filepath"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
api "github.com/containerd/cri/pkg/api/v1"
"github.com/containerd/cri/pkg/containerd/importer"
@ -33,20 +33,20 @@ import (
func (c *criContainerdService) LoadImage(ctx context.Context, r *api.LoadImageRequest) (*api.LoadImageResponse, error) {
path := r.GetFilePath()
if !filepath.IsAbs(path) {
return nil, fmt.Errorf("path %q is not an absolute path", path)
return nil, errors.Errorf("path %q is not an absolute path", path)
}
f, err := os.Open(path)
if err != nil {
return nil, fmt.Errorf("failed to open file: %v", err)
return nil, errors.Wrap(err, "failed to open file")
}
repoTags, err := importer.Import(ctx, c.client, f)
if err != nil {
return nil, fmt.Errorf("failed to import image: %v", err)
return nil, errors.Wrap(err, "failed to import image")
}
for _, repoTag := range repoTags {
image, err := c.client.GetImage(ctx, repoTag)
if err != nil {
return nil, fmt.Errorf("failed to get image %q: %v", repoTag, err)
return nil, errors.Wrapf(err, "failed to get image %q", repoTag)
}
if err := image.Unpack(ctx, c.config.ContainerdConfig.Snapshotter); err != nil {
logrus.WithError(err).Warnf("Failed to unpack image %q", repoTag)
@ -54,12 +54,12 @@ func (c *criContainerdService) LoadImage(ctx context.Context, r *api.LoadImageRe
}
info, err := getImageInfo(ctx, image)
if err != nil {
return nil, fmt.Errorf("failed to get image %q info: %v", repoTag, err)
return nil, errors.Wrapf(err, "failed to get image %q info", repoTag)
}
id := info.id
if err := c.createImageReference(ctx, id, image.Target()); err != nil {
return nil, fmt.Errorf("failed to create image reference %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to create image reference %q", id)
}
img := imagestore.Image{
@ -72,7 +72,7 @@ func (c *criContainerdService) LoadImage(ctx context.Context, r *api.LoadImageRe
}
if err := c.imageStore.Add(img); err != nil {
return nil, fmt.Errorf("failed to add image %q into store: %v", id, err)
return nil, errors.Wrapf(err, "failed to add image %q into store", id)
}
logrus.Debugf("Imported image with id %q, repo tag %q", id, repoTag)
}

18
pkg/server/image_pull.go
View File

@ -18,7 +18,6 @@ package server
import (
"encoding/base64"
"fmt"
"net/http"
"strings"
@ -26,6 +25,7 @@ import (
"github.com/containerd/containerd/errdefs"
containerdimages "github.com/containerd/containerd/images"
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -81,7 +81,7 @@ func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullIma
imageRef := r.GetImage().GetImage()
namedRef, err := util.NormalizeImageRef(imageRef)
if err != nil {
return nil, fmt.Errorf("failed to parse image reference %q: %v", imageRef, err)
return nil, errors.Wrapf(err, "failed to parse image reference %q", imageRef)
}
ref := namedRef.String()
if ref != imageRef {
@ -94,7 +94,7 @@ func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullIma
})
_, desc, err := resolver.Resolve(ctx, ref)
if err != nil {
return nil, fmt.Errorf("failed to resolve image %q: %v", ref, err)
return nil, errors.Wrapf(err, "failed to resolve image %q", ref)
}
// We have to check schema1 here, because after `Pull`, schema1
// image has already been converted.
@ -106,7 +106,7 @@ func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullIma
containerd.WithResolver(resolver),
)
if err != nil {
return nil, fmt.Errorf("failed to pull image %q: %v", ref, err)
return nil, errors.Wrapf(err, "failed to pull image %q", ref)
}
// Do best effort unpack.
@ -119,7 +119,7 @@ func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullIma
// Get image information.
info, err := getImageInfo(ctx, image)
if err != nil {
return nil, fmt.Errorf("failed to get image information: %v", err)
return nil, errors.Wrap(err, "failed to get image information")
}
imageID := info.id
@ -129,7 +129,7 @@ func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullIma
continue
}
if err := c.createImageReference(ctx, r, image.Target()); err != nil {
return nil, fmt.Errorf("failed to update image reference %q: %v", r, err)
return nil, errors.Wrapf(err, "failed to update image reference %q", r)
}
}
@ -150,7 +150,7 @@ func (c *criContainerdService) PullImage(ctx context.Context, r *runtime.PullIma
}
if err := c.imageStore.Add(img); err != nil {
return nil, fmt.Errorf("failed to add image %q into store: %v", img.ID, err)
return nil, errors.Wrapf(err, "failed to add image %q into store", img.ID)
}
// NOTE(random-liu): the actual state in containerd is the source of truth, even we maintain
@ -181,13 +181,13 @@ func ParseAuth(auth *runtime.AuthConfig) (string, string, error) {
}
fields := strings.SplitN(string(decoded), ":", 2)
if len(fields) != 2 {
return "", "", fmt.Errorf("invalid decoded auth: %q", decoded)
return "", "", errors.Errorf("invalid decoded auth: %q", decoded)
}
user, passwd := fields[0], fields[1]
return user, strings.Trim(passwd, "\x00"), nil
}
// TODO(random-liu): Support RegistryToken.
return "", "", fmt.Errorf("invalid auth config")
return "", "", errors.New("invalid auth config")
}
// createImageReference creates image reference inside containerd image store.

11
pkg/server/image_remove.go
View File

@ -17,10 +17,9 @@ limitations under the License.
package server
import (
"fmt"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/images"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -35,7 +34,7 @@ import (
func (c *criContainerdService) RemoveImage(ctx context.Context, r *runtime.RemoveImageRequest) (*runtime.RemoveImageResponse, error) {
image, err := c.localResolve(ctx, r.GetImage().GetImage())
if err != nil {
return nil, fmt.Errorf("can not resolve %q locally: %v", r.GetImage().GetImage(), err)
return nil, errors.Wrapf(err, "can not resolve %q locally", r.GetImage().GetImage())
}
if image == nil {
// return empty without error when image not found.
@ -49,7 +48,7 @@ func (c *criContainerdService) RemoveImage(ctx context.Context, r *runtime.Remov
if errdefs.IsNotFound(err) {
continue
}
return nil, fmt.Errorf("failed to get image %q: %v", tag, err)
return nil, errors.Wrapf(err, "failed to get image %q", tag)
}
desc, err := cImage.Config(ctx)
if err != nil {
@ -82,12 +81,12 @@ func (c *criContainerdService) RemoveImage(ctx context.Context, r *runtime.Remov
if err == nil || errdefs.IsNotFound(err) {
continue
}
return nil, fmt.Errorf("failed to delete image reference %q for image %q: %v", ref, image.ID, err)
return nil, errors.Wrapf(err, "failed to delete image reference %q for image %q", ref, image.ID)
}
// Delete image id synchronously to trigger garbage collection.
err = c.client.ImageService().Delete(ctx, image.ID, images.SynchronousDelete())
if err != nil && !errdefs.IsNotFound(err) {
return nil, fmt.Errorf("failed to delete image id %q: %v", image.ID, err)
return nil, errors.Wrapf(err, "failed to delete image id %q", image.ID)
}
c.imageStore.Delete(image.ID)
return &runtime.RemoveImageResponse{}, nil

6
pkg/server/image_status.go
View File

@ -18,8 +18,8 @@ package server
import (
"encoding/json"
"fmt"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -34,7 +34,7 @@ import (
func (c *criContainerdService) ImageStatus(ctx context.Context, r *runtime.ImageStatusRequest) (*runtime.ImageStatusResponse, error) {
image, err := c.localResolve(ctx, r.GetImage().GetImage())
if err != nil {
return nil, fmt.Errorf("can not resolve %q locally: %v", r.GetImage().GetImage(), err)
return nil, errors.Wrapf(err, "can not resolve %q locally", r.GetImage().GetImage())
}
if image == nil {
// return empty without error when image not found.
@ -46,7 +46,7 @@ func (c *criContainerdService) ImageStatus(ctx context.Context, r *runtime.Image
runtimeImage := toCRIRuntimeImage(image)
info, err := c.toCRIImageInfo(ctx, image, r.GetVerbose())
if err != nil {
return nil, fmt.Errorf("failed to generate image info: %v", err)
return nil, errors.Wrap(err, "failed to generate image info")
}
return &runtime.ImageStatusResponse{

4
pkg/server/io/logger.go
View File

@ -19,12 +19,12 @@ package io
import (
"bufio"
"bytes"
"fmt"
"io"
"io/ioutil"
"os"
"time"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -57,7 +57,7 @@ func NewCRILogger(path string, stream StreamType) (io.WriteCloser, error) {
prc, pwc := io.Pipe()
f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0640)
if err != nil {
return nil, fmt.Errorf("failed to open log file: %v", err)
return nil, errors.Wrap(err, "failed to open log file")
}
go redirectLogs(path, prc, f, stream)
return pwc, nil

66
pkg/server/restart.go
View File

@ -17,7 +17,6 @@ limitations under the License.
package server
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
@ -31,6 +30,7 @@ import (
"github.com/containerd/typeurl"
"github.com/docker/distribution/reference"
"github.com/docker/docker/pkg/system"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -55,7 +55,7 @@ func (c *criContainerdService) recover(ctx context.Context) error {
// Recover all sandboxes.
sandboxes, err := c.client.Containers(ctx, filterLabel(containerKindLabel, containerKindSandbox))
if err != nil {
return fmt.Errorf("failed to list sandbox containers: %v", err)
return errors.Wrap(err, "failed to list sandbox containers")
}
for _, sandbox := range sandboxes {
sb, err := loadSandbox(ctx, sandbox)
@ -65,17 +65,17 @@ func (c *criContainerdService) recover(ctx context.Context) error {
}
logrus.Debugf("Loaded sandbox %+v", sb)
if err := c.sandboxStore.Add(sb); err != nil {
return fmt.Errorf("failed to add sandbox %q to store: %v", sandbox.ID(), err)
return errors.Wrapf(err, "failed to add sandbox %q to store", sandbox.ID())
}
if err := c.sandboxNameIndex.Reserve(sb.Name, sb.ID); err != nil {
return fmt.Errorf("failed to reserve sandbox name %q: %v", sb.Name, err)
return errors.Wrapf(err, "failed to reserve sandbox name %q", sb.Name)
}
}
// Recover all containers.
containers, err := c.client.Containers(ctx, filterLabel(containerKindLabel, containerKindContainer))
if err != nil {
return fmt.Errorf("failed to list containers: %v", err)
return errors.Wrap(err, "failed to list containers")
}
for _, container := range containers {
containerDir := getContainerRootDir(c.config.RootDir, container.ID())
@ -86,26 +86,26 @@ func (c *criContainerdService) recover(ctx context.Context) error {
}
logrus.Debugf("Loaded container %+v", cntr)
if err := c.containerStore.Add(cntr); err != nil {
return fmt.Errorf("failed to add container %q to store: %v", container.ID(), err)
return errors.Wrapf(err, "failed to add container %q to store", container.ID())
}
if err := c.containerNameIndex.Reserve(cntr.Name, cntr.ID); err != nil {
return fmt.Errorf("failed to reserve container name %q: %v", cntr.Name, err)
return errors.Wrapf(err, "failed to reserve container name %q", cntr.Name)
}
}
// Recover all images.
cImages, err := c.client.ListImages(ctx)
if err != nil {
return fmt.Errorf("failed to list images: %v", err)
return errors.Wrap(err, "failed to list images")
}
images, err := loadImages(ctx, cImages, c.config.ContainerdConfig.Snapshotter)
if err != nil {
return fmt.Errorf("failed to load images: %v", err)
return errors.Wrap(err, "failed to load images")
}
for _, image := range images {
logrus.Debugf("Loaded image %+v", image)
if err := c.imageStore.Add(image); err != nil {
return fmt.Errorf("failed to add image %q to store: %v", image.ID, err)
return errors.Wrapf(err, "failed to add image %q to store", image.ID)
}
}
@ -115,12 +115,12 @@ func (c *criContainerdService) recover(ctx context.Context) error {
// Cleanup orphaned sandbox directories without corresponding containerd container.
if err := cleanupOrphanedSandboxDirs(sandboxes, filepath.Join(c.config.RootDir, "sandboxes")); err != nil {
return fmt.Errorf("failed to cleanup orphaned sandbox directories: %v", err)
return errors.Wrap(err, "failed to cleanup orphaned sandbox directories")
}
// Cleanup orphaned container directories without corresponding containerd container.
if err := cleanupOrphanedContainerDirs(containers, filepath.Join(c.config.RootDir, "containers")); err != nil {
return fmt.Errorf("failed to cleanup orphaned container directories: %v", err)
return errors.Wrap(err, "failed to cleanup orphaned container directories")
}
return nil
@ -133,15 +133,15 @@ func loadContainer(ctx context.Context, cntr containerd.Container, containerDir
// Load container metadata.
exts, err := cntr.Extensions(ctx)
if err != nil {
return container, fmt.Errorf("failed to get container extensions: %v", err)
return container, errors.Wrap(err, "failed to get container extensions")
}
ext, ok := exts[containerMetadataExtension]
if !ok {
return container, fmt.Errorf("metadata extension %q not found", containerMetadataExtension)
return container, errors.Errorf("metadata extension %q not found", containerMetadataExtension)
}
data, err := typeurl.UnmarshalAny(&ext)
if err != nil {
return container, fmt.Errorf("failed to unmarshal metadata extension %q: %v", ext, err)
return container, errors.Wrapf(err, "failed to unmarshal metadata extension %q", ext)
}
meta := data.(*containerstore.Metadata)
@ -170,7 +170,7 @@ func loadContainer(ctx context.Context, cntr containerd.Container, containerDir
return containerIO, nil
})
if err != nil && !errdefs.IsNotFound(err) {
return container, fmt.Errorf("failed to load task: %v", err)
return container, errors.Wrap(err, "failed to load task")
}
var s containerd.Status
var notFound bool
@ -183,7 +183,7 @@ func loadContainer(ctx context.Context, cntr containerd.Container, containerDir
if err != nil {
// It's still possible that task is deleted during this window.
if !errdefs.IsNotFound(err) {
return container, fmt.Errorf("failed to get task status: %v", err)
return container, errors.Wrap(err, "failed to get task status")
}
notFound = true
}
@ -200,7 +200,7 @@ func loadContainer(ctx context.Context, cntr containerd.Container, containerDir
cio.WithNewFIFOs(containerDir, meta.Config.GetTty(), meta.Config.GetStdin()),
)
if err != nil {
return container, fmt.Errorf("failed to create container io: %v", err)
return container, errors.Wrap(err, "failed to create container io")
}
case runtime.ContainerState_CONTAINER_RUNNING:
// Container was in running state, but its task has been deleted,
@ -219,17 +219,17 @@ func loadContainer(ctx context.Context, cntr containerd.Container, containerDir
// gets restarted during container start.
// Container must be in `CREATED` state.
if _, err := t.Delete(ctx, containerd.WithProcessKill); err != nil && !errdefs.IsNotFound(err) {
return container, fmt.Errorf("failed to delete task: %v", err)
return container, errors.Wrap(err, "failed to delete task")
}
if status.State() != runtime.ContainerState_CONTAINER_CREATED {
return container, fmt.Errorf("unexpected container state for created task: %q", status.State())
return container, errors.Errorf("unexpected container state for created task: %q", status.State())
}
case containerd.Running:
// Task is running. Container must be in `RUNNING` state, based on our assuption that
// "task should not be started when containerd is down".
switch status.State() {
case runtime.ContainerState_CONTAINER_EXITED:
return container, fmt.Errorf("unexpected container state for running task: %q", status.State())
return container, errors.Errorf("unexpected container state for running task: %q", status.State())
case runtime.ContainerState_CONTAINER_RUNNING:
default:
// This may happen if containerd gets restarted after task is started, but
@ -240,12 +240,12 @@ func loadContainer(ctx context.Context, cntr containerd.Container, containerDir
case containerd.Stopped:
// Task is stopped. Updata status and delete the task.
if _, err := t.Delete(ctx, containerd.WithProcessKill); err != nil && !errdefs.IsNotFound(err) {
return container, fmt.Errorf("failed to delete task: %v", err)
return container, errors.Wrap(err, "failed to delete task")
}
status.FinishedAt = s.ExitTime.UnixNano()
status.ExitCode = int32(s.ExitStatus)
default:
return container, fmt.Errorf("unexpected task status %q", s.Status)
return container, errors.Errorf("unexpected task status %q", s.Status)
}
}
opts := []containerstore.Opts{
@ -282,29 +282,29 @@ func loadSandbox(ctx context.Context, cntr containerd.Container) (sandboxstore.S
// Load sandbox metadata.
exts, err := cntr.Extensions(ctx)
if err != nil {
return sandbox, fmt.Errorf("failed to get sandbox container extensions: %v", err)
return sandbox, errors.Wrap(err, "failed to get sandbox container extensions")
}
ext, ok := exts[sandboxMetadataExtension]
if !ok {
return sandbox, fmt.Errorf("metadata extension %q not found", sandboxMetadataExtension)
return sandbox, errors.Errorf("metadata extension %q not found", sandboxMetadataExtension)
}
data, err := typeurl.UnmarshalAny(&ext)
if err != nil {
return sandbox, fmt.Errorf("failed to unmarshal metadata extension %q: %v", ext, err)
return sandbox, errors.Wrapf(err, "failed to unmarshal metadata extension %q", ext)
}
meta := data.(*sandboxstore.Metadata)
// Load sandbox created timestamp.
info, err := cntr.Info(ctx)
if err != nil {
return sandbox, fmt.Errorf("failed to get sandbox container info: %v", err)
return sandbox, errors.Wrap(err, "failed to get sandbox container info")
}
createdAt := info.CreatedAt
// Load sandbox status.
t, err := cntr.Task(ctx, nil)
if err != nil && !errdefs.IsNotFound(err) {
return sandbox, fmt.Errorf("failed to load task: %v", err)
return sandbox, errors.Wrap(err, "failed to load task")
}
var s containerd.Status
var notFound bool
@ -317,7 +317,7 @@ func loadSandbox(ctx context.Context, cntr containerd.Container) (sandboxstore.S
if err != nil {
// It's still possible that task is deleted during this window.
if !errdefs.IsNotFound(err) {
return sandbox, fmt.Errorf("failed to get task status: %v", err)
return sandbox, errors.Wrap(err, "failed to get task status")
}
notFound = true
}
@ -335,7 +335,7 @@ func loadSandbox(ctx context.Context, cntr containerd.Container) (sandboxstore.S
} else {
// Task is not running. Delete the task and set sandbox state as NOTREADY.
if _, err := t.Delete(ctx, containerd.WithProcessKill); err != nil && !errdefs.IsNotFound(err) {
return sandbox, fmt.Errorf("failed to delete task: %v", err)
return sandbox, errors.Wrap(err, "failed to delete task")
}
state = sandboxstore.StateNotReady
}
@ -359,7 +359,7 @@ func loadSandbox(ctx context.Context, cntr containerd.Container) (sandboxstore.S
netNS, err := sandboxstore.LoadNetNS(meta.NetNSPath)
if err != nil {
if err != sandboxstore.ErrClosedNetNS {
return sandbox, fmt.Errorf("failed to load netns %q: %v", meta.NetNSPath, err)
return sandbox, errors.Wrapf(err, "failed to load netns %q", meta.NetNSPath)
}
netNS = nil
}
@ -452,7 +452,7 @@ func cleanupOrphanedSandboxDirs(cntrs []containerd.Container, sandboxesRoot stri
// Cleanup orphaned sandbox directories.
dirs, err := ioutil.ReadDir(sandboxesRoot)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to read pod sandboxes directory %q: %v", sandboxesRoot, err)
return errors.Wrapf(err, "failed to read pod sandboxes directory %q", sandboxesRoot)
}
cntrsMap := make(map[string]containerd.Container)
for _, cntr := range cntrs {
@ -481,7 +481,7 @@ func cleanupOrphanedContainerDirs(cntrs []containerd.Container, containersRoot s
// Cleanup orphaned container directories.
dirs, err := ioutil.ReadDir(containersRoot)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to read containers directory %q: %v", containersRoot, err)
return errors.Wrapf(err, "failed to read containers directory %q", containersRoot)
}
cntrsMap := make(map[string]containerd.Container)
for _, cntr := range cntrs {

16
pkg/server/sandbox_portforward.go
View File

@ -18,12 +18,12 @@ package server
import (
"bytes"
"errors"
"fmt"
"io"
"os/exec"
"strings"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -37,7 +37,7 @@ func (c *criContainerdService) PortForward(ctx context.Context, r *runtime.PortF
// TODO(random-liu): Run a socat container inside the sandbox to do portforward.
sandbox, err := c.sandboxStore.Get(r.GetPodSandboxId())
if err != nil {
return nil, fmt.Errorf("failed to find sandbox %q: %v", r.GetPodSandboxId(), err)
return nil, errors.Wrapf(err, "failed to find sandbox %q", r.GetPodSandboxId())
}
if sandbox.Status.Get().State != sandboxstore.StateReady {
return nil, errors.New("sandbox container is not running")
@ -52,17 +52,17 @@ func (c *criContainerdService) PortForward(ctx context.Context, r *runtime.PortF
func (c *criContainerdService) portForward(id string, port int32, stream io.ReadWriteCloser) error {
s, err := c.sandboxStore.Get(id)
if err != nil {
return fmt.Errorf("failed to find sandbox %q in store: %v", id, err)
return errors.Wrapf(err, "failed to find sandbox %q in store", id)
}
t, err := s.Container.Task(ctrdutil.NamespacedContext(), nil)
if err != nil {
return fmt.Errorf("failed to get sandbox container task: %v", err)
return errors.Wrap(err, "failed to get sandbox container task")
}
pid := t.Pid()
socat, err := exec.LookPath("socat")
if err != nil {
return fmt.Errorf("failed to find socat: %v", err)
return errors.Wrap(err, "failed to find socat")
}
// Check following links for meaning of the options:
@ -73,7 +73,7 @@ func (c *criContainerdService) portForward(id string, port int32, stream io.Read
nsenter, err := exec.LookPath("nsenter")
if err != nil {
return fmt.Errorf("failed to find nsenter: %v", err)
return errors.Wrap(err, "failed to find nsenter")
}
logrus.Infof("Executing port forwarding command: %s %s", nsenter, strings.Join(args, " "))
@ -95,7 +95,7 @@ func (c *criContainerdService) portForward(id string, port int32, stream io.Read
// when the command (socat) exits.
in, err := cmd.StdinPipe()
if err != nil {
return fmt.Errorf("failed to create stdin pipe: %v", err)
return errors.Wrap(err, "failed to create stdin pipe")
}
go func() {
if _, err := io.Copy(in, stream); err != nil {
@ -106,7 +106,7 @@ func (c *criContainerdService) portForward(id string, port int32, stream io.Read
}()
if err := cmd.Run(); err != nil {
return fmt.Errorf("nsenter command returns error: %v, stderr: %q", err, stderr.String())
return errors.Errorf("nsenter command returns error: %v, stderr: %q", err, stderr.String())
}
logrus.Infof("Finish port forwarding for %q port %d", id, port)

19
pkg/server/sandbox_remove.go
View File

@ -17,11 +17,10 @@ limitations under the License.
package server
import (
"fmt"
"github.com/containerd/containerd"
"github.com/containerd/containerd/errdefs"
"github.com/docker/docker/pkg/system"
"github.com/pkg/errors"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -36,8 +35,8 @@ func (c *criContainerdService) RemovePodSandbox(ctx context.Context, r *runtime.
sandbox, err := c.sandboxStore.Get(r.GetPodSandboxId())
if err != nil {
if err != store.ErrNotExist {
return nil, fmt.Errorf("an error occurred when try to find sandbox %q: %v",
r.GetPodSandboxId(), err)
return nil, errors.Wrapf(err, "an error occurred when try to find sandbox %q",
r.GetPodSandboxId())
}
// Do not return error if the id doesn't exist.
log.Tracef("RemovePodSandbox called for sandbox %q that does not exist",
@ -49,12 +48,12 @@ func (c *criContainerdService) RemovePodSandbox(ctx context.Context, r *runtime.
// Return error if sandbox container is still running.
if sandbox.Status.Get().State == sandboxstore.StateReady {
return nil, fmt.Errorf("sandbox container %q is not fully stopped", id)
return nil, errors.Errorf("sandbox container %q is not fully stopped", id)
}
// Return error if sandbox network namespace is not closed yet.
if sandbox.NetNS != nil && !sandbox.NetNS.Closed() {
return nil, fmt.Errorf("sandbox network namespace %q is not fully closed", sandbox.NetNS.GetPath())
return nil, errors.Errorf("sandbox network namespace %q is not fully closed", sandbox.NetNS.GetPath())
}
// Remove all containers inside the sandbox.
@ -69,21 +68,21 @@ func (c *criContainerdService) RemovePodSandbox(ctx context.Context, r *runtime.
}
_, err = c.RemoveContainer(ctx, &runtime.RemoveContainerRequest{ContainerId: cntr.ID})
if err != nil {
return nil, fmt.Errorf("failed to remove container %q: %v", cntr.ID, err)
return nil, errors.Wrapf(err, "failed to remove container %q", cntr.ID)
}
}
// Cleanup the sandbox root directory.
sandboxRootDir := getSandboxRootDir(c.config.RootDir, id)
if err := system.EnsureRemoveAll(sandboxRootDir); err != nil {
return nil, fmt.Errorf("failed to remove sandbox root directory %q: %v",
sandboxRootDir, err)
return nil, errors.Wrapf(err, "failed to remove sandbox root directory %q",
sandboxRootDir)
}
// Delete sandbox container.
if err := sandbox.Container.Delete(ctx, containerd.WithSnapshotCleanup); err != nil {
if !errdefs.IsNotFound(err) {
return nil, fmt.Errorf("failed to delete sandbox container %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to delete sandbox container %q", id)
}
log.Tracef("Remove called for sandbox container %q that does not exist", id)
}

58
pkg/server/sandbox_run.go
View File

@ -30,6 +30,7 @@ import (
"github.com/containerd/typeurl"
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
runtimespec "github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
"golang.org/x/sys/unix"
@ -60,7 +61,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// Reserve the sandbox name to avoid concurrent `RunPodSandbox` request starting the
// same sandbox.
if err := c.sandboxNameIndex.Reserve(name, id); err != nil {
return nil, fmt.Errorf("failed to reserve sandbox name %q: %v", name, err)
return nil, errors.Wrapf(err, "failed to reserve sandbox name %q", name)
}
defer func() {
// Release the name if the function returns with an error.
@ -84,7 +85,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// Ensure sandbox container image snapshot.
image, err := c.ensureImageExists(ctx, c.config.SandboxImage)
if err != nil {
return nil, fmt.Errorf("failed to get sandbox image %q: %v", c.config.SandboxImage, err)
return nil, errors.Wrapf(err, "failed to get sandbox image %q", c.config.SandboxImage)
}
securityContext := config.GetLinux().GetSecurityContext()
//Create Network Namespace if it is not in host network
@ -96,7 +97,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// be used.
sandbox.NetNS, err = sandboxstore.NewNetNS()
if err != nil {
return nil, fmt.Errorf("failed to create network namespace for sandbox %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to create network namespace for sandbox %q", id)
}
sandbox.NetNSPath = sandbox.NetNS.GetPath()
defer func() {
@ -117,7 +118,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// SandboxStatus request.
sandbox.IP, err = c.setupPod(id, sandbox.NetNSPath, config)
if err != nil {
return nil, fmt.Errorf("failed to setup network for sandbox %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to setup network for sandbox %q", id)
}
defer func() {
if retErr != nil {
@ -132,7 +133,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// Create sandbox container.
spec, err := c.generateSandboxContainerSpec(id, config, &image.ImageSpec.Config, sandbox.NetNSPath)
if err != nil {
return nil, fmt.Errorf("failed to generate sandbox container spec: %v", err)
return nil, errors.Wrap(err, "failed to generate sandbox container spec")
}
logrus.Debugf("Sandbox container spec: %+v", spec)
@ -146,7 +147,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
securityContext.GetPrivileged(),
c.seccompEnabled)
if err != nil {
return nil, fmt.Errorf("failed to generate seccomp spec opts: %v", err)
return nil, errors.Wrap(err, "failed to generate seccomp spec opts")
}
if seccompSpecOpts != nil {
specOpts = append(specOpts, seccompSpecOpts)
@ -169,7 +170,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
container, err := c.client.NewContainer(ctx, id, opts...)
if err != nil {
return nil, fmt.Errorf("failed to create containerd container: %v", err)
return nil, errors.Wrap(err, "failed to create containerd container")
}
defer func() {
if retErr != nil {
@ -184,8 +185,8 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// Create sandbox container root directory.
sandboxRootDir := getSandboxRootDir(c.config.RootDir, id)
if err := c.os.MkdirAll(sandboxRootDir, 0755); err != nil {
return nil, fmt.Errorf("failed to create sandbox root directory %q: %v",
sandboxRootDir, err)
return nil, errors.Wrapf(err, "failed to create sandbox root directory %q",
sandboxRootDir)
}
defer func() {
if retErr != nil {
@ -199,7 +200,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// Setup sandbox /dev/shm, /etc/hosts and /etc/resolv.conf.
if err = c.setupSandboxFiles(sandboxRootDir, config); err != nil {
return nil, fmt.Errorf("failed to setup sandbox files: %v", err)
return nil, errors.Wrapf(err, "failed to setup sandbox files")
}
defer func() {
if retErr != nil {
@ -213,19 +214,19 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// Update sandbox created timestamp.
info, err := container.Info(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get sandbox container info: %v", err)
return nil, errors.Wrap(err, "failed to get sandbox container info")
}
if err := sandbox.Status.Update(func(status sandboxstore.Status) (sandboxstore.Status, error) {
status.CreatedAt = info.CreatedAt
return status, nil
}); err != nil {
return nil, fmt.Errorf("failed to update sandbox created timestamp: %v", err)
return nil, errors.Wrap(err, "failed to update sandbox created timestamp")
}
// Add sandbox into sandbox store in UNKNOWN state.
sandbox.Container = container
if err := c.sandboxStore.Add(sandbox); err != nil {
return nil, fmt.Errorf("failed to add sandbox %+v into store: %v", sandbox, err)
return nil, errors.Wrapf(err, "failed to add sandbox %+v into store", sandbox)
}
defer func() {
// Delete sandbox from sandbox store if there is an error.
@ -266,7 +267,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
// We don't need stdio for sandbox container.
task, err := container.NewTask(ctx, containerdio.NullIO)
if err != nil {
return status, fmt.Errorf("failed to create containerd task: %v", err)
return status, errors.Wrap(err, "failed to create containerd task")
}
defer func() {
if retErr != nil {
@ -281,8 +282,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
}()
if err := task.Start(ctx); err != nil {
return status, fmt.Errorf("failed to start sandbox container task %q: %v",
id, err)
return status, errors.Wrapf(err, "failed to start sandbox container task %q", id)
}
// Set the pod sandbox as ready after successfully start sandbox container.
@ -290,7 +290,7 @@ func (c *criContainerdService) RunPodSandbox(ctx context.Context, r *runtime.Run
status.State = sandboxstore.StateReady
return status, nil
}); err != nil {
return nil, fmt.Errorf("failed to start sandbox container: %v", err)
return nil, errors.Wrap(err, "failed to start sandbox container")
}
return &runtime.RunPodSandboxResponse{PodSandboxId: id}, nil
@ -317,7 +317,7 @@ func (c *criContainerdService) generateSandboxContainerSpec(id string, config *r
if len(imageConfig.Entrypoint) == 0 {
// Pause image must have entrypoint.
return nil, fmt.Errorf("invalid empty entrypoint in image config %+v", imageConfig)
return nil, errors.Errorf("invalid empty entrypoint in image config %+v", imageConfig)
}
// Set process commands.
g.SetProcessArgs(append(imageConfig.Entrypoint, imageConfig.Cmd...))
@ -362,7 +362,7 @@ func (c *criContainerdService) generateSandboxContainerSpec(id string, config *r
selinuxOpt := securityContext.GetSelinuxOptions()
processLabel, mountLabel, err := initSelinuxOpts(selinuxOpt)
if err != nil {
return nil, fmt.Errorf("failed to init selinux options %+v: %v", securityContext.GetSelinuxOptions(), err)
return nil, errors.Wrapf(err, "failed to init selinux options %+v", securityContext.GetSelinuxOptions())
}
g.SetProcessSelinuxLabel(processLabel)
g.SetLinuxMountLabel(mountLabel)
@ -395,7 +395,7 @@ func (c *criContainerdService) setupSandboxFiles(rootDir string, config *runtime
// TODO(random-liu): Consider whether we should maintain /etc/hosts and /etc/resolv.conf in kubelet.
sandboxEtcHosts := getSandboxHosts(rootDir)
if err := c.os.CopyFile(etcHosts, sandboxEtcHosts, 0644); err != nil {
return fmt.Errorf("failed to generate sandbox hosts file %q: %v", sandboxEtcHosts, err)
return errors.Wrapf(err, "failed to generate sandbox hosts file %q", sandboxEtcHosts)
}
// Set DNS options. Maintain a resolv.conf for the sandbox.
@ -404,7 +404,7 @@ func (c *criContainerdService) setupSandboxFiles(rootDir string, config *runtime
if dnsConfig := config.GetDnsConfig(); dnsConfig != nil {
resolvContent, err = parseDNSOptions(dnsConfig.Servers, dnsConfig.Searches, dnsConfig.Options)
if err != nil {
return fmt.Errorf("failed to parse sandbox DNSConfig %+v: %v", dnsConfig, err)
return errors.Wrapf(err, "failed to parse sandbox DNSConfig %+v", dnsConfig)
}
}
resolvPath := getResolvPath(rootDir)
@ -412,28 +412,28 @@ func (c *criContainerdService) setupSandboxFiles(rootDir string, config *runtime
// copy host's resolv.conf to resolvPath
err = c.os.CopyFile(resolvConfPath, resolvPath, 0644)
if err != nil {
return fmt.Errorf("failed to copy host's resolv.conf to %q: %v", resolvPath, err)
return errors.Wrapf(err, "failed to copy host's resolv.conf to %q", resolvPath)
}
} else {
err = c.os.WriteFile(resolvPath, []byte(resolvContent), 0644)
if err != nil {
return fmt.Errorf("failed to write resolv content to %q: %v", resolvPath, err)
return errors.Wrapf(err, "failed to write resolv content to %q", resolvPath)
}
}
// Setup sandbox /dev/shm.
if config.GetLinux().GetSecurityContext().GetNamespaceOptions().GetIpc() == runtime.NamespaceMode_NODE {
if _, err := c.os.Stat(devShm); err != nil {
return fmt.Errorf("host %q is not available for host ipc: %v", devShm, err)
return errors.Wrapf(err, "host %q is not available for host ipc", devShm)
}
} else {
sandboxDevShm := getSandboxDevShm(rootDir)
if err := c.os.MkdirAll(sandboxDevShm, 0700); err != nil {
return fmt.Errorf("failed to create sandbox shm: %v", err)
return errors.Wrap(err, "failed to create sandbox shm")
}
shmproperty := fmt.Sprintf("mode=1777,size=%d", defaultShmSize)
if err := c.os.Mount("shm", sandboxDevShm, "tmpfs", uintptr(unix.MS_NOEXEC|unix.MS_NOSUID|unix.MS_NODEV), shmproperty); err != nil {
return fmt.Errorf("failed to mount sandbox shm: %v", err)
return errors.Wrap(err, "failed to mount sandbox shm")
}
}
@ -446,7 +446,7 @@ func parseDNSOptions(servers, searches, options []string) (string, error) {
resolvContent := ""
if len(searches) > maxDNSSearches {
return "", fmt.Errorf("DNSOption.Searches has more than 6 domains")
return "", errors.New("DNSOption.Searches has more than 6 domains")
}
if len(searches) > 0 {
@ -480,7 +480,7 @@ func (c *criContainerdService) unmountSandboxFiles(rootDir string, config *runti
// setupPod setups up the network for a pod
func (c *criContainerdService) setupPod(id string, path string, config *runtime.PodSandboxConfig) (string, error) {
if c.netPlugin == nil {
return "", fmt.Errorf("cni config not intialized")
return "", errors.New("cni config not intialized")
}
labels := getPodCNILabels(id, config)
@ -499,7 +499,7 @@ func (c *criContainerdService) setupPod(id string, path string, config *runtime.
if err := c.teardownPod(id, path, config); err != nil {
logrus.WithError(err).Errorf("Failed to destroy network for sandbox %q", id)
}
return "", fmt.Errorf("failed to find network info for sandbox %q", id)
return "", errors.Errorf("failed to find network info for sandbox %q", id)
}
// toCNIPortMappings converts CRI port mappings to CNI.

16
pkg/server/sandbox_status.go
View File

@ -18,11 +18,11 @@ package server
import (
"encoding/json"
"fmt"
"github.com/containerd/containerd"
"github.com/containerd/containerd/errdefs"
runtimespec "github.com/opencontainers/runtime-spec/specs-go"
"github.com/pkg/errors"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -33,7 +33,7 @@ import (
func (c *criContainerdService) PodSandboxStatus(ctx context.Context, r *runtime.PodSandboxStatusRequest) (*runtime.PodSandboxStatusResponse, error) {
sandbox, err := c.sandboxStore.Get(r.GetPodSandboxId())
if err != nil {
return nil, fmt.Errorf("an error occurred when try to find sandbox: %v", err)
return nil, errors.Wrap(err, "an error occurred when try to find sandbox")
}
ip := c.getIP(sandbox)
@ -45,7 +45,7 @@ func (c *criContainerdService) PodSandboxStatus(ctx context.Context, r *runtime.
// Generate verbose information.
info, err := toCRISandboxInfo(ctx, sandbox)
if err != nil {
return nil, fmt.Errorf("failed to get verbose sandbox container info: %v", err)
return nil, errors.Wrap(err, "failed to get verbose sandbox container info")
}
return &runtime.PodSandboxStatusResponse{
@ -116,14 +116,14 @@ func toCRISandboxInfo(ctx context.Context, sandbox sandboxstore.Sandbox) (map[st
container := sandbox.Container
task, err := container.Task(ctx, nil)
if err != nil && !errdefs.IsNotFound(err) {
return nil, fmt.Errorf("failed to get sandbox container task: %v", err)
return nil, errors.Wrap(err, "failed to get sandbox container task")
}
var processStatus containerd.ProcessStatus
if task != nil {
taskStatus, err := task.Status(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get task status: %v", err)
return nil, errors.Wrap(err, "failed to get task status")
}
processStatus = taskStatus.Status
@ -148,13 +148,13 @@ func toCRISandboxInfo(ctx context.Context, sandbox sandboxstore.Sandbox) (map[st
spec, err := container.Spec(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get sandbox container runtime spec: %v", err)
return nil, errors.Wrap(err, "failed to get sandbox container runtime spec")
}
si.RuntimeSpec = spec
ctrInfo, err := container.Info(ctx)
if err != nil {
return nil, fmt.Errorf("failed to get sandbox container info: %v", err)
return nil, errors.Wrap(err, "failed to get sandbox container info")
}
// Do not use config.SandboxImage because the configuration might
// be changed during restart. It may not reflect the actual image
@ -165,7 +165,7 @@ func toCRISandboxInfo(ctx context.Context, sandbox sandboxstore.Sandbox) (map[st
infoBytes, err := json.Marshal(si)
if err != nil {
return nil, fmt.Errorf("failed to marshal info %v: %v", si, err)
return nil, errors.Wrapf(err, "failed to marshal info %v", si)
}
return map[string]string{
"info": string(infoBytes),

28
pkg/server/sandbox_stop.go
View File

@ -17,13 +17,13 @@ limitations under the License.
package server
import (
"fmt"
"os"
"time"
"github.com/containerd/containerd"
"github.com/containerd/containerd/errdefs"
cni "github.com/containerd/go-cni"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/net/context"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -36,8 +36,8 @@ import (
func (c *criContainerdService) StopPodSandbox(ctx context.Context, r *runtime.StopPodSandboxRequest) (*runtime.StopPodSandboxResponse, error) {
sandbox, err := c.sandboxStore.Get(r.GetPodSandboxId())
if err != nil {
return nil, fmt.Errorf("an error occurred when try to find sandbox %q: %v",
r.GetPodSandboxId(), err)
return nil, errors.Wrapf(err, "an error occurred when try to find sandbox %q",
r.GetPodSandboxId())
}
// Use the full sandbox id.
id := sandbox.ID
@ -54,7 +54,7 @@ func (c *criContainerdService) StopPodSandbox(ctx context.Context, r *runtime.St
// Forcibly stop the container. Do not use `StopContainer`, because it introduces a race
// if a container is removed after list.
if err = c.stopContainer(ctx, container, 0); err != nil {
return nil, fmt.Errorf("failed to stop container %q: %v", container.ID, err)
return nil, errors.Wrapf(err, "failed to stop container %q", container.ID)
}
}
@ -62,11 +62,11 @@ func (c *criContainerdService) StopPodSandbox(ctx context.Context, r *runtime.St
if sandbox.NetNSPath != "" && sandbox.NetNS != nil {
if _, err := os.Stat(sandbox.NetNSPath); err != nil {
if !os.IsNotExist(err) {
return nil, fmt.Errorf("failed to stat network namespace path %s :%v", sandbox.NetNSPath, err)
return nil, errors.Wrapf(err, "failed to stat network namespace path %s", sandbox.NetNSPath)
}
} else {
if teardownErr := c.teardownPod(id, sandbox.NetNSPath, sandbox.Config); teardownErr != nil {
return nil, fmt.Errorf("failed to destroy network for sandbox %q: %v", id, teardownErr)
return nil, errors.Wrapf(teardownErr, "failed to destroy network for sandbox %q", id)
}
}
/*TODO:It is still possible that containerd crashes after we teardown the network, but before we remove the network namespace.
@ -75,7 +75,7 @@ func (c *criContainerdService) StopPodSandbox(ctx context.Context, r *runtime.St
//Close the sandbox network namespace if it was created
if err = sandbox.NetNS.Remove(); err != nil {
return nil, fmt.Errorf("failed to remove network namespace for sandbox %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to remove network namespace for sandbox %q", id)
}
}
@ -83,13 +83,13 @@ func (c *criContainerdService) StopPodSandbox(ctx context.Context, r *runtime.St
sandboxRoot := getSandboxRootDir(c.config.RootDir, id)
if err := c.unmountSandboxFiles(sandboxRoot, sandbox.Config); err != nil {
return nil, fmt.Errorf("failed to unmount sandbox files in %q: %v", sandboxRoot, err)
return nil, errors.Wrapf(err, "failed to unmount sandbox files in %q", sandboxRoot)
}
// Only stop sandbox container when it's running.
if sandbox.Status.Get().State == sandboxstore.StateReady {
if err := c.stopSandboxContainer(ctx, sandbox); err != nil {
return nil, fmt.Errorf("failed to stop sandbox container %q: %v", id, err)
return nil, errors.Wrapf(err, "failed to stop sandbox container %q", id)
}
}
return &runtime.StopPodSandboxResponse{}, nil
@ -103,13 +103,13 @@ func (c *criContainerdService) stopSandboxContainer(ctx context.Context, sandbox
if errdefs.IsNotFound(err) {
return nil
}
return fmt.Errorf("failed to get sandbox container: %v", err)
return errors.Wrap(err, "failed to get sandbox container")
}
// Delete the sandbox container from containerd.
_, err = task.Delete(ctx, containerd.WithProcessKill)
if err != nil && !errdefs.IsNotFound(err) {
return fmt.Errorf("failed to delete sandbox container: %v", err)
return errors.Wrap(err, "failed to delete sandbox container")
}
return c.waitSandboxStop(ctx, sandbox, killContainerTimeout)
@ -121,9 +121,9 @@ func (c *criContainerdService) waitSandboxStop(ctx context.Context, sandbox sand
defer timeoutTimer.Stop()
select {
case <-ctx.Done():
return fmt.Errorf("wait sandbox container %q is cancelled", sandbox.ID)
return errors.Errorf("wait sandbox container %q is cancelled", sandbox.ID)
case <-timeoutTimer.C:
return fmt.Errorf("wait sandbox container %q stop timeout", sandbox.ID)
return errors.Errorf("wait sandbox container %q stop timeout", sandbox.ID)
case <-sandbox.Stopped():
return nil
}
@ -132,7 +132,7 @@ func (c *criContainerdService) waitSandboxStop(ctx context.Context, sandbox sand
// teardownPod removes the network from the pod
func (c *criContainerdService) teardownPod(id string, path string, config *runtime.PodSandboxConfig) error {
if c.netPlugin == nil {
return fmt.Errorf("cni config not intialized")
return errors.New("cni config not intialized")
}
labels := getPodCNILabels(id, config)

13
pkg/server/service.go
View File

@ -28,6 +28,7 @@ import (
runcapparmor "github.com/opencontainers/runc/libcontainer/apparmor"
runcseccomp "github.com/opencontainers/runc/libcontainer/seccomp"
"github.com/opencontainers/selinux/go-selinux"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"google.golang.org/grpc"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
@ -137,7 +138,7 @@ func NewCRIContainerdService(config criconfig.Config, client *containerd.Client)
cni.WithPluginConfDir(config.NetworkPluginConfDir),
cni.WithPluginDir([]string{config.NetworkPluginBinDir}))
if err != nil {
return nil, fmt.Errorf("failed to initialize cni: %v", err)
return nil, errors.Wrap(err, "failed to initialize cni")
}
// Try to load the config if it exists. Just log the error if load fails
@ -148,7 +149,7 @@ func NewCRIContainerdService(config criconfig.Config, client *containerd.Client)
// prepare streaming server
c.streamServer, err = newStreamServer(c, config.StreamServerAddress, config.StreamServerPort)
if err != nil {
return nil, fmt.Errorf("failed to create stream server: %v", err)
return nil, errors.Wrap(err, "failed to create stream server")
}
c.eventMonitor = newEventMonitor(c.containerStore, c.sandboxStore)
@ -173,14 +174,14 @@ func (c *criContainerdService) Run() error {
logrus.Infof("Start recovering state")
if err := c.recover(ctrdutil.NamespacedContext()); err != nil {
return fmt.Errorf("failed to recover state: %v", err)
return errors.Wrap(err, "failed to recover state")
}
// Start event handler.
logrus.Info("Start event monitor")
eventMonitorCloseCh, err := c.eventMonitor.start()
if err != nil {
return fmt.Errorf("failed to start event monitor: %v", err)
return errors.Wrap(err, "failed to start event monitor")
}
// Start snapshot stats syncer, it doesn't need to be stopped.
@ -211,7 +212,7 @@ func (c *criContainerdService) Run() error {
case <-streamServerCloseCh:
}
if err := c.Close(); err != nil {
return fmt.Errorf("failed to stop cri service: %v", err)
return errors.Wrap(err, "failed to stop cri service")
}
<-eventMonitorCloseCh
@ -240,7 +241,7 @@ func (c *criContainerdService) Close() error {
// TODO(random-liu): Make event monitor stop synchronous.
c.eventMonitor.stop()
if err := c.streamServer.Stop(); err != nil {
return fmt.Errorf("failed to stop stream server: %v", err)
return errors.Wrap(err, "failed to stop stream server")
}
return nil
}

4
pkg/server/snapshots.go
View File

@ -18,11 +18,11 @@ package server
import (
"context"
"fmt"
"time"
"github.com/containerd/containerd/errdefs"
snapshot "github.com/containerd/containerd/snapshots"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
ctrdutil "github.com/containerd/cri/pkg/containerd/util"
@ -80,7 +80,7 @@ func (s *snapshotsSyncer) sync() error {
snapshots = append(snapshots, info)
return nil
}); err != nil {
return fmt.Errorf("walk all snapshots failed: %v", err)
return errors.Wrap(err, "walk all snapshots failed")
}
for _, info := range snapshots {
sn, err := s.store.Get(info.Name)

10
pkg/server/streaming.go
View File

@ -17,11 +17,11 @@ limitations under the License.
package server
import (
"fmt"
"io"
"math"
"net"
"github.com/pkg/errors"
k8snet "k8s.io/apimachinery/pkg/util/net"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/client-go/tools/remotecommand"
@ -35,7 +35,7 @@ func newStreamServer(c *criContainerdService, addr, port string) (streaming.Serv
if addr == "" {
a, err := k8snet.ChooseBindAddress(nil)
if err != nil {
return nil, fmt.Errorf("failed to get stream server address: %v", err)
return nil, errors.Wrap(err, "failed to get stream server address")
}
addr = a.String()
}
@ -66,13 +66,13 @@ func (s *streamRuntime) Exec(containerID string, cmd []string, stdin io.Reader,
resize: resize,
})
if err != nil {
return fmt.Errorf("failed to exec in container: %v", err)
return errors.Wrap(err, "failed to exec in container")
}
if *exitCode == 0 {
return nil
}
return &exec.CodeExitError{
Err: fmt.Errorf("error executing command %v, exit code %d", cmd, *exitCode),
Err: errors.Errorf("error executing command %v, exit code %d", cmd, *exitCode),
Code: int(*exitCode),
}
}
@ -84,7 +84,7 @@ func (s *streamRuntime) Attach(containerID string, in io.Reader, out, err io.Wri
func (s *streamRuntime) PortForward(podSandboxID string, port int32, stream io.ReadWriteCloser) error {
if port <= 0 || port > math.MaxUint16 {
return fmt.Errorf("invalid port %d", port)
return errors.Errorf("invalid port %d", port)
}
return s.c.portForward(podSandboxID, port, stream)
}

4
pkg/store/container/metadata.go
View File

@ -18,8 +18,8 @@ package container
import (
"encoding/json"
"fmt"
"github.com/pkg/errors"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
)
@ -80,5 +80,5 @@ func (c *Metadata) UnmarshalJSON(data []byte) error {
*c = Metadata(versioned.Metadata)
return nil
}
return fmt.Errorf("unsupported version: %q", versioned.Version)
return errors.Errorf("unsupported version: %q", versioned.Version)
}

16
pkg/store/container/status.go
View File

@ -18,13 +18,13 @@ package container
import (
"encoding/json"
"fmt"
"io/ioutil"
"os"
"path/filepath"
"sync"
"github.com/docker/docker/pkg/ioutils"
"github.com/pkg/errors"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
)
@ -95,7 +95,7 @@ func (s *Status) decode(data []byte) error {
*s = versioned.Status
return nil
}
return fmt.Errorf("unsupported version")
return errors.New("unsupported version")
}
// UpdateFunc is function used to update the container status. If there
@ -125,11 +125,11 @@ type StatusStorage interface {
func StoreStatus(root, id string, status Status) (StatusStorage, error) {
data, err := status.encode()
if err != nil {
return nil, fmt.Errorf("failed to encode status: %v", err)
return nil, errors.Wrap(err, "failed to encode status")
}
path := filepath.Join(root, "status")
if err := ioutils.AtomicWriteFile(path, data, 0600); err != nil {
return nil, fmt.Errorf("failed to checkpoint status to %q: %v", path, err)
return nil, errors.Wrapf(err, "failed to checkpoint status to %q", path)
}
return &statusStorage{
path: path,
@ -143,11 +143,11 @@ func LoadStatus(root, id string) (Status, error) {
path := filepath.Join(root, "status")
data, err := ioutil.ReadFile(path)
if err != nil {
return Status{}, fmt.Errorf("failed to read status from %q: %v", path, err)
return Status{}, errors.Wrapf(err, "failed to read status from %q", path)
}
var status Status
if err := status.decode(data); err != nil {
return Status{}, fmt.Errorf("failed to decode status %q: %v", data, err)
return Status{}, errors.Wrapf(err, "failed to decode status %q", data)
}
return status, nil
}
@ -175,10 +175,10 @@ func (s *statusStorage) UpdateSync(u UpdateFunc) error {
}
data, err := newStatus.encode()
if err != nil {
return fmt.Errorf("failed to encode status: %v", err)
return errors.Wrap(err, "failed to encode status")
}
if err := ioutils.AtomicWriteFile(s.path, data, 0600); err != nil {
return fmt.Errorf("failed to checkpoint status to %q: %v", s.path, err)
return errors.Wrapf(err, "failed to checkpoint status to %q", s.path)
}
s.status = newStatus
return nil

4
pkg/store/sandbox/metadata.go
View File

@ -18,8 +18,8 @@ package sandbox
import (
"encoding/json"
"fmt"
"github.com/pkg/errors"
runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
)
@ -76,5 +76,5 @@ func (c *Metadata) UnmarshalJSON(data []byte) error {
*c = Metadata(versioned.Metadata)
return nil
}
return fmt.Errorf("unsupported version: %q", versioned.Version)
return errors.Errorf("unsupported version: %q", versioned.Version)
}

19
pkg/store/sandbox/netns.go
View File

@ -17,14 +17,13 @@ limitations under the License.
package sandbox
import (
"errors"
"fmt"
"os"
"sync"
cnins "github.com/containernetworking/plugins/pkg/ns"
"github.com/docker/docker/pkg/mount"
"github.com/docker/docker/pkg/symlink"
"github.com/pkg/errors"
"golang.org/x/sys/unix"
)
@ -43,7 +42,7 @@ type NetNS struct {
func NewNetNS() (*NetNS, error) {
netns, err := cnins.NewNS()
if err != nil {
return nil, fmt.Errorf("failed to setup network namespace %v", err)
return nil, errors.Wrap(err, "failed to setup network namespace")
}
n := new(NetNS)
n.ns = netns
@ -63,7 +62,7 @@ func LoadNetNS(path string) (*NetNS, error) {
os.RemoveAll(path) // nolint: errcheck
return nil, ErrClosedNetNS
}
return nil, fmt.Errorf("failed to load network namespace %v", err)
return nil, errors.Wrap(err, "failed to load network namespace")
}
return &NetNS{ns: ns, restored: true}, nil
}
@ -76,7 +75,7 @@ func (n *NetNS) Remove() error {
if !n.closed {
err := n.ns.Close()
if err != nil {
return fmt.Errorf("failed to close network namespace: %v", err)
return errors.Wrap(err, "failed to close network namespace")
}
n.closed = true
}
@ -88,24 +87,24 @@ func (n *NetNS) Remove() error {
if os.IsNotExist(err) {
return nil
}
return fmt.Errorf("failed to stat netns: %v", err)
return errors.Wrap(err, "failed to stat netns")
}
path, err := symlink.FollowSymlinkInScope(path, "/")
if err != nil {
return fmt.Errorf("failed to follow symlink: %v", err)
return errors.Wrap(err, "failed to follow symlink")
}
mounted, err := mount.Mounted(path)
if err != nil {
return fmt.Errorf("failed to check netns mounted: %v", err)
return errors.Wrap(err, "failed to check netns mounted")
}
if mounted {
err := unix.Unmount(path, unix.MNT_DETACH)
if err != nil && !os.IsNotExist(err) {
return fmt.Errorf("failed to umount netns: %v", err)
return errors.Wrap(err, "failed to umount netns")
}
}
if err := os.RemoveAll(path); err != nil {
return fmt.Errorf("failed to remove netns: %v", err)
return errors.Wrap(err, "failed to remove netns")
}
n.restored = false
}

11
pkg/util/deep_copy.go
View File

@ -18,24 +18,25 @@ package util
import (
"encoding/json"
"fmt"
"github.com/pkg/errors"
)
// DeepCopy makes a deep copy from src into dst.
func DeepCopy(dst interface{}, src interface{}) error {
if dst == nil {
return fmt.Errorf("dst cannot be nil")
return errors.New("dst cannot be nil")
}
if src == nil {
return fmt.Errorf("src cannot be nil")
return errors.New("src cannot be nil")
}
bytes, err := json.Marshal(src)
if err != nil {
return fmt.Errorf("unable to marshal src: %s", err)
return errors.Wrap(err, "unable to marshal src")
}
err = json.Unmarshal(bytes, dst)
if err != nil {
return fmt.Errorf("unable to unmarshal into dst: %s", err)
return errors.Wrap(err, "unable to unmarshal into dst")
}
return nil
}