TrueCloudLab/frostfs-s3-gw
18
3
Fork 20
Code Issues 29 Pull requests 8 Projects Releases 23 Packages Activity Actions

[#104] app: Reload copies numbers on SIGHUP

Browse source 
Signed-off-by: Artem Tataurov <a.tataurov@yadro.com>
This commit is contained in:
Artem Tataurov 2023-05-18 16:20:00 +03:00
parent 11f30a037b
commit 9f186d9aba
7 changed files with 105 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
CHANGELOG.md
View file

@ -11,6 +11,7 @@ This document outlines major changes between releases.
- Don't create unnecessary delete-markers (#83)
### Added
- Reload default and custom copies numbers on SIGHUP (#104)
- Add `copies_numbers` section to `placement_policy` in config file and support vectors of copies numbers (#70)
- Return `X-Owner-Id` in `head-bucket` response (#79)
- Return container name in `head-bucket` response (TrueCloudLab#18)

12
api/handler/api.go
View file

@ -34,16 +34,16 @@ type (
XMLDecoder XMLDecoderProvider
DefaultMaxAge int
NotificatorEnabled bool
DefaultCopiesNumbers []uint32
CopiesNumbers map[string][]uint32
ResolveZoneList []string
IsResolveListAllow bool // True if ResolveZoneList contains allowed zones
CompleteMultipartKeepalive time.Duration
}
PlacementPolicy interface {
Default() netmap.PlacementPolicy
Get(string) (netmap.PlacementPolicy, bool)
DefaultPlacementPolicy() netmap.PlacementPolicy
PlacementPolicy(string) (netmap.PlacementPolicy, bool)
CopiesNumbers(string) ([]uint32, bool)
DefaultCopiesNumbers() []uint32
}
XMLDecoderProvider interface {
@ -97,12 +97,12 @@ func (h *handler) pickCopiesNumbers(metadata map[string]string, locationConstrai
return result, nil
}
copiesNumbers, ok := h.cfg.CopiesNumbers[locationConstraint]
copiesNumbers, ok := h.cfg.Policy.CopiesNumbers(locationConstraint)
if ok {
return copiesNumbers, nil
}
return h.cfg.DefaultCopiesNumbers, nil
return h.cfg.Policy.DefaultCopiesNumbers(), nil
}
func parseCopiesNumbers(copiesNumbersStr string) ([]uint32, error) {

10
api/handler/api_test.go
View file

@ -7,14 +7,16 @@ import (
)
func TestCopiesNumberPicker(t *testing.T) {
var locConstraints = map[string][]uint32{}
var locationConstraints = map[string][]uint32{}
locationConstraint1 := "one"
locationConstraint2 := "two"
locConstraints[locationConstraint1] = []uint32{2, 3, 4}
locationConstraints[locationConstraint1] = []uint32{2, 3, 4}
config := &Config{
DefaultCopiesNumbers: []uint32{1},
CopiesNumbers: locConstraints,
Policy: &placementPolicyMock{
copiesNumbers: locationConstraints,
defaultCopiesNumbers: []uint32{1},
},
}
h := handler{
cfg: config,

17
api/handler/handlers_test.go
View file

@ -54,17 +54,28 @@ func (hc *handlerContext) Context() context.Context {
}
type placementPolicyMock struct {
defaultPolicy netmap.PlacementPolicy
defaultPolicy netmap.PlacementPolicy
copiesNumbers map[string][]uint32
defaultCopiesNumbers []uint32
}
func (p *placementPolicyMock) Default() netmap.PlacementPolicy {
func (p *placementPolicyMock) DefaultPlacementPolicy() netmap.PlacementPolicy {
return p.defaultPolicy
}
func (p *placementPolicyMock) Get(string) (netmap.PlacementPolicy, bool) {
func (p *placementPolicyMock) PlacementPolicy(string) (netmap.PlacementPolicy, bool) {
return netmap.PlacementPolicy{}, false
}
func (p *placementPolicyMock) CopiesNumbers(locationConstraint string) ([]uint32, bool) {
result, ok := p.copiesNumbers[locationConstraint]
return result, ok
}
func (p *placementPolicyMock) DefaultCopiesNumbers() []uint32 {
return p.defaultCopiesNumbers
}
type xmlDecoderProviderMock struct{}
func (p *xmlDecoderProviderMock) NewCompleteMultipartDecoder(r io.Reader) *xml.Decoder {

4
api/handler/put.go
View file

@ -737,7 +737,7 @@ func (h *handler) CreateBucketHandler(w http.ResponseWriter, r *http.Request) {
}
func (h handler) setPolicy(prm *layer.CreateBucketParams, locationConstraint string, userPolicies []*accessbox.ContainerPolicy) error {
prm.Policy = h.cfg.Policy.Default()
prm.Policy = h.cfg.Policy.DefaultPlacementPolicy()
prm.LocationConstraint = locationConstraint
if locationConstraint == "" {
@ -751,7 +751,7 @@ func (h handler) setPolicy(prm *layer.CreateBucketParams, locationConstraint str
}
}
if policy, ok := h.cfg.Policy.Get(locationConstraint); ok {
if policy, ok := h.cfg.Policy.PlacementPolicy(locationConstraint); ok {
prm.Policy = policy
return nil
}

94
cmd/s3-gw/app.go
View file

@ -73,9 +73,11 @@ type (
}
placementPolicy struct {
mu sync.RWMutex
defaultPolicy netmap.PlacementPolicy
regionMap map[string]netmap.PlacementPolicy
mu sync.RWMutex
defaultPolicy netmap.PlacementPolicy
regionMap map[string]netmap.PlacementPolicy
copiesNumbers map[string][]uint32
defaultCopiesNumbers []uint32
}
)
@ -154,7 +156,7 @@ func (a *App) initLayer(ctx context.Context) {
}
func newAppSettings(log *Logger, v *viper.Viper) *appSettings {
policies, err := newPlacementPolicy(getDefaultPolicyValue(v), v.GetString(cfgPolicyRegionMapFile))
policies, err := newPlacementPolicy(log.logger, v)
if err != nil {
log.logger.Fatal("failed to create new policy mapping", zap.Error(err))
}
@ -285,21 +287,25 @@ func getPool(ctx context.Context, logger *zap.Logger, cfg *viper.Viper) (*pool.P
return p, key
}
func newPlacementPolicy(defaultPolicy string, regionPolicyFilepath string) (*placementPolicy, error) {
func newPlacementPolicy(l *zap.Logger, v *viper.Viper) (*placementPolicy, error) {
policies := &placementPolicy{
regionMap: make(map[string]netmap.PlacementPolicy),
regionMap: make(map[string]netmap.PlacementPolicy),
defaultCopiesNumbers: []uint32{handler.DefaultCopiesNumber},
}
return policies, policies.update(defaultPolicy, regionPolicyFilepath)
policies.updateCopiesNumbers(l, v)
policies.updateDefaultCopiesNumbers(l, v)
return policies, policies.updatePolicy(getDefaultPolicyValue(v), v.GetString(cfgPolicyRegionMapFile))
}
func (p *placementPolicy) Default() netmap.PlacementPolicy {
func (p *placementPolicy) DefaultPlacementPolicy() netmap.PlacementPolicy {
p.mu.RLock()
defer p.mu.RUnlock()
return p.defaultPolicy
}
func (p *placementPolicy) Get(name string) (netmap.PlacementPolicy, bool) {
func (p *placementPolicy) PlacementPolicy(name string) (netmap.PlacementPolicy, bool) {
p.mu.RLock()
policy, ok := p.regionMap[name]
p.mu.RUnlock()
@ -307,7 +313,30 @@ func (p *placementPolicy) Get(name string) (netmap.PlacementPolicy, bool) {
return policy, ok
}
func (p *placementPolicy) update(defaultPolicy string, regionPolicyFilepath string) error {
func (p *placementPolicy) CopiesNumbers(locationConstraint string) ([]uint32, bool) {
p.mu.RLock()
copiesNumbers, ok := p.copiesNumbers[locationConstraint]
p.mu.RUnlock()
return copiesNumbers, ok
}
func (p *placementPolicy) DefaultCopiesNumbers() []uint32 {
p.mu.RLock()
defer p.mu.RUnlock()
return p.defaultCopiesNumbers
}
func (p *placementPolicy) update(l *zap.Logger, v *viper.Viper) {
if err := p.updatePolicy(getDefaultPolicyValue(v), v.GetString(cfgPolicyRegionMapFile)); err != nil {
l.Warn("policies won't be updated", zap.Error(err))
}
p.updateCopiesNumbers(l, v)
p.updateDefaultCopiesNumbers(l, v)
}
func (p *placementPolicy) updatePolicy(defaultPolicy string, regionPolicyFilepath string) error {
var defaultPlacementPolicy netmap.PlacementPolicy
if err := defaultPlacementPolicy.DecodeString(defaultPolicy); err != nil {
return fmt.Errorf("parse default policy '%s': %w", defaultPolicy, err)
@ -342,6 +371,31 @@ func (p *placementPolicy) update(defaultPolicy string, regionPolicyFilepath stri
return nil
}
func (p *placementPolicy) updateCopiesNumbers(l *zap.Logger, v *viper.Viper) {
if newCopiesNumbers, err := fetchCopiesNumbers(l, v); err != nil {
l.Warn("copies numbers won't be updated", zap.Error(err))
} else {
p.mu.Lock()
p.copiesNumbers = newCopiesNumbers
p.mu.Unlock()
}
}
func (p *placementPolicy) updateDefaultCopiesNumbers(l *zap.Logger, v *viper.Viper) {
configuredValues, err := fetchDefaultCopiesNumbers(v)
if err == nil {
p.mu.Lock()
p.defaultCopiesNumbers = configuredValues
p.mu.Unlock()
l.Info("default copies numbers", zap.Uint32s("vector", p.defaultCopiesNumbers))
return
}
l.Error("cannot parse default copies numbers", zap.Error(err))
l.Warn("default copies numbers won't be updated", zap.Uint32s("current value", p.DefaultCopiesNumbers()))
}
func remove(list []string, element string) []string {
for i, item := range list {
if item == element {
@ -465,9 +519,7 @@ func (a *App) updateSettings() {
a.settings.logLevel.SetLevel(lvl)
}
if err := a.settings.policies.update(getDefaultPolicyValue(a.cfg), a.cfg.GetString(cfgPolicyRegionMapFile)); err != nil {
a.log.Warn("policies won't be updated", zap.Error(err))
}
a.settings.policies.update(a.log, a.cfg)
a.settings.xmlDecoder.UseDefaultNamespaceForCompleteMultipart(a.cfg.GetBool(cfgKludgeUseDefaultXMLNSForCompleteMultipartUpload))
}
@ -634,11 +686,10 @@ func getAccessBoxCacheConfig(v *viper.Viper, l *zap.Logger) *cache.Config {
func (a *App) initHandler() {
cfg := &handler.Config{
Policy: a.settings.policies,
DefaultMaxAge: handler.DefaultMaxAge,
NotificatorEnabled: a.cfg.GetBool(cfgEnableNATS),
DefaultCopiesNumbers: []uint32{handler.DefaultCopiesNumber},
XMLDecoder: a.settings.xmlDecoder,
Policy: a.settings.policies,
DefaultMaxAge: handler.DefaultMaxAge,
NotificatorEnabled: a.cfg.GetBool(cfgEnableNATS),
XMLDecoder: a.settings.xmlDecoder,
}
if a.cfg.IsSet(cfgDefaultMaxAge) {
@ -652,13 +703,6 @@ func (a *App) initHandler() {
cfg.DefaultMaxAge = defaultMaxAge
}
cfg.CopiesNumbers = fetchCopiesNumbers(a.log, a.cfg)
if val := fetchDefaultCopiesNumbers(a.log, a.cfg); len(val) > 0 {
cfg.DefaultCopiesNumbers = val
}
a.log.Info("setting default copies numbers", zap.Uint32s("vector", cfg.DefaultCopiesNumbers))
cfg.ResolveZoneList = a.cfg.GetStringSlice(cfgResolveBucketAllow)
cfg.IsResolveListAllow = len(cfg.ResolveZoneList) > 0
if !cfg.IsResolveListAllow {

17
cmd/s3-gw/app_settings.go
View file

@ -152,25 +152,23 @@ var ignore = map[string]struct{}{
cmdVersion: {},
}
func fetchDefaultCopiesNumbers(l *zap.Logger, v *viper.Viper) []uint32 {
func fetchDefaultCopiesNumbers(v *viper.Viper) ([]uint32, error) {
unparsed := v.GetStringSlice(cfgSetCopiesNumber)
var result []uint32
for i := range unparsed {
parsedValue, err := strconv.ParseUint(unparsed[i], 10, 32)
if err != nil {
l.Error("cannot parse default copies numbers", zap.Error(err))
return make([]uint32, 0)
return nil, err
}
result = append(result, uint32(parsedValue))
}
return result
return result, nil
}
func fetchCopiesNumbers(l *zap.Logger, v *viper.Viper) map[string][]uint32 {
func fetchCopiesNumbers(l *zap.Logger, v *viper.Viper) (map[string][]uint32, error) {
var copiesNums = make(map[string][]uint32)
LOOP:
for i := 0; ; i++ {
key := cfgCopiesNumbers + "." + strconv.Itoa(i) + "."
constraint := v.GetString(key + "location_constraint")
@ -184,16 +182,15 @@ LOOP:
for j := range vector {
parsedValue, err := strconv.ParseUint(vector[j], 10, 32)
if err != nil {
l.Error("cannot parse copies numbers", zap.Error(err))
break LOOP
return nil, err
}
vector32[j] = uint32(parsedValue)
}
copiesNums[constraint] = vector32
l.Debug("added constraint", zap.String("location", constraint), zap.Strings("copies numbers", vector))
l.Info("constraint added", zap.String("location", constraint), zap.Strings("copies numbers", vector))
}
return copiesNums
return copiesNums, nil
}
func fetchPeers(l *zap.Logger, v *viper.Viper) []pool.NodeParam {