Skip to content

Commit 564b6d0

Browse files
matthewpeterkortmatthewpeterkort
committed
update to modern syfon client
1 parent b391239 commit 564b6d0

32 files changed

Lines changed: 507 additions & 1211 deletions

‎cmd/configure.go‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@ import (
44
"context"
55
"fmt"
66

7-
"github.com/calypr/data-client/common"
87
"github.com/calypr/data-client/conf"
98
"github.com/calypr/data-client/g3client"
109
"github.com/calypr/data-client/logs"
@@ -66,7 +65,7 @@ func init() {
6665
configureCmd.Flags().StringVar(&fenceToken, "fenceToken", "", "Specify the fence token to use as a substitute for credential file")
6766
configureCmd.Flags().StringVar(&apiEndpoint, "apiendpoint", "", "Specify the API endpoint of the data commons")
6867
configureCmd.MarkFlagRequired("apiendpoint") //nolint:errcheck
69-
configureCmd.Flags().StringVar(&useShepherd, "use-shepherd", "", fmt.Sprintf("Enables or disables support for the Shepherd API. If enabled, gen3client will use the Shepherd API if available. (Default: %v)", common.DefaultUseShepherd))
70-
configureCmd.Flags().StringVar(&minShepherdVersion, "min-shepherd-version", "", fmt.Sprintf("Specify the minimum version of Shepherd that the gen3client will use if Shepherd is enabled. (Default: %v)", common.DefaultMinShepherdVersion))
68+
configureCmd.Flags().StringVar(&useShepherd, "use-shepherd", "", fmt.Sprintf("Enables or disables support for the Shepherd API. If enabled, gen3client will use the Shepherd API if available. (Default: %v)", false))
69+
configureCmd.Flags().StringVar(&minShepherdVersion, "min-shepherd-version", "", fmt.Sprintf("Specify the minimum version of Shepherd that the gen3client will use if Shepherd is enabled. (Default: %v)", "2.0.0"))
7170
RootCmd.AddCommand(configureCmd)
7271
}

‎cmd/download-multiple.go‎

Lines changed: 34 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -3,21 +3,13 @@ package cmd
33
import (
44
"context"
55
"encoding/json"
6-
"io"
6+
"fmt"
77
"log"
88
"os"
99

10-
"github.com/calypr/data-client/common"
11-
"github.com/calypr/data-client/conf"
1210
"github.com/calypr/data-client/g3client"
1311
"github.com/calypr/data-client/logs"
14-
sydrs "github.com/calypr/syfon/client/drs"
15-
sylogs "github.com/calypr/syfon/client/pkg/logs"
16-
syrequest "github.com/calypr/syfon/client/pkg/request"
1712
sydownload "github.com/calypr/syfon/client/xfer/download"
18-
"github.com/vbauerster/mpb/v8"
19-
"github.com/vbauerster/mpb/v8/decor"
20-
2113
"github.com/spf13/cobra"
2214
)
2315

@@ -36,75 +28,21 @@ func init() {
3628
logger, logCloser := logs.New(profile, logs.WithConsole(), logs.WithFailedLog(), logs.WithScoreboard(), logs.WithSucceededLog())
3729
defer logCloser()
3830

39-
var dc sydrs.Client
40-
if backendType == "drs" {
41-
config := conf.NewConfigure(logger.Logger)
42-
cred, err := config.Load(profile)
43-
if err != nil {
44-
log.Fatalf("Failed to parse config on profile %s, %v", profile, err)
45-
}
46-
req := syrequest.NewRequestInterface(
47-
sylogs.NewGen3Logger(logger.Logger, "", ""),
48-
cred,
49-
config,
50-
)
51-
dc = sydrs.NewLocalDrsClient(req, cred.APIEndpoint, sylogs.NewGen3Logger(logger.Logger, "", ""))
52-
} else {
53-
g3i, err := g3client.NewGen3Interface(profile, logger)
54-
if err != nil {
55-
log.Fatalf("Failed to parse config on profile %s, %v", profile, err)
56-
}
57-
dc = g3i.DRSClient()
58-
}
59-
60-
manifestPath, _ = common.GetAbsolutePath(manifestPath)
61-
manifestFile, err := os.Open(manifestPath)
62-
if err != nil {
63-
logger.Fatalf("Failed to open manifest file %s, %v\n", manifestPath, err)
64-
}
65-
defer manifestFile.Close()
66-
manifestFileStat, err := manifestFile.Stat()
31+
g3i, err := g3client.NewGen3Interface(profile, logger)
6732
if err != nil {
68-
logger.Fatalf("Failed to get manifest file stats %s, %v\n", manifestPath, err)
33+
log.Fatalf("Failed to parse config on profile %s, %v", profile, err)
6934
}
70-
logger.Println("Reading manifest...")
71-
manifestFileSize := manifestFileStat.Size()
72-
manifestProgress := mpb.New(mpb.WithOutput(os.Stdout))
73-
manifestFileBar := manifestProgress.AddBar(manifestFileSize,
74-
mpb.PrependDecorators(
75-
decor.Name("Manifest "),
76-
decor.CountersKibiByte("% .1f / % .1f"),
77-
),
78-
mpb.AppendDecorators(decor.Percentage()),
79-
)
8035

81-
manifestFileReader := manifestFileBar.ProxyReader(manifestFile)
82-
83-
manifestBytes, err := io.ReadAll(manifestFileReader)
36+
guids, err := loadManifestGuids(manifestPath)
8437
if err != nil {
85-
logger.Fatalf("Failed reading manifest %s, %v\n", manifestPath, err)
38+
log.Fatalf("Failed to read manifest %s: %v", manifestPath, err)
8639
}
87-
manifestProgress.Wait()
8840

89-
var objects []common.ManifestObject
90-
err = json.Unmarshal(manifestBytes, &objects)
91-
if err != nil {
92-
logger.Fatalf("Error has occurred during unmarshalling manifest object: %v\n", err)
41+
syfon := g3i.SyfonClient()
42+
if syfon == nil {
43+
logger.Fatal("failed to initialize syfon client")
9344
}
94-
95-
err = sydownload.DownloadMultiple(
96-
context.Background(),
97-
dc,
98-
dc,
99-
objects,
100-
downloadPath,
101-
"original",
102-
true,
103-
false,
104-
"",
105-
numParallel,
106-
false,
107-
)
45+
err = sydownload.DownloadMultiple(context.Background(), syfon.DRS(), syfon.Data(), guids, downloadPath, numParallel, false)
10846
if err != nil {
10947
logger.Fatal(err.Error())
11048
}
@@ -119,3 +57,28 @@ func init() {
11957
downloadMultipleCmd.Flags().IntVar(&numParallel, "numparallel", 1, "Number of downloads to run in parallel")
12058
RootCmd.AddCommand(downloadMultipleCmd)
12159
}
60+
61+
func loadManifestGuids(path string) ([]string, error) {
62+
data, err := os.ReadFile(path)
63+
if err != nil {
64+
return nil, err
65+
}
66+
67+
var entries []struct {
68+
GUID string `json:"guid"`
69+
}
70+
if err := json.Unmarshal(data, &entries); err != nil {
71+
return nil, fmt.Errorf("parse manifest: %w", err)
72+
}
73+
74+
guids := make([]string, 0, len(entries))
75+
for _, entry := range entries {
76+
if entry.GUID != "" {
77+
guids = append(guids, entry.GUID)
78+
}
79+
}
80+
if len(guids) == 0 {
81+
return nil, fmt.Errorf("manifest %s did not contain any GUIDs", path)
82+
}
83+
return guids, nil
84+
}

‎cmd/download-single.go‎

Lines changed: 7 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -4,13 +4,8 @@ import (
44
"context"
55
"log"
66

7-
"github.com/calypr/data-client/common"
8-
"github.com/calypr/data-client/conf"
97
"github.com/calypr/data-client/g3client"
108
"github.com/calypr/data-client/logs"
11-
sydrs "github.com/calypr/syfon/client/drs"
12-
sylogs "github.com/calypr/syfon/client/pkg/logs"
13-
syrequest "github.com/calypr/syfon/client/pkg/request"
149
sydownload "github.com/calypr/syfon/client/xfer/download"
1510
"github.com/spf13/cobra"
1611
)
@@ -29,45 +24,16 @@ func init() {
2924
logger, logCloser := logs.New(profile, logs.WithConsole(), logs.WithFailedLog(), logs.WithSucceededLog(), logs.WithScoreboard())
3025
defer logCloser()
3126

32-
var dc sydrs.Client
33-
if backendType == "drs" {
34-
config := conf.NewConfigure(logger.Logger)
35-
cred, err := config.Load(profile)
36-
if err != nil {
37-
log.Fatalf("Failed to parse config on profile %s, %v", profile, err)
38-
}
39-
req := syrequest.NewRequestInterface(
40-
sylogs.NewGen3Logger(logger.Logger, "", ""),
41-
cred,
42-
config,
43-
)
44-
dc = sydrs.NewLocalDrsClient(req, cred.APIEndpoint, sylogs.NewGen3Logger(logger.Logger, "", ""))
45-
} else {
46-
g3I, err := g3client.NewGen3Interface(profile, logger)
47-
if err != nil {
48-
log.Fatalf("Failed to parse config on profile %s, %v", profile, err)
49-
}
50-
dc = g3I.DRSClient()
27+
g3I, err := g3client.NewGen3Interface(profile, logger)
28+
if err != nil {
29+
log.Fatalf("Failed to parse config on profile %s, %v", profile, err)
5130
}
5231

53-
objects := []common.ManifestObject{
54-
{
55-
GUID: guid,
56-
},
32+
syfon := g3I.SyfonClient()
33+
if syfon == nil {
34+
logger.Fatal("failed to initialize syfon client")
5735
}
58-
err := sydownload.DownloadMultiple(
59-
context.Background(),
60-
dc,
61-
dc,
62-
objects,
63-
downloadPath,
64-
"original",
65-
true,
66-
false,
67-
"",
68-
1,
69-
false,
70-
)
36+
err = sydownload.DownloadSingleWithProgress(context.Background(), syfon.DRS(), syfon.Data(), guid, downloadPath, "original")
7137
if err != nil {
7238
logger.Println(err.Error())
7339
}

‎cmd/retry-upload.go‎

Lines changed: 12 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,10 @@ package cmd
33
import (
44
"context"
55

6-
"github.com/calypr/data-client/common"
76
"github.com/calypr/data-client/g3client"
87
"github.com/calypr/data-client/logs"
9-
sylogs "github.com/calypr/syfon/client/pkg/logs"
10-
sytransfer "github.com/calypr/syfon/client/transfer"
11-
syupload "github.com/calypr/syfon/client/xfer/upload"
8+
syclient "github.com/calypr/syfon/client"
9+
sycommon "github.com/calypr/syfon/client/common"
1210

1311
"github.com/spf13/cobra"
1412
)
@@ -34,26 +32,19 @@ func init() {
3432
if err != nil {
3533
Logger.Fatalf("Failed to initialize client: %v", err)
3634
}
37-
bk := g3.DRSClient()
38-
uploader, ok := bk.(sytransfer.Uploader)
39-
if !ok {
40-
Logger.Fatalf("DRS client does not implement transfer.Uploader")
35+
if _, err := sycommon.LoadFailedLog(failedLogPath); err != nil {
36+
Logger.Fatalf("Cannot read failed log: %v", err)
4137
}
42-
43-
logger := g3.Logger()
44-
45-
// Create scoreboard with our logger injected
46-
sb := logs.NewSB(common.MaxRetryCount, logger.Logger)
47-
48-
// Load failed log
49-
failedMap, err := common.LoadFailedLog(failedLogPath)
38+
syfon := g3.SyfonClient()
39+
if syfon == nil {
40+
Logger.Fatal("failed to initialize syfon client")
41+
}
42+
err = syclient.Upload(context.Background(), syfon.Data(), "", syclient.UploadOptions{
43+
RetryFailedLogPath: failedLogPath,
44+
})
5045
if err != nil {
51-
logger.Fatalf("Cannot read failed log: %v", err)
46+
Logger.Fatalf("Retry upload failed: %v", err)
5247
}
53-
54-
// Unified DRS Client serves as both logical resolver and technical movement writer Across S3, GCS, and Azure.
55-
syupload.RetryFailedUploads(context.Background(), uploader, sylogs.NewGen3Logger(Logger.Logger, "", ""), failedMap)
56-
sb.PrintSB()
5748
},
5849
}
5950

‎cmd/upload-multipart.go‎

Lines changed: 10 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -2,14 +2,10 @@ package cmd
22

33
import (
44
"context"
5-
"os"
6-
"path/filepath"
75

8-
"github.com/calypr/data-client/common"
96
"github.com/calypr/data-client/g3client"
107
"github.com/calypr/data-client/logs"
11-
sytransfer "github.com/calypr/syfon/client/transfer"
12-
syupload "github.com/calypr/syfon/client/xfer/upload"
8+
syclient "github.com/calypr/syfon/client"
139
"github.com/spf13/cobra"
1410
)
1511

@@ -44,42 +40,20 @@ func init() {
4440
if err != nil {
4541
logger.Fatalf("failed to initialize Gen3 interface: %v", err)
4642
}
47-
bk := g3.DRSClient()
48-
uploader, ok := bk.(sytransfer.Uploader)
49-
if !ok {
50-
logger.Fatal("DRS client does not implement transfer.Uploader")
51-
}
52-
53-
absPath, err := common.GetAbsolutePath(filePath)
54-
if err != nil {
55-
logger.Fatalf("invalid file path: %v", err)
56-
}
57-
58-
fileInfo := common.FileUploadRequestObject{
59-
SourcePath: absPath,
60-
ObjectKey: filepath.Base(absPath),
61-
GUID: guid,
62-
FileMetadata: common.FileMetadata{},
63-
}
6443

65-
if fileInfo.Bucket == "" {
66-
fileInfo.Bucket = bucketName
44+
syfon := g3.SyfonClient()
45+
if syfon == nil {
46+
logger.Fatal("failed to initialize syfon client")
6747
}
68-
if fileInfo.Bucket == "" {
69-
fileInfo.Bucket = bk.GetBucketName()
70-
}
71-
72-
// Force multipart path by using direct multipart entrypoint.
73-
file, err := os.Open(fileInfo.SourcePath)
74-
if err != nil {
75-
logger.Fatal(err)
76-
}
77-
defer file.Close()
78-
err = syupload.MultipartUpload(context.Background(), uploader, fileInfo, file, true)
48+
err = syclient.Upload(context.Background(), syfon.Data(), filePath, syclient.UploadOptions{
49+
Bucket: bucketName,
50+
GUID: guid,
51+
ShowProgress: true,
52+
ForceMultipart: true,
53+
})
7954
if err != nil {
8055
logger.Fatal(err)
8156
}
82-
8357
},
8458
}
8559

0 commit comments

Comments
 (0)