Compare commits

...

33 Commits

Author SHA1 Message Date
ebozduman 882a1a1ccc Stops listing objects for write-only access (#6396) 2018-08-31 13:20:27 -07:00
Harshavardhana 8690d62146 Allow fallback listen if first listener fails (#6380)
On linux listen() uses kernel features TCP_FASTOPEN, DEFER_ACCEPT

Fixes #6379
2018-08-31 13:17:05 -07:00
Anis Elleuch 85117d554f xl: Avoid removing a directory if it is not an object dir (#6395)
DeleteObject should not remove any directory unless the latter
is an empty directory.

Fixes #6394
2018-08-31 13:16:35 -07:00
Harshavardhana 4487f70f08 Revert all GetObjectNInfo related PRs (#6398)
* Revert "Encrypted reader wrapped in NewGetObjectReader should be closed (#6383)"

This reverts commit 53a0bbeb5b.

* Revert "Change SelectAPI to use new GetObjectNInfo API (#6373)"

This reverts commit 5b05df215a.

* Revert "Implement GetObjectNInfo object layer call (#6290)"

This reverts commit e6d740ce09.
2018-08-31 13:10:12 -07:00
Harshavardhana fb27388101 HTTP headers are case insensitive handle them appropriately (#6390)
An issue was reproduced when minio-js client functional
tests are setting lower case http headers, in our current
master branch we specifically look for canonical host header
which may be not necessarily true for all http clients.
This leads to a perpetual hang on the *net.Conn*.

This PR fixes regression caused by #6206 by handling the
case insensitivity.
2018-08-31 17:00:32 +05:30
Kaan Kabalak 5e7ccc983d UI: Handle policies listed in mc as 'none' by not showing them (#6353) 2018-08-30 15:20:27 -07:00
Bala FA 72fa2b4537 Add RPC counters for HTTP stats. (#6206)
This patch introduces separate counters for HTTP stats for minio
reserved bucket.

Fixes #6158
2018-08-30 14:17:58 +05:30
Praveen raj Mani 5399d91965 Add version for appveyor to use go1.10.4 (#6384) 2018-08-30 14:09:31 +05:30
Harshavardhana 53a0bbeb5b Encrypted reader wrapped in NewGetObjectReader should be closed (#6383) 2018-08-29 19:18:00 -07:00
Harshavardhana 384a862940 Return quorum error based on disks in abortMultipartUpload (#6362)
Fixes #4980
2018-08-29 13:36:19 -07:00
Harshavardhana 029f52880b With no read quorum config should be treated as notFound (#6374)
This will allow the config subsystem to initialize properly
in situations where many servers are coming up in a rolling
fashion.
2018-08-28 14:23:22 -07:00
Harshavardhana 5b05df215a Change SelectAPI to use new GetObjectNInfo API (#6373)
This PR also removes some double checks
2018-08-28 13:08:30 -07:00
Harshavardhana a13cd7b7c4 Separate build functional tests from Makefile (#6351)
Recently travis seems to have issues with builds after
merge, so this PR is an experiment to fix this.

Failed: https://travis-ci.org/minio/minio/builds/419769285#L2310

Succeeded: https://travis-ci.org/minio/minio/builds/419565606#L2322

This PR was tested to be working properly on my fork to build
on travis, would need to do the same in minio/minio and see
how it goes through.
2018-08-28 13:57:01 +05:30
Anis Elleuch d524924b80 Fix gateway s3 doc to run custom S3 endpoint (#6369)
To pass a custom S3 endpoint in S3 gateway, the user needs
to specify it as an argument after 'minio gateway s3' and not
as '--address' option since this latter specifies the address
to which the gateway should listen.
2018-08-28 10:09:07 +05:30
Aditya Manthramurthy e6d740ce09 Implement GetObjectNInfo object layer call (#6290)
This combines calling GetObjectInfo and GetObject while returning a
io.ReadCloser for the object's body. This allows the two operations to
be under a single lock, fixing a race between getting object info and
reading the object body.
2018-08-27 15:28:23 +05:30
Harshavardhana cea4f82586 Revert "Remove curl after the usage (#6347)" (#6366)
This reverts commit dafa5073cb.

This commit revert also fixes #6365
2018-08-26 23:40:21 -07:00
Minio Trusted 1d6ce115da Update yaml files to latest version RELEASE.2018-08-25T01-56-38Z 2018-08-25 02:04:10 +00:00
Anis Elleuch 06d2dfa31c Fix WORM and BROWSER status calculation (#6360)
One typo introduced in a recent commit miscalculates if worm and browser
are enabled or not. A simple test is also added to detect this issue
in the future if it ever happens again.
2018-08-24 14:36:14 -07:00
poornas d547873b17 webhandler - display encryption errors properly (#6339)
For encrypted objects, download errors need to be
displayed in web response format instead of xml format.

Fixes #6327
2018-08-24 07:56:24 -07:00
Praveen raj Mani 01721a840a Build fix for go1.11rc1 (#6354)
Vendorized the following packages

- "github.com/rjeczalik/notify"
- "github.com/minio/highwayhash"

Fixes #6315
2018-08-24 13:29:17 +05:30
Krishna Srinivas 52f6d5aafc Rename of structs and methods (#6230)
Rename of ErasureStorage to Erasure (and rename of related variables and methods)
2018-08-23 23:35:37 -07:00
Harshavardhana 2211a5f1b8 Avoid ListenBucket targets to be listed in ServerInfo (#6340)
In current master when you do `mc watch` you can see a
dynamic ARN being listed which exposes the remote IP as well

```
mc watch play/airlines
```

On another terminal
```
mc admin info play
●  play.minio.io:9000
   Uptime : online since 11 hours ago
  Version : 2018-08-22T07:50:45Z
   Region :
 SQS ARNs : arn:minio:sqs::httpclient+51c39c3f-131d-42d9-b212-c5eb1450b9ee+73.222.245.195:33408
    Stats : Incoming 30GiB, Outgoing 7.6GiB
  Storage : Used 7.7GiB
```

SQS ARNs listed as part of ServerInfo should be only external targets,
since listing an ARN here is not useful and it cannot be re-purposed in
any manner.

This PR fixes this issue by filtering out httpclient from the ARN list.

This is a regression introduced in #5294 0e4431725c
2018-08-23 23:31:14 -07:00
Harshavardhana 1ffa6adcd4 Ignore io.EOF returned by ReadFrom for zero byte fs.json (#6346)
Fixes #6256
2018-08-24 11:34:21 +05:30
kannappanr add57a6938 Add content-length as part of event notification structure (#6341)
Fixes #6321
2018-08-23 14:40:54 -07:00
Manjunath A Kumatagi dafa5073cb Remove curl after the usage (#6347) 2018-08-23 10:48:29 -07:00
Praveen raj Mani 65e05a06fb Remove notifications Fix (#6082)
Remove all the notifications for an empty rulesMap

Fixes #6053
2018-08-23 22:53:18 +05:30
Dee Koder 5c9354894b Update Issue & PR Templates to have regression information when available (#6342) 2018-08-23 01:41:22 -07:00
Harshavardhana b01e69e08f Initialize global object layer after all subsystems have initialized (#6333)
This is to ensure that object API operations are not performed
on a server on which subsystems are yet to be initialized.
2018-08-22 23:11:17 -07:00
Raphael Randschau 8601f29d95 select: fix int overflow of math.MaxInt64 on ARM (#6317) 2018-08-22 16:16:04 +05:30
Nitish Tiwari 0aee722e3f Fix Minio browser screenshots in docs (#6334)
Fixes #6308
2018-08-22 13:15:36 +05:30
Harshavardhana beb6d40ce6 Avoid crash when policy subsystem is not initialized (#6326)
Fixes #6324
2018-08-21 15:38:51 -07:00
poornas 19db921555 CopyObject: fix regression in key rotation (#6331)
After key rotation, metadata was not being replaced with new sealed key.
Regression introduced in commit e71ef905f9
2018-08-21 15:12:00 -07:00
Minio Trusted 68b9e9e7e7 Update yaml files to latest version RELEASE.2018-08-21T00-37-20Z 2018-08-21 00:44:41 +00:00
82 changed files with 860 additions and 675 deletions
+4
View File
@@ -24,6 +24,10 @@
<!--- How has this issue affected you? What are you trying to accomplish? -->
<!--- Providing context helps us come up with a solution that is most useful in the real world -->
## Regression
<!-- Is this issue a regression? (Yes / No) -->
<!-- If Yes, optionally please include minio version or commit id or PR# that caused this regression, if you have these details. -->
## Your Environment
<!--- Include as many relevant details about the environment you experienced the bug in -->
* Version used (`minio version`):
+4
View File
@@ -7,6 +7,10 @@
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here. -->
## Regression
<!-- Is this PR fixing a regression? (Yes / No) -->
<!-- If Yes, optionally please include minio version or commit id or PR# that caused this regression, if you have these details. -->
## How Has This Been Tested?
<!--- Please describe in detail how you tested your changes. -->
<!--- Include details of your testing environment, and the tests you ran to -->
+2 -2
View File
@@ -20,14 +20,14 @@ matrix:
- os: linux
env:
- ARCH=x86_64
go: 1.10.1
go: 1.10.3
script:
- make
- diff -au <(gofmt -s -d cmd) <(printf "")
- diff -au <(gofmt -s -d pkg) <(printf "")
- make test GOFLAGS="-timeout 15m -race -v"
- make verify
- make coverage
- node --version
- cd browser && yarn && yarn test && cd ..
before_install:
+1 -1
View File
@@ -14,7 +14,7 @@ WORKDIR /go/src/github.com/minio/
COPY dockerscripts/docker-entrypoint.sh dockerscripts/healthcheck.sh /usr/bin/
RUN \
apk add --no-cache ca-certificates curl && \
apk add --no-cache ca-certificates 'curl>7.61.0' && \
apk add --no-cache --virtual .build-deps git && \
echo 'hosts: files mdns4_minimal [NOTFOUND=return] dns mdns4' >> /etc/nsswitch.conf && \
go get -v -d github.com/minio/minio && \
+1 -1
View File
@@ -16,7 +16,7 @@ COPY dockerscripts/docker-entrypoint.sh dockerscripts/healthcheck.sh /usr/bin/
COPY . /go/src/github.com/minio/minio
RUN \
apk add --no-cache ca-certificates curl && \
apk add --no-cache ca-certificates 'curl>7.61.0' && \
apk add --no-cache --virtual .build-deps git && \
echo 'hosts: files mdns4_minimal [NOTFOUND=return] dns mdns4' >> /etc/nsswitch.conf && \
cd /go/src/github.com/minio/minio && \
+1 -2
View File
@@ -9,8 +9,7 @@ ENV MINIO_ACCESS_KEY_FILE=access_key \
MINIO_SECRET_KEY_FILE=secret_key
RUN \
apk add --no-cache ca-certificates && \
apk add --no-cache --virtual .build-deps curl && \
apk add --no-cache ca-certificates 'curl>7.61.0' && \
echo 'hosts: files mdns4_minimal [NOTFOUND=return] dns mdns4' >> /etc/nsswitch.conf && \
curl https://dl.minio.io/server/minio/release/linux-amd64/minio > /usr/bin/minio && \
chmod +x /usr/bin/minio && \
+2
View File
@@ -61,6 +61,8 @@ check: test
test: verifiers build
@echo "Running unit tests"
@go test $(GOFLAGS) -tags kqueue ./...
verify: build
@echo "Verifying build"
@(env bash $(PWD)/buildscripts/verify-build.sh)
+1 -1
View File
@@ -12,7 +12,7 @@ clone_folder: c:\gopath\src\github.com\minio\minio
# Environment variables
environment:
GOPATH: c:\gopath
GOROOT: c:\go
GOROOT: c:\go110
# scripts that run after cloning repository
install:
+31 -28
View File
@@ -14,7 +14,7 @@
* limitations under the License.
*/
import { READ_ONLY, WRITE_ONLY, READ_WRITE } from '../constants'
import { READ_ONLY, WRITE_ONLY, READ_WRITE, NONE } from '../constants'
import React from "react"
import { connect } from "react-redux"
@@ -42,37 +42,40 @@ export class Policy extends React.Component {
render() {
const {policy, prefix} = this.props
let newPrefix = prefix
if (newPrefix === '')
newPrefix = '*'
return (
<div className="pmb-list">
<div className="pmbl-item">
{ newPrefix }
if (policy === NONE) {
return <noscript />
} else {
return (
<div className="pmb-list">
<div className="pmbl-item">
{ newPrefix }
</div>
<div className="pmbl-item">
<select className="form-control"
disabled
value={ policy }>
<option value={ READ_ONLY }>
Read Only
</option>
<option value={ WRITE_ONLY }>
Write Only
</option>
<option value={ READ_WRITE }>
Read and Write
</option>
</select>
</div>
<div className="pmbl-item">
<button className="btn btn-block btn-danger" onClick={ this.removePolicy.bind(this) }>
Remove
</button>
</div>
</div>
<div className="pmbl-item">
<select className="form-control"
disabled
value={ policy }>
<option value={ READ_ONLY }>
Read Only
</option>
<option value={ WRITE_ONLY }>
Write Only
</option>
<option value={ READ_WRITE }>
Read and Write
</option>
</select>
</div>
<div className="pmbl-item">
<button className="btn btn-block btn-danger" onClick={ this.removePolicy.bind(this) }>
Remove
</button>
</div>
</div>
)
)
}
}
}
@@ -17,7 +17,7 @@
import React from "react"
import { shallow, mount } from "enzyme"
import { Policy } from "../Policy"
import { READ_ONLY, WRITE_ONLY, READ_WRITE } from "../../constants"
import { READ_ONLY, WRITE_ONLY, READ_WRITE, NONE } from "../../constants"
import web from "../../web"
jest.mock("../../web", () => ({
@@ -31,6 +31,11 @@ describe("Policy", () => {
shallow(<Policy currentBucket={"bucket"} prefix={"foo"} policy={READ_ONLY} />)
})
it("should not render when policy is listed as 'none'", () => {
const wrapper = shallow(<Policy currentBucket={"bucket"} prefix={"foo"} policy={NONE} />)
expect(wrapper.find(".pmb-list").length).toBe(0)
})
it("should call web.setBucketPolicy and fetchPolicies on submit", () => {
const fetchPolicies = jest.fn()
const wrapper = shallow(
+1
View File
@@ -23,6 +23,7 @@ export const minioBrowserPrefix = p.slice(0, p.indexOf("/", 1))
export const READ_ONLY = "readonly"
export const WRITE_ONLY = "writeonly"
export const READ_WRITE = "readwrite"
export const NONE = "none"
export const SHARE_OBJECT_EXPIRY_DAYS = 5
export const SHARE_OBJECT_EXPIRY_HOURS = 0
+23 -34
View File
File diff suppressed because one or more lines are too long
+2 -2
View File
@@ -23,7 +23,7 @@ import (
"github.com/minio/highwayhash"
"github.com/minio/minio/cmd/logger"
"github.com/minio/sha256-simd"
sha256 "github.com/minio/sha256-simd"
"golang.org/x/crypto/blake2b"
)
@@ -70,7 +70,7 @@ func (a BitrotAlgorithm) New() hash.Hash {
}
}
// Available reports whether the given algorihm is available.
// Available reports whether the given algorithm is available.
func (a BitrotAlgorithm) Available() bool {
_, ok := bitrotAlgorithms[a]
return ok
+4 -4
View File
@@ -128,8 +128,8 @@ func (s *serverConfig) GetStorageClass() (storageClass, storageClass) {
// GetBrowser get current credentials.
func (s *serverConfig) GetBrowser() bool {
if globalIsEnvWORM {
return globalWORMEnabled
if globalIsEnvBrowser {
return globalIsBrowserEnabled
}
if s == nil {
return true
@@ -139,8 +139,8 @@ func (s *serverConfig) GetBrowser() bool {
// GetWorm get current credentials.
func (s *serverConfig) GetWorm() bool {
if globalIsEnvBrowser {
return globalIsBrowserEnabled
if globalIsEnvWORM {
return globalWORMEnabled
}
if s == nil {
return false
+17 -8
View File
@@ -66,6 +66,9 @@ func TestServerConfigWithEnvs(t *testing.T) {
os.Setenv("MINIO_BROWSER", "off")
defer os.Unsetenv("MINIO_BROWSER")
os.Setenv("MINIO_WORM", "on")
defer os.Unsetenv("MINIO_WORM")
os.Setenv("MINIO_ACCESS_KEY", "minio")
defer os.Unsetenv("MINIO_ACCESS_KEY")
@@ -99,29 +102,35 @@ func TestServerConfigWithEnvs(t *testing.T) {
// Init config
initConfig(objLayer)
// Check if serverConfig has
// Check if serverConfig has browser disabled
if globalServerConfig.GetBrowser() {
t.Errorf("Expecting browser is set to false found %v", globalServerConfig.GetBrowser())
t.Error("Expected browser to be disabled but it is not")
}
// Check if serverConfig has
// Check if serverConfig returns WORM config from the env
if !globalServerConfig.GetWorm() {
t.Error("Expected WORM to be enabled but it is not")
}
// Check if serverConfig has region from the environment
if globalServerConfig.GetRegion() != "us-west-1" {
t.Errorf("Expecting region to be \"us-west-1\" found %v", globalServerConfig.GetRegion())
t.Errorf("Expected region to be \"us-west-1\", found %v", globalServerConfig.GetRegion())
}
// Check if serverConfig has
// Check if serverConfig has credentials from the environment
cred := globalServerConfig.GetCredential()
if cred.AccessKey != "minio" {
t.Errorf("Expecting access key to be `minio` found %s", cred.AccessKey)
t.Errorf("Expected access key to be `minio`, found %s", cred.AccessKey)
}
if cred.SecretKey != "minio123" {
t.Errorf("Expecting access key to be `minio123` found %s", cred.SecretKey)
t.Errorf("Expected access key to be `minio123`, found %s", cred.SecretKey)
}
// Check if serverConfig has the correct domain
if globalServerConfig.Domain != "domain.com" {
t.Errorf("Expecting Domain to be `domain.com` found " + globalServerConfig.Domain)
t.Errorf("Expected Domain to be `domain.com`, found " + globalServerConfig.Domain)
}
}
+2 -6
View File
@@ -174,10 +174,6 @@ func TestServerConfigMigrateV2toV28(t *testing.T) {
}
defer os.RemoveAll(fsDir)
globalObjLayerMutex.Lock()
globalObjectAPI = objLayer
globalObjLayerMutex.Unlock()
configPath := rootPath + "/" + minioConfigFile
// Create a corrupted config file
@@ -203,12 +199,12 @@ func TestServerConfigMigrateV2toV28(t *testing.T) {
t.Fatal("Unexpected error: ", err)
}
if err := migrateConfigToMinioSys(); err != nil {
if err := migrateConfigToMinioSys(objLayer); err != nil {
t.Fatal("Unexpected error: ", err)
}
// Initialize server config and check again if everything is fine
if err := loadConfig(newObjectLayerFn()); err != nil {
if err := loadConfig(objLayer); err != nil {
t.Fatalf("Unable to initialize from updated config file %s", err)
}
+7 -6
View File
@@ -136,7 +136,8 @@ func checkServerConfig(ctx context.Context, objAPI ObjectLayer) error {
}
if _, err := objAPI.GetObjectInfo(ctx, minioMetaBucket, configFile); err != nil {
if isErrObjectNotFound(err) {
// Convert ObjectNotFound, Quorum errors into errConfigNotFound
if isErrObjectNotFound(err) || isInsufficientReadQuorum(err) {
return errConfigNotFound
}
logger.GetReqInfo(ctx).AppendTags("configFile", configFile)
@@ -162,7 +163,7 @@ func readConfig(ctx context.Context, objAPI ObjectLayer, configFile string) (*by
var buffer bytes.Buffer
// Read entire content by setting size to -1
if err := objAPI.GetObject(ctx, minioMetaBucket, configFile, 0, -1, &buffer, ""); err != nil {
// Ignore if err is ObjectNotFound or IncompleteBody when bucket is not configured with notification
// Convert ObjectNotFound, IncompleteBody and Quorum errors into errConfigNotFound
if isErrObjectNotFound(err) || isErrIncompleteBody(err) || isInsufficientReadQuorum(err) {
return nil, errConfigNotFound
}
@@ -202,9 +203,9 @@ func NewConfigSys() *ConfigSys {
}
// Migrates ${HOME}/.minio/config.json to '<export_path>/.minio.sys/config/config.json'
func migrateConfigToMinioSys() error {
func migrateConfigToMinioSys(objAPI ObjectLayer) error {
// Verify if backend already has the file.
if err := checkServerConfig(context.Background(), newObjectLayerFn()); err != errConfigNotFound {
if err := checkServerConfig(context.Background(), objAPI); err != errConfigNotFound {
return err
} // if errConfigNotFound proceed to migrate..
@@ -213,7 +214,7 @@ func migrateConfigToMinioSys() error {
return err
}
return saveServerConfig(newObjectLayerFn(), config)
return saveServerConfig(objAPI, config)
}
// Initialize and load config from remote etcd or local config directory
@@ -236,7 +237,7 @@ func initConfig(objAPI ObjectLayer) error {
}
// Migrates ${HOME}/.minio/config.json to '<export_path>/.minio.sys/config/config.json'
if err := migrateConfigToMinioSys(); err != nil {
if err := migrateConfigToMinioSys(objAPI); err != nil {
return err
}
}
+10 -10
View File
@@ -43,6 +43,8 @@ var (
errKMSNotConfigured = errors.New("KMS not configured for a server side encrypted object")
// Additional Minio errors for SSE-C requests.
errObjectTampered = errors.New("The requested object was modified and may be compromised")
// error returned when invalid encryption parameters are specified
errInvalidEncryptionParameters = errors.New("The encryption parameters are not applicable to this object")
)
const (
@@ -147,6 +149,7 @@ func rotateKey(oldKey []byte, newKey []byte, bucket, object string, metadata map
}
copy(extKey[:], newKey)
sealedKey = objectKey.Seal(extKey, sealedKey.IV, crypto.SSEC.String(), bucket, object)
crypto.SSEC.CreateMetadata(metadata, sealedKey)
return nil
}
}
@@ -713,28 +716,25 @@ func DecryptCopyObjectInfo(info *ObjectInfo, headers http.Header) (apiErr APIErr
// decryption succeeded.
//
// DecryptObjectInfo also returns whether the object is encrypted or not.
func DecryptObjectInfo(info *ObjectInfo, headers http.Header) (apiErr APIErrorCode, encrypted bool) {
func DecryptObjectInfo(info *ObjectInfo, headers http.Header) (encrypted bool, err error) {
// Directories are never encrypted.
if info.IsDir {
return ErrNone, false
return false, nil
}
// disallow X-Amz-Server-Side-Encryption header on HEAD and GET
if crypto.S3.IsRequested(headers) {
apiErr = ErrInvalidEncryptionParameters
err = errInvalidEncryptionParameters
return
}
if apiErr, encrypted = ErrNone, crypto.IsEncrypted(info.UserDefined); !encrypted && crypto.SSEC.IsRequested(headers) {
apiErr = ErrInvalidEncryptionParameters
if err, encrypted = nil, crypto.IsEncrypted(info.UserDefined); !encrypted && crypto.SSEC.IsRequested(headers) {
err = errInvalidEncryptionParameters
} else if encrypted {
if (crypto.SSEC.IsEncrypted(info.UserDefined) && !crypto.SSEC.IsRequested(headers)) ||
(crypto.S3.IsEncrypted(info.UserDefined) && crypto.SSEC.IsRequested(headers)) {
apiErr = ErrSSEEncryptedObject
err = errEncryptedObject
return
}
var err error
if info.Size, err = info.DecryptedSize(); err != nil {
apiErr = toAPIErrorCode(err)
}
info.Size, err = info.DecryptedSize()
}
return
}
+8 -8
View File
@@ -521,43 +521,43 @@ func TestDecryptRequest(t *testing.T) {
var decryptObjectInfoTests = []struct {
info ObjectInfo
headers http.Header
expErr APIErrorCode
expErr error
}{
{
info: ObjectInfo{Size: 100},
headers: http.Header{},
expErr: ErrNone,
expErr: nil,
},
{
info: ObjectInfo{Size: 100, UserDefined: map[string]string{crypto.SSESealAlgorithm: SSESealAlgorithmDareSha256}},
headers: http.Header{crypto.SSECAlgorithm: []string{crypto.SSEAlgorithmAES256}},
expErr: ErrNone,
expErr: nil,
},
{
info: ObjectInfo{Size: 0, UserDefined: map[string]string{crypto.SSESealAlgorithm: SSESealAlgorithmDareSha256}},
headers: http.Header{crypto.SSECAlgorithm: []string{crypto.SSEAlgorithmAES256}},
expErr: ErrNone,
expErr: nil,
},
{
info: ObjectInfo{Size: 100, UserDefined: map[string]string{crypto.SSECSealedKey: "EAAfAAAAAAD7v1hQq3PFRUHsItalxmrJqrOq6FwnbXNarxOOpb8jTWONPPKyM3Gfjkjyj6NCf+aB/VpHCLCTBA=="}},
headers: http.Header{},
expErr: ErrSSEEncryptedObject,
expErr: errEncryptedObject,
},
{
info: ObjectInfo{Size: 100, UserDefined: map[string]string{}},
headers: http.Header{crypto.SSECAlgorithm: []string{crypto.SSEAlgorithmAES256}},
expErr: ErrInvalidEncryptionParameters,
expErr: errInvalidEncryptionParameters,
},
{
info: ObjectInfo{Size: 31, UserDefined: map[string]string{crypto.SSESealAlgorithm: SSESealAlgorithmDareSha256}},
headers: http.Header{crypto.SSECAlgorithm: []string{crypto.SSEAlgorithmAES256}},
expErr: ErrObjectTampered,
expErr: errObjectTampered,
},
}
func TestDecryptObjectInfo(t *testing.T) {
for i, test := range decryptObjectInfoTests {
if err, encrypted := DecryptObjectInfo(&test.info, test.headers); err != test.expErr {
if encrypted, err := DecryptObjectInfo(&test.info, test.headers); err != test.expErr {
t.Errorf("Test %d: Decryption returned wrong error code: got %d , want %d", i, err, test.expErr)
} else if enc := crypto.IsEncrypted(test.info.UserDefined); encrypted && enc != encrypted {
t.Errorf("Test %d: Decryption thinks object is encrypted but it is not", i)
@@ -127,8 +127,8 @@ func (p *parallelReader) Read() ([][]byte, error) {
return nil, errXLReadQuorum
}
// ReadFile reads from readers, reconstructs data if needed and writes the data to the writer.
func (s ErasureStorage) ReadFile(ctx context.Context, writer io.Writer, readers []*bitrotReader, offset, length, totalLength int64) error {
// Decode reads from readers, reconstructs data if needed and writes the data to the writer.
func (e Erasure) Decode(ctx context.Context, writer io.Writer, readers []*bitrotReader, offset, length, totalLength int64) error {
if offset < 0 || length < 0 {
logger.LogIf(ctx, errInvalidArgument)
return errInvalidArgument
@@ -141,27 +141,27 @@ func (s ErasureStorage) ReadFile(ctx context.Context, writer io.Writer, readers
return nil
}
reader := newParallelReader(readers, s.dataBlocks, offset, totalLength, s.blockSize)
reader := newParallelReader(readers, e.dataBlocks, offset, totalLength, e.blockSize)
startBlock := offset / s.blockSize
endBlock := (offset + length) / s.blockSize
startBlock := offset / e.blockSize
endBlock := (offset + length) / e.blockSize
var bytesWritten int64
for block := startBlock; block <= endBlock; block++ {
var blockOffset, blockLength int64
switch {
case startBlock == endBlock:
blockOffset = offset % s.blockSize
blockOffset = offset % e.blockSize
blockLength = length
case block == startBlock:
blockOffset = offset % s.blockSize
blockLength = s.blockSize - blockOffset
blockOffset = offset % e.blockSize
blockLength = e.blockSize - blockOffset
case block == endBlock:
blockOffset = 0
blockLength = (offset + length) % s.blockSize
blockLength = (offset + length) % e.blockSize
default:
blockOffset = 0
blockLength = s.blockSize
blockLength = e.blockSize
}
if blockLength == 0 {
break
@@ -170,11 +170,11 @@ func (s ErasureStorage) ReadFile(ctx context.Context, writer io.Writer, readers
if err != nil {
return err
}
if err = s.ErasureDecodeDataBlocks(bufs); err != nil {
if err = e.DecodeDataBlocks(bufs); err != nil {
logger.LogIf(ctx, err)
return err
}
n, err := writeDataBlocks(ctx, writer, bufs, s.dataBlocks, blockOffset, blockLength)
n, err := writeDataBlocks(ctx, writer, bufs, e.dataBlocks, blockOffset, blockLength)
if err != nil {
return err
}
@@ -32,7 +32,7 @@ func (d badDisk) ReadFile(volume string, path string, offset int64, buf []byte,
return 0, errFaultyDisk
}
var erasureReadFileTests = []struct {
var erasureDecodeTests = []struct {
dataBlocks int
onDisks, offDisks int
blocksize, data int64
@@ -81,13 +81,13 @@ var erasureReadFileTests = []struct {
{dataBlocks: 8, onDisks: 12, offDisks: 4, blocksize: int64(blockSizeV1), data: int64(2 * blockSizeV1), offset: 11, length: int64(blockSizeV1) + 2*1024, algorithm: DefaultBitrotAlgorithm, shouldFail: false, shouldFailQuorum: false}, // 37
}
func TestErasureReadFile(t *testing.T) {
for i, test := range erasureReadFileTests {
func TestErasureDecode(t *testing.T) {
for i, test := range erasureDecodeTests {
setup, err := newErasureTestSetup(test.dataBlocks, test.onDisks-test.dataBlocks, test.blocksize)
if err != nil {
t.Fatalf("Test %d: failed to create test setup: %v", i, err)
}
storage, err := NewErasureStorage(context.Background(), test.dataBlocks, test.onDisks-test.dataBlocks, test.blocksize)
erasure, err := NewErasure(context.Background(), test.dataBlocks, test.onDisks-test.dataBlocks, test.blocksize)
if err != nil {
setup.Remove()
t.Fatalf("Test %d: failed to create ErasureStorage: %v", i, err)
@@ -108,7 +108,7 @@ func TestErasureReadFile(t *testing.T) {
for i, disk := range disks {
writers[i] = newBitrotWriter(disk, "testbucket", "object", writeAlgorithm)
}
n, err := storage.CreateFile(context.Background(), bytes.NewReader(data[:]), writers, buffer, storage.dataBlocks+1)
n, err := erasure.Encode(context.Background(), bytes.NewReader(data[:]), writers, buffer, erasure.dataBlocks+1)
if err != nil {
setup.Remove()
t.Fatalf("Test %d: failed to create erasure test file: %v", i, err)
@@ -129,12 +129,12 @@ func TestErasureReadFile(t *testing.T) {
if disk == OfflineDisk {
continue
}
endOffset := getErasureShardFileEndOffset(test.offset, test.length, test.data, test.blocksize, storage.dataBlocks)
endOffset := getErasureShardFileEndOffset(test.offset, test.length, test.data, test.blocksize, erasure.dataBlocks)
bitrotReaders[index] = newBitrotReader(disk, "testbucket", "object", writeAlgorithm, endOffset, writers[index].Sum())
}
writer := bytes.NewBuffer(nil)
err = storage.ReadFile(context.Background(), writer, bitrotReaders, test.offset, test.length, test.data)
err = erasure.Decode(context.Background(), writer, bitrotReaders, test.offset, test.length, test.data)
if err != nil && !test.shouldFail {
t.Errorf("Test %d: should pass but failed with: %v", i, err)
}
@@ -157,7 +157,7 @@ func TestErasureReadFile(t *testing.T) {
if disk == OfflineDisk {
continue
}
endOffset := getErasureShardFileEndOffset(test.offset, test.length, test.data, test.blocksize, storage.dataBlocks)
endOffset := getErasureShardFileEndOffset(test.offset, test.length, test.data, test.blocksize, erasure.dataBlocks)
bitrotReaders[index] = newBitrotReader(disk, "testbucket", "object", writeAlgorithm, endOffset, writers[index].Sum())
}
for j := range disks[:test.offDisks] {
@@ -167,7 +167,7 @@ func TestErasureReadFile(t *testing.T) {
bitrotReaders[0] = nil
}
writer.Reset()
err = storage.ReadFile(context.Background(), writer, bitrotReaders, test.offset, test.length, test.data)
err = erasure.Decode(context.Background(), writer, bitrotReaders, test.offset, test.length, test.data)
if err != nil && !test.shouldFailQuorum {
t.Errorf("Test %d: should pass but failed with: %v", i, err)
}
@@ -184,10 +184,10 @@ func TestErasureReadFile(t *testing.T) {
}
}
// Test erasureReadFile with random offset and lengths.
// Test erasureDecode with random offset and lengths.
// This test is t.Skip()ed as it a long time to run, hence should be run
// explicitly after commenting out t.Skip()
func TestErasureReadFileRandomOffsetLength(t *testing.T) {
func TestErasureDecodeRandomOffsetLength(t *testing.T) {
// Comment the following line to run this test.
t.SkipNow()
// Initialize environment needed for the test.
@@ -201,7 +201,7 @@ func TestErasureReadFileRandomOffsetLength(t *testing.T) {
}
defer setup.Remove()
disks := setup.disks
storage, err := NewErasureStorage(context.Background(), dataBlocks, parityBlocks, blockSize)
erasure, err := NewErasure(context.Background(), dataBlocks, parityBlocks, blockSize)
if err != nil {
t.Fatalf("failed to create ErasureStorage: %v", err)
}
@@ -226,7 +226,7 @@ func TestErasureReadFileRandomOffsetLength(t *testing.T) {
// Create a test file to read from.
buffer := make([]byte, blockSize, 2*blockSize)
n, err := storage.CreateFile(context.Background(), bytes.NewReader(data), writers, buffer, storage.dataBlocks+1)
n, err := erasure.Encode(context.Background(), bytes.NewReader(data), writers, buffer, erasure.dataBlocks+1)
if err != nil {
t.Fatal(err)
}
@@ -239,7 +239,7 @@ func TestErasureReadFileRandomOffsetLength(t *testing.T) {
buf := &bytes.Buffer{}
// Verify erasureReadFile() for random offsets and lengths.
// Verify erasure.Decode() for random offsets and lengths.
for i := 0; i < iterations; i++ {
offset := r.Int63n(length)
readLen := r.Int63n(length - offset)
@@ -252,10 +252,10 @@ func TestErasureReadFileRandomOffsetLength(t *testing.T) {
if disk == OfflineDisk {
continue
}
endOffset := getErasureShardFileEndOffset(offset, readLen, length, blockSize, storage.dataBlocks)
endOffset := getErasureShardFileEndOffset(offset, readLen, length, blockSize, erasure.dataBlocks)
bitrotReaders[index] = newBitrotReader(disk, "testbucket", "object", DefaultBitrotAlgorithm, endOffset, writers[index].Sum())
}
err = storage.ReadFile(context.Background(), buf, bitrotReaders, offset, readLen, length)
err = erasure.Decode(context.Background(), buf, bitrotReaders, offset, readLen, length)
if err != nil {
t.Fatal(err, offset, readLen)
}
@@ -269,14 +269,14 @@ func TestErasureReadFileRandomOffsetLength(t *testing.T) {
// Benchmarks
func benchmarkErasureRead(data, parity, dataDown, parityDown int, size int64, b *testing.B) {
func benchmarkErasureDecode(data, parity, dataDown, parityDown int, size int64, b *testing.B) {
setup, err := newErasureTestSetup(data, parity, blockSizeV1)
if err != nil {
b.Fatalf("failed to create test setup: %v", err)
}
defer setup.Remove()
disks := setup.disks
storage, err := NewErasureStorage(context.Background(), data, parity, blockSizeV1)
erasure, err := NewErasure(context.Background(), data, parity, blockSizeV1)
if err != nil {
b.Fatalf("failed to create ErasureStorage: %v", err)
}
@@ -291,7 +291,7 @@ func benchmarkErasureRead(data, parity, dataDown, parityDown int, size int64, b
content := make([]byte, size)
buffer := make([]byte, blockSizeV1, 2*blockSizeV1)
_, err = storage.CreateFile(context.Background(), bytes.NewReader(content), writers, buffer, storage.dataBlocks+1)
_, err = erasure.Encode(context.Background(), bytes.NewReader(content), writers, buffer, erasure.dataBlocks+1)
if err != nil {
b.Fatalf("failed to create erasure test file: %v", err)
}
@@ -312,62 +312,62 @@ func benchmarkErasureRead(data, parity, dataDown, parityDown int, size int64, b
if writers[index] == nil {
continue
}
endOffset := getErasureShardFileEndOffset(0, size, size, storage.blockSize, storage.dataBlocks)
endOffset := getErasureShardFileEndOffset(0, size, size, erasure.blockSize, erasure.dataBlocks)
bitrotReaders[index] = newBitrotReader(disk, "testbucket", "object", DefaultBitrotAlgorithm, endOffset, writers[index].Sum())
}
if err = storage.ReadFile(context.Background(), bytes.NewBuffer(content[:0]), bitrotReaders, 0, size, size); err != nil {
if err = erasure.Decode(context.Background(), bytes.NewBuffer(content[:0]), bitrotReaders, 0, size, size); err != nil {
panic(err)
}
}
}
func BenchmarkErasureReadQuick(b *testing.B) {
func BenchmarkErasureDecodeQuick(b *testing.B) {
const size = 12 * 1024 * 1024
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 1, 0, size, b) })
b.Run(" X0|X0 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 1, 1, size, b) })
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 1, 0, size, b) })
b.Run(" X0|X0 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 1, 1, size, b) })
}
func BenchmarkErasureRead_4_64KB(b *testing.B) {
func BenchmarkErasureDecode_4_64KB(b *testing.B) {
const size = 64 * 1024
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 1, 0, size, b) })
b.Run(" X0|X0 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 1, 1, size, b) })
b.Run(" 00|XX ", func(b *testing.B) { benchmarkErasureRead(2, 2, 0, 2, size, b) })
b.Run(" XX|00 ", func(b *testing.B) { benchmarkErasureRead(2, 2, 2, 0, size, b) })
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 1, 0, size, b) })
b.Run(" X0|X0 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 1, 1, size, b) })
b.Run(" 00|XX ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 0, 2, size, b) })
b.Run(" XX|00 ", func(b *testing.B) { benchmarkErasureDecode(2, 2, 2, 0, size, b) })
}
func BenchmarkErasureRead_8_20MB(b *testing.B) {
func BenchmarkErasureDecode_8_20MB(b *testing.B) {
const size = 20 * 1024 * 1024
b.Run(" 0000|0000 ", func(b *testing.B) { benchmarkErasureRead(4, 4, 0, 0, size, b) })
b.Run(" 0000|X000 ", func(b *testing.B) { benchmarkErasureRead(4, 4, 0, 1, size, b) })
b.Run(" X000|0000 ", func(b *testing.B) { benchmarkErasureRead(4, 4, 1, 0, size, b) })
b.Run(" X000|X000 ", func(b *testing.B) { benchmarkErasureRead(4, 4, 1, 1, size, b) })
b.Run(" 0000|XXXX ", func(b *testing.B) { benchmarkErasureRead(4, 4, 0, 4, size, b) })
b.Run(" XX00|XX00 ", func(b *testing.B) { benchmarkErasureRead(4, 4, 2, 2, size, b) })
b.Run(" XXXX|0000 ", func(b *testing.B) { benchmarkErasureRead(4, 4, 4, 0, size, b) })
b.Run(" 0000|0000 ", func(b *testing.B) { benchmarkErasureDecode(4, 4, 0, 0, size, b) })
b.Run(" 0000|X000 ", func(b *testing.B) { benchmarkErasureDecode(4, 4, 0, 1, size, b) })
b.Run(" X000|0000 ", func(b *testing.B) { benchmarkErasureDecode(4, 4, 1, 0, size, b) })
b.Run(" X000|X000 ", func(b *testing.B) { benchmarkErasureDecode(4, 4, 1, 1, size, b) })
b.Run(" 0000|XXXX ", func(b *testing.B) { benchmarkErasureDecode(4, 4, 0, 4, size, b) })
b.Run(" XX00|XX00 ", func(b *testing.B) { benchmarkErasureDecode(4, 4, 2, 2, size, b) })
b.Run(" XXXX|0000 ", func(b *testing.B) { benchmarkErasureDecode(4, 4, 4, 0, size, b) })
}
func BenchmarkErasureRead_12_30MB(b *testing.B) {
func BenchmarkErasureDecode_12_30MB(b *testing.B) {
const size = 30 * 1024 * 1024
b.Run(" 000000|000000 ", func(b *testing.B) { benchmarkErasureRead(6, 6, 0, 0, size, b) })
b.Run(" 000000|X00000 ", func(b *testing.B) { benchmarkErasureRead(6, 6, 0, 1, size, b) })
b.Run(" X00000|000000 ", func(b *testing.B) { benchmarkErasureRead(6, 6, 1, 0, size, b) })
b.Run(" X00000|X00000 ", func(b *testing.B) { benchmarkErasureRead(6, 6, 1, 1, size, b) })
b.Run(" 000000|XXXXXX ", func(b *testing.B) { benchmarkErasureRead(6, 6, 0, 6, size, b) })
b.Run(" XXX000|XXX000 ", func(b *testing.B) { benchmarkErasureRead(6, 6, 3, 3, size, b) })
b.Run(" XXXXXX|000000 ", func(b *testing.B) { benchmarkErasureRead(6, 6, 6, 0, size, b) })
b.Run(" 000000|000000 ", func(b *testing.B) { benchmarkErasureDecode(6, 6, 0, 0, size, b) })
b.Run(" 000000|X00000 ", func(b *testing.B) { benchmarkErasureDecode(6, 6, 0, 1, size, b) })
b.Run(" X00000|000000 ", func(b *testing.B) { benchmarkErasureDecode(6, 6, 1, 0, size, b) })
b.Run(" X00000|X00000 ", func(b *testing.B) { benchmarkErasureDecode(6, 6, 1, 1, size, b) })
b.Run(" 000000|XXXXXX ", func(b *testing.B) { benchmarkErasureDecode(6, 6, 0, 6, size, b) })
b.Run(" XXX000|XXX000 ", func(b *testing.B) { benchmarkErasureDecode(6, 6, 3, 3, size, b) })
b.Run(" XXXXXX|000000 ", func(b *testing.B) { benchmarkErasureDecode(6, 6, 6, 0, size, b) })
}
func BenchmarkErasureRead_16_40MB(b *testing.B) {
func BenchmarkErasureDecode_16_40MB(b *testing.B) {
const size = 40 * 1024 * 1024
b.Run(" 00000000|00000000 ", func(b *testing.B) { benchmarkErasureRead(8, 8, 0, 0, size, b) })
b.Run(" 00000000|X0000000 ", func(b *testing.B) { benchmarkErasureRead(8, 8, 0, 1, size, b) })
b.Run(" X0000000|00000000 ", func(b *testing.B) { benchmarkErasureRead(8, 8, 1, 0, size, b) })
b.Run(" X0000000|X0000000 ", func(b *testing.B) { benchmarkErasureRead(8, 8, 1, 1, size, b) })
b.Run(" 00000000|XXXXXXXX ", func(b *testing.B) { benchmarkErasureRead(8, 8, 0, 8, size, b) })
b.Run(" XXXX0000|XXXX0000 ", func(b *testing.B) { benchmarkErasureRead(8, 8, 4, 4, size, b) })
b.Run(" XXXXXXXX|00000000 ", func(b *testing.B) { benchmarkErasureRead(8, 8, 8, 0, size, b) })
b.Run(" 00000000|00000000 ", func(b *testing.B) { benchmarkErasureDecode(8, 8, 0, 0, size, b) })
b.Run(" 00000000|X0000000 ", func(b *testing.B) { benchmarkErasureDecode(8, 8, 0, 1, size, b) })
b.Run(" X0000000|00000000 ", func(b *testing.B) { benchmarkErasureDecode(8, 8, 1, 0, size, b) })
b.Run(" X0000000|X0000000 ", func(b *testing.B) { benchmarkErasureDecode(8, 8, 1, 1, size, b) })
b.Run(" 00000000|XXXXXXXX ", func(b *testing.B) { benchmarkErasureDecode(8, 8, 0, 8, size, b) })
b.Run(" XXXX0000|XXXX0000 ", func(b *testing.B) { benchmarkErasureDecode(8, 8, 4, 4, size, b) })
b.Run(" XXXXXXXX|00000000 ", func(b *testing.B) { benchmarkErasureDecode(8, 8, 8, 0, size, b) })
}
@@ -69,8 +69,8 @@ func (p *parallelWriter) Append(ctx context.Context, blocks [][]byte) error {
return reduceWriteQuorumErrs(ctx, p.errs, objectOpIgnoredErrs, p.writeQuorum)
}
// CreateFile reads from the reader, erasure-encodes the data and writes to the writers.
func (s *ErasureStorage) CreateFile(ctx context.Context, src io.Reader, writers []*bitrotWriter, buf []byte, quorum int) (total int64, err error) {
// Encode reads from the reader, erasure-encodes the data and writes to the writers.
func (e *Erasure) Encode(ctx context.Context, src io.Reader, writers []*bitrotWriter, buf []byte, quorum int) (total int64, err error) {
writer := &parallelWriter{
writers: writers,
writeQuorum: quorum,
@@ -90,7 +90,7 @@ func (s *ErasureStorage) CreateFile(ctx context.Context, src io.Reader, writers
break
}
// We take care of the situation where if n == 0 and total == 0 by creating empty data and parity files.
blocks, err = s.ErasureEncode(ctx, buf[:n])
blocks, err = e.EncodeData(ctx, buf[:n])
if err != nil {
logger.LogIf(ctx, err)
return 0, err
@@ -34,7 +34,7 @@ func (a badDisk) AppendFile(volume string, path string, buf []byte) error {
const oneMiByte = 1 * humanize.MiByte
var erasureCreateFileTests = []struct {
var erasureEncodeTests = []struct {
dataBlocks int
onDisks, offDisks int
blocksize, data int64
@@ -64,14 +64,14 @@ var erasureCreateFileTests = []struct {
{dataBlocks: 10, onDisks: 16, offDisks: 8, blocksize: int64(blockSizeV1), data: oneMiByte, offset: 0, algorithm: DefaultBitrotAlgorithm, shouldFail: false, shouldFailQuorum: true}, // 19
}
func TestErasureCreateFile(t *testing.T) {
for i, test := range erasureCreateFileTests {
func TestErasureEncode(t *testing.T) {
for i, test := range erasureEncodeTests {
setup, err := newErasureTestSetup(test.dataBlocks, test.onDisks-test.dataBlocks, test.blocksize)
if err != nil {
t.Fatalf("Test %d: failed to create test setup: %v", i, err)
}
disks := setup.disks
storage, err := NewErasureStorage(context.Background(), test.dataBlocks, test.onDisks-test.dataBlocks, test.blocksize)
erasure, err := NewErasure(context.Background(), test.dataBlocks, test.onDisks-test.dataBlocks, test.blocksize)
if err != nil {
setup.Remove()
t.Fatalf("Test %d: failed to create ErasureStorage: %v", i, err)
@@ -90,7 +90,7 @@ func TestErasureCreateFile(t *testing.T) {
}
writers[i] = newBitrotWriter(disk, "testbucket", "object", test.algorithm)
}
n, err := storage.CreateFile(context.Background(), bytes.NewReader(data[test.offset:]), writers, buffer, storage.dataBlocks+1)
n, err := erasure.Encode(context.Background(), bytes.NewReader(data[test.offset:]), writers, buffer, erasure.dataBlocks+1)
if err != nil && !test.shouldFail {
t.Errorf("Test %d: should pass but failed with: %v", i, err)
}
@@ -119,7 +119,7 @@ func TestErasureCreateFile(t *testing.T) {
if test.offDisks > 0 {
writers[0] = nil
}
n, err = storage.CreateFile(context.Background(), bytes.NewReader(data[test.offset:]), writers, buffer, storage.dataBlocks+1)
n, err = erasure.Encode(context.Background(), bytes.NewReader(data[test.offset:]), writers, buffer, erasure.dataBlocks+1)
if err != nil && !test.shouldFailQuorum {
t.Errorf("Test %d: should pass but failed with: %v", i, err)
}
@@ -138,13 +138,13 @@ func TestErasureCreateFile(t *testing.T) {
// Benchmarks
func benchmarkErasureWrite(data, parity, dataDown, parityDown int, size int64, b *testing.B) {
func benchmarkErasureEncode(data, parity, dataDown, parityDown int, size int64, b *testing.B) {
setup, err := newErasureTestSetup(data, parity, blockSizeV1)
if err != nil {
b.Fatalf("failed to create test setup: %v", err)
}
defer setup.Remove()
storage, err := NewErasureStorage(context.Background(), data, parity, blockSizeV1)
erasure, err := NewErasure(context.Background(), data, parity, blockSizeV1)
if err != nil {
b.Fatalf("failed to create ErasureStorage: %v", err)
}
@@ -170,50 +170,50 @@ func benchmarkErasureWrite(data, parity, dataDown, parityDown int, size int64, b
}
writers[i] = newBitrotWriter(disk, "testbucket", "object", DefaultBitrotAlgorithm)
}
_, err := storage.CreateFile(context.Background(), bytes.NewReader(content), writers, buffer, storage.dataBlocks+1)
_, err := erasure.Encode(context.Background(), bytes.NewReader(content), writers, buffer, erasure.dataBlocks+1)
if err != nil {
panic(err)
}
}
}
func BenchmarkErasureWriteQuick(b *testing.B) {
func BenchmarkErasureEncodeQuick(b *testing.B) {
const size = 12 * 1024 * 1024
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureWrite(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureWrite(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureWrite(2, 2, 1, 0, size, b) })
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureEncode(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureEncode(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureEncode(2, 2, 1, 0, size, b) })
}
func BenchmarkErasureWrite_4_64KB(b *testing.B) {
func BenchmarkErasureEncode_4_64KB(b *testing.B) {
const size = 64 * 1024
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureWrite(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureWrite(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureWrite(2, 2, 1, 0, size, b) })
b.Run(" 00|00 ", func(b *testing.B) { benchmarkErasureEncode(2, 2, 0, 0, size, b) })
b.Run(" 00|X0 ", func(b *testing.B) { benchmarkErasureEncode(2, 2, 0, 1, size, b) })
b.Run(" X0|00 ", func(b *testing.B) { benchmarkErasureEncode(2, 2, 1, 0, size, b) })
}
func BenchmarkErasureWrite_8_20MB(b *testing.B) {
func BenchmarkErasureEncode_8_20MB(b *testing.B) {
const size = 20 * 1024 * 1024
b.Run(" 0000|0000 ", func(b *testing.B) { benchmarkErasureWrite(4, 4, 0, 0, size, b) })
b.Run(" 0000|X000 ", func(b *testing.B) { benchmarkErasureWrite(4, 4, 0, 1, size, b) })
b.Run(" X000|0000 ", func(b *testing.B) { benchmarkErasureWrite(4, 4, 1, 0, size, b) })
b.Run(" 0000|XXX0 ", func(b *testing.B) { benchmarkErasureWrite(4, 4, 0, 3, size, b) })
b.Run(" XXX0|0000 ", func(b *testing.B) { benchmarkErasureWrite(4, 4, 3, 0, size, b) })
b.Run(" 0000|0000 ", func(b *testing.B) { benchmarkErasureEncode(4, 4, 0, 0, size, b) })
b.Run(" 0000|X000 ", func(b *testing.B) { benchmarkErasureEncode(4, 4, 0, 1, size, b) })
b.Run(" X000|0000 ", func(b *testing.B) { benchmarkErasureEncode(4, 4, 1, 0, size, b) })
b.Run(" 0000|XXX0 ", func(b *testing.B) { benchmarkErasureEncode(4, 4, 0, 3, size, b) })
b.Run(" XXX0|0000 ", func(b *testing.B) { benchmarkErasureEncode(4, 4, 3, 0, size, b) })
}
func BenchmarkErasureWrite_12_30MB(b *testing.B) {
func BenchmarkErasureEncode_12_30MB(b *testing.B) {
const size = 30 * 1024 * 1024
b.Run(" 000000|000000 ", func(b *testing.B) { benchmarkErasureWrite(6, 6, 0, 0, size, b) })
b.Run(" 000000|X00000 ", func(b *testing.B) { benchmarkErasureWrite(6, 6, 0, 1, size, b) })
b.Run(" X00000|000000 ", func(b *testing.B) { benchmarkErasureWrite(6, 6, 1, 0, size, b) })
b.Run(" 000000|XXXXX0 ", func(b *testing.B) { benchmarkErasureWrite(6, 6, 0, 5, size, b) })
b.Run(" XXXXX0|000000 ", func(b *testing.B) { benchmarkErasureWrite(6, 6, 5, 0, size, b) })
b.Run(" 000000|000000 ", func(b *testing.B) { benchmarkErasureEncode(6, 6, 0, 0, size, b) })
b.Run(" 000000|X00000 ", func(b *testing.B) { benchmarkErasureEncode(6, 6, 0, 1, size, b) })
b.Run(" X00000|000000 ", func(b *testing.B) { benchmarkErasureEncode(6, 6, 1, 0, size, b) })
b.Run(" 000000|XXXXX0 ", func(b *testing.B) { benchmarkErasureEncode(6, 6, 0, 5, size, b) })
b.Run(" XXXXX0|000000 ", func(b *testing.B) { benchmarkErasureEncode(6, 6, 5, 0, size, b) })
}
func BenchmarkErasureWrite_16_40MB(b *testing.B) {
func BenchmarkErasureEncode_16_40MB(b *testing.B) {
const size = 40 * 1024 * 1024
b.Run(" 00000000|00000000 ", func(b *testing.B) { benchmarkErasureWrite(8, 8, 0, 0, size, b) })
b.Run(" 00000000|X0000000 ", func(b *testing.B) { benchmarkErasureWrite(8, 8, 0, 1, size, b) })
b.Run(" X0000000|00000000 ", func(b *testing.B) { benchmarkErasureWrite(8, 8, 1, 0, size, b) })
b.Run(" 00000000|XXXXXXX0 ", func(b *testing.B) { benchmarkErasureWrite(8, 8, 0, 7, size, b) })
b.Run(" XXXXXXX0|00000000 ", func(b *testing.B) { benchmarkErasureWrite(8, 8, 7, 0, size, b) })
b.Run(" 00000000|00000000 ", func(b *testing.B) { benchmarkErasureEncode(8, 8, 0, 0, size, b) })
b.Run(" 00000000|X0000000 ", func(b *testing.B) { benchmarkErasureEncode(8, 8, 0, 1, size, b) })
b.Run(" X0000000|00000000 ", func(b *testing.B) { benchmarkErasureEncode(8, 8, 1, 0, size, b) })
b.Run(" 00000000|XXXXXXX0 ", func(b *testing.B) { benchmarkErasureEncode(8, 8, 0, 7, size, b) })
b.Run(" XXXXXXX0|00000000 ", func(b *testing.B) { benchmarkErasureEncode(8, 8, 7, 0, size, b) })
}
@@ -23,20 +23,20 @@ import (
"github.com/minio/minio/cmd/logger"
)
// HealFile heals the shard files on non-nil writers. Note that the quorum passed is 1
// Heal heals the shard files on non-nil writers. Note that the quorum passed is 1
// as healing should continue even if it has been successful healing only one shard file.
func (s ErasureStorage) HealFile(ctx context.Context, readers []*bitrotReader, writers []*bitrotWriter, size int64) error {
func (e Erasure) Heal(ctx context.Context, readers []*bitrotReader, writers []*bitrotWriter, size int64) error {
r, w := io.Pipe()
go func() {
if err := s.ReadFile(ctx, w, readers, 0, size, size); err != nil {
if err := e.Decode(ctx, w, readers, 0, size, size); err != nil {
w.CloseWithError(err)
return
}
w.Close()
}()
buf := make([]byte, s.blockSize)
buf := make([]byte, e.blockSize)
// quorum is 1 because CreateFile should continue writing as long as we are writing to even 1 disk.
n, err := s.CreateFile(ctx, r, writers, buf, 1)
n, err := e.Encode(ctx, r, writers, buf, 1)
if err != nil {
return err
}
@@ -24,7 +24,7 @@ import (
"testing"
)
var erasureHealFileTests = []struct {
var erasureHealTests = []struct {
dataBlocks, disks int
// number of offline disks is also number of staleDisks for
@@ -60,8 +60,8 @@ var erasureHealFileTests = []struct {
{dataBlocks: 2, disks: 4, offDisks: 1, badDisks: 0, badStaleDisks: 0, blocksize: int64(blockSizeV1), size: oneMiByte * 64, algorithm: SHA256, shouldFail: false}, // 19
}
func TestErasureHealFile(t *testing.T) {
for i, test := range erasureHealFileTests {
func TestErasureHeal(t *testing.T) {
for i, test := range erasureHealTests {
if test.offDisks < test.badStaleDisks {
// test case sanity check
t.Fatalf("Test %d: Bad test case - number of stale disks cannot be less than number of badstale disks", i)
@@ -73,7 +73,7 @@ func TestErasureHealFile(t *testing.T) {
t.Fatalf("Test %d: failed to setup XL environment: %v", i, err)
}
disks := setup.disks
storage, err := NewErasureStorage(context.Background(), test.dataBlocks, test.disks-test.dataBlocks, test.blocksize)
erasure, err := NewErasure(context.Background(), test.dataBlocks, test.disks-test.dataBlocks, test.blocksize)
if err != nil {
setup.Remove()
t.Fatalf("Test %d: failed to create ErasureStorage: %v", i, err)
@@ -88,7 +88,7 @@ func TestErasureHealFile(t *testing.T) {
for i, disk := range disks {
writers[i] = newBitrotWriter(disk, "testbucket", "testobject", test.algorithm)
}
_, err = storage.CreateFile(context.Background(), bytes.NewReader(data), writers, buffer, storage.dataBlocks+1)
_, err = erasure.Encode(context.Background(), bytes.NewReader(data), writers, buffer, erasure.dataBlocks+1)
if err != nil {
setup.Remove()
t.Fatalf("Test %d: failed to create random test data: %v", i, err)
@@ -96,7 +96,7 @@ func TestErasureHealFile(t *testing.T) {
readers := make([]*bitrotReader, len(disks))
for i, disk := range disks {
shardFilesize := getErasureShardFileSize(test.blocksize, test.size, storage.dataBlocks)
shardFilesize := getErasureShardFileSize(test.blocksize, test.size, erasure.dataBlocks)
readers[i] = newBitrotReader(disk, "testbucket", "testobject", test.algorithm, shardFilesize, writers[i].Sum())
}
@@ -126,7 +126,7 @@ func TestErasureHealFile(t *testing.T) {
}
// test case setup is complete - now call Healfile()
err = storage.HealFile(context.Background(), readers, staleWriters, test.size)
err = erasure.Heal(context.Background(), readers, staleWriters, test.size)
if err != nil && !test.shouldFail {
t.Errorf("Test %d: should pass but it failed with: %v", i, err)
}
+20 -23
View File
@@ -23,26 +23,23 @@ import (
"github.com/minio/minio/cmd/logger"
)
// OfflineDisk represents an unavailable disk.
var OfflineDisk StorageAPI // zero value is nil
// ErasureStorage - erasure encoding details.
type ErasureStorage struct {
erasure reedsolomon.Encoder
// Erasure - erasure encoding details.
type Erasure struct {
encoder reedsolomon.Encoder
dataBlocks, parityBlocks int
blockSize int64
}
// NewErasureStorage creates a new ErasureStorage.
func NewErasureStorage(ctx context.Context, dataBlocks, parityBlocks int, blockSize int64) (s ErasureStorage, err error) {
// NewErasure creates a new ErasureStorage.
func NewErasure(ctx context.Context, dataBlocks, parityBlocks int, blockSize int64) (e Erasure, err error) {
shardsize := int(ceilFrac(blockSize, int64(dataBlocks)))
erasure, err := reedsolomon.New(dataBlocks, parityBlocks, reedsolomon.WithAutoGoroutines(shardsize))
if err != nil {
logger.LogIf(ctx, err)
return s, err
return e, err
}
s = ErasureStorage{
erasure: erasure,
e = Erasure{
encoder: erasure,
dataBlocks: dataBlocks,
parityBlocks: parityBlocks,
blockSize: blockSize,
@@ -50,30 +47,30 @@ func NewErasureStorage(ctx context.Context, dataBlocks, parityBlocks int, blockS
return
}
// ErasureEncode encodes the given data and returns the erasure-coded data.
// EncodeData encodes the given data and returns the erasure-coded data.
// It returns an error if the erasure coding failed.
func (s *ErasureStorage) ErasureEncode(ctx context.Context, data []byte) ([][]byte, error) {
func (e *Erasure) EncodeData(ctx context.Context, data []byte) ([][]byte, error) {
if len(data) == 0 {
return make([][]byte, s.dataBlocks+s.parityBlocks), nil
return make([][]byte, e.dataBlocks+e.parityBlocks), nil
}
encoded, err := s.erasure.Split(data)
encoded, err := e.encoder.Split(data)
if err != nil {
logger.LogIf(ctx, err)
return nil, err
}
if err = s.erasure.Encode(encoded); err != nil {
if err = e.encoder.Encode(encoded); err != nil {
logger.LogIf(ctx, err)
return nil, err
}
return encoded, nil
}
// ErasureDecodeDataBlocks decodes the given erasure-coded data.
// DecodeDataBlocks decodes the given erasure-coded data.
// It only decodes the data blocks but does not verify them.
// It returns an error if the decoding failed.
func (s *ErasureStorage) ErasureDecodeDataBlocks(data [][]byte) error {
func (e *Erasure) DecodeDataBlocks(data [][]byte) error {
needsReconstruction := false
for _, b := range data[:s.dataBlocks] {
for _, b := range data[:e.dataBlocks] {
if b == nil {
needsReconstruction = true
break
@@ -82,16 +79,16 @@ func (s *ErasureStorage) ErasureDecodeDataBlocks(data [][]byte) error {
if !needsReconstruction {
return nil
}
if err := s.erasure.ReconstructData(data); err != nil {
if err := e.encoder.ReconstructData(data); err != nil {
return err
}
return nil
}
// ErasureDecodeDataAndParityBlocks decodes the given erasure-coded data and verifies it.
// DecodeDataAndParityBlocks decodes the given erasure-coded data and verifies it.
// It returns an error if the decoding failed.
func (s *ErasureStorage) ErasureDecodeDataAndParityBlocks(ctx context.Context, data [][]byte) error {
if err := s.erasure.Reconstruct(data); err != nil {
func (e *Erasure) DecodeDataAndParityBlocks(ctx context.Context, data [][]byte) error {
if err := e.encoder.Reconstruct(data); err != nil {
logger.LogIf(ctx, err)
return err
}
+8 -8
View File
@@ -25,7 +25,7 @@ import (
"testing"
)
var erasureDecodeTests = []struct {
var erasureEncodeDecodeTests = []struct {
dataBlocks, parityBlocks int
missingData, missingParity int
reconstructParity bool
@@ -43,20 +43,20 @@ var erasureDecodeTests = []struct {
{dataBlocks: 8, parityBlocks: 4, missingData: 2, missingParity: 2, reconstructParity: false, shouldFail: false},
}
func TestErasureDecode(t *testing.T) {
func TestErasureEncodeDecode(t *testing.T) {
data := make([]byte, 256)
if _, err := io.ReadFull(rand.Reader, data); err != nil {
t.Fatalf("Failed to read random data: %v", err)
}
for i, test := range erasureDecodeTests {
for i, test := range erasureEncodeDecodeTests {
buffer := make([]byte, len(data), 2*len(data))
copy(buffer, data)
storage, err := NewErasureStorage(context.Background(), test.dataBlocks, test.parityBlocks, blockSizeV1)
erasure, err := NewErasure(context.Background(), test.dataBlocks, test.parityBlocks, blockSizeV1)
if err != nil {
t.Fatalf("Test %d: failed to create erasure storage: %v", i, err)
t.Fatalf("Test %d: failed to create erasure: %v", i, err)
}
encoded, err := storage.ErasureEncode(context.Background(), buffer)
encoded, err := erasure.EncodeData(context.Background(), buffer)
if err != nil {
t.Fatalf("Test %d: failed to encode data: %v", i, err)
}
@@ -69,9 +69,9 @@ func TestErasureDecode(t *testing.T) {
}
if test.reconstructParity {
err = storage.ErasureDecodeDataAndParityBlocks(context.Background(), encoded)
err = erasure.DecodeDataAndParityBlocks(context.Background(), encoded)
} else {
err = storage.ErasureDecodeDataBlocks(encoded)
err = erasure.DecodeDataBlocks(encoded)
}
if err == nil && test.shouldFail {
+3 -3
View File
@@ -261,7 +261,7 @@ func (fs *FSObjects) StorageInfo(ctx context.Context) StorageInfo {
storageInfo := StorageInfo{
Used: used,
}
storageInfo.Backend.Type = FS
storageInfo.Backend.Type = BackendFS
return storageInfo
}
@@ -456,7 +456,7 @@ func (fs *FSObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBu
// Save objects' metadata in `fs.json`.
fsMeta := newFSMetaV1()
if _, err = fsMeta.ReadFrom(ctx, wlk); err != nil {
if _, err = fsMeta.ReadFrom(ctx, wlk); err != nil && err != io.EOF {
return oi, toObjectErr(err, srcBucket, srcObject)
}
@@ -660,7 +660,7 @@ func (fs *FSObjects) getObjectInfo(ctx context.Context, bucket, object string) (
// Read from fs metadata only if it exists.
_, rerr := fsMeta.ReadFrom(ctx, rlk.LockedFile)
fs.rwPool.Close(fsMetaPath)
if rerr != nil {
if rerr != nil && rerr != io.EOF {
return oi, rerr
}
}
+9
View File
@@ -189,6 +189,15 @@ func extractReqParams(r *http.Request) map[string]string {
}
}
// Extract response elements to be sent with event notifiation.
func extractRespElements(w http.ResponseWriter) map[string]string {
return map[string]string{
"content-length": w.Header().Get("Content-Length"),
// Add more fields here.
}
}
// Trims away `aws-chunked` from the content-encoding header if present.
// Streaming signature clients can have custom content-encoding such as
// `aws-chunked,gzip` here we need to only save `gzip`.
+35 -6
View File
@@ -19,6 +19,7 @@ package cmd
import (
"fmt"
"net/http"
"strings"
"time"
"github.com/prometheus/client_golang/prometheus"
@@ -26,22 +27,50 @@ import (
"go.uber.org/atomic"
)
func getRequestResource(r *http.Request) string {
if r == nil {
// http.Request is nil when non-HTTP data (like TLS record) is read/written.
return ""
}
if globalDomainName != "" {
host := r.Header.Get("Host")
if strings.HasSuffix(host, "."+globalDomainName) {
return "/" + strings.TrimSuffix(host, "."+globalDomainName) + r.URL.Path
}
}
return r.URL.Path
}
// ConnStats - Network statistics
// Count total input/output transferred bytes during
// the server's life.
type ConnStats struct {
totalInputBytes atomic.Uint64
totalOutputBytes atomic.Uint64
totalInputBytes atomic.Uint64
totalOutputBytes atomic.Uint64
totalRPCInputBytes atomic.Uint64
totalRPCOutputBytes atomic.Uint64
}
// Increase total input bytes
func (s *ConnStats) incInputBytes(n int) {
s.totalInputBytes.Add(uint64(n))
func (s *ConnStats) incInputBytes(r *http.Request, n int) {
resource := getRequestResource(r)
if resource == minioReservedBucketPath || strings.HasPrefix(resource, minioReservedBucketPath+"/") {
s.totalRPCInputBytes.Add(uint64(n))
} else {
s.totalInputBytes.Add(uint64(n))
}
}
// Increase total output bytes
func (s *ConnStats) incOutputBytes(n int) {
s.totalOutputBytes.Add(uint64(n))
func (s *ConnStats) incOutputBytes(r *http.Request, n int) {
resource := getRequestResource(r)
if resource == minioReservedBucketPath || strings.HasPrefix(resource, minioReservedBucketPath+"/") {
s.totalRPCOutputBytes.Add(uint64(n))
} else {
s.totalOutputBytes.Add(uint64(n))
}
}
// Return total input bytes
+23 -15
View File
@@ -19,17 +19,28 @@ package http
import (
"bufio"
"net"
"net/http"
"time"
)
// BufConn - is a generic stream-oriented network connection supporting buffered reader and read/write timeout.
type BufConn struct {
QuirkConn
bufReader *bufio.Reader // buffered reader wraps reader in net.Conn.
readTimeout time.Duration // sets the read timeout in the connection.
writeTimeout time.Duration // sets the write timeout in the connection.
updateBytesReadFunc func(int) // function to be called to update bytes read.
updateBytesWrittenFunc func(int) // function to be called to update bytes written.
bufReader *bufio.Reader // buffered reader wraps reader in net.Conn.
readTimeout time.Duration // sets the read timeout in the connection.
writeTimeout time.Duration // sets the write timeout in the connection.
request *http.Request // HTTP request information.
updateBytesReadFunc func(*http.Request, int) // function to be called to update bytes read.
updateBytesWrittenFunc func(*http.Request, int) // function to be called to update bytes written.
}
func (c *BufConn) setRequest(request *http.Request) {
c.request = request
}
func (c *BufConn) setUpdateFuncs(updateBytesReadFunc, updateBytesWrittenFunc func(*http.Request, int)) {
c.updateBytesReadFunc = updateBytesReadFunc
c.updateBytesWrittenFunc = updateBytesWrittenFunc
}
// Sets read timeout
@@ -70,7 +81,7 @@ func (c *BufConn) Read(b []byte) (n int, err error) {
c.setReadTimeout()
n, err = c.bufReader.Read(b)
if err == nil && c.updateBytesReadFunc != nil {
c.updateBytesReadFunc(n)
c.updateBytesReadFunc(c.request, n)
}
return n, err
@@ -81,21 +92,18 @@ func (c *BufConn) Write(b []byte) (n int, err error) {
c.setWriteTimeout()
n, err = c.Conn.Write(b)
if err == nil && c.updateBytesWrittenFunc != nil {
c.updateBytesWrittenFunc(n)
c.updateBytesWrittenFunc(c.request, n)
}
return n, err
}
// newBufConn - creates a new connection object wrapping net.Conn.
func newBufConn(c net.Conn, readTimeout, writeTimeout time.Duration,
updateBytesReadFunc, updateBytesWrittenFunc func(int)) *BufConn {
func newBufConn(c net.Conn, readTimeout, writeTimeout time.Duration) *BufConn {
return &BufConn{
QuirkConn: QuirkConn{Conn: c},
bufReader: bufio.NewReader(c),
readTimeout: readTimeout,
writeTimeout: writeTimeout,
updateBytesReadFunc: updateBytesReadFunc,
updateBytesWrittenFunc: updateBytesWrittenFunc,
QuirkConn: QuirkConn{Conn: c},
bufReader: bufio.NewReader(c),
readTimeout: readTimeout,
writeTimeout: writeTimeout,
}
}
+1 -1
View File
@@ -48,7 +48,7 @@ func TestBuffConnReadTimeout(t *testing.T) {
if terr != nil {
t.Fatalf("failed to accept new connection. %v", terr)
}
bufconn := newBufConn(tcpConn, 1*time.Second, 1*time.Second, nil, nil)
bufconn := newBufConn(tcpConn, 1*time.Second, 1*time.Second)
defer bufconn.Close()
// Read a line
+6 -1
View File
@@ -18,7 +18,11 @@
package http
import "github.com/valyala/tcplisten"
import (
"net"
"github.com/valyala/tcplisten"
)
var cfg = &tcplisten.Config{
DeferAccept: true,
@@ -27,3 +31,4 @@ var cfg = &tcplisten.Config{
// Unix listener with special TCP options.
var listen = cfg.NewListener
var fallbackListen = net.Listen
+1
View File
@@ -22,3 +22,4 @@ import "net"
// Windows, plan9 specific listener.
var listen = net.Listen
var fallbackListen = net.Listen
+103 -12
View File
@@ -23,6 +23,7 @@ import (
"io"
"net"
"net/http"
"net/url"
"os"
"strings"
"sync"
@@ -72,6 +73,44 @@ func isHTTPMethod(s string) bool {
return false
}
func getResourceHost(bufConn *BufConn, maxHeaderBytes, methodLen int) (resource string, host string, err error) {
var data []byte
for dataLen := 0; methodLen+dataLen < maxHeaderBytes; dataLen += 8 {
if data, err = bufConn.Peek(methodLen + dataLen); err != nil {
return "", "", err
}
tokens := strings.Split(string(data), "\n")
if len(tokens) < 2 {
continue
}
if resource == "" {
resource = strings.SplitN(tokens[0][methodLen:], " ", 2)[0]
}
for _, token := range tokens[1:] {
if token == "" || !strings.HasSuffix(token, "\r") {
continue
}
// HTTP headers are case insensitive, so we should simply convert
// each tokens to their lower case form to match 'host' header.
token = strings.ToLower(token)
if strings.HasPrefix(token, "host: ") {
host = strings.TrimPrefix(strings.TrimSuffix(token, "\r"), "host: ")
return resource, host, nil
}
}
if tokens[len(tokens)-1] == "\r" {
break
}
}
return resource, host, nil
}
type acceptResult struct {
conn net.Conn
err error
@@ -87,8 +126,9 @@ type httpListener struct {
tcpKeepAliveTimeout time.Duration
readTimeout time.Duration
writeTimeout time.Duration
updateBytesReadFunc func(int) // function to be called to update bytes read in BufConn.
updateBytesWrittenFunc func(int) // function to be called to update bytes written in BufConn.
maxHeaderBytes int
updateBytesReadFunc func(*http.Request, int) // function to be called to update bytes read in BufConn.
updateBytesWrittenFunc func(*http.Request, int) // function to be called to update bytes written in BufConn.
}
// isRoutineNetErr returns true if error is due to a network timeout,
@@ -134,8 +174,7 @@ func (listener *httpListener) start() {
tcpConn.SetKeepAlive(true)
tcpConn.SetKeepAlivePeriod(listener.tcpKeepAliveTimeout)
bufconn := newBufConn(tcpConn, listener.readTimeout, listener.writeTimeout,
listener.updateBytesReadFunc, listener.updateBytesWrittenFunc)
bufconn := newBufConn(tcpConn, listener.readTimeout, listener.writeTimeout)
// Peek bytes of maximum length of all HTTP methods.
data, err := bufconn.Peek(methodMaxLen)
@@ -158,14 +197,39 @@ func (listener *httpListener) start() {
// Return bufconn if read data is a valid HTTP method.
tokens := strings.SplitN(string(data), " ", 2)
if isHTTPMethod(tokens[0]) {
if listener.tlsConfig == nil {
send(acceptResult{bufconn, nil}, doneCh)
} else {
if listener.tlsConfig != nil {
// As TLS is configured and we got plain text HTTP request,
// return 403 (forbidden) error.
bufconn.Write(sslRequiredErrMsg)
bufconn.Close()
return
}
var resource, host string
if resource, host, err = getResourceHost(bufconn, listener.maxHeaderBytes, len(tokens[0])+1); err != nil {
if !isRoutineNetErr(err) {
reqInfo := (&logger.ReqInfo{}).AppendTags("remoteAddr", bufconn.RemoteAddr().String())
reqInfo.AppendTags("localAddr", bufconn.LocalAddr().String())
ctx := logger.SetReqInfo(context.Background(), reqInfo)
logger.LogIf(ctx, err)
}
bufconn.Close()
return
}
header := make(http.Header)
if host != "" {
header.Add("Host", host)
}
bufconn.setRequest(&http.Request{
Method: tokens[0],
URL: &url.URL{Path: resource},
Host: bufconn.LocalAddr().String(),
Header: header,
})
bufconn.setUpdateFuncs(listener.updateBytesReadFunc, listener.updateBytesWrittenFunc)
send(acceptResult{bufconn, nil}, doneCh)
return
}
@@ -182,8 +246,7 @@ func (listener *httpListener) start() {
}
// Check whether the connection contains HTTP request or not.
bufconn = newBufConn(tlsConn, listener.readTimeout, listener.writeTimeout,
listener.updateBytesReadFunc, listener.updateBytesWrittenFunc)
bufconn = newBufConn(tlsConn, listener.readTimeout, listener.writeTimeout)
// Peek bytes of maximum length of all HTTP methods.
data, err = bufconn.Peek(methodMaxLen)
@@ -201,6 +264,30 @@ func (listener *httpListener) start() {
// Return bufconn if read data is a valid HTTP method.
tokens := strings.SplitN(string(data), " ", 2)
if isHTTPMethod(tokens[0]) {
var resource, host string
if resource, host, err = getResourceHost(bufconn, listener.maxHeaderBytes, len(tokens[0])+1); err != nil {
if !isRoutineNetErr(err) {
reqInfo := (&logger.ReqInfo{}).AppendTags("remoteAddr", bufconn.RemoteAddr().String())
reqInfo.AppendTags("localAddr", bufconn.LocalAddr().String())
ctx := logger.SetReqInfo(context.Background(), reqInfo)
logger.LogIf(ctx, err)
}
bufconn.Close()
return
}
header := make(http.Header)
if host != "" {
header.Add("Host", host)
}
bufconn.setRequest(&http.Request{
Method: tokens[0],
URL: &url.URL{Path: resource},
Host: bufconn.LocalAddr().String(),
Header: header,
})
bufconn.setUpdateFuncs(listener.updateBytesReadFunc, listener.updateBytesWrittenFunc)
send(acceptResult{bufconn, nil}, doneCh)
return
}
@@ -296,8 +383,9 @@ func newHTTPListener(serverAddrs []string,
tcpKeepAliveTimeout time.Duration,
readTimeout time.Duration,
writeTimeout time.Duration,
updateBytesReadFunc func(int),
updateBytesWrittenFunc func(int)) (listener *httpListener, err error) {
maxHeaderBytes int,
updateBytesReadFunc func(*http.Request, int),
updateBytesWrittenFunc func(*http.Request, int)) (listener *httpListener, err error) {
var tcpListeners []*net.TCPListener
@@ -316,7 +404,9 @@ func newHTTPListener(serverAddrs []string,
for _, serverAddr := range serverAddrs {
var l net.Listener
if l, err = listen("tcp4", serverAddr); err != nil {
return nil, err
if l, err = fallbackListen("tcp4", serverAddr); err != nil {
return nil, err
}
}
tcpListener, ok := l.(*net.TCPListener)
@@ -333,6 +423,7 @@ func newHTTPListener(serverAddrs []string,
tcpKeepAliveTimeout: tcpKeepAliveTimeout,
readTimeout: readTimeout,
writeTimeout: writeTimeout,
maxHeaderBytes: maxHeaderBytes,
updateBytesReadFunc: updateBytesReadFunc,
updateBytesWrittenFunc: updateBytesWrittenFunc,
}
+42 -64
View File
@@ -19,13 +19,11 @@ package http
import (
"bufio"
"bytes"
"context"
"crypto/tls"
"errors"
"fmt"
"io"
"net"
"runtime"
"net/http"
"strconv"
"strings"
"sync"
@@ -195,27 +193,6 @@ func TestIsHTTPMethod(t *testing.T) {
}
func TestNewHTTPListener(t *testing.T) {
errMsg := ": no such host"
remoteAddrErrMsgIP := "cannot bind to \"93.184.216.34:65432\": cannot assign requested address"
if runtime.GOOS == "windows" {
remoteAddrErrMsgIP = "listen tcp 93.184.216.34:65432: bind: The requested address is not valid in its context."
}
remoteAddrErrMsgHost := "cannot bind to \"example.org:65432\": cannot assign requested address"
if runtime.GOOS == "windows" {
remoteAddrErrMsgHost = "listen tcp 93.184.216.34:65432: bind: The requested address is not valid in its context."
}
remoteMissingErr := "address unknown-host: missing port in address"
if runtime.GOOS == "windows" {
remoteMissingErr = "listen tcp: address unknown-host: missing port in address"
}
remoteUnknownErr := "lookup unknown-host" + errMsg
if runtime.GOOS == "wpindows" {
remoteUnknownErr = "listen tcp: lookup unknown-host" + errMsg
}
tlsConfig := getTLSConfig(t)
testCases := []struct {
@@ -224,19 +201,18 @@ func TestNewHTTPListener(t *testing.T) {
tcpKeepAliveTimeout time.Duration
readTimeout time.Duration
writeTimeout time.Duration
updateBytesReadFunc func(int)
updateBytesWrittenFunc func(int)
errorLogFunc func(context.Context, error)
expectedErr error
updateBytesReadFunc func(*http.Request, int)
updateBytesWrittenFunc func(*http.Request, int)
expectedErr bool
}{
{[]string{"93.184.216.34:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, errors.New(remoteAddrErrMsgIP)},
{[]string{"example.org:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, errors.New(remoteAddrErrMsgHost)},
{[]string{"unknown-host"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, errors.New(remoteMissingErr)},
{[]string{"unknown-host:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, errors.New(remoteUnknownErr)},
{[]string{"localhost:65432", "93.184.216.34:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, errors.New(remoteAddrErrMsgIP)},
{[]string{"localhost:65432", "unknown-host:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, errors.New(remoteUnknownErr)},
{[]string{"localhost:0"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, nil},
{[]string{"localhost:0"}, tlsConfig, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, nil, nil},
{[]string{"93.184.216.34:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, true},
{[]string{"example.org:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, true},
{[]string{"unknown-host"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, true},
{[]string{"unknown-host:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, true},
{[]string{"localhost:65432", "93.184.216.34:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, true},
{[]string{"localhost:65432", "unknown-host:65432"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, true},
{[]string{"localhost:0"}, nil, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, false},
{[]string{"localhost:0"}, tlsConfig, time.Duration(0), time.Duration(0), time.Duration(0), nil, nil, false},
}
for _, testCase := range testCases {
@@ -246,26 +222,17 @@ func TestNewHTTPListener(t *testing.T) {
testCase.tcpKeepAliveTimeout,
testCase.readTimeout,
testCase.writeTimeout,
DefaultMaxHeaderBytes,
testCase.updateBytesReadFunc,
testCase.updateBytesWrittenFunc,
)
if testCase.expectedErr == nil {
if !testCase.expectedErr {
if err != nil {
t.Fatalf("error: expected = <nil>, got = %v", err)
}
} else if err == nil {
t.Fatalf("error: expected = %v, got = <nil>", testCase.expectedErr)
} else {
var match bool
if strings.HasSuffix(testCase.expectedErr.Error(), errMsg) {
match = strings.HasSuffix(err.Error(), errMsg)
} else {
match = (testCase.expectedErr.Error() == err.Error())
}
if !match {
t.Fatalf("error: expected = %v, got = %v", testCase.expectedErr, err)
}
}
if err == nil {
@@ -297,6 +264,7 @@ func TestHTTPListenerStartClose(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -344,6 +312,7 @@ func TestHTTPListenerAddr(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -388,6 +357,7 @@ func TestHTTPListenerAddrs(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -414,17 +384,20 @@ func TestHTTPListenerAccept(t *testing.T) {
nonLoopBackIP := getNonLoopBackIP(t)
testCases := []struct {
serverAddrs []string
tlsConfig *tls.Config
request string
reply string
serverAddrs []string
tlsConfig *tls.Config
request string
reply string
expectedRequestLine string
}{
{[]string{"localhost:0"}, nil, "GET / HTTP/1.0\n", "200 OK\n"},
{[]string{nonLoopBackIP + ":0"}, nil, "POST / HTTP/1.0\n", "200 OK\n"},
{[]string{"127.0.0.1:0", nonLoopBackIP + ":0"}, nil, "CONNECT \n", "200 OK\n"},
{[]string{"localhost:0"}, tlsConfig, "GET / HTTP/1.0\n", "200 OK\n"},
{[]string{nonLoopBackIP + ":0"}, tlsConfig, "POST / HTTP/1.0\n", "200 OK\n"},
{[]string{"127.0.0.1:0", nonLoopBackIP + ":0"}, tlsConfig, "CONNECT \n", "200 OK\n"},
{[]string{"localhost:0"}, nil, "GET / HTTP/1.0\r\nHost: example.org\r\n\r\n", "200 OK\r\n", "GET / HTTP/1.0\r\n"},
{[]string{nonLoopBackIP + ":0"}, nil, "POST / HTTP/1.0\r\nHost: example.org\r\n\r\n", "200 OK\r\n", "POST / HTTP/1.0\r\n"},
{[]string{nonLoopBackIP + ":0"}, nil, "HEAD / HTTP/1.0\r\nhost: example.org\r\n\r\n", "200 OK\r\n", "HEAD / HTTP/1.0\r\n"},
{[]string{"127.0.0.1:0", nonLoopBackIP + ":0"}, nil, "CONNECT \r\nHost: www.example.org\r\n\r\n", "200 OK\r\n", "CONNECT \r\n"},
{[]string{"localhost:0"}, tlsConfig, "GET / HTTP/1.0\r\nHost: example.org\r\n\r\n", "200 OK\r\n", "GET / HTTP/1.0\r\n"},
{[]string{nonLoopBackIP + ":0"}, tlsConfig, "POST / HTTP/1.0\r\nHost: example.org\r\n\r\n", "200 OK\r\n", "POST / HTTP/1.0\r\n"},
{[]string{nonLoopBackIP + ":0"}, tlsConfig, "HEAD / HTTP/1.0\r\nhost: example.org\r\n\r\n", "200 OK\r\n", "HEAD / HTTP/1.0\r\n"},
{[]string{"127.0.0.1:0", nonLoopBackIP + ":0"}, tlsConfig, "CONNECT \r\nHost: www.example.org\r\n\r\n", "200 OK\r\n", "CONNECT \r\n"},
}
for i, testCase := range testCases {
@@ -434,6 +407,7 @@ func TestHTTPListenerAccept(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -463,13 +437,13 @@ func TestHTTPListenerAccept(t *testing.T) {
t.Fatalf("Test %d: accept: expected = <nil>, got = %v", i+1, err)
}
request, err := bufio.NewReader(serverConn).ReadString('\n')
requestLine, err := bufio.NewReader(serverConn).ReadString('\n')
if err != nil {
t.Fatalf("Test %d: request read: expected = <nil>, got = %v", i+1, err)
}
if testCase.request != request {
t.Fatalf("Test %d: request: expected = %v, got = %v", i+1, testCase.request, request)
if requestLine != testCase.expectedRequestLine {
t.Fatalf("Test %d: request: expected = %v, got = %v", i+1, testCase.expectedRequestLine, requestLine)
}
if _, err = io.WriteString(serverConn, testCase.reply); err != nil {
@@ -513,6 +487,7 @@ func TestHTTPListenerAcceptPeekError(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -566,6 +541,7 @@ func TestHTTPListenerAcceptTLSError(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -632,6 +608,7 @@ func TestHTTPListenerAcceptError(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -757,6 +734,7 @@ func TestHTTPListenerAcceptParallel(t *testing.T) {
time.Duration(0),
time.Duration(0),
time.Duration(0),
DefaultMaxHeaderBytes,
nil,
nil,
)
@@ -765,8 +743,8 @@ func TestHTTPListenerAcceptParallel(t *testing.T) {
}
for _, serverAddr := range listener.Addrs() {
go connect(i, serverAddr.String(), testCase.tlsConfig != nil, true, "GET /1 HTTP/1.0\n", testCase.reply)
go connect(i, serverAddr.String(), testCase.tlsConfig != nil, false, "GET /2 HTTP/1.0\n", testCase.reply)
go connect(i, serverAddr.String(), testCase.tlsConfig != nil, true, "GET /1 HTTP/1.0\r\nHost: example.org\r\nr\n", testCase.reply)
go connect(i, serverAddr.String(), testCase.tlsConfig != nil, false, "GET /2 HTTP/1.0\r\nHost: example.org\r\n\r\n", testCase.reply)
var wg sync.WaitGroup
@@ -775,14 +753,14 @@ func TestHTTPListenerAcceptParallel(t *testing.T) {
t.Fatalf("Test %d: accept: expected = <nil>, got = %v", i+1, err)
}
wg.Add(1)
go handleConnection(i, &wg, serverConn, "GET /2 HTTP/1.0\n", testCase.reply)
go handleConnection(i, &wg, serverConn, "GET /2 HTTP/1.0\r\n", testCase.reply)
serverConn, err = listener.Accept()
if err != nil {
t.Fatalf("Test %d: accept: expected = <nil>, got = %v", i+1, err)
}
wg.Add(1)
go handleConnection(i, &wg, serverConn, "GET /1 HTTP/1.0\n", testCase.reply)
go handleConnection(i, &wg, serverConn, "GET /1 HTTP/1.0\r\n", testCase.reply)
wg.Wait()
}
+10 -9
View File
@@ -51,15 +51,15 @@ const (
// Server - extended http.Server supports multiple addresses to serve and enhanced connection handling.
type Server struct {
http.Server
Addrs []string // addresses on which the server listens for new connection.
ShutdownTimeout time.Duration // timeout used for graceful server shutdown.
TCPKeepAliveTimeout time.Duration // timeout used for underneath TCP connection.
UpdateBytesReadFunc func(int) // function to be called to update bytes read in bufConn.
UpdateBytesWrittenFunc func(int) // function to be called to update bytes written in bufConn.
listenerMutex *sync.Mutex // to guard 'listener' field.
listener *httpListener // HTTP listener for all 'Addrs' field.
inShutdown uint32 // indicates whether the server is in shutdown or not
requestCount int32 // counter holds no. of request in progress.
Addrs []string // addresses on which the server listens for new connection.
ShutdownTimeout time.Duration // timeout used for graceful server shutdown.
TCPKeepAliveTimeout time.Duration // timeout used for underneath TCP connection.
UpdateBytesReadFunc func(*http.Request, int) // function to be called to update bytes read in bufConn.
UpdateBytesWrittenFunc func(*http.Request, int) // function to be called to update bytes written in bufConn.
listenerMutex *sync.Mutex // to guard 'listener' field.
listener *httpListener // HTTP listener for all 'Addrs' field.
inShutdown uint32 // indicates whether the server is in shutdown or not
requestCount int32 // counter holds no. of request in progress.
}
// GetRequestCount - returns number of request in progress.
@@ -91,6 +91,7 @@ func (srv *Server) Start() (err error) {
tcpKeepAliveTimeout,
readTimeout,
writeTimeout,
srv.MaxHeaderBytes,
updateBytesReadFunc,
updateBytesWrittenFunc,
)
+1 -1
View File
@@ -124,7 +124,7 @@ func (c *minioCollector) Collect(ch chan<- prometheus.Metric) {
var totalDisks, offlineDisks int
// Setting totalDisks to 1 and offlineDisks to 0 in FS mode
if s.Backend.Type == FS {
if s.Backend.Type == BackendFS {
totalDisks = 1
offlineDisks = 0
} else {
+38 -16
View File
@@ -23,6 +23,7 @@ import (
"fmt"
"net/url"
"path"
"strings"
"sync"
"time"
@@ -46,7 +47,13 @@ func (sys *NotificationSys) GetARNList() []string {
arns := []string{}
region := globalServerConfig.GetRegion()
for _, targetID := range sys.targetList.List() {
arns = append(arns, targetID.ToARN(region).String())
// httpclient target is part of ListenBucketNotification
// which doesn't need to be listed as part of the ARN list
// This list is only meant for external targets, filter
// this out pro-actively.
if !strings.HasPrefix(targetID.ID, "httpclient+") {
arns = append(arns, targetID.ToARN(region).String())
}
}
return arns
@@ -192,11 +199,17 @@ func (sys *NotificationSys) AddRemoteTarget(bucketName string, target event.Targ
if targetMap == nil {
targetMap = make(map[event.TargetID]event.RulesMap)
}
targetMap[target.ID()] = rulesMap.Clone()
rulesMap = rulesMap.Clone()
targetMap[target.ID()] = rulesMap
sys.bucketRemoteTargetRulesMap[bucketName] = targetMap
rulesMap = rulesMap.Clone()
rulesMap.Add(sys.bucketRulesMap[bucketName])
sys.bucketRulesMap[bucketName] = rulesMap
sys.Unlock()
sys.AddRulesMap(bucketName, rulesMap)
return nil
}
@@ -341,8 +354,12 @@ func (sys *NotificationSys) AddRulesMap(bucketName string, rulesMap event.RulesM
rulesMap.Add(targetRulesMap)
}
rulesMap.Add(sys.bucketRulesMap[bucketName])
sys.bucketRulesMap[bucketName] = rulesMap
// Do not add for an empty rulesMap.
if len(rulesMap) == 0 {
delete(sys.bucketRulesMap, bucketName)
} else {
sys.bucketRulesMap[bucketName] = rulesMap
}
}
// RemoveRulesMap - removes rules map for bucket name.
@@ -439,13 +456,14 @@ func NewNotificationSys(config *serverConfig, endpoints EndpointList) *Notificat
}
type eventArgs struct {
EventName event.Name
BucketName string
Object ObjectInfo
ReqParams map[string]string
Host string
Port string
UserAgent string
EventName event.Name
BucketName string
Object ObjectInfo
ReqParams map[string]string
RespElements map[string]string
Host string
Port string
UserAgent string
}
// ToEvent - converts to notification event.
@@ -464,6 +482,13 @@ func (args eventArgs) ToEvent() event.Event {
eventTime := UTCNow()
uniqueID := fmt.Sprintf("%X", eventTime.UnixNano())
respElements := map[string]string{
"x-amz-request-id": uniqueID,
"x-minio-origin-endpoint": getOriginEndpoint(), // Minio specific custom elements.
}
if args.RespElements["content-length"] != "" {
respElements["content-length"] = args.RespElements["content-length"]
}
newEvent := event.Event{
EventVersion: "2.0",
EventSource: "minio:s3",
@@ -472,10 +497,7 @@ func (args eventArgs) ToEvent() event.Event {
EventName: args.EventName,
UserIdentity: event.Identity{creds.AccessKey},
RequestParameters: args.ReqParams,
ResponseElements: map[string]string{
"x-amz-request-id": uniqueID,
"x-minio-origin-endpoint": getOriginEndpoint(), // Minio specific custom elements.
},
ResponseElements: respElements,
S3: event.Metadata{
SchemaVersion: "1.0",
ConfigurationID: "Config",
+3 -3
View File
@@ -31,9 +31,9 @@ type BackendType int
const (
Unknown BackendType = iota
// Filesystem backend.
FS
// Multi disk Erasure (single, distributed) backend.
Erasure
BackendFS
// Multi disk BackendErasure (single, distributed) backend.
BackendErasure
// Add your own backend.
)
+21 -19
View File
@@ -306,8 +306,8 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
}
if objectAPI.IsEncryptionSupported() {
if apiErr, _ := DecryptObjectInfo(&objInfo, r.Header); apiErr != ErrNone {
writeErrorResponse(w, apiErr, r.URL)
if _, err = DecryptObjectInfo(&objInfo, r.Header); err != nil {
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
return
}
}
@@ -405,13 +405,14 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
// Notify object accessed via a GET request.
sendEvent(eventArgs{
EventName: event.ObjectAccessedGet,
BucketName: bucket,
Object: objInfo,
ReqParams: extractReqParams(r),
UserAgent: r.UserAgent(),
Host: host,
Port: port,
EventName: event.ObjectAccessedGet,
BucketName: bucket,
Object: objInfo,
ReqParams: extractReqParams(r),
RespElements: extractRespElements(w),
UserAgent: r.UserAgent(),
Host: host,
Port: port,
})
}
@@ -467,10 +468,10 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
writeErrorResponseHeadersOnly(w, toAPIErrorCode(err))
return
}
var encrypted bool
if objectAPI.IsEncryptionSupported() {
if apiErr, encrypted := DecryptObjectInfo(&objInfo, r.Header); apiErr != ErrNone {
writeErrorResponse(w, apiErr, r.URL)
if encrypted, err = DecryptObjectInfo(&objInfo, r.Header); err != nil {
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
return
} else if encrypted {
s3Encrypted := crypto.S3.IsEncrypted(objInfo.UserDefined)
@@ -509,13 +510,14 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
// Notify object accessed via a HEAD request.
sendEvent(eventArgs{
EventName: event.ObjectAccessedHead,
BucketName: bucket,
Object: objInfo,
ReqParams: extractReqParams(r),
UserAgent: r.UserAgent(),
Host: host,
Port: port,
EventName: event.ObjectAccessedHead,
BucketName: bucket,
Object: objInfo,
ReqParams: extractReqParams(r),
RespElements: extractRespElements(w),
UserAgent: r.UserAgent(),
Host: host,
Port: port,
})
}
+7 -7
View File
@@ -313,10 +313,6 @@ func serverMain(ctx *cli.Context) {
logger.FatalIf(err, "Unable to initialize backend")
}
globalObjLayerMutex.Lock()
globalObjectAPI = newObject
globalObjLayerMutex.Unlock()
// Populate existing buckets to the etcd backend
if globalDNSConfig != nil {
initFederatorBackend(newObject)
@@ -326,7 +322,7 @@ func serverMain(ctx *cli.Context) {
globalConfigSys = NewConfigSys()
// Initialize config system.
if err = globalConfigSys.Init(newObjectLayerFn()); err != nil {
if err = globalConfigSys.Init(newObject); err != nil {
logger.Fatal(err, "Unable to initialize config system")
}
@@ -347,7 +343,7 @@ func serverMain(ctx *cli.Context) {
globalPolicySys = NewPolicySys()
// Initialize policy system.
if err := globalPolicySys.Init(newObjectLayerFn()); err != nil {
if err := globalPolicySys.Init(newObject); err != nil {
logger.Fatal(err, "Unable to initialize policy system")
}
@@ -355,10 +351,14 @@ func serverMain(ctx *cli.Context) {
globalNotificationSys = NewNotificationSys(globalServerConfig, globalEndpoints)
// Initialize notification system.
if err := globalNotificationSys.Init(newObjectLayerFn()); err != nil {
if err := globalNotificationSys.Init(newObject); err != nil {
logger.Fatal(err, "Unable to initialize notification system")
}
globalObjLayerMutex.Lock()
globalObjectAPI = newObject
globalObjLayerMutex.Unlock()
// Prints the formatted startup message once object layer is initialized.
apiEndpoints := getAPIEndpoints(globalMinioAddr)
printStartupMessage(apiEndpoints)
+1 -1
View File
@@ -174,7 +174,7 @@ func printObjectAPIMsg() {
// Get formatted disk/storage info message.
func getStorageInfoMsg(storageInfo StorageInfo) string {
var msg string
if storageInfo.Backend.Type == Erasure {
if storageInfo.Backend.Type == BackendErasure {
diskInfo := fmt.Sprintf(" %d Online, %d Offline. ", storageInfo.Backend.OnlineDisks, storageInfo.Backend.OfflineDisks)
msg += colorBlue("Status:") + fmt.Sprintf(getFormatStr(len(diskInfo), 8), diskInfo)
}
+1 -1
View File
@@ -30,7 +30,7 @@ import (
// Tests if we generate storage info.
func TestStorageInfoMsg(t *testing.T) {
infoStorage := StorageInfo{}
infoStorage.Backend.Type = Erasure
infoStorage.Backend.Type = BackendErasure
infoStorage.Backend.OnlineDisks = 7
infoStorage.Backend.OfflineDisks = 1
-2
View File
@@ -17,7 +17,6 @@
package cmd
import (
"fmt"
"io/ioutil"
"net/http"
"net/http/httptest"
@@ -355,7 +354,6 @@ func testStorageAPIReadFile(t *testing.T, storage StorageAPI) {
expectErr := (err != nil)
if expectErr != testCase.expectErr {
fmt.Println(err)
t.Fatalf("case %v: error: expected: %v, got: %v", i+1, testCase.expectErr, expectErr)
}
+1
View File
@@ -434,6 +434,7 @@ func resetGlobalIsXL() {
func resetGlobalIsEnvs() {
globalIsEnvCreds = false
globalIsEnvWORM = false
globalIsEnvBrowser = false
globalIsEnvRegion = false
globalIsStorageClass = false
+18 -8
View File
@@ -328,12 +328,16 @@ func (web *webAPIHandlers) ListObjects(r *http.Request, args *ListObjectsArgs, r
return toJSONError(authErr)
}
// Error out anonymous (non-owner) has no access download or upload objects.
if !readable && !writable {
return errAuthentication
reply.Writable = writable
if !readable {
// Error out if anonymous user (non-owner) has no access to download or upload objects
if !writable {
return errAuthentication
}
// return empty object list if access is write only
return nil
}
reply.Writable = writable
}
lo, err := listObjects(context.Background(), args.BucketName, args.Prefix, args.Marker, slashSeparator, 1000)
@@ -715,8 +719,8 @@ func (web *webAPIHandlers) Download(w http.ResponseWriter, r *http.Request) {
}
if objectAPI.IsEncryptionSupported() {
if apiErr, _ := DecryptObjectInfo(&objInfo, r.Header); apiErr != ErrNone {
writeErrorResponse(w, apiErr, r.URL)
if _, err = DecryptObjectInfo(&objInfo, r.Header); err != nil {
writeWebErrorResponse(w, err)
return
}
}
@@ -818,8 +822,8 @@ func (web *webAPIHandlers) DownloadZip(w http.ResponseWriter, r *http.Request) {
return err
}
if objectAPI.IsEncryptionSupported() {
if apiErr, _ := DecryptObjectInfo(&info, r.Header); apiErr != ErrNone {
writeErrorResponse(w, apiErr, r.URL)
if _, err = DecryptObjectInfo(&info, r.Header); err != nil {
writeWebErrorResponse(w, err)
return err
}
}
@@ -1235,6 +1239,12 @@ func toWebAPIError(err error) APIError {
HTTPStatusCode: http.StatusBadRequest,
Description: err.Error(),
}
} else if err == errEncryptedObject {
return getAPIError(ErrSSEEncryptedObject)
} else if err == errInvalidEncryptionParameters {
return getAPIError(ErrInvalidEncryptionParameters)
} else if err == errObjectTampered {
return getAPIError(ErrObjectTampered)
} else if err == errMethodNotAllowed {
return getAPIError(ErrMethodNotAllowed)
}
+1 -1
View File
@@ -272,7 +272,7 @@ func newXLSets(endpoints EndpointList, format *formatXLV3, setCount int, drivesP
// StorageInfo - combines output of StorageInfo across all erasure coded object sets.
func (s *xlSets) StorageInfo(ctx context.Context) StorageInfo {
var storageInfo StorageInfo
storageInfo.Backend.Type = Erasure
storageInfo.Backend.Type = BackendErasure
for _, set := range s.sets {
lstorageInfo := set.StorageInfo(ctx)
storageInfo.Used = storageInfo.Used + lstorageInfo.Used
+4 -4
View File
@@ -446,7 +446,7 @@ func healObject(ctx context.Context, storageDisks []StorageAPI, bucket string, o
// Heal each part. erasureHealFile() will write the healed
// part to .minio/tmp/uuid/ which needs to be renamed later to
// the final location.
storage, err := NewErasureStorage(ctx, latestMeta.Erasure.DataBlocks,
erasure, err := NewErasure(ctx, latestMeta.Erasure.DataBlocks,
latestMeta.Erasure.ParityBlocks, latestMeta.Erasure.BlockSize)
if err != nil {
return result, toObjectErr(err, bucket, object)
@@ -455,7 +455,7 @@ func healObject(ctx context.Context, storageDisks []StorageAPI, bucket string, o
for partIndex := 0; partIndex < len(latestMeta.Parts); partIndex++ {
partName := latestMeta.Parts[partIndex].Name
partSize := latestMeta.Parts[partIndex].Size
erasure := latestMeta.Erasure
erasureInfo := latestMeta.Erasure
var algorithm BitrotAlgorithm
bitrotReaders := make([]*bitrotReader, len(latestDisks))
for i, disk := range latestDisks {
@@ -464,7 +464,7 @@ func healObject(ctx context.Context, storageDisks []StorageAPI, bucket string, o
}
info := partsMetadata[i].Erasure.GetChecksumInfo(partName)
algorithm = info.Algorithm
endOffset := getErasureShardFileEndOffset(0, partSize, partSize, erasure.BlockSize, storage.dataBlocks)
endOffset := getErasureShardFileEndOffset(0, partSize, partSize, erasureInfo.BlockSize, erasure.dataBlocks)
bitrotReaders[i] = newBitrotReader(disk, bucket, pathJoin(object, partName), algorithm, endOffset, info.Hash)
}
bitrotWriters := make([]*bitrotWriter, len(outDatedDisks))
@@ -474,7 +474,7 @@ func healObject(ctx context.Context, storageDisks []StorageAPI, bucket string, o
}
bitrotWriters[i] = newBitrotWriter(disk, minioMetaTmpBucket, pathJoin(tmpID, partName), algorithm)
}
hErr := storage.HealFile(ctx, bitrotReaders, bitrotWriters, partSize)
hErr := erasure.Heal(ctx, bitrotReaders, bitrotWriters, partSize)
if hErr != nil {
return result, toObjectErr(hErr, bucket, object)
}
+7 -40
View File
@@ -223,7 +223,7 @@ func (xl xlObjects) newMultipartUpload(ctx context.Context, bucket string, objec
// delete the tmp path later in case we fail to rename (ignore
// returned errors) - this will be a no-op in case of a rename
// success.
defer xl.deleteObject(ctx, minioMetaTmpBucket, tempUploadIDPath)
defer xl.deleteObject(ctx, minioMetaTmpBucket, tempUploadIDPath, writeQuorum, false)
// Attempt to rename temp upload object to actual upload path object
_, rErr := renameObject(ctx, disks, minioMetaTmpBucket, tempUploadIDPath, minioMetaMultipartBucket, uploadIDPath, writeQuorum)
@@ -361,7 +361,7 @@ func (xl xlObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID
tmpPartPath := path.Join(tmpPart, partSuffix)
// Delete the temporary object part. If PutObjectPart succeeds there would be nothing to delete.
defer xl.deleteObject(ctx, minioMetaTmpBucket, tmpPart)
defer xl.deleteObject(ctx, minioMetaTmpBucket, tmpPart, writeQuorum, false)
if data.Size() > 0 {
if pErr := xl.prepareFile(ctx, minioMetaTmpBucket, tmpPartPath, data.Size(), onlineDisks, xlMeta.Erasure.BlockSize, xlMeta.Erasure.DataBlocks, writeQuorum); err != nil {
return pi, toObjectErr(pErr, bucket, object)
@@ -369,7 +369,7 @@ func (xl xlObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID
}
}
storage, err := NewErasureStorage(ctx, xlMeta.Erasure.DataBlocks, xlMeta.Erasure.ParityBlocks, xlMeta.Erasure.BlockSize)
erasure, err := NewErasure(ctx, xlMeta.Erasure.DataBlocks, xlMeta.Erasure.ParityBlocks, xlMeta.Erasure.BlockSize)
if err != nil {
return pi, toObjectErr(err, bucket, object)
}
@@ -397,7 +397,7 @@ func (xl xlObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID
}
writers[i] = newBitrotWriter(disk, minioMetaTmpBucket, tmpPartPath, DefaultBitrotAlgorithm)
}
n, err := storage.CreateFile(ctx, data, writers, buffer, storage.dataBlocks+1)
n, err := erasure.Encode(ctx, data, writers, buffer, erasure.dataBlocks+1)
if err != nil {
return pi, toObjectErr(err, bucket, object)
}
@@ -752,7 +752,7 @@ func (xl xlObjects) CompleteMultipartUpload(ctx context.Context, bucket string,
newUniqueID := mustGetUUID()
// Delete success renamed object.
defer xl.deleteObject(ctx, minioMetaTmpBucket, newUniqueID)
defer xl.deleteObject(ctx, minioMetaTmpBucket, newUniqueID, writeQuorum, false)
// NOTE: Do not use online disks slice here.
// The reason is that existing object should be purged
@@ -792,34 +792,6 @@ func (xl xlObjects) CompleteMultipartUpload(ctx context.Context, bucket string,
return xlMeta.ToObjectInfo(bucket, object), nil
}
// Wrapper which removes all the uploaded parts.
func (xl xlObjects) cleanupUploadedParts(ctx context.Context, uploadIDPath string, writeQuorum int) error {
var errs = make([]error, len(xl.getDisks()))
var wg = &sync.WaitGroup{}
// Cleanup uploadID for all disks.
for index, disk := range xl.getDisks() {
if disk == nil {
errs[index] = errDiskNotFound
continue
}
wg.Add(1)
// Cleanup each uploadID in a routine.
go func(index int, disk StorageAPI) {
defer wg.Done()
err := cleanupDir(ctx, disk, minioMetaMultipartBucket, uploadIDPath)
if err != nil {
errs[index] = err
}
}(index, disk)
}
// Wait for all the cleanups to finish.
wg.Wait()
return reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
}
// AbortMultipartUpload - aborts an ongoing multipart operation
// signified by the input uploadID. This is an atomic operation
// doesn't require clients to initiate multiple such requests.
@@ -859,7 +831,7 @@ func (xl xlObjects) AbortMultipartUpload(ctx context.Context, bucket, object, up
}
// Cleanup all uploaded parts.
if err = xl.cleanupUploadedParts(ctx, uploadIDPath, writeQuorum); err != nil {
if err = xl.deleteObject(ctx, minioMetaMultipartBucket, uploadIDPath, writeQuorum, false); err != nil {
return toObjectErr(err, bucket, object)
}
@@ -911,12 +883,7 @@ func (xl xlObjects) cleanupStaleMultipartUploadsOnDisk(ctx context.Context, disk
continue
}
if now.Sub(fi.ModTime) > expiry {
// Quorum value will need to be figured out using readAllXLMetadata() and objectQuorumFromMeta()
// But we can avoid these calls as we do not care if xl.cleanupUploadedParts() meets quorum
// when it removes files. We igore the error message from xl.cleanupUploadedParts() as we can't
// return it to any client. Hence we set quorum to 0.
quorum := 0
xl.cleanupUploadedParts(ctx, uploadIDPath, quorum)
xl.deleteObject(ctx, minioMetaMultipartBucket, uploadIDPath, len(xl.getDisks())/2+1, false)
}
}
}
+45 -42
View File
@@ -262,7 +262,7 @@ func (xl xlObjects) getObject(ctx context.Context, bucket, object string, startO
}
var totalBytesRead int64
storage, err := NewErasureStorage(ctx, xlMeta.Erasure.DataBlocks, xlMeta.Erasure.ParityBlocks, xlMeta.Erasure.BlockSize)
erasure, err := NewErasure(ctx, xlMeta.Erasure.DataBlocks, xlMeta.Erasure.ParityBlocks, xlMeta.Erasure.BlockSize)
if err != nil {
return toObjectErr(err, bucket, object)
}
@@ -292,7 +292,7 @@ func (xl xlObjects) getObject(ctx context.Context, bucket, object string, startO
bitrotReaders[index] = newBitrotReader(disk, bucket, pathJoin(object, partName), checksumInfo.Algorithm, endOffset, checksumInfo.Hash)
}
err := storage.ReadFile(ctx, writer, bitrotReaders, partOffset, partLength, partSize)
err := erasure.Decode(ctx, writer, bitrotReaders, partOffset, partLength, partSize)
if err != nil {
return toObjectErr(err, bucket, object)
}
@@ -546,7 +546,7 @@ func (xl xlObjects) putObject(ctx context.Context, bucket string, object string,
// Delete temporary object in the event of failure.
// If PutObject succeeded there would be no temporary
// object to delete.
defer xl.deleteObject(ctx, minioMetaTmpBucket, tempObj)
defer xl.deleteObject(ctx, minioMetaTmpBucket, tempObj, writeQuorum, false)
// This is a special case with size as '0' and object ends with
// a slash separator, we treat it like a valid operation and
@@ -608,7 +608,7 @@ func (xl xlObjects) putObject(ctx context.Context, bucket string, object string,
// Total size of the written object
var sizeWritten int64
storage, err := NewErasureStorage(ctx, xlMeta.Erasure.DataBlocks, xlMeta.Erasure.ParityBlocks, xlMeta.Erasure.BlockSize)
erasure, err := NewErasure(ctx, xlMeta.Erasure.DataBlocks, xlMeta.Erasure.ParityBlocks, xlMeta.Erasure.BlockSize)
if err != nil {
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
@@ -667,7 +667,7 @@ func (xl xlObjects) putObject(ctx context.Context, bucket string, object string,
}
writers[i] = newBitrotWriter(disk, minioMetaTmpBucket, tempErasureObj, DefaultBitrotAlgorithm)
}
n, erasureErr := storage.CreateFile(ctx, curPartReader, writers, buffer, storage.dataBlocks+1)
n, erasureErr := erasure.Encode(ctx, curPartReader, writers, buffer, erasure.dataBlocks+1)
if erasureErr != nil {
return ObjectInfo{}, toObjectErr(erasureErr, minioMetaTmpBucket, tempErasureObj)
}
@@ -715,7 +715,7 @@ func (xl xlObjects) putObject(ctx context.Context, bucket string, object string,
newUniqueID := mustGetUUID()
// Delete successfully renamed object.
defer xl.deleteObject(ctx, minioMetaTmpBucket, newUniqueID)
defer xl.deleteObject(ctx, minioMetaTmpBucket, newUniqueID, writeQuorum, false)
// NOTE: Do not use online disks slice here.
// The reason is that existing object should be purged
@@ -774,36 +774,32 @@ func (xl xlObjects) putObject(ctx context.Context, bucket string, object string,
// deleteObject - wrapper for delete object, deletes an object from
// all the disks in parallel, including `xl.json` associated with the
// object.
func (xl xlObjects) deleteObject(ctx context.Context, bucket, object string) error {
func (xl xlObjects) deleteObject(ctx context.Context, bucket, object string, writeQuorum int, isDir bool) error {
var disks []StorageAPI
var err error
tmpObj := mustGetUUID()
if bucket == minioMetaTmpBucket {
tmpObj = object
disks = xl.getDisks()
} else {
if isDir {
disks, err = renameObjectDir(ctx, xl.getDisks(), bucket, object, minioMetaTmpBucket, tmpObj, writeQuorum)
} else {
disks, err = renameObject(ctx, xl.getDisks(), bucket, object, minioMetaTmpBucket, tmpObj, writeQuorum)
}
if err != nil {
return toObjectErr(err, bucket, object)
}
}
// Initialize sync waitgroup.
var wg = &sync.WaitGroup{}
var writeQuorum int
var err error
isDir := hasSuffix(object, slashSeparator)
if !isDir {
// Read metadata associated with the object from all disks.
metaArr, errs := readAllXLMetadata(ctx, xl.getDisks(), bucket, object)
// get Quorum for this object
_, writeQuorum, err = objectQuorumFromMeta(ctx, xl, metaArr, errs)
if err != nil {
return err
}
err = reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
if err != nil {
return err
}
} else {
// WriteQuorum is defaulted to N/2 + 1 for directories
writeQuorum = len(xl.getDisks())/2 + 1
}
// Initialize list of errors.
var dErrs = make([]error, len(xl.getDisks()))
var dErrs = make([]error, len(disks))
for index, disk := range xl.getDisks() {
for index, disk := range disks {
if disk == nil {
dErrs[index] = errDiskNotFound
continue
@@ -815,9 +811,9 @@ func (xl xlObjects) deleteObject(ctx context.Context, bucket, object string) err
if isDir {
// DeleteFile() simply tries to remove a directory
// and will succeed only if that directory is empty.
e = disk.DeleteFile(bucket, object)
e = disk.DeleteFile(minioMetaTmpBucket, tmpObj)
} else {
e = cleanupDir(ctx, disk, bucket, object)
e = cleanupDir(ctx, disk, minioMetaTmpBucket, tmpObj)
}
if e != nil && e != errVolumeNotFound {
dErrs[index] = e
@@ -847,20 +843,27 @@ func (xl xlObjects) DeleteObject(ctx context.Context, bucket, object string) (er
return err
}
if hasSuffix(object, slashSeparator) {
// Delete the object on all disks.
if err = xl.deleteObject(ctx, bucket, object); err != nil {
if !xl.isObject(bucket, object) {
return ObjectNotFound{bucket, object}
}
var writeQuorum int
var isObjectDir = hasSuffix(object, slashSeparator)
if isObjectDir {
writeQuorum = len(xl.getDisks())/2 + 1
} else {
// Read metadata associated with the object from all disks.
partsMetadata, errs := readAllXLMetadata(ctx, xl.getDisks(), bucket, object)
// get Quorum for this object
_, writeQuorum, err = objectQuorumFromMeta(ctx, xl, partsMetadata, errs)
if err != nil {
return toObjectErr(err, bucket, object)
}
}
// Validate object exists.
if !xl.isObject(bucket, object) {
return ObjectNotFound{bucket, object}
} // else proceed to delete the object.
// Delete the object on all disks.
if err = xl.deleteObject(ctx, bucket, object); err != nil {
if err = xl.deleteObject(ctx, bucket, object, writeQuorum, isObjectDir); err != nil {
return toObjectErr(err, bucket, object)
}
+8 -6
View File
@@ -71,11 +71,14 @@ func TestXLDeleteObjectBasic(t *testing.T) {
object string
expectedErr error
}{
{".test", "obj", BucketNameInvalid{Bucket: ".test"}},
{"----", "obj", BucketNameInvalid{Bucket: "----"}},
{".test", "dir/obj", BucketNameInvalid{Bucket: ".test"}},
{"----", "dir/obj", BucketNameInvalid{Bucket: "----"}},
{"bucket", "", ObjectNameInvalid{Bucket: "bucket", Object: ""}},
{"bucket", "doesnotexist", ObjectNotFound{Bucket: "bucket", Object: "doesnotexist"}},
{"bucket", "obj", nil},
{"bucket", "dir/doesnotexist", ObjectNotFound{Bucket: "bucket", Object: "dir/doesnotexist"}},
{"bucket", "dir", ObjectNotFound{Bucket: "bucket", Object: "dir"}},
{"bucket", "dir/", ObjectNotFound{Bucket: "bucket", Object: "dir/"}},
{"bucket", "dir/obj", nil},
}
// Create an instance of xl backend
@@ -84,14 +87,13 @@ func TestXLDeleteObjectBasic(t *testing.T) {
t.Fatal(err)
}
// Make bucket for Test 7 to pass
err = xl.MakeBucketWithLocation(context.Background(), "bucket", "")
if err != nil {
t.Fatal(err)
}
// Create object "obj" under bucket "bucket" for Test 7 to pass
_, err = xl.PutObject(context.Background(), "bucket", "obj", mustGetHashReader(t, bytes.NewReader([]byte("abcd")), int64(len("abcd")), "", ""), nil)
// Create object "dir/obj" under bucket "bucket" for Test 7 to pass
_, err = xl.PutObject(context.Background(), "bucket", "dir/obj", mustGetHashReader(t, bytes.NewReader([]byte("abcd")), int64(len("abcd")), "", ""), nil)
if err != nil {
t.Fatalf("XL Object upload failed: <ERROR> %s", err)
}
+4 -1
View File
@@ -30,6 +30,9 @@ const (
xlMetaJSONFile = "xl.json"
)
// OfflineDisk represents an unavailable disk.
var OfflineDisk StorageAPI // zero value is nil
// xlObjects - Implements XL object layer.
type xlObjects struct {
// name space mutex for object layer.
@@ -137,7 +140,7 @@ func getStorageInfo(disks []StorageAPI) StorageInfo {
}
storageInfo.Used = used
storageInfo.Backend.Type = Erasure
storageInfo.Backend.Type = BackendErasure
storageInfo.Backend.OnlineDisks = onlineDisks
storageInfo.Backend.OfflineDisks = offlineDisks
+2 -2
View File
@@ -33,7 +33,7 @@ As a prerequisite to run Minio S3 gateway on an AWS S3 compatible service, you n
docker run -p 9000:9000 --name minio-s3 \
-e "MINIO_ACCESS_KEY=access_key" \
-e "MINIO_SECRET_KEY=secret_key" \
minio/minio gateway s3 --address https://s3_compatible_service_endpoint:port
minio/minio gateway s3 https://s3_compatible_service_endpoint:port
```
### Using Binary
@@ -41,7 +41,7 @@ docker run -p 9000:9000 --name minio-s3 \
```
export MINIO_ACCESS_KEY=access_key
export MINIO_SECRET_KEY=secret_key
minio gateway s3 --address https://s3_compatible_service_endpoint:port
minio gateway s3 https://s3_compatible_service_endpoint:port
```
## Minio Caching
@@ -5,7 +5,7 @@ version: '2'
# 9001 through 9004.
services:
minio1:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- data1:/data
ports:
@@ -15,7 +15,7 @@ services:
MINIO_SECRET_KEY: minio123
command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data
minio2:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- data2:/data
ports:
@@ -25,7 +25,7 @@ services:
MINIO_SECRET_KEY: minio123
command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data
minio3:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- data3:/data
ports:
@@ -35,7 +35,7 @@ services:
MINIO_SECRET_KEY: minio123
command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data
minio4:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- data4:/data
ports:
@@ -2,7 +2,7 @@ version: '3.1'
services:
minio1:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio1-data:/export
ports:
@@ -20,7 +20,7 @@ services:
- access_key
minio2:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio2-data:/export
ports:
@@ -38,7 +38,7 @@ services:
- access_key
minio3:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio3-data:/export
ports:
@@ -56,7 +56,7 @@ services:
- access_key
minio4:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio4-data:/export
ports:
@@ -2,7 +2,7 @@ version: '3'
services:
minio1:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio1-data:/export
ports:
@@ -20,7 +20,7 @@ services:
command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
minio2:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio2-data:/export
ports:
@@ -38,7 +38,7 @@ services:
command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
minio3:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio3-data:/export
ports:
@@ -56,7 +56,7 @@ services:
command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
minio4:
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
volumes:
- minio4-data:/export
ports:
+3 -3
View File
@@ -124,7 +124,7 @@ spec:
- name: data
mountPath: "/data"
# Pulls the lastest Minio image from Docker Hub
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
args:
- server
- /data
@@ -303,7 +303,7 @@ spec:
value: "minio"
- name: MINIO_SECRET_KEY
value: "minio123"
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
args:
- server
- http://minio-0.minio.default.svc.cluster.local/data
@@ -514,7 +514,7 @@ spec:
containers:
- name: minio
# Pulls the default Minio image from Docker Hub
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
args:
- gateway
- gcs
@@ -21,7 +21,7 @@ spec:
value: "minio"
- name: MINIO_SECRET_KEY
value: "minio123"
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
args:
- server
- http://minio-0.minio.default.svc.cluster.local/data
@@ -21,7 +21,7 @@ spec:
containers:
- name: minio
# Pulls the default Minio image from Docker Hub
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
args:
- gateway
- gcs
@@ -29,7 +29,7 @@ spec:
- name: data
mountPath: "/data"
# Pulls the lastest Minio image from Docker Hub
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
args:
- server
- /data
+1 -1
View File
@@ -30,7 +30,7 @@ Above command deploys Minio on the Kubernetes cluster in the default configurati
| Parameter | Description | Default |
|----------------------------|-------------------------------------|---------------------------------------------------------|
| `image` | Minio image name | `minio/minio` |
| `imageTag` | Minio image tag. Possible values listed [here](https://hub.docker.com/r/minio/minio/tags/).| `RELEASE.2018-08-18T03-49-57Z`|
| `imageTag` | Minio image tag. Possible values listed [here](https://hub.docker.com/r/minio/minio/tags/).| `RELEASE.2018-08-25T01-56-38Z`|
| `imagePullPolicy` | Image pull policy | `Always` |
| `mode` | Minio server mode (`standalone`, `shared` or `distributed`)| `standalone` |
| `numberOfNodes` | Number of nodes (applicable only for Minio distributed mode). Should be 4 <= x <= 16 | `4` |
+1 -1
View File
@@ -54,7 +54,7 @@ spec:
value: "minio"
- name: MINIO_SECRET_KEY
value: "minio123"
image: minio/minio:RELEASE.2018-08-18T03-49-57Z
image: minio/minio:RELEASE.2018-08-25T01-56-38Z
imagePullPolicy: IfNotPresent
args: ["server", "http://minio-0.minio.default.svc.cluster.local/data", "http://minio-1.minio.default.svc.cluster.local/data", "http://minio-2.minio.default.svc.cluster.local/data", "http://minio-3.minio.default.svc.cluster.local/data"]
ports:
Binary file not shown.

Before

Width:  |  Height:  |  Size: 127 KiB

After

Width:  |  Height:  |  Size: 111 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 121 KiB

After

Width:  |  Height:  |  Size: 106 KiB

+6 -5
View File
@@ -48,13 +48,13 @@ func (reader *Input) runSelectParser(selectExpression string, myRow chan *Row) {
// ParseSelect parses the SELECT expression, and effectively tokenizes it into
// its separate parts. It returns the requested column names,alias,limit of
// records, and the where clause.
func (reader *Input) ParseSelect(sqlInput string) ([]string, string, int, interface{}, []string, *SelectFuncs, error) {
func (reader *Input) ParseSelect(sqlInput string) ([]string, string, int64, interface{}, []string, *SelectFuncs, error) {
// return columnNames, alias, limitOfRecords, whereclause,coalStore, nil
stmt, err := sqlparser.Parse(sqlInput)
var whereClause interface{}
var alias string
var limit int
var limit int64
myFuncs := &SelectFuncs{}
// TODO Maybe can parse their errors a bit to return some more of the s3 errors
if err != nil {
@@ -137,7 +137,8 @@ func (reader *Input) ParseSelect(sqlInput string) ([]string, string, int, interf
switch expr := stmt.Limit.Rowcount.(type) {
case *sqlparser.SQLVal:
// The Value of how many rows we're going to limit by
limit, _ = strconv.Atoi(string(expr.Val[:]))
parsedLimit, _ := strconv.Atoi(string(expr.Val[:]))
limit = int64(parsedLimit)
}
}
if stmt.GroupBy != nil {
@@ -157,7 +158,7 @@ func (reader *Input) ParseSelect(sqlInput string) ([]string, string, int, interf
// This is the main function, It goes row by row and for records which validate
// the where clause it currently prints the appropriate row given the requested
// columns.
func (reader *Input) processSelectReq(reqColNames []string, alias string, whereClause interface{}, limitOfRecords int, functionNames []string, myRow chan *Row, myFunc *SelectFuncs) {
func (reader *Input) processSelectReq(reqColNames []string, alias string, whereClause interface{}, limitOfRecords int64, functionNames []string, myRow chan *Row, myFunc *SelectFuncs) {
counter := -1
filtrCount := 0
functionFlag := false
@@ -203,7 +204,7 @@ func (reader *Input) processSelectReq(reqColNames []string, alias string, whereC
}
// When we have reached our limit, on what the user specified as the number
// of rows they wanted, we terminate our interpreter.
if filtrCount == limitOfRecords && limitOfRecords != 0 {
if int64(filtrCount) == limitOfRecords && limitOfRecords != 0 {
close(myRow)
return
}
+1 -1
View File
@@ -296,7 +296,7 @@ func TestMyParser(t *testing.T) {
if alias != table.alias {
t.Error()
}
if myLimit != table.myLimit {
if myLimit != int64(table.myLimit) {
t.Error()
}
if !reflect.DeepEqual(table.aggFuncs, aggFunctionNames) {
+16 -17
View File
@@ -7,18 +7,14 @@
package highwayhash
import "golang.org/x/sys/cpu"
var (
useSSE4 = supportsSSE4()
useAVX2 = supportsAVX2()
useSSE4 = cpu.X86.HasSSE41
useAVX2 = cpu.X86.HasAVX2
useNEON = false
)
//go:noescape
func supportsSSE4() bool
//go:noescape
func supportsAVX2() bool
//go:noescape
func initializeSSE4(state *[16]uint64, key []byte)
@@ -38,31 +34,34 @@ func finalizeSSE4(out []byte, state *[16]uint64)
func finalizeAVX2(out []byte, state *[16]uint64)
func initialize(state *[16]uint64, key []byte) {
if useAVX2 {
switch {
case useAVX2:
initializeAVX2(state, key)
} else if useSSE4 {
case useSSE4:
initializeSSE4(state, key)
} else {
default:
initializeGeneric(state, key)
}
}
func update(state *[16]uint64, msg []byte) {
if useAVX2 {
switch {
case useAVX2:
updateAVX2(state, msg)
} else if useSSE4 {
case useSSE4:
updateSSE4(state, msg)
} else {
default:
updateGeneric(state, msg)
}
}
func finalize(out []byte, state *[16]uint64) {
if useAVX2 {
switch {
case useAVX2:
finalizeAVX2(out, state)
} else if useSSE4 {
case useSSE4:
finalizeSSE4(out, state)
} else {
default:
finalizeGeneric(out, state)
}
}
-5
View File
@@ -248,8 +248,3 @@ hash64:
MOVQ DX, 0(BX)
RET
// func supportsAVX2() bool
TEXT ·supportsAVX2(SB), 4, $0-1
MOVQ runtime·support_avx2(SB), AX
MOVB AX, ret+0(FP)
RET
+3 -4
View File
@@ -7,15 +7,14 @@
package highwayhash
import "golang.org/x/sys/cpu"
var (
useSSE4 = supportsSSE4()
useSSE4 = cpu.X86.HasSSE41
useAVX2 = false
useNEON = false
)
//go:noescape
func supportsSSE4() bool
//go:noescape
func initializeSSE4(state *[16]uint64, key []byte)
-9
View File
@@ -292,12 +292,3 @@ hash64:
MOVQ m10, DX
MOVQ DX, 0(BX)
RET
// func supportsSSE4() bool
TEXT ·supportsSSE4(SB), 4, $0-1
MOVL $1, AX
CPUID
SHRL $19, CX // Bit 19 indicates SSE4 support
ANDL $1, CX // CX != 0 if support SSE4
MOVB CX, ret+0(FP)
RET
+8 -4
View File
@@ -7,16 +7,20 @@ clone_folder: c:\projects\src\github.com\rjeczalik\notify
environment:
PATH: c:\projects\bin;%PATH%
GOPATH: c:\projects
NOTIFY_TIMEOUT: 5s
NOTIFY_TIMEOUT: 10s
GOVERSION: 1.10.3
install:
- rmdir c:\go /s /q
- appveyor DownloadFile https://storage.googleapis.com/golang/go%GOVERSION%.windows-amd64.zip
- 7z x go%GOVERSION%.windows-amd64.zip -y -oC:\ > NUL
- cd %APPVEYOR_BUILD_FOLDER%
- go version
- go get -v -t ./...
build_script:
- go tool vet -all .
- go build ./...
- go test -v -timeout 60s -race ./...
- go test -v -timeout 120s -race ./...
test: off
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright (c) 2014-2015 The Notify Authors. All rights reserved.
// Copyright (c) 2014-2018 The Notify Authors. All rights reserved.
// Use of this source code is governed by the MIT license that can be
// found in the LICENSE file.
@@ -6,4 +6,4 @@
package notify
var debugTag bool = true
var debugTag = true
+2 -2
View File
@@ -1,4 +1,4 @@
// Copyright (c) 2014-2015 The Notify Authors. All rights reserved.
// Copyright (c) 2014-2018 The Notify Authors. All rights reserved.
// Use of this source code is governed by the MIT license that can be
// found in the LICENSE file.
@@ -6,4 +6,4 @@
package notify
var debugTag bool = false
var debugTag = false
+8 -4
View File
@@ -48,7 +48,7 @@ var wg sync.WaitGroup // used to wait until the runloop starts
// started and is ready via the wg. It also serves purpose of a dummy source,
// thanks to it the runloop does not return as it also has at least one source
// registered.
var source = C.CFRunLoopSourceCreate(nil, 0, &C.CFRunLoopSourceContext{
var source = C.CFRunLoopSourceCreate(refZero, 0, &C.CFRunLoopSourceContext{
perform: (C.CFRunLoopPerformCallBack)(C.gosource),
})
@@ -90,6 +90,10 @@ func gostream(_, info uintptr, n C.size_t, paths, flags, ids uintptr) {
if n == 0 {
return
}
fn := streamFuncs.get(info)
if fn == nil {
return
}
ev := make([]FSEvent, 0, int(n))
for i := uintptr(0); i < uintptr(n); i++ {
switch flags := *(*uint32)(unsafe.Pointer((flags + i*offflag))); {
@@ -104,7 +108,7 @@ func gostream(_, info uintptr, n C.size_t, paths, flags, ids uintptr) {
}
}
streamFuncs.get(info)(ev)
fn(ev)
}
// StreamFunc is a callback called when stream receives file events.
@@ -162,8 +166,8 @@ func (s *stream) Start() error {
return nil
}
wg.Wait()
p := C.CFStringCreateWithCStringNoCopy(nil, C.CString(s.path), C.kCFStringEncodingUTF8, nil)
path := C.CFArrayCreate(nil, (*unsafe.Pointer)(unsafe.Pointer(&p)), 1, nil)
p := C.CFStringCreateWithCStringNoCopy(refZero, C.CString(s.path), C.kCFStringEncodingUTF8, refZero)
path := C.CFArrayCreate(refZero, (*unsafe.Pointer)(unsafe.Pointer(&p)), 1, nil)
ctx := C.FSEventStreamContext{}
ref := C.EventStreamCreate(&ctx, C.uintptr_t(s.info), path, C.FSEventStreamEventId(atomic.LoadUint64(&since)), latency, flags)
if ref == nilstream {
+14
View File
@@ -0,0 +1,14 @@
// Copyright (c) 2018 The Notify Authors. All rights reserved.
// Use of this source code is governed by the MIT license that can be
// found in the LICENSE file.
// +build darwin,!kqueue,cgo,!go1.11
package notify
/*
#include <CoreServices/CoreServices.h>
*/
import "C"
var refZero = (*C.struct___CFAllocator)(nil)
+9
View File
@@ -0,0 +1,9 @@
// Copyright (c) 2018 The Notify Authors. All rights reserved.
// Use of this source code is governed by the MIT license that can be
// found in the LICENSE file.
// +build darwin,!kqueue,go1.11
package notify
const refZero = 0
+15
View File
@@ -0,0 +1,15 @@
// Copyright (c) 2014-2018 The Notify Authors. All rights reserved.
// Use of this source code is governed by the MIT license that can be
// found in the LICENSE file.
// +build !darwin,!linux,!freebsd,!dragonfly,!netbsd,!openbsd,!windows
// +build !kqueue,!solaris
package notify
import "errors"
// newWatcher stub.
func newWatcher(chan<- EventInfo) watcher {
return watcherStub{errors.New("notify: not implemented")}
}
+73 -50
View File
@@ -1,4 +1,4 @@
// Copyright (c) 2014-2015 The Notify Authors. All rights reserved.
// Copyright (c) 2014-2018 The Notify Authors. All rights reserved.
// Use of this source code is governed by the MIT license that can be
// found in the LICENSE file.
@@ -22,7 +22,7 @@ import (
const readBufferSize = 4096
// Since all operations which go through the Windows completion routine are done
// asynchronously, filter may set one of the constants belor. They were defined
// asynchronously, filter may set one of the constants below. They were defined
// in order to distinguish whether current folder should be re-registered in
// ReadDirectoryChangesW function or some control operations need to be executed.
const (
@@ -109,8 +109,13 @@ func (g *grip) register(cph syscall.Handle) (err error) {
// buffer. Directory changes that occur between calls to this function are added
// to the buffer and then, returned with the next call.
func (g *grip) readDirChanges() error {
handle := syscall.Handle(atomic.LoadUintptr((*uintptr)(&g.handle)))
if handle == syscall.InvalidHandle {
return nil // Handle was closed.
}
return syscall.ReadDirectoryChanges(
g.handle,
handle,
&g.buffer[0],
uint32(unsafe.Sizeof(g.buffer)),
g.recursive,
@@ -220,12 +225,27 @@ func (wd *watched) updateGrip(idx int, cph syscall.Handle, reset bool,
// returned from the operating system kernel.
func (wd *watched) closeHandle() (err error) {
for _, g := range wd.digrip {
if g != nil && g.handle != syscall.InvalidHandle {
switch suberr := syscall.CloseHandle(g.handle); {
case suberr == nil:
g.handle = syscall.InvalidHandle
case err == nil:
err = suberr
if g == nil {
continue
}
for {
handle := syscall.Handle(atomic.LoadUintptr((*uintptr)(&g.handle)))
if handle == syscall.InvalidHandle {
break // Already closed.
}
e := syscall.CloseHandle(handle)
if e != nil && err == nil {
err = e
}
// Set invalid handle even when CloseHandle fails. This will leak
// the handle but, since we can't close it anyway, there won't be
// any difference.
if atomic.CompareAndSwapUintptr((*uintptr)(&g.handle),
(uintptr)(handle), (uintptr)(syscall.InvalidHandle)) {
break
}
}
}
@@ -272,50 +292,49 @@ func (r *readdcw) RecursiveWatch(path string, event Event) error {
// watch inserts a directory to the group of watched folders. If watched folder
// already exists, function tries to rewatch it with new filters(NOT VALID). Moreover,
// watch starts the main event loop goroutine when called for the first time.
func (r *readdcw) watch(path string, event Event, recursive bool) (err error) {
func (r *readdcw) watch(path string, event Event, recursive bool) error {
if event&^(All|fileNotifyChangeAll) != 0 {
return errors.New("notify: unknown event")
}
r.Lock()
wd, ok := r.m[path]
r.Unlock()
if !ok {
if err = r.lazyinit(); err != nil {
return
}
r.Lock()
defer r.Unlock()
if wd, ok = r.m[path]; ok {
dbgprint("watch: exists already")
return
}
if wd, err = newWatched(r.cph, uint32(event), recursive, path); err != nil {
return
}
r.m[path] = wd
dbgprint("watch: new watch added")
} else {
dbgprint("watch: exists already")
defer r.Unlock()
if wd, ok := r.m[path]; ok {
dbgprint("watch: already exists")
wd.filter &^= stateUnwatch
return nil
}
if err := r.lazyinit(); err != nil {
return err
}
wd, err := newWatched(r.cph, uint32(event), recursive, path)
if err != nil {
return err
}
r.m[path] = wd
dbgprint("watch: new watch added")
return nil
}
// lazyinit creates an I/O completion port and starts the main event processing
// loop. This method uses Double-Checked Locking optimization.
// lazyinit creates an I/O completion port and starts the main event loop.
func (r *readdcw) lazyinit() (err error) {
invalid := uintptr(syscall.InvalidHandle)
if atomic.LoadUintptr((*uintptr)(&r.cph)) == invalid {
r.Lock()
defer r.Unlock()
if atomic.LoadUintptr((*uintptr)(&r.cph)) == invalid {
cph := syscall.InvalidHandle
if cph, err = syscall.CreateIoCompletionPort(cph, 0, 0, 0); err != nil {
return
}
r.cph, r.start = cph, true
go r.loop()
cph := syscall.InvalidHandle
if cph, err = syscall.CreateIoCompletionPort(cph, 0, 0, 0); err != nil {
return
}
r.cph, r.start = cph, true
go r.loop()
}
return
}
@@ -364,6 +383,7 @@ func (r *readdcw) loopstate(overEx *overlappedEx) {
overEx.parent.parent.recreate(r.cph)
case stateUnwatch:
dbgprint("loopstate unwatch")
overEx.parent.parent.closeHandle()
delete(r.m, syscall.UTF16ToString(overEx.parent.pathw))
case stateCPClose:
default:
@@ -495,27 +515,30 @@ func (r *readdcw) RecursiveUnwatch(path string) error {
// TODO : pknap
func (r *readdcw) unwatch(path string) (err error) {
var wd *watched
r.Lock()
defer r.Unlock()
if wd, err = r.nonStateWatchedLocked(path); err != nil {
return
}
wd.filter |= stateUnwatch
if err = wd.closeHandle(); err != nil {
wd.filter &^= stateUnwatch
return
}
dbgprint("unwatch: set unwatch state")
if _, attrErr := syscall.GetFileAttributes(&wd.pathw[0]); attrErr != nil {
for _, g := range wd.digrip {
if g != nil {
dbgprint("unwatch: posting")
if err = syscall.PostQueuedCompletionStatus(r.cph, 0, 0, (*syscall.Overlapped)(unsafe.Pointer(g.ovlapped))); err != nil {
wd.filter &^= stateUnwatch
return
}
if g == nil {
continue
}
dbgprint("unwatch: posting")
if err = syscall.PostQueuedCompletionStatus(r.cph, 0, 0, (*syscall.Overlapped)(unsafe.Pointer(g.ovlapped))); err != nil {
wd.filter &^= stateUnwatch
return
}
}
}
return
}
+6 -16
View File
@@ -1,23 +1,13 @@
// Copyright (c) 2014-2015 The Notify Authors. All rights reserved.
// Copyright (c) 2014-2018 The Notify Authors. All rights reserved.
// Use of this source code is governed by the MIT license that can be
// found in the LICENSE file.
// +build !darwin,!linux,!freebsd,!dragonfly,!netbsd,!openbsd,!windows
// +build !kqueue,!solaris
package notify
import "errors"
type stub struct{ error }
// newWatcher stub.
func newWatcher(chan<- EventInfo) watcher {
return stub{errors.New("notify: not implemented")}
}
type watcherStub struct{ error }
// Following methods implement notify.watcher interface.
func (s stub) Watch(string, Event) error { return s }
func (s stub) Rewatch(string, Event, Event) error { return s }
func (s stub) Unwatch(string) (err error) { return s }
func (s stub) Close() error { return s }
func (s watcherStub) Watch(string, Event) error { return s }
func (s watcherStub) Rewatch(string, Event, Event) error { return s }
func (s watcherStub) Unwatch(string) (err error) { return s }
func (s watcherStub) Close() error { return s }
+2 -1
View File
@@ -106,7 +106,8 @@ func newWatcher(c chan<- EventInfo) watcher {
}
t.t = newTrigger(t.pthLkp)
if err := t.t.Init(); err != nil {
panic(err)
t.Close()
return watcherStub{fmt.Errorf("failed setting up watcher: %v", err)}
}
go t.monitor()
return t
+6 -6
View File
@@ -628,10 +628,10 @@
"revisionTime": "2018-01-23T12:12:34Z"
},
{
"checksumSHA1": "w7ykRf31om2J6832Py4Kz/PEnSI=",
"checksumSHA1": "2Fu1GmLwDo6FFdahjnlWnPkwJTE=",
"path": "github.com/minio/highwayhash",
"revision": "1ea1b5fce73b1d7abbd822a194b51f3736b1f23b",
"revisionTime": "2018-01-09T22:40:08Z"
"revision": "85fc8a2dacad36a6beb2865793cd81363a496696",
"revisionTime": "2018-05-01T08:09:13Z"
},
{
"checksumSHA1": "7/Hdd23/j4/yt4BXa+h0kqz1yjw=",
@@ -830,10 +830,10 @@
"revisionTime": "2018-04-08T09:29:02Z"
},
{
"checksumSHA1": "T2BgxGcsgrldYNFPQcmw46/K4qM=",
"checksumSHA1": "D8AVDI39CJ+jvw0HOotYU2gz54c=",
"path": "github.com/rjeczalik/notify",
"revision": "d152f3ce359a5464dc41e84a8919fc67e55bbbf0",
"revisionTime": "2018-03-12T21:30:58Z"
"revision": "4e54e7fd043e865c50bda93359fb78813a8d165b",
"revisionTime": "2018-08-08T20:39:25Z"
},
{
"path": "github.com/rs/cors",