Skip to content

Commit c758d9c

Browse files
dines-rldines-rl
committed
cp dines
1 parent c882e3a commit c758d9c

3 files changed

Lines changed: 411 additions & 21 deletions

File tree

‎pkg/convertor/convertor.go‎

Lines changed: 127 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,22 @@ type contentLoader struct {
114114
}
115115

116116
func (loader *contentLoader) Load(ctx context.Context, cs content.Store) (l Layer, err error) {
117+
startTime := time.Now()
118+
ctx, span := tracer.Start(ctx, "content_load",
119+
trace.WithAttributes(
120+
attribute.Bool("isAccelLayer", loader.isAccelLayer),
121+
attribute.String("fsType", loader.fsType),
122+
attribute.Int("fileCount", len(loader.files)),
123+
))
124+
defer func() {
125+
if err != nil {
126+
addErrorEvent(span, err, "content_load", l.Desc)
127+
}
128+
span.SetAttributes(
129+
attribute.Int64("load_duration_ms", time.Since(startTime).Milliseconds()),
130+
)
131+
span.End()
132+
}()
117133
refName := fmt.Sprintf(ConvContentNameFormat, UniquePart())
118134
contentWriter, err := content.OpenWriter(ctx, cs, content.WithRef(refName))
119135
if err != nil {
@@ -187,6 +203,9 @@ func (loader *contentLoader) Load(ctx context.Context, cs content.Store) (l Laye
187203
return emptyLayer, errors.Wrapf(err, "failed to close tar file")
188204
}
189205

206+
// Add progress event before commit
207+
addProgressEvent(span, countWriter.c, countWriter.c, len(loader.files))
208+
190209
labels := map[string]string{
191210
labelBuildLayerFrom: strings.Join(srcPathList, ","),
192211
}
@@ -480,12 +499,22 @@ func (c *overlaybdConvertor) sentToRemote(ctx context.Context, desc ocispec.Desc
480499
// convertLayers applys image layers on overlaybd with specified filesystem and
481500
// exports the layers based on zfile.
482501
func (c *overlaybdConvertor) convertLayers(ctx context.Context, srcDescs []ocispec.Descriptor, srcDiffIDs []digest.Digest, fsType string) ([]Layer, error) {
502+
startTime := time.Now()
483503
ctx, span := tracer.Start(ctx, "convertLayers",
484504
trace.WithAttributes(
485505
attribute.String("fsType", fsType),
486506
attribute.Int("layerCount", len(srcDescs)),
507+
attribute.Int64("total_size", calculateTotalSize(srcDescs)),
487508
))
488-
defer span.End()
509+
defer func() {
510+
if root, ok := c.sn.(interface{ Root() string }); ok {
511+
addResourceAttributes(span, root.Root())
512+
}
513+
span.SetAttributes(
514+
attribute.Int64("conversion_duration_ms", time.Since(startTime).Milliseconds()),
515+
)
516+
span.End()
517+
}()
489518

490519
fmt.Printf("convertLayers: Starting conversion of %d layers\n", len(srcDescs))
491520
var (
@@ -514,27 +543,47 @@ func (c *overlaybdConvertor) convertLayers(ctx context.Context, srcDescs []ocisp
514543
log.G(ctx).Infof("Skipping provenance layer: %s", desc.MediaType)
515544
continue
516545
}
546+
547+
addLayerAttributes(span, desc, idx)
548+
span.AddEvent("processing_layer", trace.WithAttributes(
549+
attribute.Int("layer_index", idx),
550+
attribute.String("digest", desc.Digest.String()),
551+
attribute.Int64("size", desc.Size),
552+
))
553+
517554
fmt.Printf("convertLayers: Processing layer %d/%d (digest: %s)\n", idx+1, len(srcDescs), desc.Digest)
518555
chain = append(chain, srcDiffIDs[idx])
519556
chainID := identity.ChainID(chain).String()
520557
fmt.Printf("convertLayers: Layer chainID: %s\n", chainID)
521558

522559
var remoteDesc ocispec.Descriptor
523560

524-
if c.remote {
525-
fmt.Printf("convertLayers: Looking for remote layer\n")
526-
remoteDesc, err = c.findRemote(ctx, chainID)
527-
if err != nil {
528-
if !errdefs.IsNotFound(err) {
529-
fmt.Printf("convertLayers: ERROR finding remote: %v\n", err)
530-
return nil, err
531-
}
532-
fmt.Printf("convertLayers: Remote layer not found, will process locally\n")
533-
} else {
534-
fmt.Printf("convertLayers: Found remote layer\n")
561+
if c.remote {
562+
fmt.Printf("convertLayers: Looking for remote layer\n")
563+
span.AddEvent("remote_check_start", trace.WithAttributes(
564+
attribute.String("host", c.host),
565+
attribute.String("repo", c.repo),
566+
attribute.String("chainID", chainID),
567+
))
568+
569+
remoteDesc, err = c.findRemote(ctx, chainID)
570+
if err != nil {
571+
if !errdefs.IsNotFound(err) {
572+
fmt.Printf("convertLayers: ERROR finding remote: %v\n", err)
573+
addErrorEvent(span, err, "remote_check", desc)
574+
return nil, err
535575
}
576+
fmt.Printf("convertLayers: Remote layer not found, will process locally\n")
577+
} else {
578+
fmt.Printf("convertLayers: Found remote layer\n")
536579
}
537580

581+
span.AddEvent("remote_check_complete", trace.WithAttributes(
582+
attribute.Bool("found", err == nil),
583+
attribute.String("chainID", chainID),
584+
))
585+
}
586+
538587
if c.remote && err == nil {
539588
key := fmt.Sprintf(convSnapshotNameFormat, chainID)
540589
opts := []snapshots.Opt{
@@ -628,14 +677,22 @@ func (c *overlaybdConvertor) applyOCIV1LayerInObd(
628677
snOpts []snapshots.Opt, // apply for the commit snapshotter
629678
afterApply func(root string) error, // do something after apply tar stream
630679
) (string, error) {
680+
startTime := time.Now()
631681
ctx, span := tracer.Start(ctx, "applyOCIV1LayerInObd",
632682
trace.WithAttributes(
633-
attribute.String("layerDigest", desc.Digest.String()),
634-
attribute.Int64("layerSize", desc.Size),
635-
attribute.String("parentID", parentID),
683+
attribute.String("parent_id", parentID),
684+
attribute.String("digest", desc.Digest.String()),
685+
attribute.Int64("size", desc.Size),
686+
attribute.String("media_type", desc.MediaType),
636687
))
637-
defer span.End()
638-
688+
defer func() {
689+
addConfigAttributes(span, c.zfileCfg, c.vsize)
690+
span.SetAttributes(
691+
attribute.Int64("apply_duration_ms", time.Since(startTime).Milliseconds()),
692+
)
693+
span.End()
694+
}()
695+
// Start applying the layer
639696
fmt.Printf("applyOCIV1LayerInObd: Starting layer application (digest: %s, size: %d)\n", desc.Digest, desc.Size)
640697

641698
fmt.Printf("applyOCIV1LayerInObd: Getting reader for layer content\n")
@@ -690,12 +747,50 @@ func (c *overlaybdConvertor) applyOCIV1LayerInObd(
690747
}
691748

692749
if err = mount.WithTempMount(ctx, mounts, func(root string) error {
693-
_, err := archive.Apply(ctx, root, rc)
694-
if err == nil && afterApply != nil {
695-
err = afterApply(root)
750+
// Create a wrapper to track progress
751+
var bytesProcessed int64
752+
progressReader := &readCountWrapper{
753+
r: rc,
754+
c: 0,
755+
}
756+
757+
// Start a goroutine to report progress
758+
done := make(chan struct{})
759+
go func() {
760+
ticker := time.NewTicker(time.Second)
761+
defer ticker.Stop()
762+
for {
763+
select {
764+
case <-ticker.C:
765+
addProgressEvent(span, bytesProcessed, desc.Size, 0)
766+
case <-done:
767+
return
768+
}
769+
}
770+
}()
771+
772+
// Apply the layer and track progress
773+
_, err := archive.Apply(ctx, root, progressReader)
774+
close(done)
775+
bytesProcessed = progressReader.c
776+
777+
if err != nil {
778+
addErrorEvent(span, err, "layer_apply", desc)
779+
return errors.Wrapf(err, "failed to extract layer into snapshot")
696780
}
697-
return err
781+
782+
// Add final progress event
783+
addProgressEvent(span, bytesProcessed, desc.Size, 0)
784+
785+
if afterApply != nil {
786+
if err := afterApply(root); err != nil {
787+
addErrorEvent(span, err, "after_apply", desc)
788+
return err
789+
}
790+
}
791+
return nil
698792
}); err != nil {
793+
addErrorEvent(span, err, "mount_operation", desc)
699794
return emptyString, errors.Wrapf(err, "failed to apply layer in snapshot %s", key)
700795
}
701796

@@ -724,6 +819,17 @@ func UniquePart() string {
724819
return fmt.Sprintf("%d-%s", t.Nanosecond(), strings.Replace(base64.URLEncoding.EncodeToString(b[:]), "_", "-", -1))
725820
}
726821

822+
type readCountWrapper struct {
823+
r io.Reader
824+
c int64
825+
}
826+
827+
func (rc *readCountWrapper) Read(p []byte) (n int, err error) {
828+
n, err = rc.r.Read(p)
829+
rc.c += int64(n)
830+
return
831+
}
832+
727833
type writeCountWrapper struct {
728834
w io.Writer
729835
c int64

‎pkg/convertor/trace_helpers.go‎

Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,133 @@
1+
/*
2+
Copyright The Accelerated Container Image Authors
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package convertor
18+
19+
import (
20+
"runtime"
21+
"syscall"
22+
"time"
23+
24+
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
25+
"go.opentelemetry.io/otel/attribute"
26+
"go.opentelemetry.io/otel/trace"
27+
)
28+
29+
// calculateTotalSize returns the total size of all descriptors
30+
func calculateTotalSize(descs []ocispec.Descriptor) int64 {
31+
var total int64
32+
for _, desc := range descs {
33+
total += desc.Size
34+
}
35+
return total
36+
}
37+
38+
// getCompressionType determines the compression type from media type
39+
func getCompressionType(desc ocispec.Descriptor) string {
40+
switch desc.MediaType {
41+
case ocispec.MediaTypeImageLayerGzip:
42+
return "gzip"
43+
case ocispec.MediaTypeImageLayerZstd:
44+
return "zstd"
45+
default:
46+
return "none"
47+
}
48+
}
49+
50+
// getMemoryUsage returns the current memory usage in bytes
51+
func getMemoryUsage() int64 {
52+
var m runtime.MemStats
53+
runtime.ReadMemStats(&m)
54+
return int64(m.Alloc)
55+
}
56+
57+
// getDiskUsage returns the current disk usage for the given path
58+
func getDiskUsage(path string) int64 {
59+
var stat syscall.Statfs_t
60+
if err := syscall.Statfs(path, &stat); err != nil {
61+
return 0
62+
}
63+
return int64(stat.Blocks-stat.Bfree) * int64(stat.Bsize)
64+
}
65+
66+
// calculateThroughput calculates bytes per second
67+
func calculateThroughput(bytesProcessed int64, duration time.Duration) int64 {
68+
if duration.Seconds() == 0 {
69+
return 0
70+
}
71+
return int64(float64(bytesProcessed) / duration.Seconds())
72+
}
73+
74+
// getErrorType returns a string representation of the error type
75+
func getErrorType(err error) string {
76+
if err == nil {
77+
return ""
78+
}
79+
return err.Error()
80+
}
81+
82+
// addLayerAttributes adds common layer attributes to a span
83+
func addLayerAttributes(span trace.Span, desc ocispec.Descriptor, idx int) {
84+
span.SetAttributes(
85+
attribute.Int("layer_index", idx),
86+
attribute.String("digest", desc.Digest.String()),
87+
attribute.Int64("size", desc.Size),
88+
attribute.String("media_type", desc.MediaType),
89+
attribute.String("compression", getCompressionType(desc)),
90+
)
91+
}
92+
93+
// addResourceAttributes adds resource usage attributes to a span
94+
func addResourceAttributes(span trace.Span, path string) {
95+
span.SetAttributes(
96+
attribute.Int64("memory_used_bytes", getMemoryUsage()),
97+
attribute.Int64("disk_used_bytes", getDiskUsage(path)),
98+
)
99+
}
100+
101+
// addConfigAttributes adds configuration attributes to a span
102+
func addConfigAttributes(span trace.Span, cfg ZFileConfig, vsize int) {
103+
span.SetAttributes(
104+
attribute.String("algorithm", cfg.Algorithm),
105+
attribute.Int("block_size", cfg.BlockSize),
106+
attribute.Int("vsize", vsize),
107+
)
108+
}
109+
110+
// addProgressEvent adds a progress event to a span
111+
func addProgressEvent(span trace.Span, processed, total int64, blocks int) {
112+
var percent float64
113+
if total > 0 {
114+
percent = float64(processed) / float64(total) * 100
115+
}
116+
span.AddEvent("conversion_progress", trace.WithAttributes(
117+
attribute.Int64("bytes_processed", processed),
118+
attribute.Float64("percent_complete", percent),
119+
attribute.Int("blocks_converted", blocks),
120+
))
121+
}
122+
123+
// addErrorEvent adds detailed error information to a span
124+
func addErrorEvent(span trace.Span, err error, operation string, desc ocispec.Descriptor) {
125+
if err == nil {
126+
return
127+
}
128+
span.RecordError(err, trace.WithAttributes(
129+
attribute.String("operation_type", operation),
130+
attribute.String("layer_digest", desc.Digest.String()),
131+
attribute.String("error_type", getErrorType(err)),
132+
))
133+
}

0 commit comments

Comments
 (0)