1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
package s3
import (
"context"
"errors"
"fmt"
neturl "net/url"
"os"
"strings"
"time"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/spf13/afero"
)
var (
ErrUnsupported = errors.New("unsupported operation")
ErrInvalidProtocol = errors.New("invalid protocol")
)
type Fs struct {
client *minio.Client
bucket string
// Configurables
Timeout time.Duration
}
func NewFs(endpoint, bucket, accessKeyID, secretAccessKey string, useSSL bool) (*Fs, error) {
// Initialize minio client object.
client, err := minio.New(endpoint, &minio.Options{
Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
Secure: useSSL,
})
if err != nil {
return nil, err
}
return &Fs{
client: client,
bucket: bucket,
Timeout: 10 * time.Second,
}, nil
}
func NewFsFromURL(url string) (*Fs, error) {
r, err := neturl.ParseRequestURI(url)
if err != nil {
return nil, fmt.Errorf("failed to parse url: %w", err)
}
useSSL := false
switch r.Scheme {
case "http":
case "https":
useSSL = true
default:
return nil, ErrInvalidProtocol
}
endpoint := r.Host
bucket := r.Path[1:]
accessKeyId := r.User.Username()
secretAccessKey, _ := r.User.Password()
return NewFs(endpoint, bucket, accessKeyId, secretAccessKey, useSSL)
}
// The name of this FileSystem
func (fs *Fs) Name() string {
return "s3"
}
// Create creates a file in the filesystem, returning the file and an
// error, if any happens.
func (fs *Fs) Create(name string) (File, error) {
panic("not implemented")
}
// Mkdir creates a directory in the filesystem, return an error if any
// happens.
func (fs *Fs) Mkdir(name string, perm os.FileMode) error {
return ErrUnsupported
}
// MkdirAll creates a directory path and all parents that does not exist
// yet.
func (fs *Fs) MkdirAll(path string, perm os.FileMode) error {
return ErrUnsupported
}
// Open opens a file, returning it or an error, if any happens.
func (fs *Fs) Open(name string) (File, error) {
panic("not implemented")
}
// OpenFile opens a file using the given flags and the given mode.
func (fs *Fs) OpenFile(name string, flag int, perm os.FileMode) (File, error) {
panic("not implemented")
}
// Remove removes a file identified by name, returning an error, if any
// happens.
func (fs *Fs) Remove(name string) error {
ctx, cancel := fs.contextWithTimeout()
defer cancel()
err := fs.client.RemoveObject(ctx, fs.bucket, name, minio.RemoveObjectOptions{})
if err != nil {
return transformError(err)
}
return nil
}
// RemoveAll removes a directory path and any children it contains. It
// does not fail if the path does not exist (return nil).
func (fs *Fs) RemoveAll(path string) error {
ctx, cancel := fs.contextWithTimeout()
defer cancel()
if !strings.HasSuffix(path, "/") {
path += "/"
}
objectsCh := fs.client.ListObjects(ctx, fs.bucket, minio.ListObjectsOptions{Prefix: path})
errc := fs.client.RemoveObjects(ctx, fs.bucket, objectsCh, minio.RemoveObjectsOptions{})
for err := range errc {
if err.Err != nil {
return transformError(err.Err)
}
}
return nil
}
// Rename renames a file.
func (fs *Fs) Rename(oldname, newname string) error {
ctx, cancel := fs.contextWithTimeout()
defer cancel()
_, err := fs.client.CopyObject(ctx,
minio.CopyDestOptions{Bucket: fs.bucket, Object: newname},
minio.CopySrcOptions{Bucket: fs.bucket, Object: oldname})
if err != nil {
return transformError(err)
}
err = fs.client.RemoveObject(ctx, fs.bucket, oldname, minio.RemoveObjectOptions{})
if err != nil {
return transformError(err)
}
return nil
}
// Stat returns a FileInfo describing the named file, or an error, if any
// happens.
func (fs *Fs) Stat(name string) (os.FileInfo, error) {
ctx, cancel := fs.contextWithTimeout()
defer cancel()
info, err := fs.client.StatObject(ctx, fs.bucket, name, minio.GetObjectOptions{})
if err != nil {
return nil, transformError(err)
}
return transformObjectInfo(info), nil
}
// Chmod changes the mode of the named file to mode.
func (fs *Fs) Chmod(name string, mode os.FileMode) error {
return ErrUnsupported
}
// Chown changes the uid and gid of the named file.
func (fs *Fs) Chown(name string, uid, gid int) error {
return ErrUnsupported
}
// Chtimes changes the access and modification times of the named file
func (fs *Fs) Chtimes(name string, atime time.Time, mtime time.Time) error {
return ErrUnsupported
}
func (fs *Fs) contextWithTimeout() (context.Context, context.CancelFunc) {
return context.WithTimeout(context.Background(), fs.Timeout)
}
func transformError(err error) error {
resp, ok := err.(minio.ErrorResponse)
if !ok {
return err
}
switch resp.Code {
case "NoSuchKey":
return afero.ErrFileNotFound
case "EntityTooLarge":
return afero.ErrTooLarge
}
return err
}
|
