From f183f67f78e009f4c48a58ab0eb341402b34c1b5 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Wed, 29 Jul 2026 11:09:58 -0700 Subject: [PATCH 01/15] feat(convertor): add --output-dir flag for local artifact export Adds a new --output-dir flag that writes converted overlaybd artifacts to a local directory instead of pushing to a registry. This enables the blueprint-uploader tool to push blobs directly to S3 via discoball's prepare/confirm API, bypassing the convertor's registry proxy path. Output directory layout: /manifest.json - OCI manifest JSON /config.json - OCI image config JSON /config.digest - "sha256:" of config.json /blobs/ - one file per layer blob Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/dir_exporter.go | 173 ++++++++++++++++++++++++++ cmd/convertor/main.go | 27 +++- 2 files changed, 196 insertions(+), 4 deletions(-) create mode 100644 cmd/convertor/builder/dir_exporter.go diff --git a/cmd/convertor/builder/dir_exporter.go b/cmd/convertor/builder/dir_exporter.go new file mode 100644 index 00000000..215d66a0 --- /dev/null +++ b/cmd/convertor/builder/dir_exporter.go @@ -0,0 +1,173 @@ +/* + Copyright The Accelerated Container Image Authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package builder + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + + "github.com/containerd/containerd/v2/core/content" + "github.com/containerd/containerd/v2/core/images" + "github.com/containerd/log" + "github.com/opencontainers/go-digest" + v1 "github.com/opencontainers/image-spec/specs-go/v1" + "github.com/pkg/errors" +) + +// ExportContentStoreToDir writes converted OCI image artifacts from the content +// store to outputDir in a flat layout for consumption by blueprint-uploader. +// +// Output layout: +// +// manifest.json - OCI manifest JSON +// config.json - OCI image config JSON +// config.digest - "sha256:" of config.json +// blobs/ - layer blobs, one file per layer (named by digest with +// ':' replaced by '-', e.g. "sha256-abc123...") +// +// Only single-arch OCI manifests are supported; multi-arch index entries are skipped. +func ExportContentStoreToDir(ctx context.Context, store content.Store, imageStore images.Store, outputDir string) error { + if err := os.MkdirAll(outputDir, 0755); err != nil { + return errors.Wrapf(err, "failed to create output directory %q", outputDir) + } + + imgs, err := imageStore.List(ctx) + if err != nil { + return errors.Wrapf(err, "failed to list images in output store") + } + if len(imgs) == 0 { + return errors.New("no images in output store after conversion") + } + + // Find the first non-index image (for overlaybd OCI base builds there is one per arch). + var manifestDesc v1.Descriptor + for _, img := range imgs { + if img.Target.MediaType != v1.MediaTypeImageIndex && + img.Target.MediaType != "application/vnd.docker.distribution.manifest.list.v2+json" { + manifestDesc = img.Target + break + } + } + if manifestDesc.Digest == "" { + return errors.New("no single-arch manifest found in output store") + } + + // Read and write manifest.json. + manifestBytes, err := content.ReadBlob(ctx, store, manifestDesc) + if err != nil { + return errors.Wrapf(err, "failed to read manifest blob") + } + if err := os.WriteFile(filepath.Join(outputDir, "manifest.json"), manifestBytes, 0644); err != nil { + return errors.Wrapf(err, "failed to write manifest.json") + } + log.G(ctx).Debugf("dir exporter: wrote manifest.json (%d bytes)", len(manifestBytes)) + + // Parse manifest to find config and layer descriptors. + var manifest struct { + Config struct { + MediaType string `json:"mediaType"` + Digest string `json:"digest"` + Size int64 `json:"size"` + } `json:"config"` + Layers []struct { + MediaType string `json:"mediaType"` + Digest string `json:"digest"` + Size int64 `json:"size"` + } `json:"layers"` + } + if err := json.Unmarshal(manifestBytes, &manifest); err != nil { + return errors.Wrapf(err, "failed to parse manifest JSON") + } + + // Write config.json and config.digest. + configDigestStr := manifest.Config.Digest + cd, err := digest.Parse(configDigestStr) + if err != nil { + return errors.Wrapf(err, "failed to parse config digest %q", configDigestStr) + } + configSpec := v1.Descriptor{ + MediaType: manifest.Config.MediaType, + Digest: cd, + Size: manifest.Config.Size, + } + configBytes, err := content.ReadBlob(ctx, store, configSpec) + if err != nil { + return errors.Wrapf(err, "failed to read config blob") + } + if err := os.WriteFile(filepath.Join(outputDir, "config.json"), configBytes, 0644); err != nil { + return errors.Wrapf(err, "failed to write config.json") + } + if err := os.WriteFile(filepath.Join(outputDir, "config.digest"), []byte(configDigestStr), 0644); err != nil { + return errors.Wrapf(err, "failed to write config.digest") + } + log.G(ctx).Debugf("dir exporter: wrote config.json (digest: %s)", configDigestStr) + + // Write each layer blob to blobs/. + if len(manifest.Layers) == 0 { + return errors.New("manifest has no layers") + } + blobsDir := filepath.Join(outputDir, "blobs") + if err := os.MkdirAll(blobsDir, 0755); err != nil { + return errors.Wrapf(err, "failed to create blobs directory") + } + + for i, layer := range manifest.Layers { + layerDigestStr := layer.Digest + ld, err := digest.Parse(layerDigestStr) + if err != nil { + return errors.Wrapf(err, "failed to parse layer[%d] digest %q", i, layerDigestStr) + } + layerSpec := v1.Descriptor{ + MediaType: layer.MediaType, + Digest: ld, + Size: layer.Size, + } + // Name the file by digest with ':' replaced by '-' for filesystem safety. + blobFilename := strings.ReplaceAll(layerDigestStr, ":", "-") + blobPath := filepath.Join(blobsDir, blobFilename) + if err := copyBlobToFile(ctx, store, layerSpec, blobPath); err != nil { + return errors.Wrapf(err, "failed to write layer[%d] blob (digest: %s)", i, layerDigestStr) + } + log.G(ctx).Debugf("dir exporter: wrote blobs/%s (%s bytes)", blobFilename, fmt.Sprintf("%d", layer.Size)) + } + + log.G(ctx).Infof("dir exporter: wrote %d layer blob(s) to %s", len(manifest.Layers), outputDir) + return nil +} + +// copyBlobToFile streams the blob identified by desc from store to the file at path. +func copyBlobToFile(ctx context.Context, store content.Store, desc v1.Descriptor, path string) error { + ra, err := store.ReaderAt(ctx, desc) + if err != nil { + return errors.Wrapf(err, "failed to open blob %s for reading", desc.Digest) + } + defer ra.Close() + + f, err := os.Create(path) + if err != nil { + return errors.Wrapf(err, "failed to create file %q", path) + } + defer f.Close() + + _, err = io.Copy(f, content.NewReader(ra)) + return errors.Wrapf(err, "failed to stream blob to %q", path) +} diff --git a/cmd/convertor/main.go b/cmd/convertor/main.go index 3cabeacf..6fefa450 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -64,6 +64,7 @@ var ( // tar import/export importTar string exportTar string + outputDir string tarExportRepo string // certification @@ -98,6 +99,14 @@ Version: ` + commitID, logrus.Error("import-tar cannot be used with input-tag or input-digest") os.Exit(1) } + if outputDir != "" && importTar == "" { + logrus.Error("--output-dir requires --import-tar") + os.Exit(1) + } + if outputDir != "" && exportTar != "" { + logrus.Error("--output-dir and --export-tar are mutually exclusive") + os.Exit(1) + } if importTar == "" && repo == "" { logrus.Error("repository is required when not using import-tar") os.Exit(1) @@ -175,9 +184,9 @@ Version: ` + commitID, // Choose resolver based on export mode var customResolver remotes.Resolver - if exportTar != "" { - // For tar export, use FileBasedResolver to capture converted layers locally - logrus.Debugf("tar export mode: using file-based resolver to capture converted layers") + if exportTar != "" || outputDir != "" { + // For tar export or dir export, use FileBasedResolver to capture converted layers locally. + logrus.Debugf("local export mode: using file-based resolver to capture converted layers") var err error exportResolver, err = builder.NewFileBasedResolver(importResolver.Store(), importResolver.ImageStore()) if err != nil { @@ -187,7 +196,7 @@ Version: ` + commitID, repo = tarExportRepo customResolver = exportResolver - // Setup cleanup for export resolver temporary directory + // Setup cleanup for export resolver temporary directory. defer func() { if !reserve && exportResolver != nil { if err := exportResolver.CleanupTempDir(); err != nil { @@ -349,6 +358,15 @@ Version: ` + commitID, } logrus.Info("tar export finished") } + // Handle dir export if requested + if outputDir != "" && exportResolver != nil { + logrus.Debugf("exporting converted overlaybd artifacts to directory: %s", outputDir) + if err := builder.ExportContentStoreToDir(ctx, exportResolver.OutputStore(), exportResolver.OutputImageStore(), outputDir); err != nil { + logrus.Errorf("failed to export to directory: %v", err) + os.Exit(1) + } + logrus.Info("dir export finished") + } } if tb != "" { logrus.Info("building [Overlaybd - Turbo OCIv1] image...") @@ -401,6 +419,7 @@ func init() { // tar import/export rootCmd.Flags().StringVar(&importTar, "import-tar", "", "import image from tar file (OCI layout format)") rootCmd.Flags().StringVar(&exportTar, "export-tar", "", "export converted image to tar file (OCI layout format)") + rootCmd.Flags().StringVar(&outputDir, "output-dir", "", "export converted artifacts to a local directory (requires --import-tar; mutually exclusive with --export-tar)") rootCmd.Flags().StringVar(&tarExportRepo, "tar-export-repo", "localhost/converted", "repository name used in exported tar file (only used with --export-tar)") // certification From e956762f8fe981275f6d5eca574766f284bc5f21 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Wed, 29 Jul 2026 14:44:55 -0700 Subject: [PATCH 02/15] feat(convertor): add --direct-upload flag for discoball prepare/confirm API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds --direct-upload and --registry-url flags to convertor. When set, convertor converts OCI layers → overlaybd locally (via FileBasedResolver) then uploads blobs directly to S3 via discoball's two-phase prepare/confirm API, bypassing the normal registry push path. Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/direct_upload.go | 466 +++++++++++++++++++++++++ cmd/convertor/main.go | 55 ++- 2 files changed, 517 insertions(+), 4 deletions(-) create mode 100644 cmd/convertor/builder/direct_upload.go diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go new file mode 100644 index 00000000..a5c7c278 --- /dev/null +++ b/cmd/convertor/builder/direct_upload.go @@ -0,0 +1,466 @@ +package builder + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/binary" + "encoding/json" + "fmt" + "hash/crc64" + "io" + "net/http" + "net/url" + "sort" + "strings" + "sync" + + "github.com/containerd/containerd/v2/core/content" + "github.com/containerd/containerd/v2/core/images" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" + "github.com/sirupsen/logrus" +) + +const ( + ociManifestV1MediaType = "application/vnd.oci.image.manifest.v1+json" + partUploadConcurrency = 4 +) + +// CRC64/NVME table (Jones polynomial, reflected form). +var crc64NVMETable = crc64.MakeTable(0xAD93D23594C935A9) + +// ---- wire types (mirror discoball/registry/handlers/directupload.go) -------- + +// prepareDirectUploadRequest is sent to the prepare endpoint. +// The manifest field is Go []byte, which JSON-encodes as base64. +type prepareDirectUploadRequest struct { + Manifest []byte `json:"manifest"` + MediaType string `json:"media_type"` +} + +type blobUploadInstruction struct { + Digest string `json:"digest"` + Exists bool `json:"exists"` + Token *string `json:"token,omitempty"` + Parts []uploadPartInfo `json:"parts,omitempty"` + PartSize *int64 `json:"part_size,omitempty"` +} + +type uploadPartInfo struct { + Number int `json:"number"` + URL string `json:"url"` +} + +type prepareDirectUploadResponse struct { + Blobs []blobUploadInstruction `json:"blobs"` +} + +type completedPart struct { + Number int `json:"number"` + ETag string `json:"etag"` +} + +type blobConfirmEntry struct { + Digest string `json:"digest"` + Token string `json:"token"` + Parts []completedPart `json:"parts"` + CRC64NVME *string `json:"crc64nvme,omitempty"` +} + +type confirmDirectUploadRequest struct { + Manifest []byte `json:"manifest"` + MediaType string `json:"media_type"` + Tag *string `json:"tag,omitempty"` + Blobs []blobConfirmEntry `json:"blobs"` +} + +// ---- DirectUploadFromStore -------------------------------------------------- + +// DirectUploadFromStore reads the converted OCI image from the content store +// captured by FileBasedResolver and pushes blobs directly to S3 via discoball's +// prepare/confirm API. Returns the manifest digest stored by discoball. +// +// imageRef must be the full reference (e.g. "disco.runloop.pro/account:bpt_xxx"). +// registryURL is used only to determine the scheme +// (e.g. "https://disco.runloop.pro"). +func DirectUploadFromStore( + ctx context.Context, + store content.Store, + imageStore images.Store, + imageRef string, + registryURL string, +) (string, error) { + scheme := "https" + if strings.HasPrefix(registryURL, "http://") { + scheme = "http" + } + + registry, repository, tag, err := parseImageRef(imageRef) + if err != nil { + return "", fmt.Errorf("invalid image ref %q: %w", imageRef, err) + } + logrus.Debugf("direct-upload: registry=%s repository=%s tag=%s", registry, repository, tag) + + // Find the converted image in the store. + imgs, err := imageStore.List(ctx, "") + if err != nil { + return "", fmt.Errorf("listing image store: %w", err) + } + var manifestDesc *ocispec.Descriptor + for _, img := range imgs { + if img.Target.MediaType == ocispec.MediaTypeImageIndex { + continue + } + manifestDesc = &img.Target + break + } + if manifestDesc == nil { + return "", fmt.Errorf("no non-index image found in output store") + } + + // Read manifest bytes verbatim — reusing the on-disk bytes ensures the + // manifest digest stored by discoball matches what the content store holds. + manifestBytes, err := content.ReadBlob(ctx, store, *manifestDesc) + if err != nil { + return "", fmt.Errorf("reading manifest: %w", err) + } + + var manifest ocispec.Manifest + if err := json.Unmarshal(manifestBytes, &manifest); err != nil { + return "", fmt.Errorf("parsing manifest: %w", err) + } + + client := &http.Client{} + baseURL := fmt.Sprintf("%s://%s/gitlab/v1/repositories/%s/direct-upload", scheme, registry, repository) + + // Phase 1: prepare. + prep, err := prepareUpload(ctx, client, baseURL, manifestBytes) + if err != nil { + return "", fmt.Errorf("prepare_direct_upload failed: %w", err) + } + logrus.Infof("direct-upload prepare: total=%d missing=%d", len(prep.Blobs), countMissing(prep.Blobs)) + + // Phase 2: upload missing blobs. + var confirmBlobs []blobConfirmEntry + for _, instruction := range prep.Blobs { + if instruction.Exists { + continue + } + token := *instruction.Token + parts := instruction.Parts + var partSize int64 + if instruction.PartSize != nil { + partSize = *instruction.PartSize + } + + // Determine which blob this is. + instDigest := digest.Digest(instruction.Digest) + var completed []completedPart + var crc64nvme *string + + if instDigest == manifest.Config.Digest { + // Config blob: small, load into memory. + configDesc := manifest.Config + configBytes, err := content.ReadBlob(ctx, store, configDesc) + if err != nil { + return "", fmt.Errorf("reading config blob: %w", err) + } + completed, err = uploadPartsFromBytes(client, configBytes, parts, partSize) + if err != nil { + return "", fmt.Errorf("uploading config parts: %w", err) + } + crc := computeCRC64NVME(configBytes) + crc64nvme = &crc + } else { + // Layer blob: stream from content store. + var layerDesc *ocispec.Descriptor + for i := range manifest.Layers { + if manifest.Layers[i].Digest == instDigest { + layerDesc = &manifest.Layers[i] + break + } + } + if layerDesc == nil { + return "", fmt.Errorf("discoball requested unknown blob: %s", instruction.Digest) + } + completed, crc64nvme, err = uploadPartsFromStore(ctx, client, store, *layerDesc, parts, partSize) + if err != nil { + return "", fmt.Errorf("uploading layer parts (digest=%s): %w", instruction.Digest, err) + } + } + + confirmBlobs = append(confirmBlobs, blobConfirmEntry{ + Digest: instruction.Digest, + Token: token, + Parts: completed, + CRC64NVME: crc64nvme, + }) + } + + // Phase 3: confirm. + manifestDigest, err := confirmUpload(ctx, client, baseURL, manifestBytes, tag, confirmBlobs) + if err != nil { + return "", fmt.Errorf("confirm_direct_upload failed: %w", err) + } + logrus.Infof("direct-upload complete: manifest_digest=%s", manifestDigest) + return manifestDigest, nil +} + +// ---- HTTP helpers ----------------------------------------------------------- + +func prepareUpload(ctx context.Context, client *http.Client, baseURL string, manifestBytes []byte) (*prepareDirectUploadResponse, error) { + reqBody, err := json.Marshal(prepareDirectUploadRequest{ + Manifest: manifestBytes, + MediaType: ociManifestV1MediaType, + }) + if err != nil { + return nil, err + } + resp, err := doPost(ctx, client, baseURL+"/prepare/", reqBody) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("reading prepare response: %w", err) + } + if resp.StatusCode/100 != 2 { + return nil, fmt.Errorf("prepare returned HTTP %d: %s", resp.StatusCode, body) + } + var parsed prepareDirectUploadResponse + if err := json.Unmarshal(body, &parsed); err != nil { + return nil, fmt.Errorf("parsing prepare response: %w", err) + } + return &parsed, nil +} + +func confirmUpload(ctx context.Context, client *http.Client, baseURL string, manifestBytes []byte, tag string, blobs []blobConfirmEntry) (string, error) { + var tagPtr *string + if tag != "" { + tagPtr = &tag + } + reqBody, err := json.Marshal(confirmDirectUploadRequest{ + Manifest: manifestBytes, + MediaType: ociManifestV1MediaType, + Tag: tagPtr, + Blobs: blobs, + }) + if err != nil { + return "", err + } + resp, err := doPost(ctx, client, baseURL+"/confirm/", reqBody) + if err != nil { + return "", err + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", fmt.Errorf("reading confirm response: %w", err) + } + if resp.StatusCode/100 != 2 { + return "", fmt.Errorf("confirm returned HTTP %d: %s", resp.StatusCode, body) + } + + // Prefer the digest from the response header; fall back to computing it locally. + manifestDigest := resp.Header.Get("Docker-Content-Digest") + if manifestDigest == "" { + manifestDigest = digest.FromBytes(manifestBytes).String() + } + return manifestDigest, nil +} + +func doPost(ctx context.Context, client *http.Client, rawURL string, body []byte) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, rawURL, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + return client.Do(req) +} + +// ---- Part upload helpers ---------------------------------------------------- + +// uploadPartsFromBytes uploads an in-memory blob via the presigned part URLs. +func uploadPartsFromBytes(client *http.Client, data []byte, parts []uploadPartInfo, partSize int64) ([]completedPart, error) { + completed := make([]completedPart, 0, len(parts)) + for _, p := range parts { + offset := int64(p.Number-1) * partSize + end := offset + partSize + if end > int64(len(data)) { + end = int64(len(data)) + } + etag, err := putPart(client, p.URL, data[offset:end]) + if err != nil { + return nil, fmt.Errorf("part %d: %w", p.Number, err) + } + completed = append(completed, completedPart{Number: p.Number, ETag: etag}) + } + sort.Slice(completed, func(i, j int) bool { return completed[i].Number < completed[j].Number }) + return completed, nil +} + +// uploadPartsFromStore streams a blob from the content store to S3 presigned URLs. +// Returns completed parts and the base64-encoded CRC64/NVME of the full blob. +func uploadPartsFromStore( + ctx context.Context, + client *http.Client, + store content.Store, + desc ocispec.Descriptor, + parts []uploadPartInfo, + partSize int64, +) ([]completedPart, *string, error) { + ra, err := store.ReaderAt(ctx, desc) + if err != nil { + return nil, nil, fmt.Errorf("opening blob %s: %w", desc.Digest, err) + } + defer ra.Close() + + totalSize := ra.Size() + type result struct { + part completedPart + err error + } + + results := make([]result, len(parts)) + sem := make(chan struct{}, partUploadConcurrency) + var wg sync.WaitGroup + + for i, p := range parts { + wg.Add(1) + go func(idx int, p uploadPartInfo) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + + offset := int64(p.Number-1) * partSize + size := partSize + if offset+size > totalSize { + size = totalSize - offset + } + buf := make([]byte, size) + if _, err := ra.ReadAt(buf, offset); err != nil { + results[idx] = result{err: fmt.Errorf("reading part %d from store: %w", p.Number, err)} + return + } + etag, err := putPart(client, p.URL, buf) + if err != nil { + results[idx] = result{err: fmt.Errorf("part %d: %w", p.Number, err)} + return + } + results[idx] = result{part: completedPart{Number: p.Number, ETag: etag}} + }(i, p) + } + wg.Wait() + + completed := make([]completedPart, 0, len(parts)) + for _, r := range results { + if r.err != nil { + return nil, nil, r.err + } + completed = append(completed, r.part) + } + sort.Slice(completed, func(i, j int) bool { return completed[i].Number < completed[j].Number }) + + // Compute CRC64/NVME by reading the full blob sequentially. + crcHash := crc64.New(crc64NVMETable) + if _, err := io.Copy(crcHash, io.NewSectionReader(ra, 0, totalSize)); err != nil { + return nil, nil, fmt.Errorf("computing CRC64/NVME: %w", err) + } + var crcBuf [8]byte + binary.BigEndian.PutUint64(crcBuf[:], crcHash.Sum64()) + crcStr := base64.StdEncoding.EncodeToString(crcBuf[:]) + + return completed, &crcStr, nil +} + +// putPart PUTs buf to the presigned URL and returns the ETag. +// Retries up to 3 times on transient (5xx) errors. +func putPart(client *http.Client, presignedURL string, buf []byte) (string, error) { + const maxRetries = 3 + var lastErr error + for attempt := 0; attempt <= maxRetries; attempt++ { + req, err := http.NewRequest(http.MethodPut, presignedURL, bytes.NewReader(buf)) + if err != nil { + return "", err + } + req.ContentLength = int64(len(buf)) + + resp, err := client.Do(req) + if err != nil { + lastErr = err + continue + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + + if resp.StatusCode/100 == 2 { + etag := resp.Header.Get("ETag") + if etag == "" { + return "", fmt.Errorf("S3 part upload returned no ETag") + } + return etag, nil + } + if resp.StatusCode/100 == 5 && attempt < maxRetries { + lastErr = fmt.Errorf("HTTP %d: %s", resp.StatusCode, body) + continue + } + return "", fmt.Errorf("S3 part upload HTTP %d: %s", resp.StatusCode, body) + } + return "", fmt.Errorf("S3 part upload failed after %d retries: %w", maxRetries, lastErr) +} + +// ---- Utilities -------------------------------------------------------------- + +func countMissing(blobs []blobUploadInstruction) int { + n := 0 + for _, b := range blobs { + if !b.Exists { + n++ + } + } + return n +} + +func computeCRC64NVME(data []byte) string { + crcVal := crc64.Checksum(data, crc64NVMETable) + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], crcVal) + return base64.StdEncoding.EncodeToString(buf[:]) +} + +// parseImageRef parses "registry/repository:tag" into its components. +// The repository may contain slashes (e.g. "host/org/repo:tag"). +func parseImageRef(ref string) (registry, repository, tag string, err error) { + // Strip scheme if present (shouldn't be, but be defensive). + ref = strings.TrimPrefix(ref, "https://") + ref = strings.TrimPrefix(ref, "http://") + + slashIdx := strings.Index(ref, "/") + if slashIdx < 0 { + return "", "", "", fmt.Errorf("missing repository in ref %q", ref) + } + registry = ref[:slashIdx] + rest := ref[slashIdx+1:] + + // Tag is after the last ':' in rest, but only if ':' comes after any '/'. + colonIdx := strings.LastIndex(rest, ":") + if colonIdx >= 0 { + repository = rest[:colonIdx] + tag = rest[colonIdx+1:] + } else { + repository = rest + } + if repository == "" { + return "", "", "", fmt.Errorf("empty repository in ref %q", ref) + } + + // Validate: no URL characters that would break the path segment. + if _, parseErr := url.ParseRequestURI("https://" + registry + "/" + repository); parseErr != nil { + return "", "", "", fmt.Errorf("invalid ref %q: %w", ref, parseErr) + } + return registry, repository, tag, nil +} diff --git a/cmd/convertor/main.go b/cmd/convertor/main.go index 6fefa450..eb9d5e06 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -67,6 +67,10 @@ var ( outputDir string tarExportRepo string + // direct upload + directUpload bool + registryURL string + // certification certDirs []string rootCAs []string @@ -107,6 +111,18 @@ Version: ` + commitID, logrus.Error("--output-dir and --export-tar are mutually exclusive") os.Exit(1) } + if directUpload && importTar == "" { + logrus.Error("--direct-upload requires --import-tar") + os.Exit(1) + } + if directUpload && (exportTar != "" || outputDir != "") { + logrus.Error("--direct-upload is mutually exclusive with --export-tar and --output-dir") + os.Exit(1) + } + if directUpload && repo == "" { + logrus.Error("--direct-upload requires -r/--repository") + os.Exit(1) + } if importTar == "" && repo == "" { logrus.Error("repository is required when not using import-tar") os.Exit(1) @@ -184,16 +200,20 @@ Version: ` + commitID, // Choose resolver based on export mode var customResolver remotes.Resolver - if exportTar != "" || outputDir != "" { - // For tar export or dir export, use FileBasedResolver to capture converted layers locally. - logrus.Debugf("local export mode: using file-based resolver to capture converted layers") + if exportTar != "" || outputDir != "" || directUpload { + // For local export or direct upload, use FileBasedResolver to capture converted layers locally. + logrus.Debugf("local capture mode: using file-based resolver to capture converted layers") var err error exportResolver, err = builder.NewFileBasedResolver(importResolver.Store(), importResolver.ImageStore()) if err != nil { logrus.Errorf("failed to create file-based resolver: %v", err) os.Exit(1) } - repo = tarExportRepo + if !directUpload { + // For tar/dir export, override repo to a synthetic local value so the builder + // does not attempt a real registry push. + repo = tarExportRepo + } customResolver = exportResolver // Setup cleanup for export resolver temporary directory. @@ -367,6 +387,29 @@ Version: ` + commitID, } logrus.Info("dir export finished") } + // Handle direct upload if requested + if directUpload && exportResolver != nil { + imageRef := repo + ":" + overlaybd + logrus.Debugf("uploading converted overlaybd artifacts directly to discoball: %s", imageRef) + regURL := registryURL + if regURL == "" { + // Derive from the registry portion of repo (first path segment). + registry := strings.SplitN(repo, "/", 2)[0] + regURL = "https://" + registry + } + manifestDigest, err := builder.DirectUploadFromStore( + ctx, + exportResolver.OutputStore(), + exportResolver.OutputImageStore(), + imageRef, + regURL, + ) + if err != nil { + logrus.Errorf("direct upload failed: %v", err) + os.Exit(1) + } + logrus.Infof("direct upload complete: manifest_digest=%s", manifestDigest) + } } if tb != "" { logrus.Info("building [Overlaybd - Turbo OCIv1] image...") @@ -422,6 +465,10 @@ func init() { rootCmd.Flags().StringVar(&outputDir, "output-dir", "", "export converted artifacts to a local directory (requires --import-tar; mutually exclusive with --export-tar)") rootCmd.Flags().StringVar(&tarExportRepo, "tar-export-repo", "localhost/converted", "repository name used in exported tar file (only used with --export-tar)") + // direct upload + rootCmd.Flags().BoolVar(&directUpload, "direct-upload", false, "upload converted artifacts directly to discoball (requires --import-tar and -r/--repository)") + rootCmd.Flags().StringVar(®istryURL, "registry-url", "", "registry base URL for direct upload (e.g. https://disco.runloop.pro); defaults to https://{registry from -r}") + // certification rootCmd.Flags().StringArrayVar(&certDirs, "cert-dir", nil, "In these directories, root CA should be named as *.crt and client cert should be named as *.cert, *.key") rootCmd.Flags().StringArrayVar(&rootCAs, "root-ca", nil, "root CA certificates") From 9af07632ff0d197966627ce18d44d97e8af5e8eb Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Wed, 29 Jul 2026 16:38:10 -0700 Subject: [PATCH 03/15] fix(convertor): add exponential backoff on S3 part upload retries Matches Rust upload_one_part: 100ms, 200ms, 400ms delays before each retry. Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/direct_upload.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index a5c7c278..f3354013 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -14,6 +14,7 @@ import ( "sort" "strings" "sync" + "time" "github.com/containerd/containerd/v2/core/content" "github.com/containerd/containerd/v2/core/images" @@ -383,6 +384,12 @@ func putPart(client *http.Client, presignedURL string, buf []byte) (string, erro const maxRetries = 3 var lastErr error for attempt := 0; attempt <= maxRetries; attempt++ { + if attempt > 0 { + delay := time.Duration(100<<(attempt-1)) * time.Millisecond // 100ms, 200ms, 400ms + logrus.Warnf("retrying S3 part upload (attempt=%d delay=%s): %v", attempt, delay, lastErr) + time.Sleep(delay) + } + req, err := http.NewRequest(http.MethodPut, presignedURL, bytes.NewReader(buf)) if err != nil { return "", err From b201d88bbe7ce75d624f739c42f545abf0bfda40 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Thu, 30 Jul 2026 14:37:34 -0700 Subject: [PATCH 04/15] fix(direct-upload): correct CRC-64/NVME init and xorout CRC-64/NVME requires init=0xFFFFFFFFFFFFFFFF and xorout=0xFFFFFFFFFFFFFFFF. The previous code used crc64.New(table) which starts with init=0 and no final XOR, producing a wrong checksum that S3 rejects with BadDigest. Fix by using crc64.Update starting from 0xFFFFFFFFFFFFFFFF and XORing the result with 0xFFFFFFFFFFFFFFFF before encoding, matching the Rust crc64fast_nvme::Digest::new() behavior. --- cmd/convertor/builder/direct_upload.go | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index f3354013..68c4af32 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -367,12 +367,25 @@ func uploadPartsFromStore( sort.Slice(completed, func(i, j int) bool { return completed[i].Number < completed[j].Number }) // Compute CRC64/NVME by reading the full blob sequentially. - crcHash := crc64.New(crc64NVMETable) - if _, err := io.Copy(crcHash, io.NewSectionReader(ra, 0, totalSize)); err != nil { - return nil, nil, fmt.Errorf("computing CRC64/NVME: %w", err) + // CRC-64/NVME: init=0xFFFFFFFFFFFFFFFF, xorout=0xFFFFFFFFFFFFFFFF. + var crcRunning uint64 = 0xFFFFFFFFFFFFFFFF + chunk := make([]byte, 64*1024) + sr := io.NewSectionReader(ra, 0, totalSize) + for { + n, readErr := sr.Read(chunk) + if n > 0 { + crcRunning = crc64.Update(crcRunning, crc64NVMETable, chunk[:n]) + } + if readErr == io.EOF { + break + } + if readErr != nil { + return nil, nil, fmt.Errorf("computing CRC64/NVME: %w", readErr) + } } + crcRunning ^= 0xFFFFFFFFFFFFFFFF var crcBuf [8]byte - binary.BigEndian.PutUint64(crcBuf[:], crcHash.Sum64()) + binary.BigEndian.PutUint64(crcBuf[:], crcRunning) crcStr := base64.StdEncoding.EncodeToString(crcBuf[:]) return completed, &crcStr, nil @@ -433,7 +446,8 @@ func countMissing(blobs []blobUploadInstruction) int { } func computeCRC64NVME(data []byte) string { - crcVal := crc64.Checksum(data, crc64NVMETable) + // CRC-64/NVME: init=0xFFFFFFFFFFFFFFFF, xorout=0xFFFFFFFFFFFFFFFF. + crcVal := crc64.Update(0xFFFFFFFFFFFFFFFF, crc64NVMETable, data) ^ 0xFFFFFFFFFFFFFFFF var buf [8]byte binary.BigEndian.PutUint64(buf[:], crcVal) return base64.StdEncoding.EncodeToString(buf[:]) From 7b2d2e7b9dd7488c179fa3a0ad9643a73d2dcad8 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Thu, 30 Jul 2026 20:10:06 -0700 Subject: [PATCH 05/15] fix(direct-upload): use reflected CRC-64/NVME polynomial for Go's MakeTable Go's crc64.MakeTable expects the polynomial in reflected (bit-reversed) form, like crc64.ECMA = 0xC96C5795D7870F42 (the reflection of 0x42F0E1EBA9EA3693). The previous code passed 0xAD93D23594C935A9 (the normal polynomial from the NVMe spec) directly, producing a wrong lookup table. The correct reflected form is 0x95AC9329AC4BC9B5 = bit_reverse_64(0xAD93D23594C935A9). --- cmd/convertor/builder/direct_upload.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index 68c4af32..34664207 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -28,8 +28,10 @@ const ( partUploadConcurrency = 4 ) -// CRC64/NVME table (Jones polynomial, reflected form). -var crc64NVMETable = crc64.MakeTable(0xAD93D23594C935A9) +// CRC-64/NVME table. Go's MakeTable expects the reflected (bit-reversed) +// polynomial. The normal polynomial from the NVMe spec is 0xAD93D23594C935A9; +// its 64-bit reflection is 0x95AC9329AC4BC9B5. +var crc64NVMETable = crc64.MakeTable(0x95AC9329AC4BC9B5) // ---- wire types (mirror discoball/registry/handlers/directupload.go) -------- From ca2ae96050e79a8176a9be09a409ef99d21f58d7 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Thu, 30 Jul 2026 20:26:54 -0700 Subject: [PATCH 06/15] fix(direct-upload): use correct CRC-64/NVME polynomial 0x9A6C9329AC4BC9B5 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Go's crc64 package builds reflected CRC tables and applies init/xorout internally (same as the NVMe spec: init=0xFFFFFFFFFFFFFFFF, xorout=0xFFFFFFFFFFFFFFFF). crc64.New/Checksum are correct as-is; the only bug was the polynomial. The original 0xAD93D23594C935A9 is the normal form; the previous "fix" 0x95AC9329AC4BC9B5 is its bit-reversal but still wrong. The correct reflected polynomial verified against the CRC RevEng check value ("123456789" → 0xAE8B14860A799888) is 0x9A6C9329AC4BC9B5. Also reverts the incorrect init/xorout manual application from the previous two commits, which was double-applying the complement. --- cmd/convertor/builder/direct_upload.go | 33 ++++++++------------------ 1 file changed, 10 insertions(+), 23 deletions(-) diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index 34664207..64c41fcb 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -28,10 +28,11 @@ const ( partUploadConcurrency = 4 ) -// CRC-64/NVME table. Go's MakeTable expects the reflected (bit-reversed) -// polynomial. The normal polynomial from the NVMe spec is 0xAD93D23594C935A9; -// its 64-bit reflection is 0x95AC9329AC4BC9B5. -var crc64NVMETable = crc64.MakeTable(0x95AC9329AC4BC9B5) +// CRC-64/NVME table. Go's crc64 package requires the polynomial in reflected +// form and applies init=0xFFFFFFFFFFFFFFFF / xorout=0xFFFFFFFFFFFFFFFF +// internally (matching the NVMe spec). The correct reflected polynomial is +// 0x9A6C9329AC4BC9B5. +var crc64NVMETable = crc64.MakeTable(0x9A6C9329AC4BC9B5) // ---- wire types (mirror discoball/registry/handlers/directupload.go) -------- @@ -369,25 +370,12 @@ func uploadPartsFromStore( sort.Slice(completed, func(i, j int) bool { return completed[i].Number < completed[j].Number }) // Compute CRC64/NVME by reading the full blob sequentially. - // CRC-64/NVME: init=0xFFFFFFFFFFFFFFFF, xorout=0xFFFFFFFFFFFFFFFF. - var crcRunning uint64 = 0xFFFFFFFFFFFFFFFF - chunk := make([]byte, 64*1024) - sr := io.NewSectionReader(ra, 0, totalSize) - for { - n, readErr := sr.Read(chunk) - if n > 0 { - crcRunning = crc64.Update(crcRunning, crc64NVMETable, chunk[:n]) - } - if readErr == io.EOF { - break - } - if readErr != nil { - return nil, nil, fmt.Errorf("computing CRC64/NVME: %w", readErr) - } + crcHash := crc64.New(crc64NVMETable) + if _, err := io.Copy(crcHash, io.NewSectionReader(ra, 0, totalSize)); err != nil { + return nil, nil, fmt.Errorf("computing CRC64/NVME: %w", err) } - crcRunning ^= 0xFFFFFFFFFFFFFFFF var crcBuf [8]byte - binary.BigEndian.PutUint64(crcBuf[:], crcRunning) + binary.BigEndian.PutUint64(crcBuf[:], crcHash.Sum64()) crcStr := base64.StdEncoding.EncodeToString(crcBuf[:]) return completed, &crcStr, nil @@ -448,8 +436,7 @@ func countMissing(blobs []blobUploadInstruction) int { } func computeCRC64NVME(data []byte) string { - // CRC-64/NVME: init=0xFFFFFFFFFFFFFFFF, xorout=0xFFFFFFFFFFFFFFFF. - crcVal := crc64.Update(0xFFFFFFFFFFFFFFFF, crc64NVMETable, data) ^ 0xFFFFFFFFFFFFFFFF + crcVal := crc64.Checksum(data, crc64NVMETable) var buf [8]byte binary.BigEndian.PutUint64(buf[:], crcVal) return base64.StdEncoding.EncodeToString(buf[:]) From a14a2ec343ecf77eac0b47f9925953ce40b68602 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Fri, 31 Jul 2026 06:56:40 -0700 Subject: [PATCH 07/15] docs(direct-upload): clarify CRC-64/NVME polynomial comment --- cmd/convertor/builder/direct_upload.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index 64c41fcb..3cffb0d1 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -28,10 +28,15 @@ const ( partUploadConcurrency = 4 ) -// CRC-64/NVME table. Go's crc64 package requires the polynomial in reflected -// form and applies init=0xFFFFFFFFFFFFFFFF / xorout=0xFFFFFFFFFFFFFFFF -// internally (matching the NVMe spec). The correct reflected polynomial is -// 0x9A6C9329AC4BC9B5. +// CRC-64/NVME lookup table. Go's crc64.MakeTable takes the polynomial in +// reflected form. Go also applies init=0xFFFFFFFFFFFFFFFF and +// xorout=0xFFFFFFFFFFFFFFFF internally (same as the NVMe spec), so +// crc64.New / crc64.Checksum are correct as-is without manual init/xorout. +// +// The polynomial 0x9A6C9329AC4BC9B5 was verified against the CRC RevEng +// catalog check value: crc64.Checksum([]byte("123456789"), table) == +// 0xAE8B14860A799888, and produces the same output as the Rust +// crc64fast_nvme crate used on the server side. var crc64NVMETable = crc64.MakeTable(0x9A6C9329AC4BC9B5) // ---- wire types (mirror discoball/registry/handlers/directupload.go) -------- From dae7e4bd4caa45089793c977f4dfa9a6ee8124da Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Fri, 31 Jul 2026 06:59:19 -0700 Subject: [PATCH 08/15] test(direct-upload): add CRC-64/NVME check vector test --- cmd/convertor/builder/direct_upload_test.go | 27 +++++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 cmd/convertor/builder/direct_upload_test.go diff --git a/cmd/convertor/builder/direct_upload_test.go b/cmd/convertor/builder/direct_upload_test.go new file mode 100644 index 00000000..e12812ef --- /dev/null +++ b/cmd/convertor/builder/direct_upload_test.go @@ -0,0 +1,27 @@ +package builder + +import ( + "hash/crc64" + "testing" +) + +func TestCRC64NVMECheckVector(t *testing.T) { + tests := []struct { + name string + input []byte + wantHex uint64 + }{ + // Standard check value from the CRC RevEng catalog for CRC-64/NVME. + {"check vector", []byte("123456789"), 0xAE8B14860A799888}, + // Empty input: init XOR xorout = 0xFFFF... XOR 0xFFFF... = 0. + {"empty", []byte{}, 0x0000000000000000}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := crc64.Checksum(tt.input, crc64NVMETable) + if got != tt.wantHex { + t.Errorf("CRC-64/NVME(%q) = 0x%016X, want 0x%016X", tt.input, got, tt.wantHex) + } + }) + } +} From dbae9f73c0cf1bf6f6595163b1b388aa7ad99ea8 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Fri, 31 Jul 2026 11:03:58 -0700 Subject: [PATCH 09/15] test(direct-upload): add unit tests for wire types and URL parsing Adds unit tests mirroring the Rust tests in container-registry/src/direct_upload.rs: - manifest field base64-encodes in prepare request - confirm request omits tag when nil, includes it when set - prepare response parses existing and pending blobs correctly - parseImageRef handles registry/repository/tag splitting Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/direct_upload_test.go | 154 ++++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/cmd/convertor/builder/direct_upload_test.go b/cmd/convertor/builder/direct_upload_test.go index e12812ef..9976c54e 100644 --- a/cmd/convertor/builder/direct_upload_test.go +++ b/cmd/convertor/builder/direct_upload_test.go @@ -1,10 +1,164 @@ package builder import ( + "encoding/base64" + "encoding/json" "hash/crc64" "testing" ) +func TestPrepareRequestManifestBase64Encodes(t *testing.T) { + body := prepareDirectUploadRequest{ + Manifest: []byte("hello"), + MediaType: ociManifestV1MediaType, + } + raw, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("unmarshal: %v", err) + } + // Go encodes []byte as base64; base64("hello") = "aGVsbG8=" + var got string + if err := json.Unmarshal(m["manifest"], &got); err != nil { + t.Fatalf("unmarshal manifest field: %v", err) + } + if want := base64.StdEncoding.EncodeToString([]byte("hello")); got != want { + t.Errorf("manifest = %q, want %q", got, want) + } + var mediaType string + if err := json.Unmarshal(m["media_type"], &mediaType); err != nil { + t.Fatalf("unmarshal media_type: %v", err) + } + if mediaType != ociManifestV1MediaType { + t.Errorf("media_type = %q, want %q", mediaType, ociManifestV1MediaType) + } +} + +func TestConfirmRequestOmitsTagWhenEmpty(t *testing.T) { + body := confirmDirectUploadRequest{ + Manifest: []byte("m"), + MediaType: ociManifestV1MediaType, + Tag: nil, + Blobs: nil, + } + raw, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var m map[string]json.RawMessage + json.Unmarshal(raw, &m) + if _, ok := m["tag"]; ok { + t.Error("tag field should be omitted when nil") + } +} + +func TestConfirmRequestIncludesTagWhenSet(t *testing.T) { + tag := "snp_abc" + body := confirmDirectUploadRequest{ + Manifest: []byte("m"), + MediaType: ociManifestV1MediaType, + Tag: &tag, + Blobs: nil, + } + raw, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var m map[string]json.RawMessage + json.Unmarshal(raw, &m) + var got string + if err := json.Unmarshal(m["tag"], &got); err != nil { + t.Fatalf("unmarshal tag: %v", err) + } + if got != tag { + t.Errorf("tag = %q, want %q", got, tag) + } +} + +func TestPrepareResponseParsesExistingAndPendingBlobs(t *testing.T) { + raw := `{ + "blobs": [ + {"digest": "sha256:aaa", "exists": true}, + {"digest": "sha256:bbb", "size": 12345, "token": "abc.def", + "parts": [{"number": 1, "url": "https://s3/?sig=1"}], + "part_size": 5242880} + ] + }` + var resp prepareDirectUploadResponse + if err := json.Unmarshal([]byte(raw), &resp); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(resp.Blobs) != 2 { + t.Fatalf("blobs len = %d, want 2", len(resp.Blobs)) + } + if !resp.Blobs[0].Exists { + t.Error("blobs[0].exists should be true") + } + if resp.Blobs[0].Token != nil { + t.Error("blobs[0].token should be nil") + } + if resp.Blobs[1].Exists { + t.Error("blobs[1].exists should be false") + } + if resp.Blobs[1].Token == nil || *resp.Blobs[1].Token != "abc.def" { + t.Errorf("blobs[1].token = %v, want \"abc.def\"", resp.Blobs[1].Token) + } + if resp.Blobs[1].PartSize == nil || *resp.Blobs[1].PartSize != 5_242_880 { + t.Errorf("blobs[1].part_size = %v, want 5242880", resp.Blobs[1].PartSize) + } + if len(resp.Blobs[1].Parts) != 1 { + t.Fatalf("blobs[1].parts len = %d, want 1", len(resp.Blobs[1].Parts)) + } + if resp.Blobs[1].Parts[0].Number != 1 || resp.Blobs[1].Parts[0].URL != "https://s3/?sig=1" { + t.Errorf("blobs[1].parts[0] = %+v", resp.Blobs[1].Parts[0]) + } +} + +func TestParseImageRef(t *testing.T) { + tests := []struct { + input string + registry string + repository string + tag string + wantErr bool + }{ + { + "disco.runloop.ai/repo/foo:snp_x", + "disco.runloop.ai", "repo/foo", "snp_x", false, + }, + { + "disco.runloop.ai/account:bpt_abc", + "disco.runloop.ai", "account", "bpt_abc", false, + }, + { + "disco.runloop.ai/repo/foo", + "disco.runloop.ai", "repo/foo", "", false, + }, + {"no-slash", "", "", "", true}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + reg, repo, tag, err := parseImageRef(tt.input) + if tt.wantErr { + if err == nil { + t.Error("expected error, got nil") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if reg != tt.registry || repo != tt.repository || tag != tt.tag { + t.Errorf("got (%q, %q, %q), want (%q, %q, %q)", + reg, repo, tag, tt.registry, tt.repository, tt.tag) + } + }) + } +} + func TestCRC64NVMECheckVector(t *testing.T) { tests := []struct { name string From dceca7d8f30d6a76f73a0efa9ec9080296ebf375 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Fri, 31 Jul 2026 11:20:14 -0700 Subject: [PATCH 10/15] refactor(convertor): remove --output-dir flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --output-dir was an intermediate step toward direct upload — it wrote converted blobs to disk so a separate blueprint-uploader tool could upload them. That tool was removed in favor of --direct-upload, which handles the full convert-and-upload pipeline in one step. Nothing calls --output-dir in the current build flow. Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/dir_exporter.go | 173 -------------------------- cmd/convertor/main.go | 35 ++---- 2 files changed, 7 insertions(+), 201 deletions(-) delete mode 100644 cmd/convertor/builder/dir_exporter.go diff --git a/cmd/convertor/builder/dir_exporter.go b/cmd/convertor/builder/dir_exporter.go deleted file mode 100644 index 215d66a0..00000000 --- a/cmd/convertor/builder/dir_exporter.go +++ /dev/null @@ -1,173 +0,0 @@ -/* - Copyright The Accelerated Container Image Authors - - Licensed under the Apache License, Version 2.0 (the "License"); - you may not use this file except in compliance with the License. - You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, software - distributed under the License is distributed on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - See the License for the specific language governing permissions and - limitations under the License. -*/ - -package builder - -import ( - "context" - "encoding/json" - "fmt" - "io" - "os" - "path/filepath" - "strings" - - "github.com/containerd/containerd/v2/core/content" - "github.com/containerd/containerd/v2/core/images" - "github.com/containerd/log" - "github.com/opencontainers/go-digest" - v1 "github.com/opencontainers/image-spec/specs-go/v1" - "github.com/pkg/errors" -) - -// ExportContentStoreToDir writes converted OCI image artifacts from the content -// store to outputDir in a flat layout for consumption by blueprint-uploader. -// -// Output layout: -// -// manifest.json - OCI manifest JSON -// config.json - OCI image config JSON -// config.digest - "sha256:" of config.json -// blobs/ - layer blobs, one file per layer (named by digest with -// ':' replaced by '-', e.g. "sha256-abc123...") -// -// Only single-arch OCI manifests are supported; multi-arch index entries are skipped. -func ExportContentStoreToDir(ctx context.Context, store content.Store, imageStore images.Store, outputDir string) error { - if err := os.MkdirAll(outputDir, 0755); err != nil { - return errors.Wrapf(err, "failed to create output directory %q", outputDir) - } - - imgs, err := imageStore.List(ctx) - if err != nil { - return errors.Wrapf(err, "failed to list images in output store") - } - if len(imgs) == 0 { - return errors.New("no images in output store after conversion") - } - - // Find the first non-index image (for overlaybd OCI base builds there is one per arch). - var manifestDesc v1.Descriptor - for _, img := range imgs { - if img.Target.MediaType != v1.MediaTypeImageIndex && - img.Target.MediaType != "application/vnd.docker.distribution.manifest.list.v2+json" { - manifestDesc = img.Target - break - } - } - if manifestDesc.Digest == "" { - return errors.New("no single-arch manifest found in output store") - } - - // Read and write manifest.json. - manifestBytes, err := content.ReadBlob(ctx, store, manifestDesc) - if err != nil { - return errors.Wrapf(err, "failed to read manifest blob") - } - if err := os.WriteFile(filepath.Join(outputDir, "manifest.json"), manifestBytes, 0644); err != nil { - return errors.Wrapf(err, "failed to write manifest.json") - } - log.G(ctx).Debugf("dir exporter: wrote manifest.json (%d bytes)", len(manifestBytes)) - - // Parse manifest to find config and layer descriptors. - var manifest struct { - Config struct { - MediaType string `json:"mediaType"` - Digest string `json:"digest"` - Size int64 `json:"size"` - } `json:"config"` - Layers []struct { - MediaType string `json:"mediaType"` - Digest string `json:"digest"` - Size int64 `json:"size"` - } `json:"layers"` - } - if err := json.Unmarshal(manifestBytes, &manifest); err != nil { - return errors.Wrapf(err, "failed to parse manifest JSON") - } - - // Write config.json and config.digest. - configDigestStr := manifest.Config.Digest - cd, err := digest.Parse(configDigestStr) - if err != nil { - return errors.Wrapf(err, "failed to parse config digest %q", configDigestStr) - } - configSpec := v1.Descriptor{ - MediaType: manifest.Config.MediaType, - Digest: cd, - Size: manifest.Config.Size, - } - configBytes, err := content.ReadBlob(ctx, store, configSpec) - if err != nil { - return errors.Wrapf(err, "failed to read config blob") - } - if err := os.WriteFile(filepath.Join(outputDir, "config.json"), configBytes, 0644); err != nil { - return errors.Wrapf(err, "failed to write config.json") - } - if err := os.WriteFile(filepath.Join(outputDir, "config.digest"), []byte(configDigestStr), 0644); err != nil { - return errors.Wrapf(err, "failed to write config.digest") - } - log.G(ctx).Debugf("dir exporter: wrote config.json (digest: %s)", configDigestStr) - - // Write each layer blob to blobs/. - if len(manifest.Layers) == 0 { - return errors.New("manifest has no layers") - } - blobsDir := filepath.Join(outputDir, "blobs") - if err := os.MkdirAll(blobsDir, 0755); err != nil { - return errors.Wrapf(err, "failed to create blobs directory") - } - - for i, layer := range manifest.Layers { - layerDigestStr := layer.Digest - ld, err := digest.Parse(layerDigestStr) - if err != nil { - return errors.Wrapf(err, "failed to parse layer[%d] digest %q", i, layerDigestStr) - } - layerSpec := v1.Descriptor{ - MediaType: layer.MediaType, - Digest: ld, - Size: layer.Size, - } - // Name the file by digest with ':' replaced by '-' for filesystem safety. - blobFilename := strings.ReplaceAll(layerDigestStr, ":", "-") - blobPath := filepath.Join(blobsDir, blobFilename) - if err := copyBlobToFile(ctx, store, layerSpec, blobPath); err != nil { - return errors.Wrapf(err, "failed to write layer[%d] blob (digest: %s)", i, layerDigestStr) - } - log.G(ctx).Debugf("dir exporter: wrote blobs/%s (%s bytes)", blobFilename, fmt.Sprintf("%d", layer.Size)) - } - - log.G(ctx).Infof("dir exporter: wrote %d layer blob(s) to %s", len(manifest.Layers), outputDir) - return nil -} - -// copyBlobToFile streams the blob identified by desc from store to the file at path. -func copyBlobToFile(ctx context.Context, store content.Store, desc v1.Descriptor, path string) error { - ra, err := store.ReaderAt(ctx, desc) - if err != nil { - return errors.Wrapf(err, "failed to open blob %s for reading", desc.Digest) - } - defer ra.Close() - - f, err := os.Create(path) - if err != nil { - return errors.Wrapf(err, "failed to create file %q", path) - } - defer f.Close() - - _, err = io.Copy(f, content.NewReader(ra)) - return errors.Wrapf(err, "failed to stream blob to %q", path) -} diff --git a/cmd/convertor/main.go b/cmd/convertor/main.go index eb9d5e06..7fac8deb 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -64,8 +64,6 @@ var ( // tar import/export importTar string exportTar string - outputDir string - tarExportRepo string // direct upload directUpload bool @@ -103,20 +101,12 @@ Version: ` + commitID, logrus.Error("import-tar cannot be used with input-tag or input-digest") os.Exit(1) } - if outputDir != "" && importTar == "" { - logrus.Error("--output-dir requires --import-tar") - os.Exit(1) - } - if outputDir != "" && exportTar != "" { - logrus.Error("--output-dir and --export-tar are mutually exclusive") - os.Exit(1) - } if directUpload && importTar == "" { logrus.Error("--direct-upload requires --import-tar") os.Exit(1) } - if directUpload && (exportTar != "" || outputDir != "") { - logrus.Error("--direct-upload is mutually exclusive with --export-tar and --output-dir") + if directUpload && exportTar != "" { + logrus.Error("--direct-upload is mutually exclusive with --export-tar") os.Exit(1) } if directUpload && repo == "" { @@ -200,7 +190,7 @@ Version: ` + commitID, // Choose resolver based on export mode var customResolver remotes.Resolver - if exportTar != "" || outputDir != "" || directUpload { + if exportTar != "" || directUpload { // For local export or direct upload, use FileBasedResolver to capture converted layers locally. logrus.Debugf("local capture mode: using file-based resolver to capture converted layers") var err error @@ -209,10 +199,10 @@ Version: ` + commitID, logrus.Errorf("failed to create file-based resolver: %v", err) os.Exit(1) } - if !directUpload { - // For tar/dir export, override repo to a synthetic local value so the builder + if exportTar != "" { + // For tar export, override repo to a synthetic local value so the builder // does not attempt a real registry push. - repo = tarExportRepo + repo = "localhost/converted" } customResolver = exportResolver @@ -378,16 +368,7 @@ Version: ` + commitID, } logrus.Info("tar export finished") } - // Handle dir export if requested - if outputDir != "" && exportResolver != nil { - logrus.Debugf("exporting converted overlaybd artifacts to directory: %s", outputDir) - if err := builder.ExportContentStoreToDir(ctx, exportResolver.OutputStore(), exportResolver.OutputImageStore(), outputDir); err != nil { - logrus.Errorf("failed to export to directory: %v", err) - os.Exit(1) - } - logrus.Info("dir export finished") - } - // Handle direct upload if requested + // Handle direct upload if requested if directUpload && exportResolver != nil { imageRef := repo + ":" + overlaybd logrus.Debugf("uploading converted overlaybd artifacts directly to discoball: %s", imageRef) @@ -462,8 +443,6 @@ func init() { // tar import/export rootCmd.Flags().StringVar(&importTar, "import-tar", "", "import image from tar file (OCI layout format)") rootCmd.Flags().StringVar(&exportTar, "export-tar", "", "export converted image to tar file (OCI layout format)") - rootCmd.Flags().StringVar(&outputDir, "output-dir", "", "export converted artifacts to a local directory (requires --import-tar; mutually exclusive with --export-tar)") - rootCmd.Flags().StringVar(&tarExportRepo, "tar-export-repo", "localhost/converted", "repository name used in exported tar file (only used with --export-tar)") // direct upload rootCmd.Flags().BoolVar(&directUpload, "direct-upload", false, "upload converted artifacts directly to discoball (requires --import-tar and -r/--repository)") From 505d7b1b0114bdf14f2d8713f6a9cd3fa90f3003 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Fri, 31 Jul 2026 11:24:32 -0700 Subject: [PATCH 11/15] fix(convertor): restore --tar-export-repo flag removed by mistake The flag belongs to --export-tar, not --output-dir. It was accidentally dropped when --output-dir was removed. Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/main.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cmd/convertor/main.go b/cmd/convertor/main.go index 7fac8deb..6eeab668 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -64,6 +64,7 @@ var ( // tar import/export importTar string exportTar string + tarExportRepo string // direct upload directUpload bool @@ -202,7 +203,7 @@ Version: ` + commitID, if exportTar != "" { // For tar export, override repo to a synthetic local value so the builder // does not attempt a real registry push. - repo = "localhost/converted" + repo = tarExportRepo } customResolver = exportResolver @@ -443,6 +444,7 @@ func init() { // tar import/export rootCmd.Flags().StringVar(&importTar, "import-tar", "", "import image from tar file (OCI layout format)") rootCmd.Flags().StringVar(&exportTar, "export-tar", "", "export converted image to tar file (OCI layout format)") + rootCmd.Flags().StringVar(&tarExportRepo, "tar-export-repo", "localhost/converted", "repository name used in exported tar file (only used with --export-tar)") // direct upload rootCmd.Flags().BoolVar(&directUpload, "direct-upload", false, "upload converted artifacts directly to discoball (requires --import-tar and -r/--repository)") From 4bfe4f52ecfc8b8c59a4edfc0a4fe1644fa2bd8c Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Fri, 31 Jul 2026 23:28:06 +0000 Subject: [PATCH 12/15] fix(convertor): require OCI media types for direct upload --- cmd/convertor/main.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/cmd/convertor/main.go b/cmd/convertor/main.go index 6eeab668..3f3dea9f 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -114,6 +114,10 @@ Version: ` + commitID, logrus.Error("--direct-upload requires -r/--repository") os.Exit(1) } + if directUpload && !oci { + logrus.Error("--direct-upload requires --oci") + os.Exit(1) + } if importTar == "" && repo == "" { logrus.Error("repository is required when not using import-tar") os.Exit(1) From c74abb1b606e2cd87b04db94d034398501e8fada Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Wed, 5 Aug 2026 10:08:13 -0700 Subject: [PATCH 13/15] perf(convertor): pipeline layer uploads with conversion via per-blob prepare Previously DirectUploadFromStore ran after Build() completed, serializing conversion (21s) and upload (23s) for a total of ~44s. This change overlaps them: each layer is uploaded to S3 immediately after it finishes converting, while the next layer is still being converted. Changes: - Add DirectUploadPipeline that calls a new per-blob prepare API (no manifest needed) and uploads each layer as it finishes, collecting confirm tokens. - Wire Pipeline into BuilderOptions and builderEngineBase; overlaybd_builder calls pipeline.UploadBlob at the end of UploadLayer. - main.go creates the pipeline before Build() and calls DirectUploadWithPipeline after, which handles remaining blobs (config, base layer) and confirms with all pre-collected tokens. Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/builder.go | 6 + cmd/convertor/builder/builder_engine.go | 1 + cmd/convertor/builder/direct_upload.go | 245 ++++++++++++++++++++- cmd/convertor/builder/overlaybd_builder.go | 5 + cmd/convertor/main.go | 33 ++- 5 files changed, 282 insertions(+), 8 deletions(-) diff --git a/cmd/convertor/builder/builder.go b/cmd/convertor/builder/builder.go index 5befe1b2..a6a13b4f 100644 --- a/cmd/convertor/builder/builder.go +++ b/cmd/convertor/builder/builder.go @@ -81,6 +81,11 @@ type BuilderOptions struct { // CustomResolver allows using a custom resolver instead of the default docker resolver // Used for tar import/export functionality CustomResolver remotes.Resolver + + // Pipeline, when set, uploads each converted layer blob to S3 immediately + // after it finishes converting, overlapping upload time with conversion of + // subsequent layers. + Pipeline *DirectUploadPipeline } type graphBuilder struct { @@ -300,6 +305,7 @@ func (b *graphBuilder) buildOne(ctx context.Context, src v1.Descriptor, tag bool engineBase.noUpload = b.NoUpload engineBase.dumpManifest = b.DumpManifest engineBase.retryCount = b.RetryCount + engineBase.pipeline = b.Pipeline if _, ok := b.Resolver.(*FileBasedResolver); ok { engineBase.tarExport = true } diff --git a/cmd/convertor/builder/builder_engine.go b/cmd/convertor/builder/builder_engine.go index bccb6efa..bf8dc36e 100644 --- a/cmd/convertor/builder/builder_engine.go +++ b/cmd/convertor/builder/builder_engine.go @@ -118,6 +118,7 @@ type builderEngineBase struct { referrer bool tarExport bool retryCount int + pipeline *DirectUploadPipeline } func (e *builderEngineBase) isGzipLayer(ctx context.Context, idx int) (bool, error) { diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index 3cffb0d1..d75b4a39 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -42,10 +42,17 @@ var crc64NVMETable = crc64.MakeTable(0x9A6C9329AC4BC9B5) // ---- wire types (mirror discoball/registry/handlers/directupload.go) -------- // prepareDirectUploadRequest is sent to the prepare endpoint. -// The manifest field is Go []byte, which JSON-encodes as base64. +// Either (Manifest + MediaType) or Blobs must be set. type prepareDirectUploadRequest struct { - Manifest []byte `json:"manifest"` - MediaType string `json:"media_type"` + Manifest []byte `json:"manifest,omitempty"` + MediaType string `json:"media_type,omitempty"` + Blobs []blobDescriptorReq `json:"blobs,omitempty"` +} + +// blobDescriptorReq is a single blob descriptor for the per-blob prepare path. +type blobDescriptorReq struct { + Digest string `json:"digest"` + Size int64 `json:"size"` } type blobUploadInstruction struct { @@ -216,6 +223,207 @@ func DirectUploadFromStore( return manifestDigest, nil } +// ---- DirectUploadPipeline --------------------------------------------------- + +// DirectUploadPipeline uploads each converted layer blob to S3 immediately +// after it finishes converting, overlapping upload time with conversion of +// subsequent layers. Call UploadBlob from UploadLayer, then pass the pipeline +// to DirectUploadWithPipeline after Build completes. +type DirectUploadPipeline struct { + client *http.Client + baseURL string + store content.Store + mu sync.Mutex + tokens map[digest.Digest]blobConfirmEntry +} + +// NewDirectUploadPipeline creates a pipeline that uploads blobs to the given +// discoball endpoint as they finish converting. store must be the output +// content store that layers are written to during Build. +func NewDirectUploadPipeline(store content.Store, imageRef, registryURL string) (*DirectUploadPipeline, error) { + scheme := "https" + if strings.HasPrefix(registryURL, "http://") { + scheme = "http" + } + _, repository, _, err := parseImageRef(imageRef) + if err != nil { + return nil, fmt.Errorf("invalid image ref %q: %w", imageRef, err) + } + registry := strings.SplitN(strings.TrimPrefix(strings.TrimPrefix(imageRef, "https://"), "http://"), "/", 2)[0] + baseURL := fmt.Sprintf("%s://%s/gitlab/v1/repositories/%s/direct-upload", scheme, registry, repository) + return &DirectUploadPipeline{ + client: &http.Client{}, + baseURL: baseURL, + store: store, + tokens: make(map[digest.Digest]blobConfirmEntry), + }, nil +} + +// UploadBlob prepares and uploads a single blob immediately. If the blob +// already exists in the repository it is skipped. Safe to call concurrently. +func (p *DirectUploadPipeline) UploadBlob(ctx context.Context, desc ocispec.Descriptor) error { + instr, err := prepareSingleBlob(ctx, p.client, p.baseURL, desc.Digest, desc.Size) + if err != nil { + return fmt.Errorf("prepare blob %s: %w", desc.Digest, err) + } + if instr.Exists { + logrus.Debugf("pipeline: blob %s already exists, skipping", desc.Digest) + return nil + } + var partSize int64 + if instr.PartSize != nil { + partSize = *instr.PartSize + } + completed, crc64nvme, err := uploadPartsFromStore(ctx, p.client, p.store, desc, instr.Parts, partSize) + if err != nil { + return fmt.Errorf("upload blob %s: %w", desc.Digest, err) + } + p.mu.Lock() + p.tokens[desc.Digest] = blobConfirmEntry{ + Digest: desc.Digest.String(), + Token: *instr.Token, + Parts: completed, + CRC64NVME: crc64nvme, + } + p.mu.Unlock() + logrus.Debugf("pipeline: uploaded blob %s", desc.Digest) + return nil +} + +func (p *DirectUploadPipeline) getToken(dgst digest.Digest) (blobConfirmEntry, bool) { + p.mu.Lock() + defer p.mu.Unlock() + e, ok := p.tokens[dgst] + return e, ok +} + +// DirectUploadWithPipeline is like DirectUploadFromStore but reuses tokens +// collected by pipeline for blobs already uploaded during conversion. Only +// remaining blobs (config, base layer) are uploaded in this call. +func DirectUploadWithPipeline( + ctx context.Context, + store content.Store, + imageStore images.Store, + imageRef string, + registryURL string, + pipeline *DirectUploadPipeline, +) (string, error) { + scheme := "https" + if strings.HasPrefix(registryURL, "http://") { + scheme = "http" + } + + registry, repository, tag, err := parseImageRef(imageRef) + if err != nil { + return "", fmt.Errorf("invalid image ref %q: %w", imageRef, err) + } + logrus.Debugf("direct-upload: registry=%s repository=%s tag=%s", registry, repository, tag) + + imgs, err := imageStore.List(ctx, "") + if err != nil { + return "", fmt.Errorf("listing image store: %w", err) + } + var manifestDesc *ocispec.Descriptor + for _, img := range imgs { + if img.Target.MediaType == ocispec.MediaTypeImageIndex { + continue + } + manifestDesc = &img.Target + break + } + if manifestDesc == nil { + return "", fmt.Errorf("no non-index image found in output store") + } + + manifestBytes, err := content.ReadBlob(ctx, store, *manifestDesc) + if err != nil { + return "", fmt.Errorf("reading manifest: %w", err) + } + + var manifest ocispec.Manifest + if err := json.Unmarshal(manifestBytes, &manifest); err != nil { + return "", fmt.Errorf("parsing manifest: %w", err) + } + + client := pipeline.client + baseURL := fmt.Sprintf("%s://%s/gitlab/v1/repositories/%s/direct-upload", scheme, registry, repository) + + // Phase 1: prepare with full manifest to discover any remaining blobs. + prep, err := prepareUpload(ctx, client, baseURL, manifestBytes) + if err != nil { + return "", fmt.Errorf("prepare_direct_upload failed: %w", err) + } + logrus.Infof("direct-upload prepare: total=%d missing=%d", len(prep.Blobs), countMissing(prep.Blobs)) + + // Phase 2: for each missing blob, use pipeline token if available; else upload now. + var confirmBlobs []blobConfirmEntry + for _, instruction := range prep.Blobs { + if instruction.Exists { + continue + } + instDigest := digest.Digest(instruction.Digest) + + // Reuse token from pipelined upload if this blob was already uploaded. + if entry, ok := pipeline.getToken(instDigest); ok { + confirmBlobs = append(confirmBlobs, entry) + continue + } + + token := *instruction.Token + parts := instruction.Parts + var partSize int64 + if instruction.PartSize != nil { + partSize = *instruction.PartSize + } + + var completed []completedPart + var crc64nvme *string + + if instDigest == manifest.Config.Digest { + configBytes, err := content.ReadBlob(ctx, store, manifest.Config) + if err != nil { + return "", fmt.Errorf("reading config blob: %w", err) + } + completed, err = uploadPartsFromBytes(client, configBytes, parts, partSize) + if err != nil { + return "", fmt.Errorf("uploading config parts: %w", err) + } + crc := computeCRC64NVME(configBytes) + crc64nvme = &crc + } else { + var layerDesc *ocispec.Descriptor + for i := range manifest.Layers { + if manifest.Layers[i].Digest == instDigest { + layerDesc = &manifest.Layers[i] + break + } + } + if layerDesc == nil { + return "", fmt.Errorf("discoball requested unknown blob: %s", instruction.Digest) + } + completed, crc64nvme, err = uploadPartsFromStore(ctx, client, store, *layerDesc, parts, partSize) + if err != nil { + return "", fmt.Errorf("uploading layer parts (digest=%s): %w", instruction.Digest, err) + } + } + + confirmBlobs = append(confirmBlobs, blobConfirmEntry{ + Digest: instruction.Digest, + Token: token, + Parts: completed, + CRC64NVME: crc64nvme, + }) + } + + // Phase 3: confirm. + manifestDigest, err := confirmUpload(ctx, client, baseURL, manifestBytes, tag, confirmBlobs) + if err != nil { + return "", fmt.Errorf("confirm_direct_upload failed: %w", err) + } + logrus.Infof("direct-upload complete: manifest_digest=%s", manifestDigest) + return manifestDigest, nil +} + // ---- HTTP helpers ----------------------------------------------------------- func prepareUpload(ctx context.Context, client *http.Client, baseURL string, manifestBytes []byte) (*prepareDirectUploadResponse, error) { @@ -246,6 +454,37 @@ func prepareUpload(ctx context.Context, client *http.Client, baseURL string, man return &parsed, nil } +// prepareSingleBlob calls the per-blob prepare endpoint to get a presigned +// upload session for one blob without needing the full image manifest. +func prepareSingleBlob(ctx context.Context, client *http.Client, baseURL string, dgst digest.Digest, size int64) (*blobUploadInstruction, error) { + reqBody, err := json.Marshal(prepareDirectUploadRequest{ + Blobs: []blobDescriptorReq{{Digest: dgst.String(), Size: size}}, + }) + if err != nil { + return nil, err + } + resp, err := doPost(ctx, client, baseURL+"/prepare/", reqBody) + if err != nil { + return nil, err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("reading prepare response: %w", err) + } + if resp.StatusCode/100 != 2 { + return nil, fmt.Errorf("prepare returned HTTP %d: %s", resp.StatusCode, body) + } + var parsed prepareDirectUploadResponse + if err := json.Unmarshal(body, &parsed); err != nil { + return nil, fmt.Errorf("parsing prepare response: %w", err) + } + if len(parsed.Blobs) != 1 { + return nil, fmt.Errorf("expected 1 blob in prepare response, got %d", len(parsed.Blobs)) + } + return &parsed.Blobs[0], nil +} + func confirmUpload(ctx context.Context, client *http.Client, baseURL string, manifestBytes []byte, tag string, blobs []blobConfirmEntry) (string, error) { var tagPtr *string if tag != "" { diff --git a/cmd/convertor/builder/overlaybd_builder.go b/cmd/convertor/builder/overlaybd_builder.go index f2a7f538..07ae7648 100644 --- a/cmd/convertor/builder/overlaybd_builder.go +++ b/cmd/convertor/builder/overlaybd_builder.go @@ -197,6 +197,11 @@ func (e *overlaybdBuilderEngine) UploadLayer(ctx context.Context, idx int) error } } e.overlaybdLayers[idx].desc = desc + if e.pipeline != nil { + if err := e.pipeline.UploadBlob(ctx, desc); err != nil { + return fmt.Errorf("pipeline upload layer %d: %w", idx, err) + } + } return nil } diff --git a/cmd/convertor/main.go b/cmd/convertor/main.go index 3f3dea9f..137afdc1 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -358,6 +358,27 @@ Version: ` + commitID, logrus.Debugf("falling back to no deduplication") } + // Set up pipelined direct upload: each layer is uploaded to S3 + // immediately after it finishes converting, overlapping upload + // with conversion of subsequent layers. + if directUpload && exportResolver != nil { + imageRef := repo + ":" + overlaybd + regURL := registryURL + if regURL == "" { + regURL = "https://" + strings.SplitN(repo, "/", 2)[0] + } + pipeline, err := builder.NewDirectUploadPipeline( + exportResolver.OutputStore(), + imageRef, + regURL, + ) + if err != nil { + logrus.Errorf("failed to create direct upload pipeline: %v", err) + os.Exit(1) + } + opt.Pipeline = pipeline + } + if err := builder.Build(ctx, opt); err != nil { logrus.Errorf("failed to build overlaybd: %v", err) os.Exit(1) @@ -373,22 +394,24 @@ Version: ` + commitID, } logrus.Info("tar export finished") } - // Handle direct upload if requested + + // Handle direct upload: remaining blobs (config, base layer) + // are uploaded here; layer blobs were already uploaded by the + // pipeline during Build. if directUpload && exportResolver != nil { imageRef := repo + ":" + overlaybd logrus.Debugf("uploading converted overlaybd artifacts directly to discoball: %s", imageRef) regURL := registryURL if regURL == "" { - // Derive from the registry portion of repo (first path segment). - registry := strings.SplitN(repo, "/", 2)[0] - regURL = "https://" + registry + regURL = "https://" + strings.SplitN(repo, "/", 2)[0] } - manifestDigest, err := builder.DirectUploadFromStore( + manifestDigest, err := builder.DirectUploadWithPipeline( ctx, exportResolver.OutputStore(), exportResolver.OutputImageStore(), imageRef, regURL, + opt.Pipeline, ) if err != nil { logrus.Errorf("direct upload failed: %v", err) From 00e3f1638670cec48182958a1284964ba540f85a Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Wed, 5 Aug 2026 11:18:07 -0700 Subject: [PATCH 14/15] refactor(direct-upload): remove unused DirectUploadFromStore Replaced by DirectUploadWithPipeline which pipelines layer uploads with conversion. Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/direct_upload.go | 132 ------------------------- 1 file changed, 132 deletions(-) diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index d75b4a39..50e9c88e 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -91,138 +91,6 @@ type confirmDirectUploadRequest struct { Blobs []blobConfirmEntry `json:"blobs"` } -// ---- DirectUploadFromStore -------------------------------------------------- - -// DirectUploadFromStore reads the converted OCI image from the content store -// captured by FileBasedResolver and pushes blobs directly to S3 via discoball's -// prepare/confirm API. Returns the manifest digest stored by discoball. -// -// imageRef must be the full reference (e.g. "disco.runloop.pro/account:bpt_xxx"). -// registryURL is used only to determine the scheme -// (e.g. "https://disco.runloop.pro"). -func DirectUploadFromStore( - ctx context.Context, - store content.Store, - imageStore images.Store, - imageRef string, - registryURL string, -) (string, error) { - scheme := "https" - if strings.HasPrefix(registryURL, "http://") { - scheme = "http" - } - - registry, repository, tag, err := parseImageRef(imageRef) - if err != nil { - return "", fmt.Errorf("invalid image ref %q: %w", imageRef, err) - } - logrus.Debugf("direct-upload: registry=%s repository=%s tag=%s", registry, repository, tag) - - // Find the converted image in the store. - imgs, err := imageStore.List(ctx, "") - if err != nil { - return "", fmt.Errorf("listing image store: %w", err) - } - var manifestDesc *ocispec.Descriptor - for _, img := range imgs { - if img.Target.MediaType == ocispec.MediaTypeImageIndex { - continue - } - manifestDesc = &img.Target - break - } - if manifestDesc == nil { - return "", fmt.Errorf("no non-index image found in output store") - } - - // Read manifest bytes verbatim — reusing the on-disk bytes ensures the - // manifest digest stored by discoball matches what the content store holds. - manifestBytes, err := content.ReadBlob(ctx, store, *manifestDesc) - if err != nil { - return "", fmt.Errorf("reading manifest: %w", err) - } - - var manifest ocispec.Manifest - if err := json.Unmarshal(manifestBytes, &manifest); err != nil { - return "", fmt.Errorf("parsing manifest: %w", err) - } - - client := &http.Client{} - baseURL := fmt.Sprintf("%s://%s/gitlab/v1/repositories/%s/direct-upload", scheme, registry, repository) - - // Phase 1: prepare. - prep, err := prepareUpload(ctx, client, baseURL, manifestBytes) - if err != nil { - return "", fmt.Errorf("prepare_direct_upload failed: %w", err) - } - logrus.Infof("direct-upload prepare: total=%d missing=%d", len(prep.Blobs), countMissing(prep.Blobs)) - - // Phase 2: upload missing blobs. - var confirmBlobs []blobConfirmEntry - for _, instruction := range prep.Blobs { - if instruction.Exists { - continue - } - token := *instruction.Token - parts := instruction.Parts - var partSize int64 - if instruction.PartSize != nil { - partSize = *instruction.PartSize - } - - // Determine which blob this is. - instDigest := digest.Digest(instruction.Digest) - var completed []completedPart - var crc64nvme *string - - if instDigest == manifest.Config.Digest { - // Config blob: small, load into memory. - configDesc := manifest.Config - configBytes, err := content.ReadBlob(ctx, store, configDesc) - if err != nil { - return "", fmt.Errorf("reading config blob: %w", err) - } - completed, err = uploadPartsFromBytes(client, configBytes, parts, partSize) - if err != nil { - return "", fmt.Errorf("uploading config parts: %w", err) - } - crc := computeCRC64NVME(configBytes) - crc64nvme = &crc - } else { - // Layer blob: stream from content store. - var layerDesc *ocispec.Descriptor - for i := range manifest.Layers { - if manifest.Layers[i].Digest == instDigest { - layerDesc = &manifest.Layers[i] - break - } - } - if layerDesc == nil { - return "", fmt.Errorf("discoball requested unknown blob: %s", instruction.Digest) - } - completed, crc64nvme, err = uploadPartsFromStore(ctx, client, store, *layerDesc, parts, partSize) - if err != nil { - return "", fmt.Errorf("uploading layer parts (digest=%s): %w", instruction.Digest, err) - } - } - - confirmBlobs = append(confirmBlobs, blobConfirmEntry{ - Digest: instruction.Digest, - Token: token, - Parts: completed, - CRC64NVME: crc64nvme, - }) - } - - // Phase 3: confirm. - manifestDigest, err := confirmUpload(ctx, client, baseURL, manifestBytes, tag, confirmBlobs) - if err != nil { - return "", fmt.Errorf("confirm_direct_upload failed: %w", err) - } - logrus.Infof("direct-upload complete: manifest_digest=%s", manifestDigest) - return manifestDigest, nil -} - // ---- DirectUploadPipeline --------------------------------------------------- // DirectUploadPipeline uploads each converted layer blob to S3 immediately From e2f157fcc27efc526bdb8b91fe2d61fcc245ca15 Mon Sep 17 00:00:00 2001 From: Yoon Park Date: Wed, 5 Aug 2026 14:24:24 -0700 Subject: [PATCH 15/15] perf(direct-upload): use per-blob prepare for config only in pipeline path Replace prepareUpload(manifest) with prepareSingleBlob(config) in DirectUploadWithPipeline. The full manifest prepare was opening S3 multipart sessions for all blobs including layers already uploaded by the pipeline, wasting ~5 S3 API calls. Now only the config blob gets a new session. Co-Authored-By: Claude Sonnet 4.6 --- cmd/convertor/builder/direct_upload.go | 126 ++++++------------------- 1 file changed, 29 insertions(+), 97 deletions(-) diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go index 50e9c88e..5f80a883 100644 --- a/cmd/convertor/builder/direct_upload.go +++ b/cmd/convertor/builder/direct_upload.go @@ -165,9 +165,9 @@ func (p *DirectUploadPipeline) getToken(dgst digest.Digest) (blobConfirmEntry, b return e, ok } -// DirectUploadWithPipeline is like DirectUploadFromStore but reuses tokens -// collected by pipeline for blobs already uploaded during conversion. Only -// remaining blobs (config, base layer) are uploaded in this call. +// DirectUploadWithPipeline completes a pipelined direct upload. Layer blobs +// were already uploaded to S3 during Build via UploadBlob; this call uploads +// only the config blob and confirms all blobs with discoball. func DirectUploadWithPipeline( ctx context.Context, store content.Store, @@ -216,74 +216,44 @@ func DirectUploadWithPipeline( client := pipeline.client baseURL := fmt.Sprintf("%s://%s/gitlab/v1/repositories/%s/direct-upload", scheme, registry, repository) - // Phase 1: prepare with full manifest to discover any remaining blobs. - prep, err := prepareUpload(ctx, client, baseURL, manifestBytes) - if err != nil { - return "", fmt.Errorf("prepare_direct_upload failed: %w", err) - } - logrus.Infof("direct-upload prepare: total=%d missing=%d", len(prep.Blobs), countMissing(prep.Blobs)) - - // Phase 2: for each missing blob, use pipeline token if available; else upload now. + // Collect pipeline tokens for layer blobs uploaded during Build. + // Layers not in the pipeline were already present in the registry and + // don't need a token — discoball's closure check will verify them. var confirmBlobs []blobConfirmEntry - for _, instruction := range prep.Blobs { - if instruction.Exists { - continue - } - instDigest := digest.Digest(instruction.Digest) - - // Reuse token from pipelined upload if this blob was already uploaded. - if entry, ok := pipeline.getToken(instDigest); ok { + for _, layer := range manifest.Layers { + if entry, ok := pipeline.getToken(layer.Digest); ok { confirmBlobs = append(confirmBlobs, entry) - continue } + } - token := *instruction.Token - parts := instruction.Parts + // Upload only the config blob via per-blob prepare, avoiding the overhead + // of opening S3 sessions for all blobs when layers are already uploaded. + configInstr, err := prepareSingleBlob(ctx, client, baseURL, manifest.Config.Digest, manifest.Config.Size) + if err != nil { + return "", fmt.Errorf("prepare config blob: %w", err) + } + if !configInstr.Exists { var partSize int64 - if instruction.PartSize != nil { - partSize = *instruction.PartSize + if configInstr.PartSize != nil { + partSize = *configInstr.PartSize } - - var completed []completedPart - var crc64nvme *string - - if instDigest == manifest.Config.Digest { - configBytes, err := content.ReadBlob(ctx, store, manifest.Config) - if err != nil { - return "", fmt.Errorf("reading config blob: %w", err) - } - completed, err = uploadPartsFromBytes(client, configBytes, parts, partSize) - if err != nil { - return "", fmt.Errorf("uploading config parts: %w", err) - } - crc := computeCRC64NVME(configBytes) - crc64nvme = &crc - } else { - var layerDesc *ocispec.Descriptor - for i := range manifest.Layers { - if manifest.Layers[i].Digest == instDigest { - layerDesc = &manifest.Layers[i] - break - } - } - if layerDesc == nil { - return "", fmt.Errorf("discoball requested unknown blob: %s", instruction.Digest) - } - completed, crc64nvme, err = uploadPartsFromStore(ctx, client, store, *layerDesc, parts, partSize) - if err != nil { - return "", fmt.Errorf("uploading layer parts (digest=%s): %w", instruction.Digest, err) - } + configBytes, err := content.ReadBlob(ctx, store, manifest.Config) + if err != nil { + return "", fmt.Errorf("reading config blob: %w", err) } - + completed, err := uploadPartsFromBytes(client, configBytes, configInstr.Parts, partSize) + if err != nil { + return "", fmt.Errorf("uploading config parts: %w", err) + } + crc := computeCRC64NVME(configBytes) confirmBlobs = append(confirmBlobs, blobConfirmEntry{ - Digest: instruction.Digest, - Token: token, + Digest: manifest.Config.Digest.String(), + Token: *configInstr.Token, Parts: completed, - CRC64NVME: crc64nvme, + CRC64NVME: &crc, }) } - // Phase 3: confirm. manifestDigest, err := confirmUpload(ctx, client, baseURL, manifestBytes, tag, confirmBlobs) if err != nil { return "", fmt.Errorf("confirm_direct_upload failed: %w", err) @@ -294,34 +264,6 @@ func DirectUploadWithPipeline( // ---- HTTP helpers ----------------------------------------------------------- -func prepareUpload(ctx context.Context, client *http.Client, baseURL string, manifestBytes []byte) (*prepareDirectUploadResponse, error) { - reqBody, err := json.Marshal(prepareDirectUploadRequest{ - Manifest: manifestBytes, - MediaType: ociManifestV1MediaType, - }) - if err != nil { - return nil, err - } - resp, err := doPost(ctx, client, baseURL+"/prepare/", reqBody) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - body, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("reading prepare response: %w", err) - } - if resp.StatusCode/100 != 2 { - return nil, fmt.Errorf("prepare returned HTTP %d: %s", resp.StatusCode, body) - } - var parsed prepareDirectUploadResponse - if err := json.Unmarshal(body, &parsed); err != nil { - return nil, fmt.Errorf("parsing prepare response: %w", err) - } - return &parsed, nil -} - // prepareSingleBlob calls the per-blob prepare endpoint to get a presigned // upload session for one blob without needing the full image manifest. func prepareSingleBlob(ctx context.Context, client *http.Client, baseURL string, dgst digest.Digest, size int64) (*blobUploadInstruction, error) { @@ -537,16 +479,6 @@ func putPart(client *http.Client, presignedURL string, buf []byte) (string, erro // ---- Utilities -------------------------------------------------------------- -func countMissing(blobs []blobUploadInstruction) int { - n := 0 - for _, b := range blobs { - if !b.Exists { - n++ - } - } - return n -} - func computeCRC64NVME(data []byte) string { crcVal := crc64.Checksum(data, crc64NVMETable) var buf [8]byte