Skip to content

JFMIG-88 - Replace data-transfer plugin with GET/PUT streaming in transfer-files #1629

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amithins merged 1 commit into from
Nov 4, 2026
Merged

JFMIG-88 - Replace data-transfer plugin with GET/PUT streaming in transfer-files #1629

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 7 additions & 95 deletions artifactory/commands/transferfiles/api/types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,56 +1,20 @@
package api

import (
"fmt"

"github.com/jfrog/jfrog-client-go/utils/log"
)

type ProcessStatusType string
type ChunkFileStatusType string
type ChunkId string
type NodeId string

const (
Done ProcessStatusType = "DONE"
InProgress ProcessStatusType = "IN_PROGRESS"

Success ChunkFileStatusType = "SUCCESS"
Fail ChunkFileStatusType = "FAIL"
SkippedLargeProps ChunkFileStatusType = "SKIPPED_LARGE_PROPS"
SkippedMetadataFile ChunkFileStatusType = "SKIPPED_METADATA_FILE"
SkippedNonEmptyDir ChunkFileStatusType = "SKIPPED_NON_EMPTY_DIR"
Success ChunkFileStatusType = "SUCCESS"
Fail ChunkFileStatusType = "FAIL"
SkippedLargeProps ChunkFileStatusType = "SKIPPED_LARGE_PROPS"
SkippedMetadataFile ChunkFileStatusType = "SKIPPED_METADATA_FILE"
SkippedNonEmptyDir ChunkFileStatusType = "SKIPPED_NON_EMPTY_DIR"
SkippedSourceItemGone ChunkFileStatusType = "SKIPPED_SOURCE_ITEM_GONE"

Phase1 int = 0
Phase2 int = 1
Phase3 int = 2

maxFilesInChunk = 16
// 1 GiB
maxBytesInChunk = 1 << 30
)

type TargetAuth struct {
TargetArtifactoryUrl string `json:"target_artifactory_url,omitempty"`
TargetUsername string `json:"target_username,omitempty"`
TargetPassword string `json:"target_password,omitempty"`
TargetToken string `json:"target_token,omitempty"`
TargetProxyKey string `json:"target_proxy_key,omitempty"`
}

type UploadChunk struct {
// Authentication details of the target server
TargetAuth
// Files and folders to transfer
UploadCandidates []FileRepresentation `json:"upload_candidates,omitempty"`
// True if should check for the existence of artifacts on the target filestore
CheckExistenceInFilestore bool `json:"check_existence_in_filestore,omitempty"`
// True if should skip file filtering in the Data Transfer plugin
SkipFileFiltering bool `json:"skip_file_filtering,omitempty"`
// Minimum file size in bytes for which JFrog CLI performs checksum deploy optimization
MinCheckSumDeploySize int64 `json:"min_checksum_deploy_size,omitempty"`
}

