Skip to content

Commit 11e8e84

Browse files
adrianwit@gmail.comadrianwit@gmail.com
adrianwit@gmail.com
authored and
adrianwit@gmail.com
committed
Added Google Clod Storage http proxy support
1 parent cd32834 commit 11e8e84

13 files changed

Lines changed: 206 additions & 104 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,7 @@
1+
## December 20 2019 0.12.0
2+
* Added Google Clod Storage http proxy support
3+
* Refactored retries
4+
15
## December 1 2019 0.10.8
26
* Patched object kind support
37

‎gs/client.go‎

Lines changed: 46 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,20 +2,39 @@ package gs
22

33
import (
44
"context"
5+
"fmt"
6+
"github.com/pkg/errors"
57
"github.com/viant/afs/option"
68
"github.com/viant/afs/storage"
79
"golang.org/x/oauth2"
810
"golang.org/x/oauth2/google"
911
goptions "google.golang.org/api/option"
1012
gstorage "google.golang.org/api/storage/v1"
1113
htransport "google.golang.org/api/transport/http"
14+
"net"
1215
"net/http"
16+
"net/url"
17+
"time"
1318
)
1419

1520
type client struct {
1621
projectID string
1722
ctx context.Context
1823
*http.Client
24+
25+
useProxy bool
26+
canProxyFallback bool
27+
}
28+
29+
func (c *client) disableProxy() {
30+
if !c.useProxy {
31+
return
32+
}
33+
switch val := c.Transport.(type) {
34+
case *oauth2.Transport:
35+
val.Base = nil
36+
}
37+
c.useProxy = false
1938
}
2039

