mirror of
https://github.com/octoleo/restic.git
synced 2024-11-02 11:46:36 +00:00
2184 lines
67 KiB
Go
2184 lines
67 KiB
Go
|
package swift
|
||
|
|
||
|
import (
|
||
|
"bufio"
|
||
|
"bytes"
|
||
|
"crypto/hmac"
|
||
|
"crypto/md5"
|
||
|
"crypto/sha1"
|
||
|
"encoding/hex"
|
||
|
"encoding/json"
|
||
|
"fmt"
|
||
|
"hash"
|
||
|
"io"
|
||
|
"io/ioutil"
|
||
|
"mime"
|
||
|
"net/http"
|
||
|
"net/url"
|
||
|
"os"
|
||
|
"path"
|
||
|
"strconv"
|
||
|
"strings"
|
||
|
"sync"
|
||
|
"time"
|
||
|
)
|
||
|
|
||
|
const (
|
||
|
DefaultUserAgent = "goswift/1.0" // Default user agent
|
||
|
DefaultRetries = 3 // Default number of retries on token expiry
|
||
|
TimeFormat = "2006-01-02T15:04:05" // Python date format for json replies parsed as UTC
|
||
|
UploadTar = "tar" // Data format specifier for Connection.BulkUpload().
|
||
|
UploadTarGzip = "tar.gz" // Data format specifier for Connection.BulkUpload().
|
||
|
UploadTarBzip2 = "tar.bz2" // Data format specifier for Connection.BulkUpload().
|
||
|
allContainersLimit = 10000 // Number of containers to fetch at once
|
||
|
allObjectsLimit = 10000 // Number objects to fetch at once
|
||
|
allObjectsChanLimit = 1000 // ...when fetching to a channel
|
||
|
)
|
||
|
|
||
|
// ObjectType is the type of the swift object, regular, static large,
|
||
|
// or dynamic large.
|
||
|
type ObjectType int
|
||
|
|
||
|
// Values that ObjectType can take
|
||
|
const (
|
||
|
RegularObjectType ObjectType = iota
|
||
|
StaticLargeObjectType
|
||
|
DynamicLargeObjectType
|
||
|
)
|
||
|
|
||
|
// Connection holds the details of the connection to the swift server.
|
||
|
//
|
||
|
// You need to provide UserName, ApiKey and AuthUrl when you create a
|
||
|
// connection then call Authenticate on it.
|
||
|
//
|
||
|
// The auth version in use will be detected from the AuthURL - you can
|
||
|
// override this with the AuthVersion parameter.
|
||
|
//
|
||
|
// If using v2 auth you can also set Region in the Connection
|
||
|
// structure. If you don't set Region you will get the default region
|
||
|
// which may not be what you want.
|
||
|
//
|
||
|
// For reference some common AuthUrls looks like this:
|
||
|
//
|
||
|
// Rackspace US https://auth.api.rackspacecloud.com/v1.0
|
||
|
// Rackspace UK https://lon.auth.api.rackspacecloud.com/v1.0
|
||
|
// Rackspace v2 https://identity.api.rackspacecloud.com/v2.0
|
||
|
// Memset Memstore UK https://auth.storage.memset.com/v1.0
|
||
|
// Memstore v2 https://auth.storage.memset.com/v2.0
|
||
|
//
|
||
|
// When using Google Appengine you must provide the Connection with an
|
||
|
// appengine-specific Transport:
|
||
|
//
|
||
|
// import (
|
||
|
// "appengine/urlfetch"
|
||
|
// "fmt"
|
||
|
// "github.com/ncw/swift"
|
||
|
// )
|
||
|
//
|
||
|
// func handler(w http.ResponseWriter, r *http.Request) {
|
||
|
// ctx := appengine.NewContext(r)
|
||
|
// tr := urlfetch.Transport{Context: ctx}
|
||
|
// c := swift.Connection{
|
||
|
// UserName: "user",
|
||
|
// ApiKey: "key",
|
||
|
// AuthUrl: "auth_url",
|
||
|
// Transport: tr,
|
||
|
// }
|
||
|
// _ := c.Authenticate()
|
||
|
// containers, _ := c.ContainerNames(nil)
|
||
|
// fmt.Fprintf(w, "containers: %q", containers)
|
||
|
// }
|
||
|
//
|
||
|
// If you don't supply a Transport, one is made which relies on
|
||
|
// http.ProxyFromEnvironment (http://golang.org/pkg/net/http/#ProxyFromEnvironment).
|
||
|
// This means that the connection will respect the HTTP proxy specified by the
|
||
|
// environment variables $HTTP_PROXY and $NO_PROXY.
|
||
|
type Connection struct {
|
||
|
// Parameters - fill these in before calling Authenticate
|
||
|
// They are all optional except UserName, ApiKey and AuthUrl
|
||
|
Domain string // User's domain name
|
||
|
DomainId string // User's domain Id
|
||
|
UserName string // UserName for api
|
||
|
ApiKey string // Key for api access
|
||
|
AuthUrl string // Auth URL
|
||
|
Retries int // Retries on error (default is 3)
|
||
|
UserAgent string // Http User agent (default goswift/1.0)
|
||
|
ConnectTimeout time.Duration // Connect channel timeout (default 10s)
|
||
|
Timeout time.Duration // Data channel timeout (default 60s)
|
||
|
Region string // Region to use eg "LON", "ORD" - default is use first region (v2,v3 auth only)
|
||
|
AuthVersion int // Set to 1, 2 or 3 or leave at 0 for autodetect
|
||
|
Internal bool // Set this to true to use the the internal / service network
|
||
|
Tenant string // Name of the tenant (v2,v3 auth only)
|
||
|
TenantId string // Id of the tenant (v2,v3 auth only)
|
||
|
EndpointType EndpointType // Endpoint type (v2,v3 auth only) (default is public URL unless Internal is set)
|
||
|
TenantDomain string // Name of the tenant's domain (v3 auth only), only needed if it differs from the user domain
|
||
|
TenantDomainId string // Id of the tenant's domain (v3 auth only), only needed if it differs the from user domain
|
||
|
TrustId string // Id of the trust (v3 auth only)
|
||
|
Transport http.RoundTripper `json:"-" xml:"-"` // Optional specialised http.Transport (eg. for Google Appengine)
|
||
|
// These are filled in after Authenticate is called as are the defaults for above
|
||
|
StorageUrl string
|
||
|
AuthToken string
|
||
|
client *http.Client
|
||
|
Auth Authenticator `json:"-" xml:"-"` // the current authenticator
|
||
|
authLock sync.Mutex // lock when R/W StorageUrl, AuthToken, Auth
|
||
|
// swiftInfo is filled after QueryInfo is called
|
||
|
swiftInfo SwiftInfo
|
||
|
}
|
||
|
|
||
|
// setFromEnv reads the value that param points to (it must be a
|
||
|
// pointer), if it isn't the zero value then it reads the environment
|
||
|
// variable name passed in, parses it according to the type and writes
|
||
|
// it to the pointer.
|
||
|
func setFromEnv(param interface{}, name string) (err error) {
|
||
|
val := os.Getenv(name)
|
||
|
if val == "" {
|
||
|
return
|
||
|
}
|
||
|
switch result := param.(type) {
|
||
|
case *string:
|
||
|
if *result == "" {
|
||
|
*result = val
|
||
|
}
|
||
|
case *int:
|
||
|
if *result == 0 {
|
||
|
*result, err = strconv.Atoi(val)
|
||
|
}
|
||
|
case *bool:
|
||
|
if *result == false {
|
||
|
*result, err = strconv.ParseBool(val)
|
||
|
}
|
||
|
case *time.Duration:
|
||
|
if *result == 0 {
|
||
|
*result, err = time.ParseDuration(val)
|
||
|
}
|
||
|
case *EndpointType:
|
||
|
if *result == EndpointType("") {
|
||
|
*result = EndpointType(val)
|
||
|
}
|
||
|
default:
|
||
|
return newErrorf(0, "can't set var of type %T", param)
|
||
|
}
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// ApplyEnvironment reads environment variables and applies them to
|
||
|
// the Connection structure. It won't overwrite any parameters which
|
||
|
// are already set in the Connection struct.
|
||
|
//
|
||
|
// To make a new Connection object entirely from the environment you
|
||
|
// would do:
|
||
|
//
|
||
|
// c := new(Connection)
|
||
|
// err := c.ApplyEnvironment()
|
||
|
// if err != nil { log.Fatal(err) }
|
||
|
//
|
||
|
// The naming of these variables follows the official Openstack naming
|
||
|
// scheme so it should be compatible with OpenStack rc files.
|
||
|
//
|
||
|
// For v1 authentication (obsolete)
|
||
|
// ST_AUTH - Auth URL
|
||
|
// ST_USER - UserName for api
|
||
|
// ST_KEY - Key for api access
|
||
|
//
|
||
|
// For v2 authentication
|
||
|
// OS_AUTH_URL - Auth URL
|
||
|
// OS_USERNAME - UserName for api
|
||
|
// OS_PASSWORD - Key for api access
|
||
|
// OS_TENANT_NAME - Name of the tenant
|
||
|
// OS_TENANT_ID - Id of the tenant
|
||
|
// OS_REGION_NAME - Region to use - default is use first region
|
||
|
//
|
||
|
// For v3 authentication
|
||
|
// OS_AUTH_URL - Auth URL
|
||
|
// OS_USERNAME - UserName for api
|
||
|
// OS_PASSWORD - Key for api access
|
||
|
// OS_USER_DOMAIN_NAME - User's domain name
|
||
|
// OS_USER_DOMAIN_ID - User's domain Id
|
||
|
// OS_PROJECT_NAME - Name of the project
|
||
|
// OS_PROJECT_DOMAIN_NAME - Name of the tenant's domain, only needed if it differs from the user domain
|
||
|
// OS_PROJECT_DOMAIN_ID - Id of the tenant's domain, only needed if it differs the from user domain
|
||
|
// OS_TRUST_ID - If of the trust
|
||
|
// OS_REGION_NAME - Region to use - default is use first region
|
||
|
//
|
||
|
// Other
|
||
|
// OS_ENDPOINT_TYPE - Endpoint type public, internal or admin
|
||
|
// ST_AUTH_VERSION - Choose auth version - 1, 2 or 3 or leave at 0 for autodetect
|
||
|
//
|
||
|
// For manual authentication
|
||
|
// OS_STORAGE_URL - storage URL from alternate authentication
|
||
|
// OS_AUTH_TOKEN - Auth Token from alternate authentication
|
||
|
//
|
||
|
// Library specific
|
||
|
// GOSWIFT_RETRIES - Retries on error (default is 3)
|
||
|
// GOSWIFT_USER_AGENT - HTTP User agent (default goswift/1.0)
|
||
|
// GOSWIFT_CONNECT_TIMEOUT - Connect channel timeout with unit, eg "10s", "100ms" (default "10s")
|
||
|
// GOSWIFT_TIMEOUT - Data channel timeout with unit, eg "10s", "100ms" (default "60s")
|
||
|
// GOSWIFT_INTERNAL - Set this to "true" to use the the internal network (obsolete - use OS_ENDPOINT_TYPE)
|
||
|
func (c *Connection) ApplyEnvironment() (err error) {
|
||
|
for _, item := range []struct {
|
||
|
result interface{}
|
||
|
name string
|
||
|
}{
|
||
|
// Environment variables - keep in same order as Connection
|
||
|
{&c.Domain, "OS_USER_DOMAIN_NAME"},
|
||
|
{&c.DomainId, "OS_USER_DOMAIN_ID"},
|
||
|
{&c.UserName, "OS_USERNAME"},
|
||
|
{&c.ApiKey, "OS_PASSWORD"},
|
||
|
{&c.AuthUrl, "OS_AUTH_URL"},
|
||
|
{&c.Retries, "GOSWIFT_RETRIES"},
|
||
|
{&c.UserAgent, "GOSWIFT_USER_AGENT"},
|
||
|
{&c.ConnectTimeout, "GOSWIFT_CONNECT_TIMEOUT"},
|
||
|
{&c.Timeout, "GOSWIFT_TIMEOUT"},
|
||
|
{&c.Region, "OS_REGION_NAME"},
|
||
|
{&c.AuthVersion, "ST_AUTH_VERSION"},
|
||
|
{&c.Internal, "GOSWIFT_INTERNAL"},
|
||
|
{&c.Tenant, "OS_TENANT_NAME"}, //v2
|
||
|
{&c.Tenant, "OS_PROJECT_NAME"}, // v3
|
||
|
{&c.TenantId, "OS_TENANT_ID"},
|
||
|
{&c.EndpointType, "OS_ENDPOINT_TYPE"},
|
||
|
{&c.TenantDomain, "OS_PROJECT_DOMAIN_NAME"},
|
||
|
{&c.TenantDomainId, "OS_PROJECT_DOMAIN_ID"},
|
||
|
{&c.TrustId, "OS_TRUST_ID"},
|
||
|
{&c.StorageUrl, "OS_STORAGE_URL"},
|
||
|
{&c.AuthToken, "OS_AUTH_TOKEN"},
|
||
|
// v1 auth alternatives
|
||
|
{&c.ApiKey, "ST_KEY"},
|
||
|
{&c.UserName, "ST_USER"},
|
||
|
{&c.AuthUrl, "ST_AUTH"},
|
||
|
} {
|
||
|
err = setFromEnv(item.result, item.name)
|
||
|
if err != nil {
|
||
|
return newErrorf(0, "failed to read env var %q: %v", item.name, err)
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// Error - all errors generated by this package are of this type. Other error
|
||
|
// may be passed on from library functions though.
|
||
|
type Error struct {
|
||
|
StatusCode int // HTTP status code if relevant or 0 if not
|
||
|
Text string
|
||
|
}
|
||
|
|
||
|
// Error satisfy the error interface.
|
||
|
func (e *Error) Error() string {
|
||
|
return e.Text
|
||
|
}
|
||
|
|
||
|
// newError make a new error from a string.
|
||
|
func newError(StatusCode int, Text string) *Error {
|
||
|
return &Error{
|
||
|
StatusCode: StatusCode,
|
||
|
Text: Text,
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// newErrorf makes a new error from sprintf parameters.
|
||
|
func newErrorf(StatusCode int, Text string, Parameters ...interface{}) *Error {
|
||
|
return newError(StatusCode, fmt.Sprintf(Text, Parameters...))
|
||
|
}
|
||
|
|
||
|
// errorMap defines http error codes to error mappings.
|
||
|
type errorMap map[int]error
|
||
|
|
||
|
var (
|
||
|
// Specific Errors you might want to check for equality
|
||
|
BadRequest = newError(400, "Bad Request")
|
||
|
AuthorizationFailed = newError(401, "Authorization Failed")
|
||
|
ContainerNotFound = newError(404, "Container Not Found")
|
||
|
ContainerNotEmpty = newError(409, "Container Not Empty")
|
||
|
ObjectNotFound = newError(404, "Object Not Found")
|
||
|
ObjectCorrupted = newError(422, "Object Corrupted")
|
||
|
TimeoutError = newError(408, "Timeout when reading or writing data")
|
||
|
Forbidden = newError(403, "Operation forbidden")
|
||
|
TooLargeObject = newError(413, "Too Large Object")
|
||
|
|
||
|
// Mappings for authentication errors
|
||
|
authErrorMap = errorMap{
|
||
|
400: BadRequest,
|
||
|
401: AuthorizationFailed,
|
||
|
403: Forbidden,
|
||
|
}
|
||
|
|
||
|
// Mappings for container errors
|
||
|
ContainerErrorMap = errorMap{
|
||
|
400: BadRequest,
|
||
|
403: Forbidden,
|
||
|
404: ContainerNotFound,
|
||
|
409: ContainerNotEmpty,
|
||
|
}
|
||
|
|
||
|
// Mappings for object errors
|
||
|
objectErrorMap = errorMap{
|
||
|
400: BadRequest,
|
||
|
403: Forbidden,
|
||
|
404: ObjectNotFound,
|
||
|
413: TooLargeObject,
|
||
|
422: ObjectCorrupted,
|
||
|
}
|
||
|
)
|
||
|
|
||
|
// checkClose is used to check the return from Close in a defer
|
||
|
// statement.
|
||
|
func checkClose(c io.Closer, err *error) {
|
||
|
cerr := c.Close()
|
||
|
if *err == nil {
|
||
|
*err = cerr
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// drainAndClose discards all data from rd and closes it.
|
||
|
// If an error occurs during Read, it is discarded.
|
||
|
func drainAndClose(rd io.ReadCloser, err *error) {
|
||
|
if rd == nil {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
_, _ = io.Copy(ioutil.Discard, rd)
|
||
|
cerr := rd.Close()
|
||
|
if err != nil && *err == nil {
|
||
|
*err = cerr
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// parseHeaders checks a response for errors and translates into
|
||
|
// standard errors if necessary. If an error is returned, resp.Body
|
||
|
// has been drained and closed.
|
||
|
func (c *Connection) parseHeaders(resp *http.Response, errorMap errorMap) error {
|
||
|
if errorMap != nil {
|
||
|
if err, ok := errorMap[resp.StatusCode]; ok {
|
||
|
drainAndClose(resp.Body, nil)
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
if resp.StatusCode < 200 || resp.StatusCode > 299 {
|
||
|
drainAndClose(resp.Body, nil)
|
||
|
return newErrorf(resp.StatusCode, "HTTP Error: %d: %s", resp.StatusCode, resp.Status)
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// readHeaders returns a Headers object from the http.Response.
|
||
|
//
|
||
|
// If it receives multiple values for a key (which should never
|
||
|
// happen) it will use the first one
|
||
|
func readHeaders(resp *http.Response) Headers {
|
||
|
headers := Headers{}
|
||
|
for key, values := range resp.Header {
|
||
|
headers[key] = values[0]
|
||
|
}
|
||
|
return headers
|
||
|
}
|
||
|
|
||
|
// Headers stores HTTP headers (can only have one of each header like Swift).
|
||
|
type Headers map[string]string
|
||
|
|
||
|
// Does an http request using the running timer passed in
|
||
|
func (c *Connection) doTimeoutRequest(timer *time.Timer, req *http.Request) (*http.Response, error) {
|
||
|
// Do the request in the background so we can check the timeout
|
||
|
type result struct {
|
||
|
resp *http.Response
|
||
|
err error
|
||
|
}
|
||
|
done := make(chan result, 1)
|
||
|
go func() {
|
||
|
resp, err := c.client.Do(req)
|
||
|
done <- result{resp, err}
|
||
|
}()
|
||
|
// Wait for the read or the timeout
|
||
|
select {
|
||
|
case r := <-done:
|
||
|
return r.resp, r.err
|
||
|
case <-timer.C:
|
||
|
// Kill the connection on timeout so we don't leak sockets or goroutines
|
||
|
cancelRequest(c.Transport, req)
|
||
|
return nil, TimeoutError
|
||
|
}
|
||
|
panic("unreachable") // For Go 1.0
|
||
|
}
|
||
|
|
||
|
// Set defaults for any unset values
|
||
|
//
|
||
|
// Call with authLock held
|
||
|
func (c *Connection) setDefaults() {
|
||
|
if c.UserAgent == "" {
|
||
|
c.UserAgent = DefaultUserAgent
|
||
|
}
|
||
|
if c.Retries == 0 {
|
||
|
c.Retries = DefaultRetries
|
||
|
}
|
||
|
if c.ConnectTimeout == 0 {
|
||
|
c.ConnectTimeout = 10 * time.Second
|
||
|
}
|
||
|
if c.Timeout == 0 {
|
||
|
c.Timeout = 60 * time.Second
|
||
|
}
|
||
|
if c.Transport == nil {
|
||
|
c.Transport = &http.Transport{
|
||
|
// TLSClientConfig: &tls.Config{RootCAs: pool},
|
||
|
// DisableCompression: true,
|
||
|
Proxy: http.ProxyFromEnvironment,
|
||
|
MaxIdleConnsPerHost: 2048,
|
||
|
}
|
||
|
}
|
||
|
if c.client == nil {
|
||
|
c.client = &http.Client{
|
||
|
// CheckRedirect: redirectPolicyFunc,
|
||
|
Transport: c.Transport,
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// Authenticate connects to the Swift server.
|
||
|
//
|
||
|
// If you don't call it before calling one of the connection methods
|
||
|
// then it will be called for you on the first access.
|
||
|
func (c *Connection) Authenticate() (err error) {
|
||
|
c.authLock.Lock()
|
||
|
defer c.authLock.Unlock()
|
||
|
return c.authenticate()
|
||
|
}
|
||
|
|
||
|
// Internal implementation of Authenticate
|
||
|
//
|
||
|
// Call with authLock held
|
||
|
func (c *Connection) authenticate() (err error) {
|
||
|
c.setDefaults()
|
||
|
|
||
|
// Flush the keepalives connection - if we are
|
||
|
// re-authenticating then stuff has gone wrong
|
||
|
flushKeepaliveConnections(c.Transport)
|
||
|
|
||
|
if c.Auth == nil {
|
||
|
c.Auth, err = newAuth(c)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
|
||
|
retries := 1
|
||
|
again:
|
||
|
var req *http.Request
|
||
|
req, err = c.Auth.Request(c)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
if req != nil {
|
||
|
timer := time.NewTimer(c.ConnectTimeout)
|
||
|
var resp *http.Response
|
||
|
resp, err = c.doTimeoutRequest(timer, req)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
defer func() {
|
||
|
drainAndClose(resp.Body, &err)
|
||
|
// Flush the auth connection - we don't want to keep
|
||
|
// it open if keepalives were enabled
|
||
|
flushKeepaliveConnections(c.Transport)
|
||
|
}()
|
||
|
if err = c.parseHeaders(resp, authErrorMap); err != nil {
|
||
|
// Try again for a limited number of times on
|
||
|
// AuthorizationFailed or BadRequest. This allows us
|
||
|
// to try some alternate forms of the request
|
||
|
if (err == AuthorizationFailed || err == BadRequest) && retries > 0 {
|
||
|
retries--
|
||
|
goto again
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
err = c.Auth.Response(resp)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
if customAuth, isCustom := c.Auth.(CustomEndpointAuthenticator); isCustom && c.EndpointType != "" {
|
||
|
c.StorageUrl = customAuth.StorageUrlForEndpoint(c.EndpointType)
|
||
|
} else {
|
||
|
c.StorageUrl = c.Auth.StorageUrl(c.Internal)
|
||
|
}
|
||
|
c.AuthToken = c.Auth.Token()
|
||
|
if !c.authenticated() {
|
||
|
err = newError(0, "Response didn't have storage url and auth token")
|
||
|
return
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Get an authToken and url
|
||
|
//
|
||
|
// The Url may be updated if it needed to authenticate using the OnReAuth function
|
||
|
func (c *Connection) getUrlAndAuthToken(targetUrlIn string, OnReAuth func() (string, error)) (targetUrlOut, authToken string, err error) {
|
||
|
c.authLock.Lock()
|
||
|
defer c.authLock.Unlock()
|
||
|
targetUrlOut = targetUrlIn
|
||
|
if !c.authenticated() {
|
||
|
err = c.authenticate()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
if OnReAuth != nil {
|
||
|
targetUrlOut, err = OnReAuth()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
authToken = c.AuthToken
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// flushKeepaliveConnections is called to flush pending requests after an error.
|
||
|
func flushKeepaliveConnections(transport http.RoundTripper) {
|
||
|
if tr, ok := transport.(interface {
|
||
|
CloseIdleConnections()
|
||
|
}); ok {
|
||
|
tr.CloseIdleConnections()
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// UnAuthenticate removes the authentication from the Connection.
|
||
|
func (c *Connection) UnAuthenticate() {
|
||
|
c.authLock.Lock()
|
||
|
c.StorageUrl = ""
|
||
|
c.AuthToken = ""
|
||
|
c.authLock.Unlock()
|
||
|
}
|
||
|
|
||
|
// Authenticated returns a boolean to show if the current connection
|
||
|
// is authenticated.
|
||
|
//
|
||
|
// Doesn't actually check the credentials against the server.
|
||
|
func (c *Connection) Authenticated() bool {
|
||
|
c.authLock.Lock()
|
||
|
defer c.authLock.Unlock()
|
||
|
return c.authenticated()
|
||
|
}
|
||
|
|
||
|
// Internal version of Authenticated()
|
||
|
//
|
||
|
// Call with authLock held
|
||
|
func (c *Connection) authenticated() bool {
|
||
|
return c.StorageUrl != "" && c.AuthToken != ""
|
||
|
}
|
||
|
|
||
|
// SwiftInfo contains the JSON object returned by Swift when the /info
|
||
|
// route is queried. The object contains, among others, the Swift version,
|
||
|
// the enabled middlewares and their configuration
|
||
|
type SwiftInfo map[string]interface{}
|
||
|
|
||
|
func (i SwiftInfo) SupportsBulkDelete() bool {
|
||
|
_, val := i["bulk_delete"]
|
||
|
return val
|
||
|
}
|
||
|
|
||
|
func (i SwiftInfo) SupportsSLO() bool {
|
||
|
_, val := i["slo"]
|
||
|
return val
|
||
|
}
|
||
|
|
||
|
func (i SwiftInfo) SLOMinSegmentSize() int64 {
|
||
|
if slo, ok := i["slo"].(map[string]interface{}); ok {
|
||
|
val, _ := slo["min_segment_size"].(float64)
|
||
|
return int64(val)
|
||
|
}
|
||
|
return 1
|
||
|
}
|
||
|
|
||
|
// Discover Swift configuration by doing a request against /info
|
||
|
func (c *Connection) QueryInfo() (infos SwiftInfo, err error) {
|
||
|
infoUrl, err := url.Parse(c.StorageUrl)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
infoUrl.Path = path.Join(infoUrl.Path, "..", "..", "info")
|
||
|
resp, err := c.client.Get(infoUrl.String())
|
||
|
if err == nil {
|
||
|
if resp.StatusCode != http.StatusOK {
|
||
|
drainAndClose(resp.Body, nil)
|
||
|
return nil, fmt.Errorf("Invalid status code for info request: %d", resp.StatusCode)
|
||
|
}
|
||
|
err = readJson(resp, &infos)
|
||
|
if err == nil {
|
||
|
c.authLock.Lock()
|
||
|
c.swiftInfo = infos
|
||
|
c.authLock.Unlock()
|
||
|
}
|
||
|
return infos, err
|
||
|
}
|
||
|
return nil, err
|
||
|
}
|
||
|
|
||
|
func (c *Connection) cachedQueryInfo() (infos SwiftInfo, err error) {
|
||
|
c.authLock.Lock()
|
||
|
infos = c.swiftInfo
|
||
|
c.authLock.Unlock()
|
||
|
if infos == nil {
|
||
|
infos, err = c.QueryInfo()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
return infos, nil
|
||
|
}
|
||
|
|
||
|
// RequestOpts contains parameters for Connection.storage.
|
||
|
type RequestOpts struct {
|
||
|
Container string
|
||
|
ObjectName string
|
||
|
Operation string
|
||
|
Parameters url.Values
|
||
|
Headers Headers
|
||
|
ErrorMap errorMap
|
||
|
NoResponse bool
|
||
|
Body io.Reader
|
||
|
Retries int
|
||
|
// if set this is called on re-authentication to refresh the targetUrl
|
||
|
OnReAuth func() (string, error)
|
||
|
}
|
||
|
|
||
|
// Call runs a remote command on the targetUrl, returns a
|
||
|
// response, headers and possible error.
|
||
|
//
|
||
|
// operation is GET, HEAD etc
|
||
|
// container is the name of a container
|
||
|
// Any other parameters (if not None) are added to the targetUrl
|
||
|
//
|
||
|
// Returns a response or an error. If response is returned then
|
||
|
// the resp.Body must be read completely and
|
||
|
// resp.Body.Close() must be called on it, unless noResponse is set in
|
||
|
// which case the body will be closed in this function
|
||
|
//
|
||
|
// If "Content-Length" is set in p.Headers it will be used - this can
|
||
|
// be used to override the default chunked transfer encoding for
|
||
|
// uploads.
|
||
|
//
|
||
|
// This will Authenticate if necessary, and re-authenticate if it
|
||
|
// receives a 401 error which means the token has expired
|
||
|
//
|
||
|
// This method is exported so extensions can call it.
|
||
|
func (c *Connection) Call(targetUrl string, p RequestOpts) (resp *http.Response, headers Headers, err error) {
|
||
|
c.authLock.Lock()
|
||
|
c.setDefaults()
|
||
|
c.authLock.Unlock()
|
||
|
retries := p.Retries
|
||
|
if retries == 0 {
|
||
|
retries = c.Retries
|
||
|
}
|
||
|
var req *http.Request
|
||
|
for {
|
||
|
var authToken string
|
||
|
if targetUrl, authToken, err = c.getUrlAndAuthToken(targetUrl, p.OnReAuth); err != nil {
|
||
|
return //authentication failure
|
||
|
}
|
||
|
var URL *url.URL
|
||
|
URL, err = url.Parse(targetUrl)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
if p.Container != "" {
|
||
|
URL.Path += "/" + p.Container
|
||
|
if p.ObjectName != "" {
|
||
|
URL.Path += "/" + p.ObjectName
|
||
|
}
|
||
|
}
|
||
|
if p.Parameters != nil {
|
||
|
URL.RawQuery = p.Parameters.Encode()
|
||
|
}
|
||
|
timer := time.NewTimer(c.ConnectTimeout)
|
||
|
reader := p.Body
|
||
|
if reader != nil {
|
||
|
reader = newWatchdogReader(reader, c.Timeout, timer)
|
||
|
}
|
||
|
req, err = http.NewRequest(p.Operation, URL.String(), reader)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
if p.Headers != nil {
|
||
|
for k, v := range p.Headers {
|
||
|
// Set ContentLength in req if the user passed it in in the headers
|
||
|
if k == "Content-Length" {
|
||
|
contentLength, err := strconv.ParseInt(v, 10, 64)
|
||
|
if err != nil {
|
||
|
return nil, nil, fmt.Errorf("Invalid %q header %q: %v", k, v, err)
|
||
|
}
|
||
|
req.ContentLength = contentLength
|
||
|
} else {
|
||
|
req.Header.Add(k, v)
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
req.Header.Add("User-Agent", c.UserAgent)
|
||
|
req.Header.Add("X-Auth-Token", authToken)
|
||
|
resp, err = c.doTimeoutRequest(timer, req)
|
||
|
if err != nil {
|
||
|
if (p.Operation == "HEAD" || p.Operation == "GET") && retries > 0 {
|
||
|
retries--
|
||
|
continue
|
||
|
}
|
||
|
return nil, nil, err
|
||
|
}
|
||
|
// Check to see if token has expired
|
||
|
if resp.StatusCode == 401 && retries > 0 {
|
||
|
drainAndClose(resp.Body, nil)
|
||
|
c.UnAuthenticate()
|
||
|
retries--
|
||
|
} else {
|
||
|
break
|
||
|
}
|
||
|
}
|
||
|
|
||
|
if err = c.parseHeaders(resp, p.ErrorMap); err != nil {
|
||
|
return nil, nil, err
|
||
|
}
|
||
|
headers = readHeaders(resp)
|
||
|
if p.NoResponse {
|
||
|
var err error
|
||
|
drainAndClose(resp.Body, &err)
|
||
|
if err != nil {
|
||
|
return nil, nil, err
|
||
|
}
|
||
|
} else {
|
||
|
// Cancel the request on timeout
|
||
|
cancel := func() {
|
||
|
cancelRequest(c.Transport, req)
|
||
|
}
|
||
|
// Wrap resp.Body to make it obey an idle timeout
|
||
|
resp.Body = newTimeoutReader(resp.Body, c.Timeout, cancel)
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// storage runs a remote command on a the storage url, returns a
|
||
|
// response, headers and possible error.
|
||
|
//
|
||
|
// operation is GET, HEAD etc
|
||
|
// container is the name of a container
|
||
|
// Any other parameters (if not None) are added to the storage url
|
||
|
//
|
||
|
// Returns a response or an error. If response is returned then
|
||
|
// resp.Body.Close() must be called on it, unless noResponse is set in
|
||
|
// which case the body will be closed in this function
|
||
|
//
|
||
|
// This will Authenticate if necessary, and re-authenticate if it
|
||
|
// receives a 401 error which means the token has expired
|
||
|
func (c *Connection) storage(p RequestOpts) (resp *http.Response, headers Headers, err error) {
|
||
|
p.OnReAuth = func() (string, error) {
|
||
|
return c.StorageUrl, nil
|
||
|
}
|
||
|
c.authLock.Lock()
|
||
|
url := c.StorageUrl
|
||
|
c.authLock.Unlock()
|
||
|
return c.Call(url, p)
|
||
|
}
|
||
|
|
||
|
// readLines reads the response into an array of strings.
|
||
|
//
|
||
|
// Closes the response when done
|
||
|
func readLines(resp *http.Response) (lines []string, err error) {
|
||
|
defer drainAndClose(resp.Body, &err)
|
||
|
reader := bufio.NewReader(resp.Body)
|
||
|
buffer := bytes.NewBuffer(make([]byte, 0, 128))
|
||
|
var part []byte
|
||
|
var prefix bool
|
||
|
for {
|
||
|
if part, prefix, err = reader.ReadLine(); err != nil {
|
||
|
break
|
||
|
}
|
||
|
buffer.Write(part)
|
||
|
if !prefix {
|
||
|
lines = append(lines, buffer.String())
|
||
|
buffer.Reset()
|
||
|
}
|
||
|
}
|
||
|
if err == io.EOF {
|
||
|
err = nil
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// readJson reads the response into the json type passed in
|
||
|
//
|
||
|
// Closes the response when done
|
||
|
func readJson(resp *http.Response, result interface{}) (err error) {
|
||
|
defer drainAndClose(resp.Body, &err)
|
||
|
decoder := json.NewDecoder(resp.Body)
|
||
|
return decoder.Decode(result)
|
||
|
}
|
||
|
|
||
|
/* ------------------------------------------------------------ */
|
||
|
|
||
|
// ContainersOpts is options for Containers() and ContainerNames()
|
||
|
type ContainersOpts struct {
|
||
|
Limit int // For an integer value n, limits the number of results to at most n values.
|
||
|
Prefix string // Given a string value x, return container names matching the specified prefix.
|
||
|
Marker string // Given a string value x, return container names greater in value than the specified marker.
|
||
|
EndMarker string // Given a string value x, return container names less in value than the specified marker.
|
||
|
Headers Headers // Any additional HTTP headers - can be nil
|
||
|
}
|
||
|
|
||
|
// parse the ContainerOpts
|
||
|
func (opts *ContainersOpts) parse() (url.Values, Headers) {
|
||
|
v := url.Values{}
|
||
|
var h Headers
|
||
|
if opts != nil {
|
||
|
if opts.Limit > 0 {
|
||
|
v.Set("limit", strconv.Itoa(opts.Limit))
|
||
|
}
|
||
|
if opts.Prefix != "" {
|
||
|
v.Set("prefix", opts.Prefix)
|
||
|
}
|
||
|
if opts.Marker != "" {
|
||
|
v.Set("marker", opts.Marker)
|
||
|
}
|
||
|
if opts.EndMarker != "" {
|
||
|
v.Set("end_marker", opts.EndMarker)
|
||
|
}
|
||
|
h = opts.Headers
|
||
|
}
|
||
|
return v, h
|
||
|
}
|
||
|
|
||
|
// ContainerNames returns a slice of names of containers in this account.
|
||
|
func (c *Connection) ContainerNames(opts *ContainersOpts) ([]string, error) {
|
||
|
v, h := opts.parse()
|
||
|
resp, _, err := c.storage(RequestOpts{
|
||
|
Operation: "GET",
|
||
|
Parameters: v,
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
Headers: h,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
lines, err := readLines(resp)
|
||
|
return lines, err
|
||
|
}
|
||
|
|
||
|
// Container contains information about a container
|
||
|
type Container struct {
|
||
|
Name string // Name of the container
|
||
|
Count int64 // Number of objects in the container
|
||
|
Bytes int64 // Total number of bytes used in the container
|
||
|
}
|
||
|
|
||
|
// Containers returns a slice of structures with full information as
|
||
|
// described in Container.
|
||
|
func (c *Connection) Containers(opts *ContainersOpts) ([]Container, error) {
|
||
|
v, h := opts.parse()
|
||
|
v.Set("format", "json")
|
||
|
resp, _, err := c.storage(RequestOpts{
|
||
|
Operation: "GET",
|
||
|
Parameters: v,
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
Headers: h,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
var containers []Container
|
||
|
err = readJson(resp, &containers)
|
||
|
return containers, err
|
||
|
}
|
||
|
|
||
|
// containersAllOpts makes a copy of opts if set or makes a new one and
|
||
|
// overrides Limit and Marker
|
||
|
func containersAllOpts(opts *ContainersOpts) *ContainersOpts {
|
||
|
var newOpts ContainersOpts
|
||
|
if opts != nil {
|
||
|
newOpts = *opts
|
||
|
}
|
||
|
if newOpts.Limit == 0 {
|
||
|
newOpts.Limit = allContainersLimit
|
||
|
}
|
||
|
newOpts.Marker = ""
|
||
|
return &newOpts
|
||
|
}
|
||
|
|
||
|
// ContainersAll is like Containers but it returns all the Containers
|
||
|
//
|
||
|
// It calls Containers multiple times using the Marker parameter
|
||
|
//
|
||
|
// It has a default Limit parameter but you may pass in your own
|
||
|
func (c *Connection) ContainersAll(opts *ContainersOpts) ([]Container, error) {
|
||
|
opts = containersAllOpts(opts)
|
||
|
containers := make([]Container, 0)
|
||
|
for {
|
||
|
newContainers, err := c.Containers(opts)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
containers = append(containers, newContainers...)
|
||
|
if len(newContainers) < opts.Limit {
|
||
|
break
|
||
|
}
|
||
|
opts.Marker = newContainers[len(newContainers)-1].Name
|
||
|
}
|
||
|
return containers, nil
|
||
|
}
|
||
|
|
||
|
// ContainerNamesAll is like ContainerNamess but it returns all the Containers
|
||
|
//
|
||
|
// It calls ContainerNames multiple times using the Marker parameter
|
||
|
//
|
||
|
// It has a default Limit parameter but you may pass in your own
|
||
|
func (c *Connection) ContainerNamesAll(opts *ContainersOpts) ([]string, error) {
|
||
|
opts = containersAllOpts(opts)
|
||
|
containers := make([]string, 0)
|
||
|
for {
|
||
|
newContainers, err := c.ContainerNames(opts)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
containers = append(containers, newContainers...)
|
||
|
if len(newContainers) < opts.Limit {
|
||
|
break
|
||
|
}
|
||
|
opts.Marker = newContainers[len(newContainers)-1]
|
||
|
}
|
||
|
return containers, nil
|
||
|
}
|
||
|
|
||
|
/* ------------------------------------------------------------ */
|
||
|
|
||
|
// ObjectOpts is options for Objects() and ObjectNames()
|
||
|
type ObjectsOpts struct {
|
||
|
Limit int // For an integer value n, limits the number of results to at most n values.
|
||
|
Marker string // Given a string value x, return object names greater in value than the specified marker.
|
||
|
EndMarker string // Given a string value x, return object names less in value than the specified marker
|
||
|
Prefix string // For a string value x, causes the results to be limited to object names beginning with the substring x.
|
||
|
Path string // For a string value x, return the object names nested in the pseudo path
|
||
|
Delimiter rune // For a character c, return all the object names nested in the container
|
||
|
Headers Headers // Any additional HTTP headers - can be nil
|
||
|
}
|
||
|
|
||
|
// parse reads values out of ObjectsOpts
|
||
|
func (opts *ObjectsOpts) parse() (url.Values, Headers) {
|
||
|
v := url.Values{}
|
||
|
var h Headers
|
||
|
if opts != nil {
|
||
|
if opts.Limit > 0 {
|
||
|
v.Set("limit", strconv.Itoa(opts.Limit))
|
||
|
}
|
||
|
if opts.Marker != "" {
|
||
|
v.Set("marker", opts.Marker)
|
||
|
}
|
||
|
if opts.EndMarker != "" {
|
||
|
v.Set("end_marker", opts.EndMarker)
|
||
|
}
|
||
|
if opts.Prefix != "" {
|
||
|
v.Set("prefix", opts.Prefix)
|
||
|
}
|
||
|
if opts.Path != "" {
|
||
|
v.Set("path", opts.Path)
|
||
|
}
|
||
|
if opts.Delimiter != 0 {
|
||
|
v.Set("delimiter", string(opts.Delimiter))
|
||
|
}
|
||
|
h = opts.Headers
|
||
|
}
|
||
|
return v, h
|
||
|
}
|
||
|
|
||
|
// ObjectNames returns a slice of names of objects in a given container.
|
||
|
func (c *Connection) ObjectNames(container string, opts *ObjectsOpts) ([]string, error) {
|
||
|
v, h := opts.parse()
|
||
|
resp, _, err := c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
Operation: "GET",
|
||
|
Parameters: v,
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
Headers: h,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
return readLines(resp)
|
||
|
}
|
||
|
|
||
|
// Object contains information about an object
|
||
|
type Object struct {
|
||
|
Name string `json:"name"` // object name
|
||
|
ContentType string `json:"content_type"` // eg application/directory
|
||
|
Bytes int64 `json:"bytes"` // size in bytes
|
||
|
ServerLastModified string `json:"last_modified"` // Last modified time, eg '2011-06-30T08:20:47.736680' as a string supplied by the server
|
||
|
LastModified time.Time // Last modified time converted to a time.Time
|
||
|
Hash string `json:"hash"` // MD5 hash, eg "d41d8cd98f00b204e9800998ecf8427e"
|
||
|
PseudoDirectory bool // Set when using delimiter to show that this directory object does not really exist
|
||
|
SubDir string `json:"subdir"` // returned only when using delimiter to mark "pseudo directories"
|
||
|
ObjectType ObjectType // type of this object
|
||
|
}
|
||
|
|
||
|
// Objects returns a slice of Object with information about each
|
||
|
// object in the container.
|
||
|
//
|
||
|
// If Delimiter is set in the opts then PseudoDirectory may be set,
|
||
|
// with ContentType 'application/directory'. These are not real
|
||
|
// objects but represent directories of objects which haven't had an
|
||
|
// object created for them.
|
||
|
func (c *Connection) Objects(container string, opts *ObjectsOpts) ([]Object, error) {
|
||
|
v, h := opts.parse()
|
||
|
v.Set("format", "json")
|
||
|
resp, _, err := c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
Operation: "GET",
|
||
|
Parameters: v,
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
Headers: h,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
var objects []Object
|
||
|
err = readJson(resp, &objects)
|
||
|
// Convert Pseudo directories and dates
|
||
|
for i := range objects {
|
||
|
object := &objects[i]
|
||
|
if object.SubDir != "" {
|
||
|
object.Name = object.SubDir
|
||
|
object.PseudoDirectory = true
|
||
|
object.ContentType = "application/directory"
|
||
|
}
|
||
|
if object.ServerLastModified != "" {
|
||
|
// 2012-11-11T14:49:47.887250
|
||
|
//
|
||
|
// Remove fractional seconds if present. This
|
||
|
// then keeps it consistent with Object
|
||
|
// which can only return timestamps accurate
|
||
|
// to 1 second
|
||
|
//
|
||
|
// The TimeFormat will parse fractional
|
||
|
// seconds if desired though
|
||
|
datetime := strings.SplitN(object.ServerLastModified, ".", 2)[0]
|
||
|
object.LastModified, err = time.Parse(TimeFormat, datetime)
|
||
|
if err != nil {
|
||
|
return nil, err
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
return objects, err
|
||
|
}
|
||
|
|
||
|
// objectsAllOpts makes a copy of opts if set or makes a new one and
|
||
|
// overrides Limit and Marker
|
||
|
func objectsAllOpts(opts *ObjectsOpts, Limit int) *ObjectsOpts {
|
||
|
var newOpts ObjectsOpts
|
||
|
if opts != nil {
|
||
|
newOpts = *opts
|
||
|
}
|
||
|
if newOpts.Limit == 0 {
|
||
|
newOpts.Limit = Limit
|
||
|
}
|
||
|
newOpts.Marker = ""
|
||
|
return &newOpts
|
||
|
}
|
||
|
|
||
|
// A closure defined by the caller to iterate through all objects
|
||
|
//
|
||
|
// Call Objects or ObjectNames from here with the *ObjectOpts passed in
|
||
|
//
|
||
|
// Do whatever is required with the results then return them
|
||
|
type ObjectsWalkFn func(*ObjectsOpts) (interface{}, error)
|
||
|
|
||
|
// ObjectsWalk is uses to iterate through all the objects in chunks as
|
||
|
// returned by Objects or ObjectNames using the Marker and Limit
|
||
|
// parameters in the ObjectsOpts.
|
||
|
//
|
||
|
// Pass in a closure `walkFn` which calls Objects or ObjectNames with
|
||
|
// the *ObjectsOpts passed to it and does something with the results.
|
||
|
//
|
||
|
// Errors will be returned from this function
|
||
|
//
|
||
|
// It has a default Limit parameter but you may pass in your own
|
||
|
func (c *Connection) ObjectsWalk(container string, opts *ObjectsOpts, walkFn ObjectsWalkFn) error {
|
||
|
opts = objectsAllOpts(opts, allObjectsChanLimit)
|
||
|
for {
|
||
|
objects, err := walkFn(opts)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
var n int
|
||
|
var last string
|
||
|
switch objects := objects.(type) {
|
||
|
case []string:
|
||
|
n = len(objects)
|
||
|
if n > 0 {
|
||
|
last = objects[len(objects)-1]
|
||
|
}
|
||
|
case []Object:
|
||
|
n = len(objects)
|
||
|
if n > 0 {
|
||
|
last = objects[len(objects)-1].Name
|
||
|
}
|
||
|
default:
|
||
|
panic("Unknown type returned to ObjectsWalk")
|
||
|
}
|
||
|
if n < opts.Limit {
|
||
|
break
|
||
|
}
|
||
|
opts.Marker = last
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// ObjectsAll is like Objects but it returns an unlimited number of Objects in a slice
|
||
|
//
|
||
|
// It calls Objects multiple times using the Marker parameter
|
||
|
func (c *Connection) ObjectsAll(container string, opts *ObjectsOpts) ([]Object, error) {
|
||
|
objects := make([]Object, 0)
|
||
|
err := c.ObjectsWalk(container, opts, func(opts *ObjectsOpts) (interface{}, error) {
|
||
|
newObjects, err := c.Objects(container, opts)
|
||
|
if err == nil {
|
||
|
objects = append(objects, newObjects...)
|
||
|
}
|
||
|
return newObjects, err
|
||
|
})
|
||
|
return objects, err
|
||
|
}
|
||
|
|
||
|
// ObjectNamesAll is like ObjectNames but it returns all the Objects
|
||
|
//
|
||
|
// It calls ObjectNames multiple times using the Marker parameter
|
||
|
//
|
||
|
// It has a default Limit parameter but you may pass in your own
|
||
|
func (c *Connection) ObjectNamesAll(container string, opts *ObjectsOpts) ([]string, error) {
|
||
|
objects := make([]string, 0)
|
||
|
err := c.ObjectsWalk(container, opts, func(opts *ObjectsOpts) (interface{}, error) {
|
||
|
newObjects, err := c.ObjectNames(container, opts)
|
||
|
if err == nil {
|
||
|
objects = append(objects, newObjects...)
|
||
|
}
|
||
|
return newObjects, err
|
||
|
})
|
||
|
return objects, err
|
||
|
}
|
||
|
|
||
|
// Account contains information about this account.
|
||
|
type Account struct {
|
||
|
BytesUsed int64 // total number of bytes used
|
||
|
Containers int64 // total number of containers
|
||
|
Objects int64 // total number of objects
|
||
|
}
|
||
|
|
||
|
// getInt64FromHeader is a helper function to decode int64 from header.
|
||
|
func getInt64FromHeader(resp *http.Response, header string) (result int64, err error) {
|
||
|
value := resp.Header.Get(header)
|
||
|
result, err = strconv.ParseInt(value, 10, 64)
|
||
|
if err != nil {
|
||
|
err = newErrorf(0, "Bad Header '%s': '%s': %s", header, value, err)
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Account returns info about the account in an Account struct.
|
||
|
func (c *Connection) Account() (info Account, headers Headers, err error) {
|
||
|
var resp *http.Response
|
||
|
resp, headers, err = c.storage(RequestOpts{
|
||
|
Operation: "HEAD",
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
NoResponse: true,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
// Parse the headers into a dict
|
||
|
//
|
||
|
// {'Accept-Ranges': 'bytes',
|
||
|
// 'Content-Length': '0',
|
||
|
// 'Date': 'Tue, 05 Jul 2011 16:37:06 GMT',
|
||
|
// 'X-Account-Bytes-Used': '316598182',
|
||
|
// 'X-Account-Container-Count': '4',
|
||
|
// 'X-Account-Object-Count': '1433'}
|
||
|
if info.BytesUsed, err = getInt64FromHeader(resp, "X-Account-Bytes-Used"); err != nil {
|
||
|
return
|
||
|
}
|
||
|
if info.Containers, err = getInt64FromHeader(resp, "X-Account-Container-Count"); err != nil {
|
||
|
return
|
||
|
}
|
||
|
if info.Objects, err = getInt64FromHeader(resp, "X-Account-Object-Count"); err != nil {
|
||
|
return
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// AccountUpdate adds, replaces or remove account metadata.
|
||
|
//
|
||
|
// Add or update keys by mentioning them in the Headers.
|
||
|
//
|
||
|
// Remove keys by setting them to an empty string.
|
||
|
func (c *Connection) AccountUpdate(h Headers) error {
|
||
|
_, _, err := c.storage(RequestOpts{
|
||
|
Operation: "POST",
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
NoResponse: true,
|
||
|
Headers: h,
|
||
|
})
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// ContainerCreate creates a container.
|
||
|
//
|
||
|
// If you don't want to add Headers just pass in nil
|
||
|
//
|
||
|
// No error is returned if it already exists but the metadata if any will be updated.
|
||
|
func (c *Connection) ContainerCreate(container string, h Headers) error {
|
||
|
_, _, err := c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
Operation: "PUT",
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
NoResponse: true,
|
||
|
Headers: h,
|
||
|
})
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// ContainerDelete deletes a container.
|
||
|
//
|
||
|
// May return ContainerDoesNotExist or ContainerNotEmpty
|
||
|
func (c *Connection) ContainerDelete(container string) error {
|
||
|
_, _, err := c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
Operation: "DELETE",
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
NoResponse: true,
|
||
|
})
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// Container returns info about a single container including any
|
||
|
// metadata in the headers.
|
||
|
func (c *Connection) Container(container string) (info Container, headers Headers, err error) {
|
||
|
var resp *http.Response
|
||
|
resp, headers, err = c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
Operation: "HEAD",
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
NoResponse: true,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
// Parse the headers into the struct
|
||
|
info.Name = container
|
||
|
if info.Bytes, err = getInt64FromHeader(resp, "X-Container-Bytes-Used"); err != nil {
|
||
|
return
|
||
|
}
|
||
|
if info.Count, err = getInt64FromHeader(resp, "X-Container-Object-Count"); err != nil {
|
||
|
return
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ContainerUpdate adds, replaces or removes container metadata.
|
||
|
//
|
||
|
// Add or update keys by mentioning them in the Metadata.
|
||
|
//
|
||
|
// Remove keys by setting them to an empty string.
|
||
|
//
|
||
|
// Container metadata can only be read with Container() not with Containers().
|
||
|
func (c *Connection) ContainerUpdate(container string, h Headers) error {
|
||
|
_, _, err := c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
Operation: "POST",
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
NoResponse: true,
|
||
|
Headers: h,
|
||
|
})
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// ------------------------------------------------------------
|
||
|
|
||
|
// ObjectCreateFile represents a swift object open for writing
|
||
|
type ObjectCreateFile struct {
|
||
|
checkHash bool // whether we are checking the hash
|
||
|
pipeReader *io.PipeReader // pipe for the caller to use
|
||
|
pipeWriter *io.PipeWriter
|
||
|
hash hash.Hash // hash being build up as we go along
|
||
|
done chan struct{} // signals when the upload has finished
|
||
|
resp *http.Response // valid when done has signalled
|
||
|
err error // ditto
|
||
|
headers Headers // ditto
|
||
|
}
|
||
|
|
||
|
// Write bytes to the object - see io.Writer
|
||
|
func (file *ObjectCreateFile) Write(p []byte) (n int, err error) {
|
||
|
n, err = file.pipeWriter.Write(p)
|
||
|
if err == io.ErrClosedPipe {
|
||
|
if file.err != nil {
|
||
|
return 0, file.err
|
||
|
}
|
||
|
return 0, newError(500, "Write on closed file")
|
||
|
}
|
||
|
if err == nil && file.checkHash {
|
||
|
_, _ = file.hash.Write(p)
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Close the object and checks the md5sum if it was required.
|
||
|
//
|
||
|
// Also returns any other errors from the server (eg container not
|
||
|
// found) so it is very important to check the errors on this method.
|
||
|
func (file *ObjectCreateFile) Close() error {
|
||
|
// Close the body
|
||
|
err := file.pipeWriter.Close()
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// Wait for the HTTP operation to complete
|
||
|
<-file.done
|
||
|
|
||
|
// Check errors
|
||
|
if file.err != nil {
|
||
|
return file.err
|
||
|
}
|
||
|
if file.checkHash {
|
||
|
receivedMd5 := strings.ToLower(file.headers["Etag"])
|
||
|
calculatedMd5 := fmt.Sprintf("%x", file.hash.Sum(nil))
|
||
|
if receivedMd5 != calculatedMd5 {
|
||
|
return ObjectCorrupted
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// Headers returns the response headers from the created object if the upload
|
||
|
// has been completed. The Close() method must be called on an ObjectCreateFile
|
||
|
// before this method.
|
||
|
func (file *ObjectCreateFile) Headers() (Headers, error) {
|
||
|
// error out if upload is not complete.
|
||
|
select {
|
||
|
case <-file.done:
|
||
|
default:
|
||
|
return nil, fmt.Errorf("Cannot get metadata, object upload failed or has not yet completed.")
|
||
|
}
|
||
|
return file.headers, nil
|
||
|
}
|
||
|
|
||
|
// Check it satisfies the interface
|
||
|
var _ io.WriteCloser = &ObjectCreateFile{}
|
||
|
|
||
|
// objectPutHeaders create a set of headers for a PUT
|
||
|
//
|
||
|
// It guesses the contentType from the objectName if it isn't set
|
||
|
//
|
||
|
// checkHash may be changed
|
||
|
func objectPutHeaders(objectName string, checkHash *bool, Hash string, contentType string, h Headers) Headers {
|
||
|
if contentType == "" {
|
||
|
contentType = mime.TypeByExtension(path.Ext(objectName))
|
||
|
if contentType == "" {
|
||
|
contentType = "application/octet-stream"
|
||
|
}
|
||
|
}
|
||
|
// Meta stuff
|
||
|
extraHeaders := map[string]string{
|
||
|
"Content-Type": contentType,
|
||
|
}
|
||
|
for key, value := range h {
|
||
|
extraHeaders[key] = value
|
||
|
}
|
||
|
if Hash != "" {
|
||
|
extraHeaders["Etag"] = Hash
|
||
|
*checkHash = false // the server will do it
|
||
|
}
|
||
|
return extraHeaders
|
||
|
}
|
||
|
|
||
|
// ObjectCreate creates or updates the object in the container. It
|
||
|
// returns an io.WriteCloser you should write the contents to. You
|
||
|
// MUST call Close() on it and you MUST check the error return from
|
||
|
// Close().
|
||
|
//
|
||
|
// If checkHash is True then it will calculate the MD5 Hash of the
|
||
|
// file as it is being uploaded and check it against that returned
|
||
|
// from the server. If it is wrong then it will return
|
||
|
// ObjectCorrupted on Close()
|
||
|
//
|
||
|
// If you know the MD5 hash of the object ahead of time then set the
|
||
|
// Hash parameter and it will be sent to the server (as an Etag
|
||
|
// header) and the server will check the MD5 itself after the upload,
|
||
|
// and this will return ObjectCorrupted on Close() if it is incorrect.
|
||
|
//
|
||
|
// If you don't want any error protection (not recommended) then set
|
||
|
// checkHash to false and Hash to "".
|
||
|
//
|
||
|
// If contentType is set it will be used, otherwise one will be
|
||
|
// guessed from objectName using mime.TypeByExtension
|
||
|
func (c *Connection) ObjectCreate(container string, objectName string, checkHash bool, Hash string, contentType string, h Headers) (file *ObjectCreateFile, err error) {
|
||
|
extraHeaders := objectPutHeaders(objectName, &checkHash, Hash, contentType, h)
|
||
|
pipeReader, pipeWriter := io.Pipe()
|
||
|
file = &ObjectCreateFile{
|
||
|
hash: md5.New(),
|
||
|
checkHash: checkHash,
|
||
|
pipeReader: pipeReader,
|
||
|
pipeWriter: pipeWriter,
|
||
|
done: make(chan struct{}),
|
||
|
}
|
||
|
// Run the PUT in the background piping it data
|
||
|
go func() {
|
||
|
opts := RequestOpts{
|
||
|
Container: container,
|
||
|
ObjectName: objectName,
|
||
|
Operation: "PUT",
|
||
|
Headers: extraHeaders,
|
||
|
Body: pipeReader,
|
||
|
NoResponse: true,
|
||
|
ErrorMap: objectErrorMap,
|
||
|
}
|
||
|
file.resp, file.headers, file.err = c.storage(opts)
|
||
|
// Signal finished
|
||
|
pipeReader.Close()
|
||
|
close(file.done)
|
||
|
}()
|
||
|
return
|
||
|
}
|
||
|
|
||
|
func (c *Connection) objectPut(container string, objectName string, contents io.Reader, checkHash bool, Hash string, contentType string, h Headers, parameters url.Values) (headers Headers, err error) {
|
||
|
extraHeaders := objectPutHeaders(objectName, &checkHash, Hash, contentType, h)
|
||
|
hash := md5.New()
|
||
|
var body io.Reader = contents
|
||
|
if checkHash {
|
||
|
body = io.TeeReader(contents, hash)
|
||
|
}
|
||
|
_, headers, err = c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
ObjectName: objectName,
|
||
|
Operation: "PUT",
|
||
|
Headers: extraHeaders,
|
||
|
Body: body,
|
||
|
NoResponse: true,
|
||
|
ErrorMap: objectErrorMap,
|
||
|
Parameters: parameters,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
if checkHash {
|
||
|
receivedMd5 := strings.ToLower(headers["Etag"])
|
||
|
calculatedMd5 := fmt.Sprintf("%x", hash.Sum(nil))
|
||
|
if receivedMd5 != calculatedMd5 {
|
||
|
err = ObjectCorrupted
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectPut creates or updates the path in the container from
|
||
|
// contents. contents should be an open io.Reader which will have all
|
||
|
// its contents read.
|
||
|
//
|
||
|
// This is a low level interface.
|
||
|
//
|
||
|
// If checkHash is True then it will calculate the MD5 Hash of the
|
||
|
// file as it is being uploaded and check it against that returned
|
||
|
// from the server. If it is wrong then it will return
|
||
|
// ObjectCorrupted.
|
||
|
//
|
||
|
// If you know the MD5 hash of the object ahead of time then set the
|
||
|
// Hash parameter and it will be sent to the server (as an Etag
|
||
|
// header) and the server will check the MD5 itself after the upload,
|
||
|
// and this will return ObjectCorrupted if it is incorrect.
|
||
|
//
|
||
|
// If you don't want any error protection (not recommended) then set
|
||
|
// checkHash to false and Hash to "".
|
||
|
//
|
||
|
// If contentType is set it will be used, otherwise one will be
|
||
|
// guessed from objectName using mime.TypeByExtension
|
||
|
func (c *Connection) ObjectPut(container string, objectName string, contents io.Reader, checkHash bool, Hash string, contentType string, h Headers) (headers Headers, err error) {
|
||
|
return c.objectPut(container, objectName, contents, checkHash, Hash, contentType, h, nil)
|
||
|
}
|
||
|
|
||
|
// ObjectPutBytes creates an object from a []byte in a container.
|
||
|
//
|
||
|
// This is a simplified interface which checks the MD5.
|
||
|
func (c *Connection) ObjectPutBytes(container string, objectName string, contents []byte, contentType string) (err error) {
|
||
|
buf := bytes.NewBuffer(contents)
|
||
|
h := Headers{"Content-Length": strconv.Itoa(len(contents))}
|
||
|
_, err = c.ObjectPut(container, objectName, buf, true, "", contentType, h)
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectPutString creates an object from a string in a container.
|
||
|
//
|
||
|
// This is a simplified interface which checks the MD5
|
||
|
func (c *Connection) ObjectPutString(container string, objectName string, contents string, contentType string) (err error) {
|
||
|
buf := strings.NewReader(contents)
|
||
|
h := Headers{"Content-Length": strconv.Itoa(len(contents))}
|
||
|
_, err = c.ObjectPut(container, objectName, buf, true, "", contentType, h)
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectOpenFile represents a swift object open for reading
|
||
|
type ObjectOpenFile struct {
|
||
|
connection *Connection // stored copy of Connection used in Open
|
||
|
container string // stored copy of container used in Open
|
||
|
objectName string // stored copy of objectName used in Open
|
||
|
headers Headers // stored copy of headers used in Open
|
||
|
resp *http.Response // http connection
|
||
|
body io.Reader // read data from this
|
||
|
checkHash bool // true if checking MD5
|
||
|
hash hash.Hash // currently accumulating MD5
|
||
|
bytes int64 // number of bytes read on this connection
|
||
|
eof bool // whether we have read end of file
|
||
|
pos int64 // current position when reading
|
||
|
lengthOk bool // whether length is valid
|
||
|
length int64 // length of the object if read
|
||
|
seeked bool // whether we have seeked this file or not
|
||
|
overSeeked bool // set if we have seeked to the end or beyond
|
||
|
}
|
||
|
|
||
|
// Read bytes from the object - see io.Reader
|
||
|
func (file *ObjectOpenFile) Read(p []byte) (n int, err error) {
|
||
|
if file.overSeeked {
|
||
|
return 0, io.EOF
|
||
|
}
|
||
|
n, err = file.body.Read(p)
|
||
|
file.bytes += int64(n)
|
||
|
file.pos += int64(n)
|
||
|
if err == io.EOF {
|
||
|
file.eof = true
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Seek sets the offset for the next Read to offset, interpreted
|
||
|
// according to whence: 0 means relative to the origin of the file, 1
|
||
|
// means relative to the current offset, and 2 means relative to the
|
||
|
// end. Seek returns the new offset and an Error, if any.
|
||
|
//
|
||
|
// Seek uses HTTP Range headers which, if the file pointer is moved,
|
||
|
// will involve reopening the HTTP connection.
|
||
|
//
|
||
|
// Note that you can't seek to the end of a file or beyond; HTTP Range
|
||
|
// requests don't support the file pointer being outside the data,
|
||
|
// unlike os.File
|
||
|
//
|
||
|
// Seek(0, 1) will return the current file pointer.
|
||
|
func (file *ObjectOpenFile) Seek(offset int64, whence int) (newPos int64, err error) {
|
||
|
file.overSeeked = false
|
||
|
switch whence {
|
||
|
case 0: // relative to start
|
||
|
newPos = offset
|
||
|
case 1: // relative to current
|
||
|
newPos = file.pos + offset
|
||
|
case 2: // relative to end
|
||
|
if !file.lengthOk {
|
||
|
return file.pos, newError(0, "Length of file unknown so can't seek from end")
|
||
|
}
|
||
|
newPos = file.length + offset
|
||
|
if offset >= 0 {
|
||
|
file.overSeeked = true
|
||
|
return
|
||
|
}
|
||
|
default:
|
||
|
panic("Unknown whence in ObjectOpenFile.Seek")
|
||
|
}
|
||
|
// If at correct position (quite likely), do nothing
|
||
|
if newPos == file.pos {
|
||
|
return
|
||
|
}
|
||
|
// Close the file...
|
||
|
file.seeked = true
|
||
|
err = file.Close()
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
// ...and re-open with a Range header
|
||
|
if file.headers == nil {
|
||
|
file.headers = Headers{}
|
||
|
}
|
||
|
if newPos > 0 {
|
||
|
file.headers["Range"] = fmt.Sprintf("bytes=%d-", newPos)
|
||
|
} else {
|
||
|
delete(file.headers, "Range")
|
||
|
}
|
||
|
newFile, _, err := file.connection.ObjectOpen(file.container, file.objectName, false, file.headers)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
// Update the file
|
||
|
file.resp = newFile.resp
|
||
|
file.body = newFile.body
|
||
|
file.checkHash = false
|
||
|
file.pos = newPos
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Length gets the objects content length either from a cached copy or
|
||
|
// from the server.
|
||
|
func (file *ObjectOpenFile) Length() (int64, error) {
|
||
|
if !file.lengthOk {
|
||
|
info, _, err := file.connection.Object(file.container, file.objectName)
|
||
|
file.length = info.Bytes
|
||
|
file.lengthOk = (err == nil)
|
||
|
return file.length, err
|
||
|
}
|
||
|
return file.length, nil
|
||
|
}
|
||
|
|
||
|
// Close the object and checks the length and md5sum if it was
|
||
|
// required and all the object was read
|
||
|
func (file *ObjectOpenFile) Close() (err error) {
|
||
|
// Close the body at the end
|
||
|
defer checkClose(file.resp.Body, &err)
|
||
|
|
||
|
// If not end of file or seeked then can't check anything
|
||
|
if !file.eof || file.seeked {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Check the MD5 sum if requested
|
||
|
if file.checkHash {
|
||
|
receivedMd5 := strings.ToLower(file.resp.Header.Get("Etag"))
|
||
|
calculatedMd5 := fmt.Sprintf("%x", file.hash.Sum(nil))
|
||
|
if receivedMd5 != calculatedMd5 {
|
||
|
err = ObjectCorrupted
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// Check to see we read the correct number of bytes
|
||
|
if file.lengthOk && file.length != file.bytes {
|
||
|
err = ObjectCorrupted
|
||
|
return
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Check it satisfies the interfaces
|
||
|
var _ io.ReadCloser = &ObjectOpenFile{}
|
||
|
var _ io.Seeker = &ObjectOpenFile{}
|
||
|
|
||
|
func (c *Connection) objectOpenBase(container string, objectName string, checkHash bool, h Headers, parameters url.Values) (file *ObjectOpenFile, headers Headers, err error) {
|
||
|
var resp *http.Response
|
||
|
opts := RequestOpts{
|
||
|
Container: container,
|
||
|
ObjectName: objectName,
|
||
|
Operation: "GET",
|
||
|
ErrorMap: objectErrorMap,
|
||
|
Headers: h,
|
||
|
Parameters: parameters,
|
||
|
}
|
||
|
resp, headers, err = c.storage(opts)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
// Can't check MD5 on an object with X-Object-Manifest or X-Static-Large-Object set
|
||
|
if checkHash && headers.IsLargeObject() {
|
||
|
// log.Printf("swift: turning off md5 checking on object with manifest %v", objectName)
|
||
|
checkHash = false
|
||
|
}
|
||
|
file = &ObjectOpenFile{
|
||
|
connection: c,
|
||
|
container: container,
|
||
|
objectName: objectName,
|
||
|
headers: h,
|
||
|
resp: resp,
|
||
|
checkHash: checkHash,
|
||
|
body: resp.Body,
|
||
|
}
|
||
|
if checkHash {
|
||
|
file.hash = md5.New()
|
||
|
file.body = io.TeeReader(resp.Body, file.hash)
|
||
|
}
|
||
|
// Read Content-Length
|
||
|
if resp.Header.Get("Content-Length") != "" {
|
||
|
file.length, err = getInt64FromHeader(resp, "Content-Length")
|
||
|
file.lengthOk = (err == nil)
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
|
||
|
func (c *Connection) objectOpen(container string, objectName string, checkHash bool, h Headers, parameters url.Values) (file *ObjectOpenFile, headers Headers, err error) {
|
||
|
err = withLORetry(0, func() (Headers, int64, error) {
|
||
|
file, headers, err = c.objectOpenBase(container, objectName, checkHash, h, parameters)
|
||
|
if err != nil {
|
||
|
return headers, 0, err
|
||
|
}
|
||
|
return headers, file.length, nil
|
||
|
})
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectOpen returns an ObjectOpenFile for reading the contents of
|
||
|
// the object. This satisfies the io.ReadCloser and the io.Seeker
|
||
|
// interfaces.
|
||
|
//
|
||
|
// You must call Close() on contents when finished
|
||
|
//
|
||
|
// Returns the headers of the response.
|
||
|
//
|
||
|
// If checkHash is true then it will calculate the md5sum of the file
|
||
|
// as it is being received and check it against that returned from the
|
||
|
// server. If it is wrong then it will return ObjectCorrupted. It
|
||
|
// will also check the length returned. No checking will be done if
|
||
|
// you don't read all the contents.
|
||
|
//
|
||
|
// Note that objects with X-Object-Manifest or X-Static-Large-Object
|
||
|
// set won't ever have their md5sum's checked as the md5sum reported
|
||
|
// on the object is actually the md5sum of the md5sums of the
|
||
|
// parts. This isn't very helpful to detect a corrupted download as
|
||
|
// the size of the parts aren't known without doing more operations.
|
||
|
// If you want to ensure integrity of an object with a manifest then
|
||
|
// you will need to download everything in the manifest separately.
|
||
|
//
|
||
|
// headers["Content-Type"] will give the content type if desired.
|
||
|
func (c *Connection) ObjectOpen(container string, objectName string, checkHash bool, h Headers) (file *ObjectOpenFile, headers Headers, err error) {
|
||
|
return c.objectOpen(container, objectName, checkHash, h, nil)
|
||
|
}
|
||
|
|
||
|
// ObjectGet gets the object into the io.Writer contents.
|
||
|
//
|
||
|
// Returns the headers of the response.
|
||
|
//
|
||
|
// If checkHash is true then it will calculate the md5sum of the file
|
||
|
// as it is being received and check it against that returned from the
|
||
|
// server. If it is wrong then it will return ObjectCorrupted.
|
||
|
//
|
||
|
// headers["Content-Type"] will give the content type if desired.
|
||
|
func (c *Connection) ObjectGet(container string, objectName string, contents io.Writer, checkHash bool, h Headers) (headers Headers, err error) {
|
||
|
file, headers, err := c.ObjectOpen(container, objectName, checkHash, h)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
defer checkClose(file, &err)
|
||
|
_, err = io.Copy(contents, file)
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectGetBytes returns an object as a []byte.
|
||
|
//
|
||
|
// This is a simplified interface which checks the MD5
|
||
|
func (c *Connection) ObjectGetBytes(container string, objectName string) (contents []byte, err error) {
|
||
|
var buf bytes.Buffer
|
||
|
_, err = c.ObjectGet(container, objectName, &buf, true, nil)
|
||
|
contents = buf.Bytes()
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectGetString returns an object as a string.
|
||
|
//
|
||
|
// This is a simplified interface which checks the MD5
|
||
|
func (c *Connection) ObjectGetString(container string, objectName string) (contents string, err error) {
|
||
|
var buf bytes.Buffer
|
||
|
_, err = c.ObjectGet(container, objectName, &buf, true, nil)
|
||
|
contents = buf.String()
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectDelete deletes the object.
|
||
|
//
|
||
|
// May return ObjectNotFound if the object isn't found
|
||
|
func (c *Connection) ObjectDelete(container string, objectName string) error {
|
||
|
_, _, err := c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
ObjectName: objectName,
|
||
|
Operation: "DELETE",
|
||
|
ErrorMap: objectErrorMap,
|
||
|
})
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// ObjectTempUrl returns a temporary URL for an object
|
||
|
func (c *Connection) ObjectTempUrl(container string, objectName string, secretKey string, method string, expires time.Time) string {
|
||
|
mac := hmac.New(sha1.New, []byte(secretKey))
|
||
|
prefix, _ := url.Parse(c.StorageUrl)
|
||
|
body := fmt.Sprintf("%s\n%d\n%s/%s/%s", method, expires.Unix(), prefix.Path, container, objectName)
|
||
|
mac.Write([]byte(body))
|
||
|
sig := hex.EncodeToString(mac.Sum(nil))
|
||
|
return fmt.Sprintf("%s/%s/%s?temp_url_sig=%s&temp_url_expires=%d", c.StorageUrl, container, objectName, sig, expires.Unix())
|
||
|
}
|
||
|
|
||
|
// parseResponseStatus parses string like "200 OK" and returns Error.
|
||
|
//
|
||
|
// For status codes beween 200 and 299, this returns nil.
|
||
|
func parseResponseStatus(resp string, errorMap errorMap) error {
|
||
|
code := 0
|
||
|
reason := resp
|
||
|
t := strings.SplitN(resp, " ", 2)
|
||
|
if len(t) == 2 {
|
||
|
ncode, err := strconv.Atoi(t[0])
|
||
|
if err == nil {
|
||
|
code = ncode
|
||
|
reason = t[1]
|
||
|
}
|
||
|
}
|
||
|
if errorMap != nil {
|
||
|
if err, ok := errorMap[code]; ok {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
if 200 <= code && code <= 299 {
|
||
|
return nil
|
||
|
}
|
||
|
return newError(code, reason)
|
||
|
}
|
||
|
|
||
|
// BulkDeleteResult stores results of BulkDelete().
|
||
|
//
|
||
|
// Individual errors may (or may not) be returned by Errors.
|
||
|
// Errors is a map whose keys are a full path of where the object was
|
||
|
// to be deleted, and whose values are Error objects. A full path of
|
||
|
// object looks like "/API_VERSION/USER_ACCOUNT/CONTAINER/OBJECT_PATH".
|
||
|
type BulkDeleteResult struct {
|
||
|
NumberNotFound int64 // # of objects not found.
|
||
|
NumberDeleted int64 // # of deleted objects.
|
||
|
Errors map[string]error // Mapping between object name and an error.
|
||
|
Headers Headers // Response HTTP headers.
|
||
|
}
|
||
|
|
||
|
func (c *Connection) doBulkDelete(objects []string) (result BulkDeleteResult, err error) {
|
||
|
var buffer bytes.Buffer
|
||
|
for _, s := range objects {
|
||
|
buffer.WriteString(url.QueryEscape(s) + "\n")
|
||
|
}
|
||
|
resp, headers, err := c.storage(RequestOpts{
|
||
|
Operation: "DELETE",
|
||
|
Parameters: url.Values{"bulk-delete": []string{"1"}},
|
||
|
Headers: Headers{
|
||
|
"Accept": "application/json",
|
||
|
"Content-Type": "text/plain",
|
||
|
},
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
Body: &buffer,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
var jsonResult struct {
|
||
|
NotFound int64 `json:"Number Not Found"`
|
||
|
Status string `json:"Response Status"`
|
||
|
Errors [][]string
|
||
|
Deleted int64 `json:"Number Deleted"`
|
||
|
}
|
||
|
err = readJson(resp, &jsonResult)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
err = parseResponseStatus(jsonResult.Status, objectErrorMap)
|
||
|
result.NumberNotFound = jsonResult.NotFound
|
||
|
result.NumberDeleted = jsonResult.Deleted
|
||
|
result.Headers = headers
|
||
|
el := make(map[string]error, len(jsonResult.Errors))
|
||
|
for _, t := range jsonResult.Errors {
|
||
|
if len(t) != 2 {
|
||
|
continue
|
||
|
}
|
||
|
el[t[0]] = parseResponseStatus(t[1], objectErrorMap)
|
||
|
}
|
||
|
result.Errors = el
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// BulkDelete deletes multiple objectNames from container in one operation.
|
||
|
//
|
||
|
// Some servers may not accept bulk-delete requests since bulk-delete is
|
||
|
// an optional feature of swift - these will return the Forbidden error.
|
||
|
//
|
||
|
// See also:
|
||
|
// * http://docs.openstack.org/trunk/openstack-object-storage/admin/content/object-storage-bulk-delete.html
|
||
|
// * http://docs.rackspace.com/files/api/v1/cf-devguide/content/Bulk_Delete-d1e2338.html
|
||
|
func (c *Connection) BulkDelete(container string, objectNames []string) (result BulkDeleteResult, err error) {
|
||
|
fullPaths := make([]string, len(objectNames))
|
||
|
for i, name := range objectNames {
|
||
|
fullPaths[i] = fmt.Sprintf("/%s/%s", container, name)
|
||
|
}
|
||
|
return c.doBulkDelete(fullPaths)
|
||
|
}
|
||
|
|
||
|
// BulkUploadResult stores results of BulkUpload().
|
||
|
//
|
||
|
// Individual errors may (or may not) be returned by Errors.
|
||
|
// Errors is a map whose keys are a full path of where an object was
|
||
|
// to be created, and whose values are Error objects. A full path of
|
||
|
// object looks like "/API_VERSION/USER_ACCOUNT/CONTAINER/OBJECT_PATH".
|
||
|
type BulkUploadResult struct {
|
||
|
NumberCreated int64 // # of created objects.
|
||
|
Errors map[string]error // Mapping between object name and an error.
|
||
|
Headers Headers // Response HTTP headers.
|
||
|
}
|
||
|
|
||
|
// BulkUpload uploads multiple files in one operation.
|
||
|
//
|
||
|
// uploadPath can be empty, a container name, or a pseudo-directory
|
||
|
// within a container. If uploadPath is empty, new containers may be
|
||
|
// automatically created.
|
||
|
//
|
||
|
// Files are read from dataStream. The format of the stream is specified
|
||
|
// by the format parameter. Available formats are:
|
||
|
// * UploadTar - Plain tar stream.
|
||
|
// * UploadTarGzip - Gzip compressed tar stream.
|
||
|
// * UploadTarBzip2 - Bzip2 compressed tar stream.
|
||
|
//
|
||
|
// Some servers may not accept bulk-upload requests since bulk-upload is
|
||
|
// an optional feature of swift - these will return the Forbidden error.
|
||
|
//
|
||
|
// See also:
|
||
|
// * http://docs.openstack.org/trunk/openstack-object-storage/admin/content/object-storage-extract-archive.html
|
||
|
// * http://docs.rackspace.com/files/api/v1/cf-devguide/content/Extract_Archive-d1e2338.html
|
||
|
func (c *Connection) BulkUpload(uploadPath string, dataStream io.Reader, format string, h Headers) (result BulkUploadResult, err error) {
|
||
|
extraHeaders := Headers{"Accept": "application/json"}
|
||
|
for key, value := range h {
|
||
|
extraHeaders[key] = value
|
||
|
}
|
||
|
// The following code abuses Container parameter intentionally.
|
||
|
// The best fix might be to rename Container to UploadPath.
|
||
|
resp, headers, err := c.storage(RequestOpts{
|
||
|
Container: uploadPath,
|
||
|
Operation: "PUT",
|
||
|
Parameters: url.Values{"extract-archive": []string{format}},
|
||
|
Headers: extraHeaders,
|
||
|
ErrorMap: ContainerErrorMap,
|
||
|
Body: dataStream,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
// Detect old servers which don't support this feature
|
||
|
if headers["Content-Type"] != "application/json" {
|
||
|
err = Forbidden
|
||
|
return
|
||
|
}
|
||
|
var jsonResult struct {
|
||
|
Created int64 `json:"Number Files Created"`
|
||
|
Status string `json:"Response Status"`
|
||
|
Errors [][]string
|
||
|
}
|
||
|
err = readJson(resp, &jsonResult)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
err = parseResponseStatus(jsonResult.Status, objectErrorMap)
|
||
|
result.NumberCreated = jsonResult.Created
|
||
|
result.Headers = headers
|
||
|
el := make(map[string]error, len(jsonResult.Errors))
|
||
|
for _, t := range jsonResult.Errors {
|
||
|
if len(t) != 2 {
|
||
|
continue
|
||
|
}
|
||
|
el[t[0]] = parseResponseStatus(t[1], objectErrorMap)
|
||
|
}
|
||
|
result.Errors = el
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// Object returns info about a single object including any metadata in the header.
|
||
|
//
|
||
|
// May return ObjectNotFound.
|
||
|
//
|
||
|
// Use headers.ObjectMetadata() to read the metadata in the Headers.
|
||
|
func (c *Connection) Object(container string, objectName string) (info Object, headers Headers, err error) {
|
||
|
err = withLORetry(0, func() (Headers, int64, error) {
|
||
|
info, headers, err = c.objectBase(container, objectName)
|
||
|
if err != nil {
|
||
|
return headers, 0, err
|
||
|
}
|
||
|
return headers, info.Bytes, nil
|
||
|
})
|
||
|
return
|
||
|
}
|
||
|
|
||
|
func (c *Connection) objectBase(container string, objectName string) (info Object, headers Headers, err error) {
|
||
|
var resp *http.Response
|
||
|
resp, headers, err = c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
ObjectName: objectName,
|
||
|
Operation: "HEAD",
|
||
|
ErrorMap: objectErrorMap,
|
||
|
NoResponse: true,
|
||
|
})
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
// Parse the headers into the struct
|
||
|
// HTTP/1.1 200 OK
|
||
|
// Date: Thu, 07 Jun 2010 20:59:39 GMT
|
||
|
// Server: Apache
|
||
|
// Last-Modified: Fri, 12 Jun 2010 13:40:18 GMT
|
||
|
// ETag: 8a964ee2a5e88be344f36c22562a6486
|
||
|
// Content-Length: 512000
|
||
|
// Content-Type: text/plain; charset=UTF-8
|
||
|
// X-Object-Meta-Meat: Bacon
|
||
|
// X-Object-Meta-Fruit: Bacon
|
||
|
// X-Object-Meta-Veggie: Bacon
|
||
|
// X-Object-Meta-Dairy: Bacon
|
||
|
info.Name = objectName
|
||
|
info.ContentType = resp.Header.Get("Content-Type")
|
||
|
if resp.Header.Get("Content-Length") != "" {
|
||
|
if info.Bytes, err = getInt64FromHeader(resp, "Content-Length"); err != nil {
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
// Currently ceph doesn't return a Last-Modified header for DLO manifests without any segments
|
||
|
// See ceph http://tracker.ceph.com/issues/15812
|
||
|
if resp.Header.Get("Last-Modified") != "" {
|
||
|
info.ServerLastModified = resp.Header.Get("Last-Modified")
|
||
|
if info.LastModified, err = time.Parse(http.TimeFormat, info.ServerLastModified); err != nil {
|
||
|
return
|
||
|
}
|
||
|
}
|
||
|
|
||
|
info.Hash = resp.Header.Get("Etag")
|
||
|
if resp.Header.Get("X-Object-Manifest") != "" {
|
||
|
info.ObjectType = DynamicLargeObjectType
|
||
|
} else if resp.Header.Get("X-Static-Large-Object") != "" {
|
||
|
info.ObjectType = StaticLargeObjectType
|
||
|
}
|
||
|
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectUpdate adds, replaces or removes object metadata.
|
||
|
//
|
||
|
// Add or Update keys by mentioning them in the Metadata. Use
|
||
|
// Metadata.ObjectHeaders and Headers.ObjectMetadata to convert your
|
||
|
// Metadata to and from normal HTTP headers.
|
||
|
//
|
||
|
// This removes all metadata previously added to the object and
|
||
|
// replaces it with that passed in so to delete keys, just don't
|
||
|
// mention them the headers you pass in.
|
||
|
//
|
||
|
// Object metadata can only be read with Object() not with Objects().
|
||
|
//
|
||
|
// This can also be used to set headers not already assigned such as
|
||
|
// X-Delete-At or X-Delete-After for expiring objects.
|
||
|
//
|
||
|
// You cannot use this to change any of the object's other headers
|
||
|
// such as Content-Type, ETag, etc.
|
||
|
//
|
||
|
// Refer to copying an object when you need to update metadata or
|
||
|
// other headers such as Content-Type or CORS headers.
|
||
|
//
|
||
|
// May return ObjectNotFound.
|
||
|
func (c *Connection) ObjectUpdate(container string, objectName string, h Headers) error {
|
||
|
_, _, err := c.storage(RequestOpts{
|
||
|
Container: container,
|
||
|
ObjectName: objectName,
|
||
|
Operation: "POST",
|
||
|
ErrorMap: objectErrorMap,
|
||
|
NoResponse: true,
|
||
|
Headers: h,
|
||
|
})
|
||
|
return err
|
||
|
}
|
||
|
|
||
|
// ObjectCopy does a server side copy of an object to a new position
|
||
|
//
|
||
|
// All metadata is preserved. If metadata is set in the headers then
|
||
|
// it overrides the old metadata on the copied object.
|
||
|
//
|
||
|
// The destination container must exist before the copy.
|
||
|
//
|
||
|
// You can use this to copy an object to itself - this is the only way
|
||
|
// to update the content type of an object.
|
||
|
func (c *Connection) ObjectCopy(srcContainer string, srcObjectName string, dstContainer string, dstObjectName string, h Headers) (headers Headers, err error) {
|
||
|
// Meta stuff
|
||
|
extraHeaders := map[string]string{
|
||
|
"Destination": dstContainer + "/" + dstObjectName,
|
||
|
}
|
||
|
for key, value := range h {
|
||
|
extraHeaders[key] = value
|
||
|
}
|
||
|
_, headers, err = c.storage(RequestOpts{
|
||
|
Container: srcContainer,
|
||
|
ObjectName: srcObjectName,
|
||
|
Operation: "COPY",
|
||
|
ErrorMap: objectErrorMap,
|
||
|
NoResponse: true,
|
||
|
Headers: extraHeaders,
|
||
|
})
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ObjectMove does a server side move of an object to a new position
|
||
|
//
|
||
|
// This is a convenience method which calls ObjectCopy then ObjectDelete
|
||
|
//
|
||
|
// All metadata is preserved.
|
||
|
//
|
||
|
// The destination container must exist before the copy.
|
||
|
func (c *Connection) ObjectMove(srcContainer string, srcObjectName string, dstContainer string, dstObjectName string) (err error) {
|
||
|
_, err = c.ObjectCopy(srcContainer, srcObjectName, dstContainer, dstObjectName, nil)
|
||
|
if err != nil {
|
||
|
return
|
||
|
}
|
||
|
return c.ObjectDelete(srcContainer, srcObjectName)
|
||
|
}
|
||
|
|
||
|
// ObjectUpdateContentType updates the content type of an object
|
||
|
//
|
||
|
// This is a convenience method which calls ObjectCopy
|
||
|
//
|
||
|
// All other metadata is preserved.
|
||
|
func (c *Connection) ObjectUpdateContentType(container string, objectName string, contentType string) (err error) {
|
||
|
h := Headers{"Content-Type": contentType}
|
||
|
_, err = c.ObjectCopy(container, objectName, container, objectName, h)
|
||
|
return
|
||
|
}
|
||
|
|
||
|
// ------------------------------------------------------------
|
||
|
|
||
|
// VersionContainerCreate is a helper method for creating and enabling version controlled containers.
|
||
|
//
|
||
|
// It builds the current object container, the non-current object version container, and enables versioning.
|
||
|
//
|
||
|
// If the server doesn't support versioning then it will return
|
||
|
// Forbidden however it will have created both the containers at that point.
|
||
|
func (c *Connection) VersionContainerCreate(current, version string) error {
|
||
|
if err := c.ContainerCreate(version, nil); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
if err := c.ContainerCreate(current, nil); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
if err := c.VersionEnable(current, version); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// VersionEnable enables versioning on the current container with version as the tracking container.
|
||
|
//
|
||
|
// May return Forbidden if this isn't supported by the server
|
||
|
func (c *Connection) VersionEnable(current, version string) error {
|
||
|
h := Headers{"X-Versions-Location": version}
|
||
|
if err := c.ContainerUpdate(current, h); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
// Check to see if the header was set properly
|
||
|
_, headers, err := c.Container(current)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
// If failed to set versions header, return Forbidden as the server doesn't support this
|
||
|
if headers["X-Versions-Location"] != version {
|
||
|
return Forbidden
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// VersionDisable disables versioning on the current container.
|
||
|
func (c *Connection) VersionDisable(current string) error {
|
||
|
h := Headers{"X-Versions-Location": ""}
|
||
|
if err := c.ContainerUpdate(current, h); err != nil {
|
||
|
return err
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// VersionObjectList returns a list of older versions of the object.
|
||
|
//
|
||
|
// Objects are returned in the format <length><object_name>/<timestamp>
|
||
|
func (c *Connection) VersionObjectList(version, object string) ([]string, error) {
|
||
|
opts := &ObjectsOpts{
|
||
|
// <3-character zero-padded hexadecimal character length><object name>/
|
||
|
Prefix: fmt.Sprintf("%03x", len(object)) + object + "/",
|
||
|
}
|
||
|
return c.ObjectNames(version, opts)
|
||
|
}
|