Add tracing spans in CRI image service and pull.go
Signed-off-by: Swagat Bora <sbora@amazon.com> Add spans around image unpack operations Use image.ref to denote image name and image.id for the image config digest Add top-level spand and record errors in the CRI instrumentation service
This commit is contained in:
@@ -95,6 +95,11 @@ const (
|
||||
|
||||
// runtimeRunhcsV1 is the runtime type for runhcs.
|
||||
runtimeRunhcsV1 = "io.containerd.runhcs.v1"
|
||||
|
||||
// name prefix for CRI sbserver specific spans
|
||||
criSbServerSpanPrefix = "pkg.cri.sbserver"
|
||||
|
||||
spanDelimiter = "."
|
||||
)
|
||||
|
||||
// makeSandboxName generates sandbox name from sandbox metadata. The name
|
||||
@@ -501,3 +506,10 @@ func copyResourcesToStatus(spec *runtimespec.Spec, status containerstore.Status)
|
||||
}
|
||||
return status
|
||||
}
|
||||
|
||||
func makeSpanName(funcName string) string {
|
||||
return strings.Join([]string{
|
||||
criSbServerSpanPrefix,
|
||||
funcName,
|
||||
}, spanDelimiter)
|
||||
}
|
||||
|
||||
@@ -36,6 +36,7 @@ import (
|
||||
"github.com/containerd/imgcrypt"
|
||||
"github.com/containerd/imgcrypt/images/encryption"
|
||||
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
|
||||
"github.com/containerd/containerd"
|
||||
@@ -48,6 +49,7 @@ import (
|
||||
distribution "github.com/containerd/containerd/reference/docker"
|
||||
"github.com/containerd/containerd/remotes/docker"
|
||||
"github.com/containerd/containerd/remotes/docker/config"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
)
|
||||
|
||||
// For image management:
|
||||
@@ -93,6 +95,7 @@ import (
|
||||
|
||||
// PullImage pulls an image with authentication config.
|
||||
func (c *criService) PullImage(ctx context.Context, r *runtime.PullImageRequest) (*runtime.PullImageResponse, error) {
|
||||
span := tracing.CurrentSpan(ctx)
|
||||
imageRef := r.GetImage().GetImage()
|
||||
namedRef, err := distribution.ParseDockerRef(imageRef)
|
||||
if err != nil {
|
||||
@@ -133,6 +136,10 @@ func (c *criService) PullImage(ctx context.Context, r *runtime.PullImageRequest)
|
||||
return nil, err
|
||||
}
|
||||
log.G(ctx).Debugf("PullImage %q with snapshotter %s", ref, snapshotter)
|
||||
span.SetAttributes(
|
||||
attribute.String("image.ref", ref),
|
||||
attribute.String("snapshotter.name", snapshotter),
|
||||
)
|
||||
|
||||
pullOpts := []containerd.RemoteOpt{
|
||||
containerd.WithSchema1Conversion, //nolint:staticcheck // Ignore SA1019. Need to keep deprecated package for compatibility.
|
||||
@@ -165,6 +172,7 @@ func (c *criService) PullImage(ctx context.Context, r *runtime.PullImageRequest)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err)
|
||||
}
|
||||
span.AddEvent("Pull and unpack image complete")
|
||||
|
||||
configDesc, err := image.Config(ctx)
|
||||
if err != nil {
|
||||
|
||||
@@ -22,6 +22,8 @@ import (
|
||||
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/images"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
)
|
||||
@@ -33,15 +35,17 @@ import (
|
||||
// Remove the whole image no matter the it's image id or reference. This is the
|
||||
// semantic defined in CRI now.
|
||||
func (c *criService) RemoveImage(ctx context.Context, r *runtime.RemoveImageRequest) (*runtime.RemoveImageResponse, error) {
|
||||
span := tracing.CurrentSpan(ctx)
|
||||
image, err := c.localResolve(r.GetImage().GetImage())
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
span.AddEvent(err.Error())
|
||||
// return empty without error when image not found.
|
||||
return &runtime.RemoveImageResponse{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("can not resolve %q locally: %w", r.GetImage().GetImage(), err)
|
||||
}
|
||||
|
||||
span.SetAttributes(attribute.String("image.id", image.ID))
|
||||
// Remove all image references.
|
||||
for i, ref := range image.References {
|
||||
var opts []images.DeleteOpt
|
||||
|
||||
@@ -24,6 +24,8 @@ import (
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/log"
|
||||
imagestore "github.com/containerd/containerd/pkg/cri/store/image"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
|
||||
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
@@ -33,14 +35,17 @@ import (
|
||||
// TODO(random-liu): We should change CRI to distinguish image id and image spec. (See
|
||||
// kubernetes/kubernetes#46255)
|
||||
func (c *criService) ImageStatus(ctx context.Context, r *runtime.ImageStatusRequest) (*runtime.ImageStatusResponse, error) {
|
||||
span := tracing.CurrentSpan(ctx)
|
||||
image, err := c.localResolve(r.GetImage().GetImage())
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
span.AddEvent(err.Error())
|
||||
// return empty without error when image not found.
|
||||
return &runtime.ImageStatusResponse{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("can not resolve %q locally: %w", r.GetImage().GetImage(), err)
|
||||
}
|
||||
span.SetAttributes(attribute.String("image.id", image.ID))
|
||||
// TODO(random-liu): [P0] Make sure corresponding snapshot exists. What if snapshot
|
||||
// doesn't exist?
|
||||
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/log"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
runtime_alpha "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
|
||||
|
||||
@@ -931,6 +932,8 @@ func (in *instrumentedService) PullImage(ctx context.Context, r *runtime.PullIma
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("PullImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("PullImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -939,6 +942,7 @@ func (in *instrumentedService) PullImage(ctx context.Context, r *runtime.PullIma
|
||||
log.G(ctx).Infof("PullImage %q returns image reference %q",
|
||||
r.GetImage().GetImage(), res.GetImageRef())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.PullImage(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -948,6 +952,8 @@ func (in *instrumentedAlphaService) PullImage(ctx context.Context, r *runtime_al
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("PullImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("PullImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -956,6 +962,7 @@ func (in *instrumentedAlphaService) PullImage(ctx context.Context, r *runtime_al
|
||||
log.G(ctx).Infof("PullImage %q returns image reference %q",
|
||||
r.GetImage().GetImage(), res.GetImageRef())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.PullImageRequest
|
||||
@@ -986,6 +993,8 @@ func (in *instrumentedService) ListImages(ctx context.Context, r *runtime.ListIm
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ListImages"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ListImages with filter %+v", r.GetFilter())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -994,6 +1003,7 @@ func (in *instrumentedService) ListImages(ctx context.Context, r *runtime.ListIm
|
||||
log.G(ctx).Tracef("ListImages with filter %+v returns image list %+v",
|
||||
r.GetFilter(), res.GetImages())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.ListImages(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1003,6 +1013,8 @@ func (in *instrumentedAlphaService) ListImages(ctx context.Context, r *runtime_a
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ListImages"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ListImages with filter %+v", r.GetFilter())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1011,6 +1023,7 @@ func (in *instrumentedAlphaService) ListImages(ctx context.Context, r *runtime_a
|
||||
log.G(ctx).Tracef("ListImages with filter %+v returns image list %+v",
|
||||
r.GetFilter(), res.GetImages())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.ListImagesRequest
|
||||
@@ -1041,6 +1054,8 @@ func (in *instrumentedService) ImageStatus(ctx context.Context, r *runtime.Image
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageStatus"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ImageStatus for %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1049,6 +1064,7 @@ func (in *instrumentedService) ImageStatus(ctx context.Context, r *runtime.Image
|
||||
log.G(ctx).Tracef("ImageStatus for %q returns image status %+v",
|
||||
r.GetImage().GetImage(), res.GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.ImageStatus(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1058,6 +1074,8 @@ func (in *instrumentedAlphaService) ImageStatus(ctx context.Context, r *runtime_
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageStatus"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ImageStatus for %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1066,6 +1084,7 @@ func (in *instrumentedAlphaService) ImageStatus(ctx context.Context, r *runtime_
|
||||
log.G(ctx).Tracef("ImageStatus for %q returns image status %+v",
|
||||
r.GetImage().GetImage(), res.GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.ImageStatusRequest
|
||||
@@ -1096,6 +1115,8 @@ func (in *instrumentedService) RemoveImage(ctx context.Context, r *runtime.Remov
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("RemoveImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("RemoveImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1103,6 +1124,7 @@ func (in *instrumentedService) RemoveImage(ctx context.Context, r *runtime.Remov
|
||||
} else {
|
||||
log.G(ctx).Infof("RemoveImage %q returns successfully", r.GetImage().GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err := in.c.RemoveImage(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1112,6 +1134,8 @@ func (in *instrumentedAlphaService) RemoveImage(ctx context.Context, r *runtime_
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("RemoveImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("RemoveImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1119,6 +1143,7 @@ func (in *instrumentedAlphaService) RemoveImage(ctx context.Context, r *runtime_
|
||||
} else {
|
||||
log.G(ctx).Infof("RemoveImage %q returns successfully", r.GetImage().GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.RemoveImageRequest
|
||||
@@ -1149,6 +1174,8 @@ func (in *instrumentedService) ImageFsInfo(ctx context.Context, r *runtime.Image
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageFsInfo"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Debugf("ImageFsInfo")
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1156,6 +1183,7 @@ func (in *instrumentedService) ImageFsInfo(ctx context.Context, r *runtime.Image
|
||||
} else {
|
||||
log.G(ctx).Debugf("ImageFsInfo returns filesystem info %+v", res.ImageFilesystems)
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.ImageFsInfo(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1165,6 +1193,8 @@ func (in *instrumentedAlphaService) ImageFsInfo(ctx context.Context, r *runtime_
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageFsInfo"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Debugf("ImageFsInfo")
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1172,6 +1202,7 @@ func (in *instrumentedAlphaService) ImageFsInfo(ctx context.Context, r *runtime_
|
||||
} else {
|
||||
log.G(ctx).Debugf("ImageFsInfo returns filesystem info %+v", res.ImageFilesystems)
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.ImageFsInfoRequest
|
||||
|
||||
@@ -92,6 +92,11 @@ const (
|
||||
|
||||
// runtimeRunhcsV1 is the runtime type for runhcs.
|
||||
runtimeRunhcsV1 = "io.containerd.runhcs.v1"
|
||||
|
||||
// name prefix for CRI server specific spans
|
||||
criServerSpanPrefix = "pkg.cri.server"
|
||||
|
||||
spanDelimiter = "."
|
||||
)
|
||||
|
||||
// makeSandboxName generates sandbox name from sandbox metadata. The name
|
||||
@@ -510,3 +515,10 @@ func copyResourcesToStatus(spec *runtimespec.Spec, status containerstore.Status)
|
||||
}
|
||||
return status
|
||||
}
|
||||
|
||||
func makeSpanName(funcName string) string {
|
||||
return strings.Join([]string{
|
||||
criServerSpanPrefix,
|
||||
funcName,
|
||||
}, spanDelimiter)
|
||||
}
|
||||
|
||||
@@ -36,6 +36,7 @@ import (
|
||||
"github.com/containerd/imgcrypt"
|
||||
"github.com/containerd/imgcrypt/images/encryption"
|
||||
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
|
||||
"github.com/containerd/containerd"
|
||||
@@ -48,6 +49,7 @@ import (
|
||||
distribution "github.com/containerd/containerd/reference/docker"
|
||||
"github.com/containerd/containerd/remotes/docker"
|
||||
"github.com/containerd/containerd/remotes/docker/config"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
)
|
||||
|
||||
// For image management:
|
||||
@@ -93,6 +95,7 @@ import (
|
||||
|
||||
// PullImage pulls an image with authentication config.
|
||||
func (c *criService) PullImage(ctx context.Context, r *runtime.PullImageRequest) (*runtime.PullImageResponse, error) {
|
||||
span := tracing.CurrentSpan(ctx)
|
||||
imageRef := r.GetImage().GetImage()
|
||||
namedRef, err := distribution.ParseDockerRef(imageRef)
|
||||
if err != nil {
|
||||
@@ -133,7 +136,10 @@ func (c *criService) PullImage(ctx context.Context, r *runtime.PullImageRequest)
|
||||
return nil, err
|
||||
}
|
||||
log.G(ctx).Debugf("PullImage %q with snapshotter %s", ref, snapshotter)
|
||||
|
||||
span.SetAttributes(
|
||||
attribute.String("image.ref", ref),
|
||||
attribute.String("snapshotter.name", snapshotter),
|
||||
)
|
||||
pullOpts := []containerd.RemoteOpt{
|
||||
containerd.WithSchema1Conversion, //nolint:staticcheck // Ignore SA1019. Need to keep deprecated package for compatibility.
|
||||
containerd.WithResolver(resolver),
|
||||
@@ -165,6 +171,7 @@ func (c *criService) PullImage(ctx context.Context, r *runtime.PullImageRequest)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to pull and unpack image %q: %w", ref, err)
|
||||
}
|
||||
span.AddEvent("Pull and unpack image complete")
|
||||
|
||||
configDesc, err := image.Config(ctx)
|
||||
if err != nil {
|
||||
|
||||
@@ -22,6 +22,8 @@ import (
|
||||
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/images"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
)
|
||||
@@ -33,15 +35,17 @@ import (
|
||||
// Remove the whole image no matter the it's image id or reference. This is the
|
||||
// semantic defined in CRI now.
|
||||
func (c *criService) RemoveImage(ctx context.Context, r *runtime.RemoveImageRequest) (*runtime.RemoveImageResponse, error) {
|
||||
span := tracing.CurrentSpan(ctx)
|
||||
image, err := c.localResolve(r.GetImage().GetImage())
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
span.AddEvent(err.Error())
|
||||
// return empty without error when image not found.
|
||||
return &runtime.RemoveImageResponse{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("can not resolve %q locally: %w", r.GetImage().GetImage(), err)
|
||||
}
|
||||
|
||||
span.SetAttributes(attribute.String("image.id", image.ID))
|
||||
// Remove all image references.
|
||||
for i, ref := range image.References {
|
||||
var opts []images.DeleteOpt
|
||||
|
||||
@@ -24,6 +24,8 @@ import (
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/log"
|
||||
imagestore "github.com/containerd/containerd/pkg/cri/store/image"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
|
||||
imagespec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
@@ -33,14 +35,17 @@ import (
|
||||
// TODO(random-liu): We should change CRI to distinguish image id and image spec. (See
|
||||
// kubernetes/kubernetes#46255)
|
||||
func (c *criService) ImageStatus(ctx context.Context, r *runtime.ImageStatusRequest) (*runtime.ImageStatusResponse, error) {
|
||||
span := tracing.CurrentSpan(ctx)
|
||||
image, err := c.localResolve(r.GetImage().GetImage())
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
span.AddEvent(err.Error())
|
||||
// return empty without error when image not found.
|
||||
return &runtime.ImageStatusResponse{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("can not resolve %q locally: %w", r.GetImage().GetImage(), err)
|
||||
}
|
||||
span.SetAttributes(attribute.String("image.id", image.ID))
|
||||
// TODO(random-liu): [P0] Make sure corresponding snapshot exists. What if snapshot
|
||||
// doesn't exist?
|
||||
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/log"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
|
||||
runtime_alpha "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
|
||||
|
||||
@@ -931,6 +932,8 @@ func (in *instrumentedService) PullImage(ctx context.Context, r *runtime.PullIma
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("PullImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("PullImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -939,6 +942,7 @@ func (in *instrumentedService) PullImage(ctx context.Context, r *runtime.PullIma
|
||||
log.G(ctx).Infof("PullImage %q returns image reference %q",
|
||||
r.GetImage().GetImage(), res.GetImageRef())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.PullImage(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -948,6 +952,8 @@ func (in *instrumentedAlphaService) PullImage(ctx context.Context, r *runtime_al
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("PullImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("PullImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -956,6 +962,7 @@ func (in *instrumentedAlphaService) PullImage(ctx context.Context, r *runtime_al
|
||||
log.G(ctx).Infof("PullImage %q returns image reference %q",
|
||||
r.GetImage().GetImage(), res.GetImageRef())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.PullImageRequest
|
||||
@@ -986,6 +993,8 @@ func (in *instrumentedService) ListImages(ctx context.Context, r *runtime.ListIm
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ListImages"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ListImages with filter %+v", r.GetFilter())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -994,6 +1003,7 @@ func (in *instrumentedService) ListImages(ctx context.Context, r *runtime.ListIm
|
||||
log.G(ctx).Tracef("ListImages with filter %+v returns image list %+v",
|
||||
r.GetFilter(), res.GetImages())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.ListImages(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1003,6 +1013,8 @@ func (in *instrumentedAlphaService) ListImages(ctx context.Context, r *runtime_a
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ListImages"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ListImages with filter %+v", r.GetFilter())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1011,6 +1023,7 @@ func (in *instrumentedAlphaService) ListImages(ctx context.Context, r *runtime_a
|
||||
log.G(ctx).Tracef("ListImages with filter %+v returns image list %+v",
|
||||
r.GetFilter(), res.GetImages())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.ListImagesRequest
|
||||
@@ -1041,6 +1054,8 @@ func (in *instrumentedService) ImageStatus(ctx context.Context, r *runtime.Image
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageStatus"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ImageStatus for %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1049,6 +1064,7 @@ func (in *instrumentedService) ImageStatus(ctx context.Context, r *runtime.Image
|
||||
log.G(ctx).Tracef("ImageStatus for %q returns image status %+v",
|
||||
r.GetImage().GetImage(), res.GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.ImageStatus(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1058,6 +1074,8 @@ func (in *instrumentedAlphaService) ImageStatus(ctx context.Context, r *runtime_
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageStatus"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Tracef("ImageStatus for %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1066,6 +1084,7 @@ func (in *instrumentedAlphaService) ImageStatus(ctx context.Context, r *runtime_
|
||||
log.G(ctx).Tracef("ImageStatus for %q returns image status %+v",
|
||||
r.GetImage().GetImage(), res.GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.ImageStatusRequest
|
||||
@@ -1096,6 +1115,8 @@ func (in *instrumentedService) RemoveImage(ctx context.Context, r *runtime.Remov
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("RemoveImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("RemoveImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1103,6 +1124,7 @@ func (in *instrumentedService) RemoveImage(ctx context.Context, r *runtime.Remov
|
||||
} else {
|
||||
log.G(ctx).Infof("RemoveImage %q returns successfully", r.GetImage().GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err := in.c.RemoveImage(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1112,6 +1134,8 @@ func (in *instrumentedAlphaService) RemoveImage(ctx context.Context, r *runtime_
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("RemoveImage"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Infof("RemoveImage %q", r.GetImage().GetImage())
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1119,6 +1143,7 @@ func (in *instrumentedAlphaService) RemoveImage(ctx context.Context, r *runtime_
|
||||
} else {
|
||||
log.G(ctx).Infof("RemoveImage %q returns successfully", r.GetImage().GetImage())
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.RemoveImageRequest
|
||||
@@ -1149,6 +1174,8 @@ func (in *instrumentedService) ImageFsInfo(ctx context.Context, r *runtime.Image
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageFsInfo"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Debugf("ImageFsInfo")
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1156,6 +1183,7 @@ func (in *instrumentedService) ImageFsInfo(ctx context.Context, r *runtime.Image
|
||||
} else {
|
||||
log.G(ctx).Debugf("ImageFsInfo returns filesystem info %+v", res.ImageFilesystems)
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
res, err = in.c.ImageFsInfo(ctrdutil.WithNamespace(ctx), r)
|
||||
return res, errdefs.ToGRPC(err)
|
||||
@@ -1165,6 +1193,8 @@ func (in *instrumentedAlphaService) ImageFsInfo(ctx context.Context, r *runtime_
|
||||
if err := in.checkInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, span := tracing.StartSpan(ctx, makeSpanName("ImageFsInfo"))
|
||||
defer tracing.StopSpan(span)
|
||||
log.G(ctx).Debugf("ImageFsInfo")
|
||||
defer func() {
|
||||
if err != nil {
|
||||
@@ -1172,6 +1202,7 @@ func (in *instrumentedAlphaService) ImageFsInfo(ctx context.Context, r *runtime_
|
||||
} else {
|
||||
log.G(ctx).Debugf("ImageFsInfo returns filesystem info %+v", res.ImageFilesystems)
|
||||
}
|
||||
tracing.SetSpanStatus(span, err)
|
||||
}()
|
||||
// converts request and response for earlier CRI version to call and get response from the current version
|
||||
var v1r runtime.ImageFsInfoRequest
|
||||
|
||||
@@ -23,6 +23,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@@ -36,10 +37,12 @@ import (
|
||||
"github.com/containerd/containerd/pkg/kmutex"
|
||||
"github.com/containerd/containerd/platforms"
|
||||
"github.com/containerd/containerd/snapshots"
|
||||
"github.com/containerd/containerd/tracing"
|
||||
"github.com/opencontainers/go-digest"
|
||||
"github.com/opencontainers/image-spec/identity"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"github.com/sirupsen/logrus"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"golang.org/x/sync/semaphore"
|
||||
)
|
||||
@@ -160,6 +163,11 @@ func (u *Unpacker) Unpack(h images.Handler) images.Handler {
|
||||
layers = map[digest.Digest][]ocispec.Descriptor{}
|
||||
)
|
||||
return images.HandlerFunc(func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
|
||||
ctx, span := tracing.StartSpan(ctx, "pkg.unpack.unpacker.UnpackHandler")
|
||||
defer span.End()
|
||||
span.SetAttributes(
|
||||
attribute.String("descriptor.media.type", desc.MediaType),
|
||||
attribute.String("descriptor.digest", desc.Digest.String()))
|
||||
unlock, err := u.lockBlobDescriptor(ctx, desc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -174,10 +182,12 @@ func (u *Unpacker) Unpack(h images.Handler) images.Handler {
|
||||
case images.MediaTypeDockerSchema2Manifest, ocispec.MediaTypeImageManifest:
|
||||
var nonLayers []ocispec.Descriptor
|
||||
var manifestLayers []ocispec.Descriptor
|
||||
|
||||
// Split layers from non-layers, layers will be handled after
|
||||
// the config
|
||||
for _, child := range children {
|
||||
for i, child := range children {
|
||||
span.SetAttributes(
|
||||
attribute.StringSlice("descriptor.child."+strconv.Itoa(i), []string{child.MediaType, child.Digest.String()}),
|
||||
)
|
||||
if images.IsLayerType(child.MediaType) {
|
||||
manifestLayers = append(manifestLayers, child)
|
||||
} else {
|
||||
@@ -223,6 +233,8 @@ func (u *Unpacker) unpack(
|
||||
layers []ocispec.Descriptor,
|
||||
) error {
|
||||
ctx := u.ctx
|
||||
ctx, layerSpan := tracing.StartSpan(ctx, "pkg.unpack.unpacker.unpack")
|
||||
defer layerSpan.End()
|
||||
p, err := content.ReadBlob(ctx, u.content, config)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -398,9 +410,18 @@ func (u *Unpacker) unpack(
|
||||
}
|
||||
|
||||
for i, desc := range layers {
|
||||
_, layerSpan := tracing.StartSpan(ctx, "pkg.unpack.unpacker.unpackLayer")
|
||||
layerSpan.SetAttributes(
|
||||
attribute.String("layer.media.type", desc.MediaType),
|
||||
attribute.Int64("layer.media.size", desc.Size),
|
||||
attribute.String("layer.media.digest", desc.Digest.String()),
|
||||
)
|
||||
if err := doUnpackFn(i, desc); err != nil {
|
||||
layerSpan.RecordError(err)
|
||||
layerSpan.End()
|
||||
return err
|
||||
}
|
||||
layerSpan.End()
|
||||
}
|
||||
|
||||
chainID := identity.ChainID(chain).String()
|
||||
@@ -425,9 +446,14 @@ func (u *Unpacker) unpack(
|
||||
func (u *Unpacker) fetch(ctx context.Context, h images.Handler, layers []ocispec.Descriptor, done []chan struct{}) error {
|
||||
eg, ctx2 := errgroup.WithContext(ctx)
|
||||
for i, desc := range layers {
|
||||
ctx2, layerSpan := tracing.StartSpan(ctx2, "pkg.unpack.unpacker.fetchLayer")
|
||||
layerSpan.SetAttributes(
|
||||
attribute.String("layer.media.type", desc.MediaType),
|
||||
attribute.Int64("layer.media.size", desc.Size),
|
||||
attribute.String("layer.media.digest", desc.Digest.String()),
|
||||
)
|
||||
desc := desc
|
||||
i := i
|
||||
|
||||
if err := u.acquire(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -451,6 +477,7 @@ func (u *Unpacker) fetch(ctx context.Context, h images.Handler, layers []ocispec
|
||||
|
||||
return nil
|
||||
})
|
||||
layerSpan.End()
|
||||
}
|
||||
|
||||
return eg.Wait()
|
||||
|
||||
Reference in New Issue
Block a user