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 new file mode 100644 index 00000000..5f80a883 --- /dev/null +++ b/cmd/convertor/builder/direct_upload.go @@ -0,0 +1,520 @@ +package builder + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/binary" + "encoding/json" + "fmt" + "hash/crc64" + "io" + "net/http" + "net/url" + "sort" + "strings" + "sync" + "time" + + "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 +) + +// 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) -------- + +// prepareDirectUploadRequest is sent to the prepare endpoint. +// Either (Manifest + MediaType) or Blobs must be set. +type prepareDirectUploadRequest struct { + 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 { + 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"` +} + +// ---- 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 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, + 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) + + // 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 _, layer := range manifest.Layers { + if entry, ok := pipeline.getToken(layer.Digest); ok { + confirmBlobs = append(confirmBlobs, entry) + } + } + + // 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 configInstr.PartSize != nil { + partSize = *configInstr.PartSize + } + 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: manifest.Config.Digest.String(), + Token: *configInstr.Token, + Parts: completed, + CRC64NVME: &crc, + }) + } + + 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 ----------------------------------------------------------- + +// 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 != "" { + 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++ { + 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 + } + 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 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/builder/direct_upload_test.go b/cmd/convertor/builder/direct_upload_test.go new file mode 100644 index 00000000..9976c54e --- /dev/null +++ b/cmd/convertor/builder/direct_upload_test.go @@ -0,0 +1,181 @@ +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 + 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) + } + }) + } +} 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 3cabeacf..137afdc1 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -66,6 +66,10 @@ var ( exportTar string tarExportRepo string + // direct upload + directUpload bool + registryURL string + // certification certDirs []string rootCAs []string @@ -98,6 +102,22 @@ Version: ` + commitID, logrus.Error("import-tar cannot be used with input-tag or input-digest") os.Exit(1) } + if directUpload && importTar == "" { + logrus.Error("--direct-upload requires --import-tar") + os.Exit(1) + } + if directUpload && exportTar != "" { + logrus.Error("--direct-upload is mutually exclusive with --export-tar") + os.Exit(1) + } + if directUpload && repo == "" { + 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) @@ -175,19 +195,23 @@ 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 != "" || 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 exportTar != "" { + // For tar 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 + // Setup cleanup for export resolver temporary directory. defer func() { if !reserve && exportResolver != nil { if err := exportResolver.CleanupTempDir(); err != nil { @@ -334,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) @@ -349,6 +394,31 @@ Version: ` + commitID, } logrus.Info("tar export finished") } + + // 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 == "" { + regURL = "https://" + strings.SplitN(repo, "/", 2)[0] + } + manifestDigest, err := builder.DirectUploadWithPipeline( + ctx, + exportResolver.OutputStore(), + exportResolver.OutputImageStore(), + imageRef, + regURL, + opt.Pipeline, + ) + 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...") @@ -403,6 +473,10 @@ func init() { 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)") + 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")