diff --git a/cmd/convertor/builder/direct_upload.go b/cmd/convertor/builder/direct_upload.go new file mode 100644 index 00000000..3cffb0d1 --- /dev/null +++ b/cmd/convertor/builder/direct_upload.go @@ -0,0 +1,481 @@ +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. +// 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++ { + 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 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/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/main.go b/cmd/convertor/main.go index 3cabeacf..3f3dea9f 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 { @@ -348,6 +372,29 @@ Version: ` + commitID, os.Exit(1) } logrus.Info("tar 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 != "" { @@ -403,6 +450,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")