Skip to content

Commit 6488e25

Browse files
abhisheksatish1999abhisheksatish1999
authored
Implement s3 connector using go (#31)
1 parent 94d1be8 commit 6488e25

7 files changed

Lines changed: 595 additions & 130 deletions

File tree

‎.github/workflows/main.yml‎

Lines changed: 26 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ name: Build Release
77
jobs:
88
azure-connector:
99
name: azure blob utility
10-
runs-on: ubuntu-latest
10+
runs-on: ubuntu-20.04
1111
steps:
1212
- uses: actions/checkout@v2
1313
- name: linux i386
@@ -32,15 +32,30 @@ jobs:
3232
GOOS: linux
3333
SUBDIR: "bnr-utils/nz_azConnector"
3434
EXECUTABLE_NAME: "nz_azConnector"
35-
3635
s3-connector:
37-
name: S3 connector linux amd64
38-
runs-on: ubuntu-latest
36+
name: s3 connector using go
37+
runs-on: ubuntu-20.04
3938
steps:
40-
- uses: actions/checkout@v2
41-
- name: Create tarball and upload release
42-
run: |
43-
tar -czf bnr-utils/nz_s3connector.tgz -C bnr-utils/nz_s3Connector .
44-
gh release upload ${{ github.event.release.tag_name }} bnr-utils/nz_s3connector.tgz
45-
env:
46-
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
39+
- uses: actions/checkout@v2
40+
- name: linux i386
41+
run: |
42+
cd go-release-executables
43+
chmod +x ./*
44+
bash main.sh
45+
env:
46+
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
47+
GOARCH: "386"
48+
GOOS: linux
49+
SUBDIR: "bnr-utils/nz_s3Connector"
50+
EXECUTABLE_NAME: "nz_s3Connector"
51+
- name: linux amd64
52+
run: |
53+
cd go-release-executables
54+
chmod +x ./*
55+
bash main.sh
56+
env:
57+
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
58+
GOARCH: amd64
59+
GOOS: linux
60+
SUBDIR: "bnr-utils/nz_s3Connector"
61+
EXECUTABLE_NAME: "nz_s3Connector"

‎bnr-utils/nz_s3Connector/README.txt‎

Lines changed: 106 additions & 118 deletions
Large diffs are not rendered by default.

‎bnr-utils/nz_s3Connector/awscli-bundle-1.37.18.zip‎

-22.6 MB
Binary file not shown.

‎bnr-utils/nz_s3Connector/nz_s3Connector.go‎

Lines changed: 339 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,339 @@
1+
package main
2+
3+
import (
4+
"context"
5+
"flag"
6+
"fmt"
7+
"io/fs"
8+
"log"
9+
"os"
10+
"path"
11+
"path/filepath"
12+
"strings"
13+
"sync"
14+
"time"
15+
16+
"github.com/aws/aws-sdk-go-v2/aws"
17+
"github.com/aws/aws-sdk-go-v2/config"
18+
"github.com/aws/aws-sdk-go-v2/credentials"
19+
"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
20+
"github.com/aws/aws-sdk-go-v2/service/s3"
21+
)
22+
23+
type S3Conn struct {
24+
accessKeyId string
25+
bucketUrl string
26+
defaultRegion string
27+
secretAccessKey string
28+
endPoint string
29+
streams int64
30+
blockSize int64
31+
}
32+
33+
type BackupInfo struct {
34+
dbname string
35+
dirs string
36+
npshost string
37+
backupsetID string
38+
}
39+
40+
type OtherArgs struct {
41+
download *bool
42+
upload *bool
43+
parallelJobs int64
44+
logFileDir string
45+
uniqueId string
46+
}
47+
48+
func parseArgs(s3Conn *S3Conn, backupinfo *BackupInfo, otherArgs *OtherArgs) {
49+
flag.StringVar(&backupinfo.dbname, "db", "", "Database name")
50+
flag.StringVar(&backupinfo.dirs, "dir", "", "Full path to the directory in which the backup already exists or should be downloaded. Enclose in double quotes if there are multiple directories.")
51+
flag.StringVar(&backupinfo.npshost, "npshost", "", "Name of the NPS host as it appears in the backups")
52+
flag.StringVar(&backupinfo.backupsetID, "backupset", "", "Name of the backupset to be uploaded/downloaded.")
53+
flag.StringVar(&otherArgs.logFileDir, "logfiledir", "", "Logfile directory for this utility")
54+
55+
flag.StringVar(&s3Conn.accessKeyId, "access-key", "", "Access Key Id to access AWS s3/IBM cloud")
56+
flag.StringVar(&s3Conn.bucketUrl, "bucket-url", "", "Bucket url to access AWS s3/IBM cloud")
57+
flag.StringVar(&s3Conn.defaultRegion, "region", "", "Default region of your bucket in AWS s3/IBM cloud")
58+
flag.StringVar(&s3Conn.secretAccessKey, "secret-key", "", "Secret Access Key to access access AWS s3/IBM cloud")
59+
flag.StringVar(&s3Conn.endPoint, "endpoint", "", "URL of the entry point for an AWS s3/IBM cloud. Mandatory for IBM cloud service.")
60+
flag.Int64Var(&s3Conn.streams, "streams", 16, "Number of blocks to upload/download in parallel default 16")
61+
flag.Int64Var(&s3Conn.blockSize, "blocksize", 100, "Block size in MB to upload/download file")
62+
63+
otherArgs.download = flag.Bool("download", false, "Download from cloud")
64+
otherArgs.upload = flag.Bool("upload", false, "Upload from cloud")
65+
flag.Int64Var(&otherArgs.parallelJobs, "paralleljobs", 6, "Parallel jobs for upload/download")
66+
flag.StringVar(&otherArgs.uniqueId, "unique-id", "", "Unique ID associated with the file transfer")
67+
}
68+
69+
func main() {
70+
var conn S3Conn
71+
var backupinfo BackupInfo
72+
var otherArgs OtherArgs
73+
74+
// parse input args
75+
parseArgs(&conn, &backupinfo, &otherArgs)
76+
flag.Parse()
77+
prefixStr := fmt.Sprintf("%s ", time.Now().UTC().Format("2006-01-02 15:04:05")) + fmt.Sprintf("%-7s", "[INFO]")
78+
if otherArgs.logFileDir != "" {
79+
log.Printf("logfile dir: %s", otherArgs.logFileDir)
80+
logfilename := fmt.Sprintf("nz_s3Connector_%d_%s.log", os.Getppid(), time.Now().Format("2006-01-02"))
81+
logfilepath := path.Join(otherArgs.logFileDir, logfilename)
82+
f, err := os.OpenFile(logfilepath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
83+
if err != nil {
84+
log.Fatalf("error opening log file: %v", err)
85+
}
86+
defer f.Close()
87+
log.SetOutput(f)
88+
}
89+
log.SetFlags(0)
90+
log.SetPrefix(prefixStr)
91+
if flag.NFlag() == 0 {
92+
log.Println("No arguments passed to nz_s3Connector. Below is the list of valid args: ")
93+
flag.PrintDefaults()
94+
os.Exit(1)
95+
}
96+
97+
log.Println("Aws S3 bucket:", conn.bucketUrl)
98+
log.Println("Aws region:", conn.defaultRegion)
99+
log.Println("Backup/Restore directory:", backupinfo.dirs)
100+
log.Println("DB name :", backupinfo.dbname)
101+
log.Println("Nps hostname :", backupinfo.npshost)
102+
if backupinfo.backupsetID != "" {
103+
log.Println("BackupsetID :", backupinfo.backupsetID)
104+
} else {
105+
log.Println("BackupsetID : ALL")
106+
}
107+
log.Println("Number of files to upload/download in parallel :", otherArgs.parallelJobs)
108+
checkRequiredArguments(backupinfo, otherArgs)
109+
cfg := conn.createS3Config()
110+
if *otherArgs.download {
111+
conn.Download(cfg, backupinfo, otherArgs)
112+
log.Println("Downloading complete.")
113+
}
114+
if *otherArgs.upload {
115+
conn.Upload(cfg, backupinfo, otherArgs)
116+
log.Println("Uploading complete.")
117+
}
118+
}
119+
120+
func checkRequiredArguments(bkp BackupInfo, arg OtherArgs) {
121+
if bkp.backupsetID != "" {
122+
if bkp.dirs == "" || bkp.npshost == "" || bkp.dbname == "" {
123+
log.Fatalf("Missing required field: db, npshost or dir is not found")
124+
}
125+
} else if bkp.dbname != "" {
126+
if bkp.dirs == "" || bkp.npshost == "" {
127+
log.Fatalf("Missing required field: npshost or dir is not found")
128+
}
129+
} else if bkp.npshost != "" {
130+
if bkp.dirs == "" {
131+
log.Fatalf("Missing required field: dir is not found")
132+
}
133+
} else {
134+
if bkp.dirs == "" {
135+
log.Fatalf("Missing required field: dir is not found")
136+
}
137+
}
138+
139+
if *arg.upload || *arg.download {
140+
if arg.uniqueId == "" {
141+
log.Fatalf("Missing required field: uniqueid is not found. It is required for upload/download operation")
142+
}
143+
}
144+
}
145+
146+
func (s3Conn *S3Conn) Upload(cfg aws.Config, bkp BackupInfo, otherArgs OtherArgs) {
147+
dirlist := strings.Split(bkp.dirs, " ")
148+
for _, dir := range dirlist {
149+
backupdir := filepath.Join(dir, "Netezza", bkp.npshost, bkp.dbname, bkp.backupsetID)
150+
_, err := os.Stat(backupdir)
151+
if err != nil {
152+
log.Fatalf("Cannot access directory %s: %v. Please check if DB name, hostname are correct.", backupdir, err)
153+
}
154+
log.Printf("Uploading data to s3 bucket %s with unique-id %s from dir %s", s3Conn.bucketUrl, otherArgs.uniqueId, backupdir)
155+
156+
filesuploaded := 0
157+
var wg sync.WaitGroup
158+
var mu sync.Mutex
159+
160+
// buffered channel to limit concurrency
161+
sem := make(chan struct{}, otherArgs.parallelJobs)
162+
err = filepath.Walk(backupdir, func(path string, info fs.FileInfo, err error) error {
163+
if info.IsDir() {
164+
return nil
165+
}
166+
167+
relfilepath, patherr := filepath.Rel(dir, path)
168+
if patherr != nil {
169+
return patherr
170+
}
171+
172+
wg.Add(1)
173+
sem <- struct{}{}
174+
175+
go func() {
176+
err := s3Conn.uploadFileToS3(path, cfg, otherArgs.uniqueId, relfilepath)
177+
if err != nil {
178+
log.Println("Error while uploading file. Ensure aws s3 access-key-id, secret-access-key, bucket_url are correct.")
179+
log.Fatalf("Failed to upload file. Err: %v", err)
180+
}
181+
log.Printf("File %s uploaded successfully", path)
182+
mu.Lock()
183+
filesuploaded++
184+
mu.Unlock()
185+
wg.Done()
186+
<-sem
187+
}()
188+
return err
189+
})
190+
if err != nil {
191+
log.Fatalf("Encountered error while traversing the directory. Err: %v", err)
192+
}
193+
wg.Wait()
194+
log.Printf("Total files uploaded: %d", filesuploaded)
195+
}
196+
}
197+
198+
func (s3Conn *S3Conn) getUploader(cfg aws.Config) *manager.Uploader {
199+
return manager.NewUploader(s3.NewFromConfig(cfg), func(u *manager.Uploader) {
200+
u.PartSize = s3Conn.blockSize * 1024 * 1024
201+
u.Concurrency = int(s3Conn.streams)
202+
})
203+
}
204+
205+
func (s3Conn *S3Conn) uploadFileToS3(absFilePath string, cfg aws.Config, uniqueId string, relFilePath string) error {
206+
uploader := s3Conn.getUploader(cfg)
207+
f, err := os.Open(absFilePath)
208+
if err != nil {
209+
log.Printf("Unable to open file %s. Err: %v", absFilePath, err)
210+
return err
211+
}
212+
defer f.Close()
213+
214+
_, err = uploader.Upload(context.TODO(), &s3.PutObjectInput{
215+
Bucket: aws.String(s3Conn.bucketUrl),
216+
Body: f,
217+
Key: aws.String(filepath.Join(uniqueId, relFilePath)),
218+
})
219+
if err != nil {
220+
log.Fatalf("Failed to upload file: %s. Err: %v", absFilePath, err)
221+
}
222+
return nil
223+
}
224+
225+
func (s3Conn *S3Conn) Download(cfg aws.Config, bkp BackupInfo, otherArgs OtherArgs) {
226+
bkpath := filepath.Join(otherArgs.uniqueId, "Netezza", bkp.npshost, bkp.dbname, bkp.backupsetID)
227+
log.Printf("Backup dir path: %s", bkpath)
228+
dirlist := strings.Split(bkp.dirs, " ")
229+
230+
for _, dir := range dirlist {
231+
log.Printf("Downloading data to dir %s", dir)
232+
client := s3.NewFromConfig(cfg)
233+
filesdownloaded := 0
234+
var wg sync.WaitGroup
235+
var mu sync.Mutex
236+
237+
// buffered channel to limit concurrency
238+
sem := make(chan struct{}, otherArgs.parallelJobs)
239+
240+
paginator := s3.NewListObjectsV2Paginator(client, &s3.ListObjectsV2Input{
241+
Bucket: aws.String(s3Conn.bucketUrl),
242+
Prefix: aws.String(otherArgs.uniqueId),
243+
})
244+
245+
for paginator.HasMorePages() {
246+
page, err := paginator.NextPage(context.TODO())
247+
if err != nil {
248+
log.Fatalf("Error while listing objects: %v", err)
249+
}
250+
251+
for _, obj := range page.Contents {
252+
key := *obj.Key
253+
254+
if strings.HasPrefix(key, bkpath) {
255+
splitdir, filename := filepath.Split(key)
256+
relfilepath, err := filepath.Rel(otherArgs.uniqueId, splitdir)
257+
if err != nil {
258+
log.Fatalf("Error in fetching download relative path: %v", err)
259+
}
260+
261+
dumpdir := filepath.Join(dir, relfilepath)
262+
err = os.MkdirAll(dumpdir, 0777)
263+
if err != nil {
264+
log.Fatalf("Error in creating backup directory: %v", err)
265+
}
266+
267+
outfilepath := path.Join(dumpdir, filename)
268+
wg.Add(1)
269+
sem <- struct{}{}
270+
271+
go func() {
272+
err := s3Conn.downloadFileFromS3(outfilepath, cfg, key)
273+
if err != nil {
274+
log.Println("Error while downloading file. Ensure aws s3 access-key-id, secret-access-key, bucket_url are correct.")
275+
log.Fatalf("Failed to download file. Err: %v", err)
276+
}
277+
log.Printf("File %s downloaded successfully", key)
278+
mu.Lock()
279+
filesdownloaded++
280+
mu.Unlock()
281+
wg.Done()
282+
<-sem
283+
}()
284+
}
285+
}
286+
}
287+
wg.Wait()
288+
log.Printf("Total files downloaded: %d", filesdownloaded)
289+
}
290+
}
291+
292+
func (s3Conn *S3Conn) getDownloader(cfg aws.Config) *manager.Downloader {
293+
return manager.NewDownloader(s3.NewFromConfig(cfg), func(d *manager.Downloader) {
294+
d.PartSize = s3Conn.blockSize * 1024 * 1024
295+
d.Concurrency = int(s3Conn.streams)
296+
})
297+
}
298+
299+
func (s3Conn *S3Conn) downloadFileFromS3(absFilePath string, cfg aws.Config, relFilePath string) error {
300+
downloader := s3Conn.getDownloader(cfg)
301+
f, err := os.Create(absFilePath)
302+
if err != nil {
303+
log.Printf("Unable to create file %s. Err: %v", absFilePath, err)
304+
return err
305+
}
306+
defer f.Close()
307+
308+
bytes, err := downloader.Download(context.TODO(), f, &s3.GetObjectInput{
309+
Bucket: aws.String(s3Conn.bucketUrl),
310+
Key: aws.String(relFilePath),
311+
})
312+
if err != nil || bytes < 0 {
313+
log.Fatalf("Failed to download file: %s. Err: %v", relFilePath, err)
314+
return err
315+
}
316+
return nil
317+
}
318+
319+
func (s3Conn *S3Conn) createS3Config() aws.Config {
320+
cfg, err := config.LoadDefaultConfig(context.TODO(),
321+
config.WithRegion(s3Conn.defaultRegion),
322+
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
323+
s3Conn.accessKeyId,
324+
s3Conn.secretAccessKey,
325+
"",
326+
)),
327+
)
328+
if err != nil {
329+
log.Fatalf("Failed to create AWS config: %v", err)
330+
}
331+
cfg.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
332+
cfg.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
333+
334+
if s3Conn.endPoint != "" {
335+
cfg.BaseEndpoint = aws.String(s3Conn.endPoint)
336+
}
337+
338+
return cfg
339+
}

‎go-release-executables/main.sh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ set -x
44

55
#!/bin/bash -eux
66

7-
GO_LINUX_PACKAGE_URL="https://dl.google.com/go/go1.14.linux-amd64.tar.gz"
7+
GO_LINUX_PACKAGE_URL="https://go.dev/dl/go1.24.0.linux-amd64.tar.gz"
88
wget --no-check-certificate --progress=dot:mega ${GO_LINUX_PACKAGE_URL} -O go-linux.tar.gz
99
tar -zxf go-linux.tar.gz
1010
sudo mv go /usr/local/

0 commit comments

Comments
 (0)