type FileRepresentation struct {
Repo string `json:"repo,omitempty"`
Path string `json:"path,omitempty"`
Expand All @@ -59,26 +23,8 @@ type FileRepresentation struct {
NonEmptyDir bool `json:"non_empty_dir,omitempty"`
}

type UploadChunkResponse struct {
NodeIdResponse
UuidTokenResponse
}

type UploadChunksStatusBody struct {
AwaitingStatusChunks []ChunkId `json:"awaiting_status_chunks,omitempty"`
ChunksToDelete []ChunkId `json:"chunks_to_delete,omitempty"`
}

type UploadChunksStatusResponse struct {
NodeIdResponse
ChunksStatus []ChunkStatus `json:"chunks_status,omitempty"`
DeletedChunks []string `json:"deleted_chunks,omitempty"`
}

type ChunkStatus struct {
UuidTokenResponse
Status ProcessStatusType `json:"status,omitempty"`
Files []FileUploadStatusResponse `json:"files,omitempty"`
Files []FileUploadStatusResponse `json:"files,omitempty"`
}

type FileUploadStatusResponse struct {
Expand All @@ -89,37 +35,3 @@ type FileUploadStatusResponse struct {
StatusCode int `json:"status_code,omitempty"`
Reason string `json:"reason,omitempty"`
}

type NodeIdResponse struct {
NodeId string `json:"node_id,omitempty"`
}

type UuidTokenResponse struct {
UuidToken string `json:"uuid_token,omitempty"`
}

// Append upload candidate to the list of upload candidates. Skip empty directories in build-info repositories.
// file - The upload candidate
// buildInfoRepo - True if this is a build-info repository
func (uc *UploadChunk) AppendUploadCandidateIfNeeded(file FileRepresentation, buildInfoRepo bool) {
if buildInfoRepo && file.Name == "" {
log.Debug(fmt.Sprintf("Skipping unneeded empty dir '%s' in the build-info repository '%s'", file.Path, file.Repo))
return
}
uc.UploadCandidates = append(uc.UploadCandidates, file)
}

// Return true if the chunk contains at least 16 files or at least 1GiB in total
func (uc *UploadChunk) IsChunkFull() bool {
if len(uc.UploadCandidates) >= maxFilesInChunk {
return true
}
var totalSize int64 = 0
for _, uploadCandidate := range uc.UploadCandidates {
totalSize += uploadCandidate.Size
if totalSize > maxBytesInChunk {
return true
}
}
return false
}
51 changes: 0 additions & 51 deletions artifactory/commands/transferfiles/api/types_test.go
View file
Open in desktop

This file was deleted.

8 changes: 4 additions & 4 deletions artifactory/commands/transferfiles/delayedartifactshandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,11 +150,11 @@ func countDelayFilesContent(filePaths []string) (count int, storage int64, err e

func handleDelayedArtifactsFiles(filesToConsume []string, base phaseBase, delayUploadComparisonFunctions []shouldDelayUpload) error {
manager := newTransferManager(base, delayUploadComparisonFunctions)
action := func(pcWrapper *producerConsumerWrapper, uploadChunkChan chan UploadedChunk, delayHelper delayUploadHelper, errorsChannelMng *ErrorsChannelMng) error {
action := func(pcWrapper *producerConsumerWrapper, delayHelper delayUploadHelper, errorsChannelMng *ErrorsChannelMng) error {
if ShouldStop(&base, &delayHelper, errorsChannelMng) {
return nil
}
return consumeDelayedArtifactsFiles(pcWrapper, filesToConsume, uploadChunkChan, base, delayHelper, errorsChannelMng)
return consumeDelayedArtifactsFiles(pcWrapper, filesToConsume, base, delayHelper, errorsChannelMng)
}
delayAction := func(pBase phaseBase, addedDelayFiles []string) error {
// We call this method as a recursion in order to have inner order base on the comparison function list.
Expand All @@ -171,7 +171,7 @@ func handleDelayedArtifactsFiles(filesToConsume []string, base phaseBase, delayU
return manager.doTransferWithProducerConsumer(action, delayAction)
}

func consumeDelayedArtifactsFiles(pcWrapper *producerConsumerWrapper, filesToConsume []string, uploadChunkChan chan UploadedChunk, base phaseBase, delayHelper delayUploadHelper, errorsChannelMng *ErrorsChannelMng) error {
func consumeDelayedArtifactsFiles(pcWrapper *producerConsumerWrapper, filesToConsume []string, base phaseBase, delayHelper delayUploadHelper, errorsChannelMng *ErrorsChannelMng) error {
log.Debug(fmt.Sprintf("Starting to handle delayed artifacts files. Found %d files.", len(filesToConsume)))
for _, filePath := range filesToConsume {
log.Debug("Handling delayed artifacts file: '" + filePath + "'")
Expand All @@ -180,7 +180,7 @@ func consumeDelayedArtifactsFiles(pcWrapper *producerConsumerWrapper, filesToCon
return err
}

shouldStop, err := uploadByChunks(delayedArtifactsFile.DelayedArtifacts, uploadChunkChan, base, delayHelper, errorsChannelMng, pcWrapper)
shouldStop, err := transferFiles(delayedArtifactsFile.DelayedArtifacts, base, delayHelper, errorsChannelMng, pcWrapper)
if err != nil || shouldStop {
return err
}
Expand Down
26 changes: 23 additions & 3 deletions artifactory/commands/transferfiles/errorshandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ func getErrorsFileNamePrefix(repoKey string, phaseId int, phaseStartTime string)
func (mng *TransferErrorsMng) writeErrorContent(e ExtendedFileUploadStatusResponse) error {
var err error
switch e.Status {
case api.SkippedLargeProps:
case api.SkippedLargeProps, api.SkippedSourceItemGone:
err = mng.writeSkippedErrorContent(e)
default:
err = mng.writeRetryableErrorContent(e)
Expand Down Expand Up @@ -276,18 +276,38 @@ func createErrorsCsvSummary(sourceRepos []string, timeStarted time.Time) (string
return "", err
}

errorsFiles = append(errorsFiles, skippedErrorsFiles...)
if len(errorsFiles) == 0 {
if len(errorsFiles)+len(skippedErrorsFiles) == 0 {
return "", nil
}
// Collect all errors from the given log files
allErrors, err := parseErrorsFromLogFiles(errorsFiles)
if err != nil {
return "", err
}
skippedErrors, err := parseErrorsFromLogFiles(skippedErrorsFiles)
if err != nil {
return "", err
}
allErrors.Errors = append(allErrors.Errors, dedupeSkippedRows(skippedErrors.Errors)...)
return cmdutils.CreateCSVFile("transfer-files-logs", allErrors.Errors, timeStarted)
}

// dedupeSkippedRows keeps one row per repo/path/name/status. The same item can be skipped by more
// than one phase (or run), which would otherwise list it several times in the summary.
func dedupeSkippedRows(rows []ExtendedFileUploadStatusResponse) []ExtendedFileUploadStatusResponse {
seen := make(map[string]struct{}, len(rows))
deduped := make([]ExtendedFileUploadStatusResponse, 0, len(rows))
for _, row := range rows {
key := fmt.Sprintf("%s\x00%s\x00%s\x00%s", row.Repo, row.Path, row.Name, row.Status)
if _, exists := seen[key]; exists {
continue
}
seen[key] = struct{}{}
deduped = append(deduped, row)
}
return deduped
}

// Gets a list of all errors files from the CLI's cache.
// Errors-files contain files that were failed to upload or actions that were skipped because of known limitations.
func getErrorsFiles(repoKeys []string, isRetry bool) (filesPaths []string, err error) {
Expand Down
Loading
Loading