From 78b372816935c663f7f56b92e0e79e37e99cd4d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=B3=E6=98=8C=E6=9E=97?= Date: Wed, 15 Jun 2022 21:07:55 +0800 Subject: [PATCH] add s3 circuit breaker support for 'simultaneous request count' and 'simultaneous request bytes' limitations configure s3 circuit breaker by 'command_s3_circuitbreaker.go': usage eg: # Configure the number of simultaneous global (current s3api node) requests s3.circuit.breaker -global -type count -actions Write -values 1000 -apply # Configure the number of simultaneous requests for bucket x read and write s3.circuit.breaker -buckets -type count -actions Read,Write -values 1000 -apply # Configure the total bytes of simultaneous requests for bucket write s3.circuit.breaker -buckets -type bytes -actions Write -values 100MiB -apply # Disable circuit breaker config of bucket 'x' s3.circuit.breaker -buckets x -enable false -apply # Delete circuit breaker config of bucket 'x' s3.circuit.breaker -buckets x -delete -apply --- go.mod | 3 +- go.sum | 3 + weed/config/s3_config.go | 16 + weed/filer/s3iam_conf.go | 3 +- weed/pb/s3.proto | 10 + weed/pb/s3_pb/s3.pb.go | 211 ++++++++++-- weed/s3api/auth_credentials_subscribe.go | 35 +- weed/s3api/s3api_circuit_breaker.go | 174 ++++++++++ weed/s3api/s3api_server.go | 83 ++--- weed/s3api/s3err/s3api_errors.go | 13 + weed/shell/command_s3_circuitbreaker.go | 335 +++++++++++++++++++ weed/shell/command_s3_circuitbreaker_test.go | 7 + 12 files changed, 820 insertions(+), 73 deletions(-) create mode 100644 weed/config/s3_config.go create mode 100644 weed/s3api/s3api_circuit_breaker.go create mode 100644 weed/shell/command_s3_circuitbreaker.go create mode 100644 weed/shell/command_s3_circuitbreaker_test.go diff --git a/go.mod b/go.mod index 2fbff4d37..455f07e41 100644 --- a/go.mod +++ b/go.mod @@ -149,6 +149,7 @@ require ( require ( github.com/Jille/raft-grpc-transport v1.2.0 + github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d github.com/arangodb/go-driver v1.3.2 github.com/fluent/fluent-logger-golang v1.9.0 github.com/hanwen/go-fuse/v2 v2.1.0 @@ -156,6 +157,7 @@ require ( github.com/hashicorp/raft-boltdb v0.0.0-20220329195025-15018e9b97e0 github.com/ydb-platform/ydb-go-sdk-auth-environ v0.1.2 github.com/ydb-platform/ydb-go-sdk/v3 v3.25.3 + go.uber.org/atomic v1.9.0 ) require ( @@ -211,7 +213,6 @@ require ( github.com/ydb-platform/ydb-go-yc-metadata v0.5.2 // indirect go.etcd.io/etcd/api/v3 v3.5.4 // indirect go.etcd.io/etcd/client/pkg/v3 v3.5.4 // indirect - go.uber.org/atomic v1.9.0 // indirect go.uber.org/multierr v1.8.0 // indirect go.uber.org/zap v1.21.0 // indirect golang.org/x/mod v0.6.0-dev.0.20211013180041-c96bc1413d57 // indirect diff --git a/go.sum b/go.sum index 6fc6a2787..6ab009736 100644 --- a/go.sum +++ b/go.sum @@ -136,6 +136,7 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= +github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E= github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY= github.com/arangodb/go-driver v1.3.2 h1:07cmMqPqEl+/MlosjFtVVakEbkPqBSUvw9S9/atX0+4= @@ -1513,6 +1514,8 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.0-20220512140231-539c8e751b99/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/weed/config/s3_config.go b/weed/config/s3_config.go new file mode 100644 index 000000000..be504fb9f --- /dev/null +++ b/weed/config/s3_config.go @@ -0,0 +1,16 @@ +package config + +import "strings" + +var ( + CircuitBreakerConfigDir = "/etc/s3" + CircuitBreakerConfigFile = "circuit_breaker.json" + AllowedActions = []string{"Read", "Write", "List", "Tagging", "Admin"} + LimitTypeCount = "count" + LimitTypeBytes = "bytes" + Separator = ":" +) + +func Concat(elements ...string) string { + return strings.Join(elements, Separator) +} diff --git a/weed/filer/s3iam_conf.go b/weed/filer/s3iam_conf.go index 55c976915..891bf925b 100644 --- a/weed/filer/s3iam_conf.go +++ b/weed/filer/s3iam_conf.go @@ -2,13 +2,12 @@ package filer import ( "bytes" - "github.com/chrislusf/seaweedfs/weed/pb/iam_pb" "github.com/golang/protobuf/jsonpb" "github.com/golang/protobuf/proto" "io" ) -func ParseS3ConfigurationFromBytes(content []byte, config *iam_pb.S3ApiConfiguration) error { +func ParseS3ConfigurationFromBytes[T proto.Message](content []byte, config T) error { if err := jsonpb.Unmarshal(bytes.NewBuffer(content), config); err != nil { return err } diff --git a/weed/pb/s3.proto b/weed/pb/s3.proto index 4f129b817..b08631ae7 100644 --- a/weed/pb/s3.proto +++ b/weed/pb/s3.proto @@ -23,3 +23,13 @@ message S3ConfigureRequest { message S3ConfigureResponse { } + +message S3CircuitBreakerConfig { + CbOptions global=1; + map buckets= 2; +} + +message CbOptions { + bool enabled=1; + map actions = 2; +} diff --git a/weed/pb/s3_pb/s3.pb.go b/weed/pb/s3_pb/s3.pb.go index 53f174f02..0cac7d828 100644 --- a/weed/pb/s3_pb/s3.pb.go +++ b/weed/pb/s3_pb/s3.pb.go @@ -105,6 +105,116 @@ func (*S3ConfigureResponse) Descriptor() ([]byte, []int) { return file_s3_proto_rawDescGZIP(), []int{1} } +type S3CircuitBreakerConfig struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Global *CbOptions `protobuf:"bytes,1,opt,name=global,proto3" json:"global,omitempty"` + Buckets map[string]*CbOptions `protobuf:"bytes,2,rep,name=buckets,proto3" json:"buckets,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` +} + +func (x *S3CircuitBreakerConfig) Reset() { + *x = S3CircuitBreakerConfig{} + if protoimpl.UnsafeEnabled { + mi := &file_s3_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *S3CircuitBreakerConfig) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*S3CircuitBreakerConfig) ProtoMessage() {} + +func (x *S3CircuitBreakerConfig) ProtoReflect() protoreflect.Message { + mi := &file_s3_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use S3CircuitBreakerConfig.ProtoReflect.Descriptor instead. +func (*S3CircuitBreakerConfig) Descriptor() ([]byte, []int) { + return file_s3_proto_rawDescGZIP(), []int{2} +} + +func (x *S3CircuitBreakerConfig) GetGlobal() *CbOptions { + if x != nil { + return x.Global + } + return nil +} + +func (x *S3CircuitBreakerConfig) GetBuckets() map[string]*CbOptions { + if x != nil { + return x.Buckets + } + return nil +} + +type CbOptions struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Enabled bool `protobuf:"varint,1,opt,name=enabled,proto3" json:"enabled,omitempty"` + Actions map[string]int64 `protobuf:"bytes,2,rep,name=actions,proto3" json:"actions,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"varint,2,opt,name=value,proto3"` +} + +func (x *CbOptions) Reset() { + *x = CbOptions{} + if protoimpl.UnsafeEnabled { + mi := &file_s3_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *CbOptions) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CbOptions) ProtoMessage() {} + +func (x *CbOptions) ProtoReflect() protoreflect.Message { + mi := &file_s3_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CbOptions.ProtoReflect.Descriptor instead. +func (*CbOptions) Descriptor() ([]byte, []int) { + return file_s3_proto_rawDescGZIP(), []int{3} +} + +func (x *CbOptions) GetEnabled() bool { + if x != nil { + return x.Enabled + } + return false +} + +func (x *CbOptions) GetActions() map[string]int64 { + if x != nil { + return x.Actions + } + return nil +} + var File_s3_proto protoreflect.FileDescriptor var file_s3_proto_rawDesc = []byte{ @@ -116,18 +226,43 @@ var file_s3_proto_rawDesc = []byte{ 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x1a, 0x73, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x46, 0x69, 0x6c, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x22, 0x15, 0x0a, 0x13, 0x53, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, - 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x32, 0x5f, 0x0a, 0x09, 0x53, 0x65, 0x61, 0x77, - 0x65, 0x65, 0x64, 0x53, 0x33, 0x12, 0x52, 0x0a, 0x09, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, - 0x72, 0x65, 0x12, 0x20, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, - 0x62, 0x2e, 0x53, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x52, 0x65, 0x71, - 0x75, 0x65, 0x73, 0x74, 0x1a, 0x21, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, + 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0xeb, 0x01, 0x0a, 0x16, 0x53, 0x33, 0x43, + 0x69, 0x72, 0x63, 0x75, 0x69, 0x74, 0x42, 0x72, 0x65, 0x61, 0x6b, 0x65, 0x72, 0x43, 0x6f, 0x6e, + 0x66, 0x69, 0x67, 0x12, 0x2f, 0x0a, 0x06, 0x67, 0x6c, 0x6f, 0x62, 0x61, 0x6c, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, + 0x70, 0x62, 0x2e, 0x43, 0x62, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x06, 0x67, 0x6c, + 0x6f, 0x62, 0x61, 0x6c, 0x12, 0x4b, 0x0a, 0x07, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x18, + 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x31, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, + 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x33, 0x43, 0x69, 0x72, 0x63, 0x75, 0x69, 0x74, 0x42, 0x72, + 0x65, 0x61, 0x6b, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x2e, 0x42, 0x75, 0x63, 0x6b, + 0x65, 0x74, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x07, 0x62, 0x75, 0x63, 0x6b, 0x65, 0x74, + 0x73, 0x1a, 0x53, 0x0a, 0x0c, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74, 0x73, 0x45, 0x6e, 0x74, 0x72, + 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, + 0x6b, 0x65, 0x79, 0x12, 0x2d, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, + 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, + 0x62, 0x2e, 0x43, 0x62, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x05, 0x76, 0x61, 0x6c, + 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xa1, 0x01, 0x0a, 0x09, 0x43, 0x62, 0x4f, 0x70, 0x74, + 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x18, 0x0a, 0x07, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x08, 0x52, 0x07, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64, 0x12, 0x3e, + 0x0a, 0x07, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, + 0x24, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x43, + 0x62, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, + 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x07, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3a, + 0x0a, 0x0c, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, + 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, + 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, + 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x32, 0x5f, 0x0a, 0x09, 0x53, 0x65, + 0x61, 0x77, 0x65, 0x65, 0x64, 0x53, 0x33, 0x12, 0x52, 0x0a, 0x09, 0x43, 0x6f, 0x6e, 0x66, 0x69, + 0x67, 0x75, 0x72, 0x65, 0x12, 0x20, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x52, - 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x49, 0x0a, 0x10, 0x73, 0x65, 0x61, - 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, 0x07, 0x53, - 0x33, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, - 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, 0x65, 0x61, - 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, 0x2f, 0x73, - 0x33, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x21, 0x2e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x69, + 0x6e, 0x67, 0x5f, 0x70, 0x62, 0x2e, 0x53, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, + 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x49, 0x0a, 0x10, 0x73, + 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2e, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x42, + 0x07, 0x53, 0x33, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x5a, 0x2c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, + 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x68, 0x72, 0x69, 0x73, 0x6c, 0x75, 0x73, 0x66, 0x2f, 0x73, + 0x65, 0x61, 0x77, 0x65, 0x65, 0x64, 0x66, 0x73, 0x2f, 0x77, 0x65, 0x65, 0x64, 0x2f, 0x70, 0x62, + 0x2f, 0x73, 0x33, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -142,19 +277,27 @@ func file_s3_proto_rawDescGZIP() []byte { return file_s3_proto_rawDescData } -var file_s3_proto_msgTypes = make([]protoimpl.MessageInfo, 2) +var file_s3_proto_msgTypes = make([]protoimpl.MessageInfo, 6) var file_s3_proto_goTypes = []interface{}{ - (*S3ConfigureRequest)(nil), // 0: messaging_pb.S3ConfigureRequest - (*S3ConfigureResponse)(nil), // 1: messaging_pb.S3ConfigureResponse + (*S3ConfigureRequest)(nil), // 0: messaging_pb.S3ConfigureRequest + (*S3ConfigureResponse)(nil), // 1: messaging_pb.S3ConfigureResponse + (*S3CircuitBreakerConfig)(nil), // 2: messaging_pb.S3CircuitBreakerConfig + (*CbOptions)(nil), // 3: messaging_pb.CbOptions + nil, // 4: messaging_pb.S3CircuitBreakerConfig.BucketsEntry + nil, // 5: messaging_pb.CbOptions.ActionsEntry } var file_s3_proto_depIdxs = []int32{ - 0, // 0: messaging_pb.SeaweedS3.Configure:input_type -> messaging_pb.S3ConfigureRequest - 1, // 1: messaging_pb.SeaweedS3.Configure:output_type -> messaging_pb.S3ConfigureResponse - 1, // [1:2] is the sub-list for method output_type - 0, // [0:1] is the sub-list for method input_type - 0, // [0:0] is the sub-list for extension type_name - 0, // [0:0] is the sub-list for extension extendee - 0, // [0:0] is the sub-list for field type_name + 3, // 0: messaging_pb.S3CircuitBreakerConfig.global:type_name -> messaging_pb.CbOptions + 4, // 1: messaging_pb.S3CircuitBreakerConfig.buckets:type_name -> messaging_pb.S3CircuitBreakerConfig.BucketsEntry + 5, // 2: messaging_pb.CbOptions.actions:type_name -> messaging_pb.CbOptions.ActionsEntry + 3, // 3: messaging_pb.S3CircuitBreakerConfig.BucketsEntry.value:type_name -> messaging_pb.CbOptions + 0, // 4: messaging_pb.SeaweedS3.Configure:input_type -> messaging_pb.S3ConfigureRequest + 1, // 5: messaging_pb.SeaweedS3.Configure:output_type -> messaging_pb.S3ConfigureResponse + 5, // [5:6] is the sub-list for method output_type + 4, // [4:5] is the sub-list for method input_type + 4, // [4:4] is the sub-list for extension type_name + 4, // [4:4] is the sub-list for extension extendee + 0, // [0:4] is the sub-list for field type_name } func init() { file_s3_proto_init() } @@ -187,6 +330,30 @@ func file_s3_proto_init() { return nil } } + file_s3_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*S3CircuitBreakerConfig); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_s3_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*CbOptions); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } } type x struct{} out := protoimpl.TypeBuilder{ @@ -194,7 +361,7 @@ func file_s3_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_s3_proto_rawDesc, NumEnums: 0, - NumMessages: 2, + NumMessages: 6, NumExtensions: 0, NumServices: 1, }, diff --git a/weed/s3api/auth_credentials_subscribe.go b/weed/s3api/auth_credentials_subscribe.go index 91fd5d830..f8e7e7dd0 100644 --- a/weed/s3api/auth_credentials_subscribe.go +++ b/weed/s3api/auth_credentials_subscribe.go @@ -1,6 +1,7 @@ package s3api import ( + "github.com/chrislusf/seaweedfs/weed/config" "github.com/chrislusf/seaweedfs/weed/filer" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/pb" @@ -22,12 +23,11 @@ func (s3a *S3ApiServer) subscribeMetaEvents(clientName string, prefix string, la if message.NewParentPath != "" { dir = message.NewParentPath } - if dir == filer.IamConfigDirecotry && message.NewEntry.Name == filer.IamIdentityFile { - if err := s3a.iam.LoadS3ApiConfigurationFromBytes(message.NewEntry.Content); err != nil { - return err - } - glog.V(0).Infof("updated %s/%s", filer.IamConfigDirecotry, filer.IamIdentityFile) - } + fileName := message.NewEntry.Name + content := message.NewEntry.Content + + _ = s3a.onIamConfigUpdate(dir, fileName, content) + _ = s3a.onCbConfigUpdate(dir, fileName, content) return nil } @@ -38,5 +38,26 @@ func (s3a *S3ApiServer) subscribeMetaEvents(clientName string, prefix string, la glog.V(0).Infof("iam follow metadata changes: %v", err) return true }) - +} + +//reload iam config +func (s3a *S3ApiServer) onIamConfigUpdate(dir, filename string, content []byte) error { + if dir == filer.IamConfigDirecotry && filename == filer.IamIdentityFile { + if err := s3a.iam.LoadS3ApiConfigurationFromBytes(content); err != nil { + return err + } + glog.V(0).Infof("updated %s/%s", dir, filename) + } + return nil +} + +//reload circuit breaker config +func (s3a *S3ApiServer) onCbConfigUpdate(dir, filename string, content []byte) error { + if dir == config.CircuitBreakerConfigDir && filename == config.CircuitBreakerConfigFile { + if err := s3a.cb.LoadS3ApiConfigurationFromBytes(content); err != nil { + return err + } + glog.V(0).Infof("updated %s/%s", dir, filename) + } + return nil } diff --git a/weed/s3api/s3api_circuit_breaker.go b/weed/s3api/s3api_circuit_breaker.go new file mode 100644 index 000000000..abcca0f5c --- /dev/null +++ b/weed/s3api/s3api_circuit_breaker.go @@ -0,0 +1,174 @@ +package s3api + +import ( + "fmt" + "github.com/chrislusf/seaweedfs/weed/config" + "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/pb" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/pb/s3_pb" + "github.com/chrislusf/seaweedfs/weed/s3api/s3err" + "github.com/gorilla/mux" + "go.uber.org/atomic" + "net/http" +) + +type CircuitBreaker struct { + Enabled bool + counters map[string]*atomic.Int64 + limitations map[string]int64 +} + +func NewCircuitBreaker(option *S3ApiServerOption) *CircuitBreaker { + cb := &CircuitBreaker{ + counters: make(map[string]*atomic.Int64), + limitations: make(map[string]int64), + } + + _ = pb.WithFilerClient(false, option.Filer, option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + content, err := filer.ReadInsideFiler(client, config.CircuitBreakerConfigDir, config.CircuitBreakerConfigFile) + if err == nil { + err = cb.LoadS3ApiConfigurationFromBytes(content) + } + if err != nil { + glog.Warningf("load s3 circuit breaker config from filer: %v", err) + } else { + glog.V(2).Infof("load s3 circuit breaker config complete: %v", cb) + } + return err + }) + return cb +} + +func (cb *CircuitBreaker) LoadS3ApiConfigurationFromBytes(content []byte) error { + cbCfg := &s3_pb.S3CircuitBreakerConfig{} + if err := filer.ParseS3ConfigurationFromBytes(content, cbCfg); err != nil { + glog.Warningf("unmarshal error: %v", err) + return fmt.Errorf("unmarshal error: %v", err) + } + if err := cb.loadCbCfg(cbCfg); err != nil { + return err + } + return nil +} + +func (cb *CircuitBreaker) loadCbCfg(cfg *s3_pb.S3CircuitBreakerConfig) error { + + //global + globalEnabled := false + globalOptions := cfg.Global + limitations := make(map[string]int64) + if globalOptions != nil && globalOptions.Enabled && len(globalOptions.Actions) > 0 { + globalEnabled = globalOptions.Enabled + for action, limit := range globalOptions.Actions { + limitations[action] = limit + } + } + cb.Enabled = globalEnabled + + //buckets + for bucket, cbOptions := range cfg.Buckets { + if cbOptions.Enabled { + for action, limit := range cbOptions.Actions { + limitations[config.Concat(bucket, action)] = limit + } + } + } + + cb.limitations = limitations + return nil +} + +func (cb *CircuitBreaker) Check(f func(w http.ResponseWriter, r *http.Request), action string) (http.HandlerFunc, Action) { + return func(w http.ResponseWriter, r *http.Request) { + if !cb.Enabled { + f(w, r) + return + } + + vars := mux.Vars(r) + bucket := vars["bucket"] + + rollback, errCode := cb.check(r, bucket, action) + defer func() { + for _, rf := range rollback { + rf() + } + }() + + if errCode == s3err.ErrNone { + f(w, r) + return + } + s3err.WriteErrorResponse(w, r, errCode) + }, Action(action) +} + +func (cb *CircuitBreaker) check(r *http.Request, bucket string, action string) (rollback []func(), errCode s3err.ErrorCode) { + + //bucket simultaneous request count + bucketCountRollBack, errCode := cb.loadAndCompare(bucket, action, config.LimitTypeCount, 1, s3err.ErrTooManyRequest) + if bucketCountRollBack != nil { + rollback = append(rollback, bucketCountRollBack) + } + if errCode != s3err.ErrNone { + return + } + + //bucket simultaneous request content bytes + bucketContentLengthRollBack, errCode := cb.loadAndCompare(bucket, action, config.LimitTypeBytes, r.ContentLength, s3err.ErrRequestBytesExceed) + if bucketContentLengthRollBack != nil { + rollback = append(rollback, bucketContentLengthRollBack) + } + if errCode != s3err.ErrNone { + return + } + + //global simultaneous request count + globalCountRollBack, errCode := cb.loadAndCompare("", action, config.LimitTypeCount, 1, s3err.ErrTooManyRequest) + if globalCountRollBack != nil { + rollback = append(rollback, globalCountRollBack) + } + if errCode != s3err.ErrNone { + return + } + + //global simultaneous request content bytes + globalContentLengthRollBack, errCode := cb.loadAndCompare("", action, config.LimitTypeBytes, r.ContentLength, s3err.ErrRequestBytesExceed) + if globalContentLengthRollBack != nil { + rollback = append(rollback, globalContentLengthRollBack) + } + if errCode != s3err.ErrNone { + return + } + return +} + +func (cb CircuitBreaker) loadAndCompare(bucket, action, limitType string, inc int64, errCode s3err.ErrorCode) (f func(), e s3err.ErrorCode) { + key := config.Concat(bucket, action, limitType) + e = s3err.ErrNone + if max, ok := cb.limitations[key]; ok { + counter, exists := cb.counters[key] + if !exists { + counter = atomic.NewInt64(0) + cb.counters[key] = counter + } + current := counter.Load() + if current+inc > max { + e = errCode + return + } else { + counter.Add(inc) + f = func() { + counter.Sub(inc) + } + current = counter.Load() + if current+inc > max { + e = errCode + return + } + } + } + return +} diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 657fa8171..d439ef8f1 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -9,7 +9,6 @@ import ( "strings" "time" - "github.com/chrislusf/seaweedfs/weed/filer" "github.com/chrislusf/seaweedfs/weed/pb" . "github.com/chrislusf/seaweedfs/weed/s3api/s3_constants" "github.com/chrislusf/seaweedfs/weed/s3api/s3err" @@ -35,6 +34,7 @@ type S3ApiServer struct { s3_pb.UnimplementedSeaweedS3Server option *S3ApiServerOption iam *IdentityAccessManagement + cb *CircuitBreaker randomClientId int32 filerGuard *security.Guard client *http.Client @@ -55,6 +55,7 @@ func NewS3ApiServer(router *mux.Router, option *S3ApiServerOption) (s3ApiServer iam: NewIdentityAccessManagement(option), randomClientId: util.RandomInt32(), filerGuard: security.NewGuard([]string{}, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec), + cb: NewCircuitBreaker(option), } if option.LocalFilerSocket == nil || *option.LocalFilerSocket == "" { s3ApiServer.client = &http.Client{Transport: &http.Transport{ @@ -73,7 +74,7 @@ func NewS3ApiServer(router *mux.Router, option *S3ApiServerOption) (s3ApiServer s3ApiServer.registerRouter(router) - go s3ApiServer.subscribeMetaEvents("s3", filer.IamConfigDirecotry+"/"+filer.IamIdentityFile, time.Now().UnixNano()) + go s3ApiServer.subscribeMetaEvents("s3", "/etc", time.Now().UnixNano()) return s3ApiServer, nil } @@ -107,115 +108,115 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) { // objects with query // CopyObjectPart - bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", `.*?(\/|%2F).*?`).HandlerFunc(track(s3a.iam.Auth(s3a.CopyObjectPartHandler, ACTION_WRITE), "PUT")).Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}") + bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", `.*?(\/|%2F).*?`).HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.CopyObjectPartHandler, ACTION_WRITE)), "PUT")).Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}") // PutObjectPart - bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectPartHandler, ACTION_WRITE), "PUT")).Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}") + bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutObjectPartHandler, ACTION_WRITE)), "PUT")).Queries("partNumber", "{partNumber:[0-9]+}", "uploadId", "{uploadId:.*}") // CompleteMultipartUpload - bucket.Methods("POST").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.CompleteMultipartUploadHandler, ACTION_WRITE), "POST")).Queries("uploadId", "{uploadId:.*}") + bucket.Methods("POST").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.CompleteMultipartUploadHandler, ACTION_WRITE)), "POST")).Queries("uploadId", "{uploadId:.*}") // NewMultipartUpload - bucket.Methods("POST").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.NewMultipartUploadHandler, ACTION_WRITE), "POST")).Queries("uploads", "") + bucket.Methods("POST").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.NewMultipartUploadHandler, ACTION_WRITE)), "POST")).Queries("uploads", "") // AbortMultipartUpload - bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.AbortMultipartUploadHandler, ACTION_WRITE), "DELETE")).Queries("uploadId", "{uploadId:.*}") + bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.AbortMultipartUploadHandler, ACTION_WRITE)), "DELETE")).Queries("uploadId", "{uploadId:.*}") // ListObjectParts - bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.ListObjectPartsHandler, ACTION_READ), "GET")).Queries("uploadId", "{uploadId:.*}") + bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.ListObjectPartsHandler, ACTION_READ)), "GET")).Queries("uploadId", "{uploadId:.*}") // ListMultipartUploads - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListMultipartUploadsHandler, ACTION_READ), "GET")).Queries("uploads", "") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.ListMultipartUploadsHandler, ACTION_READ)), "GET")).Queries("uploads", "") // GetObjectTagging - bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.GetObjectTaggingHandler, ACTION_READ), "GET")).Queries("tagging", "") + bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetObjectTaggingHandler, ACTION_READ)), "GET")).Queries("tagging", "") // PutObjectTagging - bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectTaggingHandler, ACTION_TAGGING), "PUT")).Queries("tagging", "") + bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutObjectTaggingHandler, ACTION_TAGGING)), "PUT")).Queries("tagging", "") // DeleteObjectTagging - bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteObjectTaggingHandler, ACTION_TAGGING), "DELETE")).Queries("tagging", "") + bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.DeleteObjectTaggingHandler, ACTION_TAGGING)), "DELETE")).Queries("tagging", "") // PutObjectACL - bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectAclHandler, ACTION_WRITE), "PUT")).Queries("acl", "") + bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutObjectAclHandler, ACTION_WRITE)), "PUT")).Queries("acl", "") // PutObjectRetention - bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectRetentionHandler, ACTION_WRITE), "PUT")).Queries("retention", "") + bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutObjectRetentionHandler, ACTION_WRITE)), "PUT")).Queries("retention", "") // PutObjectLegalHold - bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectLegalHoldHandler, ACTION_WRITE), "PUT")).Queries("legal-hold", "") + bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutObjectLegalHoldHandler, ACTION_WRITE)), "PUT")).Queries("legal-hold", "") // PutObjectLockConfiguration - bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectLockConfigurationHandler, ACTION_WRITE), "PUT")).Queries("object-lock", "") + bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutObjectLockConfigurationHandler, ACTION_WRITE)), "PUT")).Queries("object-lock", "") // GetObjectACL - bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.GetObjectAclHandler, ACTION_READ), "GET")).Queries("acl", "") + bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetObjectAclHandler, ACTION_READ)), "GET")).Queries("acl", "") // objects with query // raw objects // HeadObject - bucket.Methods("HEAD").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.HeadObjectHandler, ACTION_READ), "GET")) + bucket.Methods("HEAD").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.HeadObjectHandler, ACTION_READ)), "GET")) // GetObject, but directory listing is not supported - bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.GetObjectHandler, ACTION_READ), "GET")) + bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetObjectHandler, ACTION_READ)), "GET")) // CopyObject - bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", ".*?(\\/|%2F).*?").HandlerFunc(track(s3a.iam.Auth(s3a.CopyObjectHandler, ACTION_WRITE), "COPY")) + bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", ".*?(\\/|%2F).*?").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.CopyObjectHandler, ACTION_WRITE)), "COPY")) // PutObject - bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectHandler, ACTION_WRITE), "PUT")) + bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutObjectHandler, ACTION_WRITE)), "PUT")) // DeleteObject - bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteObjectHandler, ACTION_WRITE), "DELETE")) + bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.DeleteObjectHandler, ACTION_WRITE)), "DELETE")) // raw objects // buckets with query // DeleteMultipleObjects - bucket.Methods("POST").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteMultipleObjectsHandler, ACTION_WRITE), "DELETE")).Queries("delete", "") + bucket.Methods("POST").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.DeleteMultipleObjectsHandler, ACTION_WRITE)), "DELETE")).Queries("delete", "") // GetBucketACL - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.GetBucketAclHandler, ACTION_READ), "GET")).Queries("acl", "") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetBucketAclHandler, ACTION_READ)), "GET")).Queries("acl", "") // PutBucketACL - bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.PutBucketAclHandler, ACTION_WRITE), "PUT")).Queries("acl", "") + bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutBucketAclHandler, ACTION_WRITE)), "PUT")).Queries("acl", "") // GetBucketPolicy - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.GetBucketPolicyHandler, ACTION_READ), "GET")).Queries("policy", "") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetBucketPolicyHandler, ACTION_READ)), "GET")).Queries("policy", "") // PutBucketPolicy - bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.PutBucketPolicyHandler, ACTION_WRITE), "PUT")).Queries("policy", "") + bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutBucketPolicyHandler, ACTION_WRITE)), "PUT")).Queries("policy", "") // DeleteBucketPolicy - bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteBucketPolicyHandler, ACTION_WRITE), "DELETE")).Queries("policy", "") + bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.DeleteBucketPolicyHandler, ACTION_WRITE)), "DELETE")).Queries("policy", "") // GetBucketCors - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.GetBucketCorsHandler, ACTION_READ), "GET")).Queries("cors", "") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetBucketCorsHandler, ACTION_READ)), "GET")).Queries("cors", "") // PutBucketCors - bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.PutBucketCorsHandler, ACTION_WRITE), "PUT")).Queries("cors", "") + bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutBucketCorsHandler, ACTION_WRITE)), "PUT")).Queries("cors", "") // DeleteBucketCors - bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteBucketCorsHandler, ACTION_WRITE), "DELETE")).Queries("cors", "") + bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.DeleteBucketCorsHandler, ACTION_WRITE)), "DELETE")).Queries("cors", "") // GetBucketLifecycleConfiguration - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.GetBucketLifecycleConfigurationHandler, ACTION_READ), "GET")).Queries("lifecycle", "") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetBucketLifecycleConfigurationHandler, ACTION_READ)), "GET")).Queries("lifecycle", "") // PutBucketLifecycleConfiguration - bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.PutBucketLifecycleConfigurationHandler, ACTION_WRITE), "PUT")).Queries("lifecycle", "") + bucket.Methods("PUT").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PutBucketLifecycleConfigurationHandler, ACTION_WRITE)), "PUT")).Queries("lifecycle", "") // DeleteBucketLifecycleConfiguration - bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteBucketLifecycleHandler, ACTION_WRITE), "DELETE")).Queries("lifecycle", "") + bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.DeleteBucketLifecycleHandler, ACTION_WRITE)), "DELETE")).Queries("lifecycle", "") // GetBucketLocation - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.GetBucketLocationHandler, ACTION_READ), "GET")).Queries("location", "") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetBucketLocationHandler, ACTION_READ)), "GET")).Queries("location", "") // GetBucketRequestPayment - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.GetBucketRequestPaymentHandler, ACTION_READ), "GET")).Queries("requestPayment", "") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.GetBucketRequestPaymentHandler, ACTION_READ)), "GET")).Queries("requestPayment", "") // ListObjectsV2 - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListObjectsV2Handler, ACTION_LIST), "LIST")).Queries("list-type", "2") + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.ListObjectsV2Handler, ACTION_LIST)), "LIST")).Queries("list-type", "2") // buckets with query // raw buckets // PostPolicy - bucket.Methods("POST").HeadersRegexp("Content-Type", "multipart/form-data*").HandlerFunc(track(s3a.iam.Auth(s3a.PostPolicyBucketHandler, ACTION_WRITE), "POST")) + bucket.Methods("POST").HeadersRegexp("Content-Type", "multipart/form-data*").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.PostPolicyBucketHandler, ACTION_WRITE)), "POST")) // HeadBucket - bucket.Methods("HEAD").HandlerFunc(track(s3a.iam.Auth(s3a.HeadBucketHandler, ACTION_READ), "GET")) + bucket.Methods("HEAD").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.HeadBucketHandler, ACTION_READ)), "GET")) // PutBucket bucket.Methods("PUT").HandlerFunc(track(s3a.PutBucketHandler, "PUT")) // DeleteBucket - bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteBucketHandler, ACTION_WRITE), "DELETE")) + bucket.Methods("DELETE").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.DeleteBucketHandler, ACTION_WRITE)), "DELETE")) // ListObjectsV1 (Legacy) - bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListObjectsV1Handler, ACTION_LIST), "LIST")) + bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.cb.Check(s3a.ListObjectsV1Handler, ACTION_LIST)), "LIST")) // raw buckets diff --git a/weed/s3api/s3err/s3api_errors.go b/weed/s3api/s3err/s3api_errors.go index 2e93f49cb..57f269a2e 100644 --- a/weed/s3api/s3err/s3api_errors.go +++ b/weed/s3api/s3err/s3api_errors.go @@ -104,6 +104,9 @@ const ( ErrExistingObjectIsDirectory ErrExistingObjectIsFile + + ErrTooManyRequest + ErrRequestBytesExceed ) // error code to APIError structure, these fields carry respective @@ -401,6 +404,16 @@ var errorCodeResponse = map[ErrorCode]APIError{ Description: "Existing Object is a file.", HTTPStatusCode: http.StatusConflict, }, + ErrTooManyRequest: { + Code: "ErrTooManyRequest", + Description: "Too many simultaneous request count", + HTTPStatusCode: http.StatusTooManyRequests, + }, + ErrRequestBytesExceed: { + Code: "ErrRequestBytesExceed", + Description: "Simultaneous request bytes exceed limitations", + HTTPStatusCode: http.StatusTooManyRequests, + }, } // GetAPIError provides API Error for input API error code. diff --git a/weed/shell/command_s3_circuitbreaker.go b/weed/shell/command_s3_circuitbreaker.go new file mode 100644 index 000000000..ec1cd033b --- /dev/null +++ b/weed/shell/command_s3_circuitbreaker.go @@ -0,0 +1,335 @@ +package shell + +import ( + "bytes" + "flag" + "fmt" + "github.com/alecthomas/units" + "github.com/chrislusf/seaweedfs/weed/config" + "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/chrislusf/seaweedfs/weed/pb/s3_pb" + "io" + "strconv" + "strings" + + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" +) + +func init() { + Commands = append(Commands, &commandS3CircuitBreaker{}) +} + +type commandS3CircuitBreaker struct { +} + +func (c *commandS3CircuitBreaker) Name() string { + return "s3.circuit.breaker" +} + +func (c *commandS3CircuitBreaker) Help() string { + return `configure and apply s3 circuit breaker options for each bucket + + # examples + # add + s3.circuit.breaker -actions Read,Write -values 500,200 -global -enable -apply -type count + s3.circuit.breaker -actions Write -values 200MiB -global -enable -apply -type bytes + s3.circuit.breaker -actions Write -values 200MiB -bucket x,y,z -enable -apply -type bytes + + #delete + s3.circuit.breaker -actions Write -bucket x,y,z -delete -apply -type bytes + s3.circuit.breaker -actions Write -bucket x,y,z -delete -apply + s3.circuit.breaker -actions Write -delete -apply + ` +} + +func (c *commandS3CircuitBreaker) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) { + dir := config.CircuitBreakerConfigDir + file := config.CircuitBreakerConfigFile + + s3CircuitBreakerCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + buckets := s3CircuitBreakerCommand.String("buckets", "", "comma separated buckets names") + global := s3CircuitBreakerCommand.Bool("global", false, "comma separated buckets names") + + actions := s3CircuitBreakerCommand.String("actions", "", "comma separated actions names: Read,Write,List,Tagging,Admin") + limitType := s3CircuitBreakerCommand.String("type", "", "count|bytes simultaneous requests count") + values := s3CircuitBreakerCommand.String("values", "", "comma separated max values,Maximum number of simultaneous requests content length, support byte unit: eg: 1k, 10m, 1g") + + enabled := s3CircuitBreakerCommand.Bool("enable", true, "enable or disable circuit breaker") + deleted := s3CircuitBreakerCommand.Bool("delete", false, "delete users, actions or access keys") + + apply := s3CircuitBreakerCommand.Bool("apply", false, "update and apply current configuration") + + if err = s3CircuitBreakerCommand.Parse(args); err != nil { + return nil + } + + var buf bytes.Buffer + if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return filer.ReadEntry(commandEnv.MasterClient, client, dir, file, &buf) + }); err != nil && err != filer_pb.ErrNotFound { + return err + } + + cbCfg := &s3_pb.S3CircuitBreakerConfig{ + Buckets: make(map[string]*s3_pb.CbOptions), + } + if buf.Len() > 0 { + if err = filer.ParseS3ConfigurationFromBytes(buf.Bytes(), cbCfg); err != nil { + return err + } + } + + if *deleted { + cmdBuckets, cmdActions, _, err := c.initActionsAndValues(buckets, actions, limitType, values, true) + if err != nil { + return err + } + + if len(cmdBuckets) <= 0 && !*global { + if len(cmdActions) > 0 { + deleteGlobalActions(cbCfg, cmdActions, limitType) + if cbCfg.Buckets != nil { + var allBuckets []string + for bucket, _ := range cbCfg.Buckets { + allBuckets = append(allBuckets, bucket) + } + deleteBucketsActions(allBuckets, cbCfg, cmdActions, limitType) + } + } else { + cbCfg.Global = nil + cbCfg.Buckets = nil + } + } else { + if len(cmdBuckets) > 0 { + deleteBucketsActions(cmdBuckets, cbCfg, cmdActions, limitType) + } + if *global { + deleteGlobalActions(cbCfg, cmdActions, nil) + } + } + } else { + cmdBuckets, cmdActions, cmdValues, err := c.initActionsAndValues(buckets, actions, limitType, values, false) + if err != nil { + return err + } + + if len(cmdActions) > 0 && len(*buckets) <= 0 && !*global { + return fmt.Errorf("one of -global and -buckets must be specified") + } + + if len(*buckets) > 0 { + for _, bucket := range cmdBuckets { + var cbOptions *s3_pb.CbOptions + var exists bool + if cbOptions, exists = cbCfg.Buckets[bucket]; !exists { + cbOptions = &s3_pb.CbOptions{} + cbCfg.Buckets[bucket] = cbOptions + } + cbOptions.Enabled = *enabled + + if len(cmdActions) > 0 { + err = insertOrUpdateValues(cbOptions, cmdActions, cmdValues, limitType) + if err != nil { + return err + } + } + + if len(cbOptions.Actions) <= 0 && !cbOptions.Enabled { + delete(cbCfg.Buckets, bucket) + } + } + } + + if *global { + globalOptions := cbCfg.Global + if globalOptions == nil { + globalOptions = &s3_pb.CbOptions{Actions: make(map[string]int64, len(cmdActions))} + cbCfg.Global = globalOptions + } + globalOptions.Enabled = *enabled + + if len(cmdActions) > 0 { + err = insertOrUpdateValues(globalOptions, cmdActions, cmdValues, limitType) + if err != nil { + return err + } + } + + if len(globalOptions.Actions) <= 0 && !globalOptions.Enabled { + cbCfg.Global = nil + } + } + } + + buf.Reset() + err = filer.ProtoToText(&buf, cbCfg) + if err != nil { + return err + } + + fmt.Fprintf(writer, string(buf.Bytes())) + fmt.Fprintln(writer) + + if *apply { + if err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return filer.SaveInsideFiler(client, dir, file, buf.Bytes()) + }); err != nil { + return err + } + + } + + return nil +} + +func insertOrUpdateValues(cbOptions *s3_pb.CbOptions, cmdActions []string, cmdValues []int64, limitType *string) error { + if len(*limitType) == 0 { + return fmt.Errorf("type not valid, only 'count' and 'bytes' are allowed") + } + + if cbOptions.Actions == nil { + cbOptions.Actions = make(map[string]int64, len(cmdActions)) + } + + if len(cmdValues) > 0 { + for i, action := range cmdActions { + cbOptions.Actions[config.Concat(action, *limitType)] = cmdValues[i] + } + } + return nil +} + +func deleteBucketsActions(cmdBuckets []string, cbCfg *s3_pb.S3CircuitBreakerConfig, cmdActions []string, limitType *string) { + if cbCfg.Buckets == nil { + return + } + + if len(cmdActions) == 0 { + for _, bucket := range cmdBuckets { + delete(cbCfg.Buckets, bucket) + } + } else { + for _, bucket := range cmdBuckets { + if cbOption, ok := cbCfg.Buckets[bucket]; ok { + if len(cmdActions) > 0 && cbOption.Actions != nil { + for _, action := range cmdActions { + delete(cbOption.Actions, config.Concat(action, *limitType)) + } + } + + if len(cbOption.Actions) == 0 && !cbOption.Enabled { + delete(cbCfg.Buckets, bucket) + } + } + } + } + + if len(cbCfg.Buckets) == 0 { + cbCfg.Buckets = nil + } +} + +func deleteGlobalActions(cbCfg *s3_pb.S3CircuitBreakerConfig, cmdActions []string, limitType *string) { + globalOptions := cbCfg.Global + if globalOptions == nil { + return + } + + if len(cmdActions) == 0 && globalOptions.Actions != nil { + globalOptions.Actions = nil + return + } else { + for _, action := range cmdActions { + delete(globalOptions.Actions, config.Concat(action, *limitType)) + } + } + + if len(globalOptions.Actions) == 0 && !globalOptions.Enabled { + cbCfg.Global = nil + } +} + +func (c *commandS3CircuitBreaker) initActionsAndValues(buckets, actions, limitType, values *string, deleteOp bool) (cmdBuckets, cmdActions []string, cmdValues []int64, err error) { + if len(*buckets) > 0 { + cmdBuckets = strings.Split(*buckets, ",") + } + + if len(*actions) > 0 { + cmdActions = strings.Split(*actions, ",") + + //check action valid + for _, action := range cmdActions { + var found bool + for _, allowedAction := range config.AllowedActions { + if allowedAction == action { + found = true + } + } + if !found { + return nil, nil, nil, fmt.Errorf("value(%s) of flag[-action] not valid, allowed actions: %v", *actions, config.AllowedActions) + } + } + } + + if !deleteOp { + if len(cmdActions) < 0 { + for _, action := range config.AllowedActions { + cmdActions = append(cmdActions, action) + } + } + + if len(*limitType) > 0 { + switch *limitType { + case config.LimitTypeCount: + elements := strings.Split(*values, ",") + if len(cmdActions) != len(elements) { + if len(elements) != 1 || len(elements) == 0 { + return nil, nil, nil, fmt.Errorf("count of flag[-actions] and flag[-counts] not equal") + } + v, err := strconv.Atoi(elements[0]) + if err != nil { + return nil, nil, nil, fmt.Errorf("value of -counts must be a legal number(s)") + } + for range cmdActions { + cmdValues = append(cmdValues, int64(v)) + } + } else { + for _, value := range elements { + v, err := strconv.Atoi(value) + if err != nil { + return nil, nil, nil, fmt.Errorf("value of -counts must be a legal number(s)") + } + cmdValues = append(cmdValues, int64(v)) + } + } + case config.LimitTypeBytes: + elements := strings.Split(*values, ",") + if len(cmdActions) != len(elements) { + if len(elements) != 1 || len(elements) == 0 { + return nil, nil, nil, fmt.Errorf("count of flag[-actions] and flag[-counts] not equal") + } + v, err := units.ParseStrictBytes(elements[0]) + if err != nil { + return nil, nil, nil, fmt.Errorf("value of -max must be a legal number(s)") + } + for range cmdActions { + cmdValues = append(cmdValues, v) + } + } else { + for _, value := range elements { + v, err := units.ParseStrictBytes(value) + if err != nil { + return nil, nil, nil, fmt.Errorf("value of -max must be a legal number(s)") + } + cmdValues = append(cmdValues, v) + } + } + default: + return nil, nil, nil, fmt.Errorf("type not valid, only 'count' and 'bytes' are allowed") + } + } else { + *limitType = "" + } + } + return cmdBuckets, cmdActions, cmdValues, nil +} diff --git a/weed/shell/command_s3_circuitbreaker_test.go b/weed/shell/command_s3_circuitbreaker_test.go new file mode 100644 index 000000000..93e9fa414 --- /dev/null +++ b/weed/shell/command_s3_circuitbreaker_test.go @@ -0,0 +1,7 @@ +package shell + +import "testing" + +func TestCircuitBreakerShell(t *testing.T) { + +}