mirror of
https://github.com/minio/minio.git
synced 2024-12-24 22:25:54 -05:00
517 lines
15 KiB
Go
517 lines
15 KiB
Go
/*
|
|
* MinIO Cloud Storage, (C) 2018 MinIO, Inc.
|
|
*
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
* you may not use this file except in compliance with the License.
|
|
* You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
* See the License for the specific language governing permissions and
|
|
* limitations under the License.
|
|
*/
|
|
|
|
package cmd
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"crypto/tls"
|
|
"io"
|
|
"io/ioutil"
|
|
"net/url"
|
|
"path"
|
|
"strconv"
|
|
|
|
"encoding/gob"
|
|
"encoding/hex"
|
|
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/minio/minio/cmd/http"
|
|
"github.com/minio/minio/cmd/rest"
|
|
xnet "github.com/minio/minio/pkg/net"
|
|
)
|
|
|
|
func isNetworkError(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
if err.Error() == errConnectionStale.Error() {
|
|
return true
|
|
}
|
|
if nerr, ok := err.(*rest.NetworkError); ok {
|
|
return xnet.IsNetworkOrHostDown(nerr.Err)
|
|
}
|
|
return false
|
|
}
|
|
|
|
// Converts rpc.ServerError to underlying error. This function is
|
|
// written so that the storageAPI errors are consistent across network
|
|
// disks as well.
|
|
func toStorageErr(err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
|
|
if isNetworkError(err) {
|
|
return errDiskNotFound
|
|
}
|
|
|
|
switch err.Error() {
|
|
case io.EOF.Error():
|
|
return io.EOF
|
|
case io.ErrUnexpectedEOF.Error():
|
|
return io.ErrUnexpectedEOF
|
|
case errFileUnexpectedSize.Error():
|
|
return errFileUnexpectedSize
|
|
case errUnexpected.Error():
|
|
return errUnexpected
|
|
case errDiskFull.Error():
|
|
return errDiskFull
|
|
case errVolumeNotFound.Error():
|
|
return errVolumeNotFound
|
|
case errVolumeExists.Error():
|
|
return errVolumeExists
|
|
case errFileNotFound.Error():
|
|
return errFileNotFound
|
|
case errFileNameTooLong.Error():
|
|
return errFileNameTooLong
|
|
case errFileAccessDenied.Error():
|
|
return errFileAccessDenied
|
|
case errIsNotRegular.Error():
|
|
return errIsNotRegular
|
|
case errVolumeNotEmpty.Error():
|
|
return errVolumeNotEmpty
|
|
case errVolumeAccessDenied.Error():
|
|
return errVolumeAccessDenied
|
|
case errCorruptedFormat.Error():
|
|
return errCorruptedFormat
|
|
case errUnformattedDisk.Error():
|
|
return errUnformattedDisk
|
|
case errInvalidAccessKeyID.Error():
|
|
return errInvalidAccessKeyID
|
|
case errAuthentication.Error():
|
|
return errAuthentication
|
|
case errRPCAPIVersionUnsupported.Error():
|
|
return errRPCAPIVersionUnsupported
|
|
case errServerTimeMismatch.Error():
|
|
return errServerTimeMismatch
|
|
}
|
|
if strings.Contains(err.Error(), "Bitrot verification mismatch") {
|
|
var expected string
|
|
var received string
|
|
fmt.Sscanf(err.Error(), "Bitrot verification mismatch - expected %s received %s", &expected, &received)
|
|
// Go's Sscanf %s scans "," that comes after the expected hash, hence remove it. Providing "," in the format string does not help.
|
|
expected = strings.TrimSuffix(expected, ",")
|
|
bitrotErr := HashMismatchError{expected, received}
|
|
return bitrotErr
|
|
}
|
|
return err
|
|
}
|
|
|
|
// Abstracts a remote disk.
|
|
type storageRESTClient struct {
|
|
endpoint Endpoint
|
|
restClient *rest.Client
|
|
connected bool
|
|
lastError error
|
|
instanceID string // REST server's instanceID which is sent with every request for validation.
|
|
}
|
|
|
|
// Wrapper to restClient.Call to handle network errors, in case of network error the connection is makred disconnected
|
|
// permanently. The only way to restore the storage connection is at the xl-sets layer by xlsets.monitorAndConnectEndpoints()
|
|
// after verifying format.json
|
|
func (client *storageRESTClient) call(method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
|
|
if !client.connected {
|
|
return nil, errDiskNotFound
|
|
}
|
|
if values == nil {
|
|
values = make(url.Values)
|
|
}
|
|
values.Set(storageRESTInstanceID, client.instanceID)
|
|
respBody, err = client.restClient.Call(method, values, body, length)
|
|
if err == nil {
|
|
return respBody, nil
|
|
}
|
|
client.lastError = err
|
|
if isNetworkError(err) {
|
|
client.connected = false
|
|
}
|
|
|
|
return nil, toStorageErr(err)
|
|
}
|
|
|
|
// Stringer provides a canonicalized representation of network device.
|
|
func (client *storageRESTClient) String() string {
|
|
return client.endpoint.String()
|
|
}
|
|
|
|
// IsOnline - returns whether RPC client failed to connect or not.
|
|
func (client *storageRESTClient) IsOnline() bool {
|
|
return client.connected
|
|
}
|
|
|
|
// LastError - returns the network error if any.
|
|
func (client *storageRESTClient) LastError() error {
|
|
return client.lastError
|
|
}
|
|
|
|
// DiskInfo - fetch disk information for a remote disk.
|
|
func (client *storageRESTClient) DiskInfo() (info DiskInfo, err error) {
|
|
respBody, err := client.call(storageRESTMethodDiskInfo, nil, nil, -1)
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
err = gob.NewDecoder(respBody).Decode(&info)
|
|
return info, err
|
|
}
|
|
|
|
// MakeVol - create a volume on a remote disk.
|
|
func (client *storageRESTClient) MakeVol(volume string) (err error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
respBody, err := client.call(storageRESTMethodMakeVol, values, nil, -1)
|
|
defer http.DrainBody(respBody)
|
|
return err
|
|
}
|
|
|
|
// ListVols - List all volumes on a remote disk.
|
|
func (client *storageRESTClient) ListVols() (volinfo []VolInfo, err error) {
|
|
respBody, err := client.call(storageRESTMethodListVols, nil, nil, -1)
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
err = gob.NewDecoder(respBody).Decode(&volinfo)
|
|
return volinfo, err
|
|
}
|
|
|
|
// StatVol - get volume info over the network.
|
|
func (client *storageRESTClient) StatVol(volume string) (volInfo VolInfo, err error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
respBody, err := client.call(storageRESTMethodStatVol, values, nil, -1)
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
err = gob.NewDecoder(respBody).Decode(&volInfo)
|
|
return volInfo, err
|
|
}
|
|
|
|
// DeleteVol - Deletes a volume over the network.
|
|
func (client *storageRESTClient) DeleteVol(volume string) (err error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
respBody, err := client.call(storageRESTMethodDeleteVol, values, nil, -1)
|
|
defer http.DrainBody(respBody)
|
|
return err
|
|
}
|
|
|
|
// AppendFile - append to a file.
|
|
func (client *storageRESTClient) AppendFile(volume, path string, buffer []byte) error {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
reader := bytes.NewBuffer(buffer)
|
|
respBody, err := client.call(storageRESTMethodAppendFile, values, reader, -1)
|
|
defer http.DrainBody(respBody)
|
|
return err
|
|
}
|
|
|
|
func (client *storageRESTClient) CreateFile(volume, path string, length int64, r io.Reader) error {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
values.Set(storageRESTLength, strconv.Itoa(int(length)))
|
|
respBody, err := client.call(storageRESTMethodCreateFile, values, ioutil.NopCloser(r), length)
|
|
defer http.DrainBody(respBody)
|
|
return err
|
|
}
|
|
|
|
// WriteAll - write all data to a file.
|
|
func (client *storageRESTClient) WriteAll(volume, path string, reader io.Reader) error {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
respBody, err := client.call(storageRESTMethodWriteAll, values, reader, -1)
|
|
defer http.DrainBody(respBody)
|
|
return err
|
|
}
|
|
|
|
// StatFile - stat a file.
|
|
func (client *storageRESTClient) StatFile(volume, path string) (info FileInfo, err error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
respBody, err := client.call(storageRESTMethodStatFile, values, nil, -1)
|
|
if err != nil {
|
|
return info, err
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
err = gob.NewDecoder(respBody).Decode(&info)
|
|
return info, err
|
|
}
|
|
|
|
// ReadAll - reads all contents of a file.
|
|
func (client *storageRESTClient) ReadAll(volume, path string) ([]byte, error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
respBody, err := client.call(storageRESTMethodReadAll, values, nil, -1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
return ioutil.ReadAll(respBody)
|
|
}
|
|
|
|
// ReadFileStream - returns a reader for the requested file.
|
|
func (client *storageRESTClient) ReadFileStream(volume, path string, offset, length int64) (io.ReadCloser, error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
values.Set(storageRESTOffset, strconv.Itoa(int(offset)))
|
|
values.Set(storageRESTLength, strconv.Itoa(int(length)))
|
|
respBody, err := client.call(storageRESTMethodReadFileStream, values, nil, -1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return respBody, nil
|
|
}
|
|
|
|
// ReadFile - reads section of a file.
|
|
func (client *storageRESTClient) ReadFile(volume, path string, offset int64, buffer []byte, verifier *BitrotVerifier) (int64, error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
values.Set(storageRESTOffset, strconv.Itoa(int(offset)))
|
|
values.Set(storageRESTLength, strconv.Itoa(len(buffer)))
|
|
if verifier != nil {
|
|
values.Set(storageRESTBitrotAlgo, verifier.algorithm.String())
|
|
values.Set(storageRESTBitrotHash, hex.EncodeToString(verifier.sum))
|
|
} else {
|
|
values.Set(storageRESTBitrotAlgo, "")
|
|
values.Set(storageRESTBitrotHash, "")
|
|
}
|
|
respBody, err := client.call(storageRESTMethodReadFile, values, nil, -1)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
n, err := io.ReadFull(respBody, buffer)
|
|
return int64(n), err
|
|
}
|
|
|
|
func (client *storageRESTClient) Walk(volume, dirPath, marker string, recursive bool, leafFile string,
|
|
readMetadataFn readMetadataFunc, endWalkCh chan struct{}) (chan FileInfo, error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTDirPath, dirPath)
|
|
values.Set(storageRESTMarkerPath, marker)
|
|
values.Set(storageRESTRecursive, strconv.FormatBool(recursive))
|
|
values.Set(storageRESTLeafFile, leafFile)
|
|
respBody, err := client.call(storageRESTMethodWalk, values, nil, -1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
ch := make(chan FileInfo)
|
|
go func() {
|
|
defer close(ch)
|
|
defer http.DrainBody(respBody)
|
|
|
|
decoder := gob.NewDecoder(respBody)
|
|
for {
|
|
var fi FileInfo
|
|
if gerr := decoder.Decode(&fi); gerr != nil {
|
|
// Upon error return
|
|
return
|
|
}
|
|
select {
|
|
case ch <- fi:
|
|
case <-endWalkCh:
|
|
return
|
|
}
|
|
|
|
}
|
|
}()
|
|
|
|
return ch, nil
|
|
}
|
|
|
|
// ListDir - lists a directory.
|
|
func (client *storageRESTClient) ListDir(volume, dirPath string, count int, leafFile string) (entries []string, err error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTDirPath, dirPath)
|
|
values.Set(storageRESTCount, strconv.Itoa(count))
|
|
values.Set(storageRESTLeafFile, leafFile)
|
|
respBody, err := client.call(storageRESTMethodListDir, values, nil, -1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
err = gob.NewDecoder(respBody).Decode(&entries)
|
|
return entries, err
|
|
}
|
|
|
|
// DeleteFile - deletes a file.
|
|
func (client *storageRESTClient) DeleteFile(volume, path string) error {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
respBody, err := client.call(storageRESTMethodDeleteFile, values, nil, -1)
|
|
defer http.DrainBody(respBody)
|
|
return err
|
|
}
|
|
|
|
// DeleteFileBulk - deletes files in bulk.
|
|
func (client *storageRESTClient) DeleteFileBulk(volume string, paths []string) (errs []error, err error) {
|
|
if len(paths) == 0 {
|
|
return errs, err
|
|
}
|
|
errs = make([]error, len(paths))
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
for _, path := range paths {
|
|
values.Add(storageRESTFilePath, path)
|
|
}
|
|
respBody, err := client.call(storageRESTMethodDeleteFileBulk, values, nil, -1)
|
|
defer http.DrainBody(respBody)
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
bulkErrs := bulkErrorsResponse{}
|
|
gob.NewDecoder(respBody).Decode(&bulkErrs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for i, dErr := range bulkErrs.Errs {
|
|
errs[i] = toStorageErr(dErr)
|
|
}
|
|
|
|
return errs, nil
|
|
}
|
|
|
|
// RenameFile - renames a file.
|
|
func (client *storageRESTClient) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) (err error) {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTSrcVolume, srcVolume)
|
|
values.Set(storageRESTSrcPath, srcPath)
|
|
values.Set(storageRESTDstVolume, dstVolume)
|
|
values.Set(storageRESTDstPath, dstPath)
|
|
respBody, err := client.call(storageRESTMethodRenameFile, values, nil, -1)
|
|
defer http.DrainBody(respBody)
|
|
return err
|
|
}
|
|
|
|
// Gets peer storage server's instanceID - to be used with every REST call for validation.
|
|
func (client *storageRESTClient) getInstanceID() (err error) {
|
|
// getInstanceID() does not use storageRESTClient.call()
|
|
// function so we need to update lastError field here.
|
|
defer func() {
|
|
if err != nil {
|
|
client.lastError = err
|
|
}
|
|
}()
|
|
|
|
respBody, err := client.restClient.Call(storageRESTMethodGetInstanceID, nil, nil, -1)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer http.DrainBody(respBody)
|
|
instanceIDBuf := make([]byte, 64)
|
|
n, err := io.ReadFull(respBody, instanceIDBuf)
|
|
if err != io.EOF && err != io.ErrUnexpectedEOF {
|
|
return err
|
|
}
|
|
client.instanceID = string(instanceIDBuf[:n])
|
|
return nil
|
|
}
|
|
|
|
func (client *storageRESTClient) VerifyFile(volume, path string, size int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
|
|
values := make(url.Values)
|
|
values.Set(storageRESTVolume, volume)
|
|
values.Set(storageRESTFilePath, path)
|
|
values.Set(storageRESTBitrotAlgo, algo.String())
|
|
values.Set(storageRESTLength, strconv.FormatInt(size, 10))
|
|
values.Set(storageRESTShardSize, strconv.Itoa(int(shardSize)))
|
|
values.Set(storageRESTBitrotHash, hex.EncodeToString(sum))
|
|
|
|
respBody, err := client.call(storageRESTMethodVerifyFile, values, nil, -1)
|
|
defer http.DrainBody(respBody)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
reader := bufio.NewReader(respBody)
|
|
for {
|
|
b, err := reader.ReadByte()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if b != ' ' {
|
|
reader.UnreadByte()
|
|
break
|
|
}
|
|
}
|
|
verifyResp := &VerifyFileResp{}
|
|
err = gob.NewDecoder(reader).Decode(verifyResp)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return toStorageErr(verifyResp.Err)
|
|
}
|
|
|
|
// Close - marks the client as closed.
|
|
func (client *storageRESTClient) Close() error {
|
|
client.connected = false
|
|
client.restClient.Close()
|
|
return nil
|
|
}
|
|
|
|
// Returns a storage rest client.
|
|
func newStorageRESTClient(endpoint Endpoint) (*storageRESTClient, error) {
|
|
host, err := xnet.ParseHost(endpoint.Host)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
scheme := "http"
|
|
if globalIsSSL {
|
|
scheme = "https"
|
|
}
|
|
|
|
serverURL := &url.URL{
|
|
Scheme: scheme,
|
|
Host: endpoint.Host,
|
|
Path: path.Join(storageRESTPath, endpoint.Path),
|
|
}
|
|
|
|
var tlsConfig *tls.Config
|
|
if globalIsSSL {
|
|
tlsConfig = &tls.Config{
|
|
ServerName: host.Name,
|
|
RootCAs: globalRootCAs,
|
|
NextProtos: []string{"http/1.1"}, // Force http1.1
|
|
}
|
|
}
|
|
|
|
restClient, err := rest.NewClient(serverURL, tlsConfig, rest.DefaultRESTTimeout, newAuthToken)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
client := &storageRESTClient{endpoint: endpoint, restClient: restClient, connected: true}
|
|
client.connected = client.getInstanceID() == nil
|
|
return client, nil
|
|
}
|