mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-29 11:15:34 +00:00
util: serialize all ViperProxy access so startup cannot hit concurrent map read/write (#11470)
* util: serialize every ViperProxy method; stop promoting unlocked viper calls ViperProxy embedded *viper.Viper, so only the five declared methods took the mutex while every promoted call — GetStringMap in backend.LoadConfiguration was the reported crash — touched viper's maps unsynchronized. `weed server` starts the volume server (SetDefault writer) and the master (GetStringMap reader) back to back, and a race build reports the pair on a plain start. The wrapped viper is now a named field: a method must be declared here to exist on the proxy, so unsynchronized access fails at compile time rather than at runtime. Every promoted use in the tree (GetStringMap, GetUint32, GetFloat64, GetDuration, IsSet, AllKeys, Set) gets a locked wrapper; NewViperProxy replaces struct literals for local vipers. GetStringMap deep-copies its result — viper hands back the internal subtree, so iterating it after the lock is released would race the next writer. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * util: take the shared lock while LoadConfiguration merges a config file viper.MergeInConfig rewrites the same maps the proxy serializes; without the lock a merge can race a concurrent SetDefault or reader exactly like the reported startup crash. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * util: deep-copy slice elements in the GetStringMap snapshot A slice of maps inside the returned subtree still shared the inner maps — copy elements recursively so nothing the caller mutates is viper's internal state. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * util: add the missing AutomaticEnv wrapper used by tests sse_reader_test reaches it through GetViper(); without the wrapper the call no longer exists once the viper field stopped being embedded. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * util: return a fresh slice from GetStringSlice A stored []string comes back uncast from viper — the backing array is shared internal state like the GetStringMap subtree, so copy it while holding the lock. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
co-authored by
Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
parent
5389f61cef
commit
80a26020d7
@@ -161,7 +161,7 @@ func LoadClientTLSFromFile(configFile string, component string) (grpc.DialOption
|
||||
v.Set(key, filepath.Join(configDir, p))
|
||||
}
|
||||
}
|
||||
return LoadClientTLS(&util.ViperProxy{Viper: v}, component), nil
|
||||
return LoadClientTLS(util.NewViperProxy(v), component), nil
|
||||
}
|
||||
|
||||
func LoadClientTLS(config *util.ViperProxy, component string) grpc.DialOption {
|
||||
|
||||
@@ -134,7 +134,7 @@ func TestLoadClientTLSPrefersClientCert(t *testing.T) {
|
||||
serverCert, serverKey := ca.issue(t, dir, "server", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
|
||||
clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
|
||||
|
||||
v := &util.ViperProxy{Viper: viper.New()}
|
||||
v := util.NewViperProxy(viper.New())
|
||||
v.Set("grpc.ca", ca.file)
|
||||
v.Set("grpc.master.cert", serverCert)
|
||||
v.Set("grpc.master.key", serverKey)
|
||||
@@ -153,7 +153,7 @@ func TestLoadClientTLSFallsBackToServingCert(t *testing.T) {
|
||||
ca := newTestCA(t, dir)
|
||||
dualCert, dualKey := ca.issue(t, dir, "dual", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth})
|
||||
|
||||
v := &util.ViperProxy{Viper: viper.New()}
|
||||
v := util.NewViperProxy(viper.New())
|
||||
v.Set("grpc.ca", ca.file)
|
||||
v.Set("grpc.master.cert", dualCert)
|
||||
v.Set("grpc.master.key", dualKey)
|
||||
@@ -171,7 +171,7 @@ func TestLoadClientTLSServerOnlyEkuRejected(t *testing.T) {
|
||||
ca := newTestCA(t, dir)
|
||||
serverCert, serverKey := ca.issue(t, dir, "server", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
|
||||
|
||||
v := &util.ViperProxy{Viper: viper.New()}
|
||||
v := util.NewViperProxy(viper.New())
|
||||
v.Set("grpc.ca", ca.file)
|
||||
v.Set("grpc.master.cert", serverCert)
|
||||
v.Set("grpc.master.key", serverKey)
|
||||
|
||||
+101
-11
@@ -3,6 +3,7 @@ package util
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/viper"
|
||||
|
||||
@@ -39,6 +40,10 @@ func LoadSecurityConfiguration() {
|
||||
}
|
||||
|
||||
func LoadConfiguration(configFileName string, required bool) (loaded bool) {
|
||||
// MergeInConfig mutates the shared viper that ViperProxy serializes;
|
||||
// take the same lock so a merge cannot race a reader or SetDefault.
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
|
||||
// find a filer store
|
||||
viper.SetConfigName(configFileName) // name of config file (without extension)
|
||||
@@ -72,8 +77,12 @@ func LoadConfiguration(configFileName string, required bool) (loaded bool) {
|
||||
return true
|
||||
}
|
||||
|
||||
// ViperProxy serializes access to the global viper. The wrapped Viper is a
|
||||
// named field, not embedded, so every method must be declared here under the
|
||||
// mutex — a promoted method would take no lock and race, e.g. GetStringMap
|
||||
// against a concurrent SetDefault during `weed server` startup.
|
||||
type ViperProxy struct {
|
||||
*viper.Viper
|
||||
v *viper.Viper
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
@@ -84,42 +93,123 @@ var (
|
||||
func (vp *ViperProxy) SetDefault(key string, value interface{}) {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
vp.Viper.SetDefault(key, value)
|
||||
vp.v.SetDefault(key, value)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) AutomaticEnv() {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
vp.v.AutomaticEnv()
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetString(key string) string {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.Viper.GetString(key)
|
||||
return vp.v.GetString(key)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetBool(key string) bool {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.Viper.GetBool(key)
|
||||
return vp.v.GetBool(key)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetInt(key string) int {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.Viper.GetInt(key)
|
||||
return vp.v.GetInt(key)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetStringSlice(key string) []string {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.Viper.GetStringSlice(key)
|
||||
// a []string value comes back uncast — the same shared subtree issue
|
||||
// GetStringMap has, so hand the caller a copy
|
||||
return append([]string(nil), vp.v.GetStringSlice(key)...)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetStringMap(key string) map[string]interface{} {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
// viper hands back its internal subtree, so a caller iterating it after
|
||||
// the lock is released would race the next SetDefault — copy it out.
|
||||
return deepCopyStringMap(vp.v.GetStringMap(key))
|
||||
}
|
||||
|
||||
func deepCopyStringMap(m map[string]interface{}) map[string]interface{} {
|
||||
out := make(map[string]interface{}, len(m))
|
||||
for k, v := range m {
|
||||
out[k] = deepCopyValue(v)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func deepCopyValue(v interface{}) interface{} {
|
||||
switch nested := v.(type) {
|
||||
case map[string]interface{}:
|
||||
return deepCopyStringMap(nested)
|
||||
case []interface{}:
|
||||
out := make([]interface{}, len(nested))
|
||||
for i, item := range nested {
|
||||
out[i] = deepCopyValue(item)
|
||||
}
|
||||
return out
|
||||
default:
|
||||
return v
|
||||
}
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetUint32(key string) uint32 {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.v.GetUint32(key)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetFloat64(key string) float64 {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.v.GetFloat64(key)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) GetDuration(key string) time.Duration {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.v.GetDuration(key)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) Set(key string, value interface{}) {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
vp.v.Set(key, value)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) IsSet(key string) bool {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.v.IsSet(key)
|
||||
}
|
||||
|
||||
func (vp *ViperProxy) AllKeys() []string {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
return vp.v.AllKeys()
|
||||
}
|
||||
|
||||
// NewViperProxy wraps a specific viper — for configuration loaded from a
|
||||
// source other than the shared instance GetViper returns.
|
||||
func NewViperProxy(v *viper.Viper) *ViperProxy {
|
||||
return &ViperProxy{v: v}
|
||||
}
|
||||
|
||||
func GetViper() *ViperProxy {
|
||||
vp.Lock()
|
||||
defer vp.Unlock()
|
||||
|
||||
if vp.Viper == nil {
|
||||
vp.Viper = viper.GetViper()
|
||||
vp.AutomaticEnv()
|
||||
vp.SetEnvPrefix("weed")
|
||||
vp.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
|
||||
if vp.v == nil {
|
||||
vp.v = viper.GetViper()
|
||||
vp.v.AutomaticEnv()
|
||||
vp.v.SetEnvPrefix("weed")
|
||||
vp.v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
|
||||
}
|
||||
|
||||
return vp
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
package util
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/spf13/viper"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
// `weed server` starts the volume server (SetDefault writers) and the master
|
||||
// (GetStringMap readers) back to back. Reads and writes must share the proxy
|
||||
// lock — a race-enabled build flags the interleaving the issue reported.
|
||||
func TestViperProxySerializesMapReadsAgainstWrites(t *testing.T) {
|
||||
p := NewViperProxy(viper.New())
|
||||
p.SetDefault("storage.backend.a.enabled", true)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 4; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
for j := 0; j < 200; j++ {
|
||||
p.SetDefault(fmt.Sprintf("storage.backend.d%d.enabled", j), i)
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for j := 0; j < 200; j++ {
|
||||
if got := p.GetStringMap("storage.backend"); len(got) == 0 {
|
||||
t.Error("GetStringMap returned no keys")
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
wg.Wait()
|
||||
|
||||
got := p.GetStringMap("storage.backend")
|
||||
assert.Contains(t, got, "a")
|
||||
}
|
||||
|
||||
// A proxy over its own viper serves the wrapped instance, for configuration
|
||||
// loaded from a source other than the shared one.
|
||||
func TestNewViperProxyWrapsGivenViper(t *testing.T) {
|
||||
v := viper.New()
|
||||
v.Set("x.y", "z")
|
||||
p := NewViperProxy(v)
|
||||
assert.Equal(t, "z", p.GetString("x.y"))
|
||||
assert.Equal(t, map[string]interface{}{"y": "z"}, p.GetStringMap("x"))
|
||||
}
|
||||
Reference in New Issue
Block a user