2140
func defaultHTTPClient(ctx context.Context, scopes Scopes) (*http.Client, error) {
@@ -30,7 +49,8 @@ func defaultHTTPClient(ctx context.Context, scopes Scopes) (*http.Client, error)
3049
func newClient(ctx context.Context, options []storage.Option) (*client, error) {
3150
var jwTProvider JWTProvider
3251
var scopes = make(Scopes, 0)
33-
option.Assign(options, &jwTProvider, &scopes)
52+
proxy := &option.Proxy{}
53+
option.Assign(options, &jwTProvider, &scopes, &proxy)
3454
if len(scopes) == 0 {
3555
scopes = NewScopes(gstorage.CloudPlatformScope, gstorage.DevstorageFullControlScope)
3656
}
@@ -48,6 +68,31 @@ func newClient(ctx context.Context, options []storage.Option) (*client, error) {
4868
} else {
4969
result.Client, err = defaultHTTPClient(ctx, scopes)
5070
}
71+
72+
if proxy.URL != "" {
73+
result.useProxy = true
74+
result.canProxyFallback = proxy.Fallback
75+
proxyUrl, err := url.Parse(proxy.URL)
76+
if err != nil {
77+
return nil, errors.Wrapf(err, "failed to parse proxy URL: %v", proxy.URL)
78+
}
79+
transport := &http.Transport{Proxy: http.ProxyURL(proxyUrl)}
80+
if proxy.TimeoutMs > 0 {
81+
timeout := time.Millisecond * time.Duration(proxy.TimeoutMs)
82+
dialContext := (&net.Dialer{
83+
Timeout: timeout,
84+
DualStack: true,
85+
}).DialContext
86+
transport.DialContext = dialContext
87+
}
88+
switch val := result.Transport.(type) {
89+
case *oauth2.Transport:
90+
val.Base = transport
91+
default:
92+
fmt.Printf("setting proxy %T\n", result.Transport)
93+
result.Transport = transport
94+
}
95+
}
5196
if result.projectID == "" {
5297
credentials, err := google.FindDefaultCredentials(ctx, scopes...)
5398
if err != nil {

‎gs/copy.go‎

Lines changed: 7 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ package gs
33
import (
44
"context"
55
"fmt"
6-
"github.com/viant/afs/base"
76
"github.com/viant/afs/file"
87
"github.com/viant/afs/option"
98
"github.com/viant/afs/storage"
@@ -12,16 +11,8 @@ import (
1211
"strings"
1312
)
1413

15-
func (s *storager) Copy(ctx context.Context, sourcePath, destBucket, destPath string, options ...storage.Option) (err error) {
16-
retry := base.NewRetry()
17-
for i := 0; i < maxRetries; i++ {
18-
err = s.copy(ctx, sourcePath, destBucket, destPath, options)
19-
if !isRetryError(err) {
20-
return err
21-
}
22-
sleepBeforeRetry(retry)
23-
}
24-
return err
14+
func (s *storager) Copy(ctx context.Context, sourcePath, destBucket, destPath string, options ...storage.Option) error {
15+
return s.copy(ctx, sourcePath, destBucket, destPath, options)
2516
}
2617

2718
func (s *storager) copy(ctx context.Context, sourcePath, destBucket, destPath string, options []storage.Option) error {
@@ -59,6 +50,9 @@ func (s *storager) copy(ctx context.Context, sourcePath, destBucket, destPath st
5950
object.Name = destPath
6051
call := s.Objects.Copy(s.bucket, sourcePath, destBucket, destPath, object)
6152
call.Context(ctx)
62-
_, err = call.Do()
63-
return err
53+
54+
return runWithRetries(ctx, func() error {
55+
_, err = call.Do()
56+
return err
57+
}, s)
6458
}

‎gs/delete.go‎

Lines changed: 20 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ package gs
22

33
import (
44
"context"
5-
"github.com/viant/afs/base"
65
"github.com/viant/afs/option"
76
"github.com/viant/afs/storage"
87
"path"
@@ -20,35 +19,30 @@ func (s *storager) Delete(ctx context.Context, location string, options ...stora
2019
}
2120
call := s.Objects.Delete(s.bucket, location)
2221
call.Context(ctx)
23-
retry := base.NewRetry()
24-
for i := 0; i < maxRetries; i++ {
25-
err = call.Do()
26-
if isNotFound(err) {
27-
objectKind := &option.ObjectKind{}
28-
if _, ok := option.Assign(options, &objectKind); ok && objectKind.File {
29-
return err
30-
}
31-
notFound := err
22+
err = runWithRetries(ctx, func() error {
23+
return call.Do()
24+
}, s)
3225

33-
infoList, err := s.List(ctx, location)
34-
if err != nil {
35-
return err
36-
}
37-
if len(infoList) > 1 {
38-
for i := 1; i < len(infoList); i++ {
39-
if err = s.Delete(ctx, path.Join(location, infoList[i].Name())); err != nil {
40-
return err
41-
}
42-
}
43-
return nil
44-
} else {
45-
return notFound
46-
}
26+
if isNotFound(err) {
27+
objectKind := &option.ObjectKind{}
28+
if _, ok := option.Assign(options, &objectKind); ok && objectKind.File {
29+
return err
4730
}
48-
if !isRetryError(err) {
31+
notFound := err
32+
infoList, err := s.List(ctx, location)
33+
if err != nil {
4934
return err
5035
}
51-
sleepBeforeRetry(retry)
36+
if len(infoList) > 1 {
37+
for i := 1; i < len(infoList); i++ {
38+
if err = s.Delete(ctx, path.Join(location, infoList[i].Name())); err != nil {
39+
return err
40+
}
41+
}
42+
return nil
43+
} else {
44+
return notFound
45+
}
5246
}
5347
return err
5448
}

‎gs/download.go‎

Lines changed: 7 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -14,14 +14,7 @@ import (
1414
)
1515

1616
func (s *storager) Download(ctx context.Context, location string, options ...storage.Option) (reader io.ReadCloser, err error) {
17-
retry := base.NewRetry()
18-
for i := 0; i < maxRetries; i++ {
19-
reader, err = s.download(ctx, location, options)
20-
if !isRetryError(err) {
21-
return reader, err
22-
}
23-
sleepBeforeRetry(retry)
24-
}
17+
reader, err = s.download(ctx, location, options)
2518
return reader, err
2619
}
2720

@@ -35,12 +28,13 @@ func (s *storager) download(ctx context.Context, location string, options []stor
3528
key := &option.AES256Key{}
3629
stream := &option.Stream{}
3730
option.Assign(options, &md5, &crc, &key, &stream)
31+
3832
if len(key.Key) != 0 {
3933
if err := SetCustomKeyHeader(key, call.Header()); err != nil {
4034
return nil, err
4135
}
4236
}
43-
object, err := call.Do()
37+
object, err := s.getObject(ctx, location, options)
4438
if err == nil {
4539
if err = md5.Decode(object.Md5Hash); err == nil {
4640
err = crc.Decode(object.Crc32c)
@@ -49,32 +43,23 @@ func (s *storager) download(ctx context.Context, location string, options []stor
4943
if err != nil {
5044
return nil, err
5145
}
52-
5346
if len(key.Key) != 0 {
5447
if err := SetCustomKeyHeader(key, call.Header()); err != nil {
5548
return nil, err
5649
}
5750
}
5851
if stream.PartSize > 0 {
5952
stream.Size = int(object.Size)
60-
readSeeker := NewReadSeeker(call, int(object.Size))
53+
readSeeker := NewReadSeeker(ctx, s, call, int(object.Size))
6154
reader := base.NewStreamReader(stream, readSeeker)
6255
return reader, nil
6356
}
6457

6558
var response *nhttp.Response
66-
retry := base.NewRetry()
67-
for i := 0; i < maxRetries; i++ {
59+
err = runWithRetries(ctx, func() error {
6860
response, err = call.Download()
69-
if err == nil {
70-
break
71-
}
72-
if !isRetryError(err) {
73-
return nil, errors.Wrapf(err, "failed to download gs://%v/%v ", s.bucket, location)
74-
}
75-
sleepBeforeRetry(retry)
76-
}
77-
61+
return err
62+
}, s)
7863
if err != nil {
7964
return nil, errors.Wrapf(err, "failed to download gs://%v/%v ", s.bucket, location)
8065
}

‎gs/get.go‎

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,22 +5,39 @@ import (
55
"github.com/pkg/errors"
66
"github.com/viant/afs/option"
77
"github.com/viant/afs/storage"
8+
gstorage "google.golang.org/api/storage/v1"
89
"os"
910
"strings"
1011
)
1112

1213
//Get returns an object for supplied location
1314
func (s *storager) get(ctx context.Context, location string, options []storage.Option) (os.FileInfo, error) {
14-
location = strings.Trim(location, "/")
15-
objectCall := s.Objects.Get(s.bucket, location)
16-
objectCall.Context(ctx)
17-
object, err := objectCall.Do()
15+
object, err := s.getObject(ctx, location, options)
1816
if object != nil {
1917
return newFileInfo(object)
2018
}
2119
return nil, err
2220
}
2321

22+
//Get returns an object for supplied location
23+
func (s *storager) getObject(ctx context.Context, location string, options []storage.Option) (object *gstorage.Object, err error) {
24+
location = strings.Trim(location, "/")
25+
objectCall := s.Objects.Get(s.bucket, location)
26+
objectCall.Context(ctx)
27+
key := &option.AES256Key{}
28+
option.Assign(options, &key)
29+
if len(key.Key) != 0 {
30+
if err := SetCustomKeyHeader(key, objectCall.Header()); err != nil {
31+
return nil, err
32+
}
33+
}
34+
err = runWithRetries(ctx, func() error {
35+
object, err = objectCall.Do()
36+
return err
37+
}, s)
38+
return object, err
39+
}
40+
2441
//Get returns an object for supplied location
2542
func (s *storager) Get(ctx context.Context, location string, options ...storage.Option) (os.FileInfo, error) {
2643
info, err := s.get(ctx, location, options)

‎gs/helper.go‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"github.com/viant/afs/base"
55
"google.golang.org/api/googleapi"
66
"net/http"
7+
"net/url"
78
"strings"
89
"sync"
910
"time"
@@ -92,6 +93,18 @@ func isNotFound(err error) bool {
9293
return strings.Contains(err.Error(), notFound)
9394
}
9495

96+
func isProxyError(err error) bool {
97+
if err == nil {
98+
return false
99+
}
100+
_, ok := err.(*url.Error)
101+
if ok {
102+
return true
103+
}
104+
105+
return strings.Contains(err.Error(), "proxy")
106+
}
107+
95108
func GetRetryCodes(reset bool) map[int]int {
96109
result := retryErrors
97110
if reset {

‎gs/list.go‎

Lines changed: 8 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ package gs
22

33
import (
44
"context"
5-
"github.com/viant/afs/base"
65
"github.com/viant/afs/file"
76
"github.com/viant/afs/option"
87
"github.com/viant/afs/storage"
@@ -19,15 +18,7 @@ var listCounter uint64
1918

2019
//List list directory or returns a file info
2120
func (s *storager) List(ctx context.Context, location string, options ...storage.Option) (files []os.FileInfo, err error) {
22-
retry := base.NewRetry()
23-
for i := 0; i < maxRetries; i++ {
24-
files, err = s.listFiles(ctx, location, options)
25-
if !isRetryError(err) {
26-
break
27-
}
28-
sleepBeforeRetry(retry)
29-
}
30-
return files, err
21+
return s.listFiles(ctx, location, options)
3122
}
3223

3324
//List list directory or returns a file info
@@ -158,7 +149,13 @@ func (s *storager) addFiles(ctx context.Context, parent string, objects *gstorag
158149

159150
func (s *storager) listObjects(ctx context.Context, location string, call *gstorage.ObjectsListCall, infoList *[]os.FileInfo, page *option.Page, matcher option.Match) (int, int, error) {
160151
atomic.AddUint64(&listCounter, 1)
161-
objects, err := call.Do()
152+
153+
var objects *gstorage.Objects
154+
var err error
155+
err = runWithRetries(ctx, func() error {
156+
objects, err = call.Do()
157+
return err
158+
}, s)
162159
if err != nil {
163160
return 0, 0, err
164161
}

0 commit comments

Comments
 (0)