Skip to content

Commit b648b80

Browse files
cmwatersstaheri14
andauthoredAug 16, 2024··
feat: re-add ability to filter tables from trace data (#1455)
Reintroduces #1374 in a non-breaking way. This would be helpful for my compact blocks test. I don't know why it got reverted --------- Co-authored-by: Sanaz Taheri <[email protected]>
1 parent c23345b commit b648b80

File tree

1 file changed

+48
-30
lines changed

1 file changed

+48
-30
lines changed
 

‎pkg/trace/fileserver.go

+48-30
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ import (
2323
"github.com/aws/aws-sdk-go/service/s3"
2424
)
2525

26+
const jsonL = ".jsonl"
27+
2628
func (lt *LocalTracer) getTableHandler() http.HandlerFunc {
2729
return func(w http.ResponseWriter, r *http.Request) {
2830
// Parse the request to get the data
@@ -76,7 +78,7 @@ func pump(table string, br *bufio.Reader) (*io.PipeReader, *multipart.Writer) {
7678
defer w.Close()
7779
defer m.Close()
7880

79-
part, err := m.CreateFormFile("filename", table+".jsonl")
81+
part, err := m.CreateFormFile("filename", table+jsonL)
8082
if err != nil {
8183
return
8284
}
@@ -133,7 +135,7 @@ func GetTable(serverURL, table, dirPath string) error {
133135
return err
134136
}
135137

136-
outputFile, err := os.Create(path.Join(dirPath, table+".jsonl"))
138+
outputFile, err := os.Create(path.Join(dirPath, table+jsonL))
137139
if err != nil {
138140
return err
139141
}
@@ -264,7 +266,9 @@ func (lt *LocalTracer) PushAll() error {
264266

265267
// S3Download downloads files that match some prefix from an S3 bucket to a
266268
// local directory dst.
267-
func S3Download(dst, prefix string, cfg S3Config) error {
269+
// fileNames is a list of traced jsonl file names to download. If it is empty, all traces are downloaded.
270+
// fileNames should not have .jsonl suffix.
271+
func S3Download(dst, prefix string, cfg S3Config, fileNames ...string) error {
268272
// Ensure local directory structure exists
269273
err := os.MkdirAll(dst, os.ModePerm)
270274
if err != nil {
@@ -293,37 +297,51 @@ func S3Download(dst, prefix string, cfg S3Config) error {
293297

294298
err = s3Svc.ListObjectsV2Pages(input, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
295299
for _, content := range page.Contents {
296-
localFilePath := filepath.Join(dst, prefix, strings.TrimPrefix(*content.Key, prefix))
297-
fmt.Printf("Downloading %s to %s\n", *content.Key, localFilePath)
298-
299-
// Create the directories in the path
300-
if err := os.MkdirAll(filepath.Dir(localFilePath), os.ModePerm); err != nil {
301-
return false
302-
}
303-
304-
// Create a file to write the S3 Object contents to.
305-
f, err := os.Create(localFilePath)
306-
if err != nil {
307-
return false
308-
}
300+
key := *content.Key
309301

310-
resp, err := s3Svc.GetObject(&s3.GetObjectInput{
311-
Bucket: aws.String(cfg.BucketName),
312-
Key: aws.String(*content.Key),
313-
})
314-
if err != nil {
315-
f.Close()
316-
continue
302+
// If no fileNames are specified, download all files
303+
if len(fileNames) == 0 {
304+
fileNames = append(fileNames, strings.TrimPrefix(key, prefix))
317305
}
318-
defer resp.Body.Close()
319306

320-
// Copy the contents of the S3 object to the local file
321-
if _, err := io.Copy(f, resp.Body); err != nil {
322-
return false
307+
for _, filename := range fileNames {
308+
// Add .jsonl suffix to the fileNames
309+
fullFilename := filename + jsonL
310+
if strings.HasSuffix(key, fullFilename) {
311+
localFilePath := filepath.Join(dst, prefix, strings.TrimPrefix(key, prefix))
312+
fmt.Printf("Downloading %s to %s\n", key, localFilePath)
313+
314+
// Create the directories in the path
315+
if err := os.MkdirAll(filepath.Dir(localFilePath), os.ModePerm); err != nil {
316+
return false
317+
}
318+
319+
// Create a file to write the S3 Object contents to.
320+
f, err := os.Create(localFilePath)
321+
if err != nil {
322+
return false
323+
}
324+
325+
resp, err := s3Svc.GetObject(&s3.GetObjectInput{
326+
Bucket: aws.String(cfg.BucketName),
327+
Key: aws.String(key),
328+
})
329+
if err != nil {
330+
f.Close()
331+
continue
332+
}
333+
defer resp.Body.Close()
334+
335+
// Copy the contents of the S3 object to the local file
336+
if _, err := io.Copy(f, resp.Body); err != nil {
337+
f.Close()
338+
return false
339+
}
340+
341+
fmt.Printf("Successfully downloaded %s to %s\n", key, localFilePath)
342+
f.Close()
343+
}
323344
}
324-
325-
fmt.Printf("Successfully downloaded %s to %s\n", *content.Key, localFilePath)
326-
f.Close()
327345
}
328346
return !lastPage // continue paging
329347
})

0 commit comments

Comments
 (0)
Please sign in to comment.