mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-21 14:02:35 +00:00
master: close etcd volume id allocator client on shutdown
This commit is contained in:
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
@@ -71,7 +72,8 @@ type MasterServer struct {
|
||||
option *MasterOption
|
||||
guard *security.Guard
|
||||
|
||||
preallocateSize int64
|
||||
preallocateSize int64
|
||||
volumeIdAllocatorCloser io.Closer
|
||||
|
||||
Topo *topology.Topology
|
||||
vg *topology.VolumeGrowth
|
||||
@@ -169,10 +171,11 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
|
||||
switch v.GetString("master.volume_id_allocator") {
|
||||
case "":
|
||||
case "etcd":
|
||||
allocator, err := newVolumeIdAllocator(v, ms.Topo)
|
||||
allocator, closer, err := newVolumeIdAllocator(v, ms.Topo)
|
||||
if err != nil {
|
||||
glog.Fatalf("volume id allocator: %v", err)
|
||||
}
|
||||
ms.volumeIdAllocatorCloser = closer
|
||||
ms.Topo.SetVolumeIdAllocator(allocator)
|
||||
default:
|
||||
glog.Fatalf("unknown master.volume_id_allocator %q", v.GetString("master.volume_id_allocator"))
|
||||
@@ -668,6 +671,9 @@ func (ms *MasterServer) Shutdown() {
|
||||
if raftServer != nil {
|
||||
raftServer.Stop()
|
||||
}
|
||||
if ms.volumeIdAllocatorCloser != nil {
|
||||
ms.volumeIdAllocatorCloser.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (ms *MasterServer) Reload() {
|
||||
|
||||
@@ -3,6 +3,7 @@ package weed_server
|
||||
import (
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -14,11 +15,13 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
)
|
||||
|
||||
func newVolumeIdAllocator(v *util.ViperProxy, topo *topology.Topology) (topology.VolumeIdAllocator, error) {
|
||||
// newVolumeIdAllocator returns the allocator and an io.Closer to release the
|
||||
// etcd client on shutdown; the closer is nil for built-in allocation.
|
||||
func newVolumeIdAllocator(v *util.ViperProxy, topo *topology.Topology) (topology.VolumeIdAllocator, io.Closer, error) {
|
||||
servers := v.GetString("master.volume_id_etcd.servers")
|
||||
timeout, err := time.ParseDuration(v.GetString("master.volume_id_etcd.timeout"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse master.volume_id_etcd.timeout: %w", err)
|
||||
return nil, nil, fmt.Errorf("parse master.volume_id_etcd.timeout: %w", err)
|
||||
}
|
||||
|
||||
var tlsConfig *tls.Config
|
||||
@@ -29,7 +32,7 @@ func newVolumeIdAllocator(v *util.ViperProxy, topo *topology.Topology) (topology
|
||||
TrustedCAFile: caFile,
|
||||
}
|
||||
if tlsConfig, err = tlsInfo.ClientConfig(); err != nil {
|
||||
return nil, fmt.Errorf("etcd TLS client configuration error: %w", err)
|
||||
return nil, nil, fmt.Errorf("etcd TLS client configuration error: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -41,13 +44,14 @@ func newVolumeIdAllocator(v *util.ViperProxy, topo *topology.Topology) (topology
|
||||
TLS: tlsConfig,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect volume id etcd: %w", err)
|
||||
return nil, nil, fmt.Errorf("connect volume id etcd: %w", err)
|
||||
}
|
||||
|
||||
glog.Infof("volume id allocator: etcd %s", servers)
|
||||
return topology.NewEtcdVolumeIdAllocator(
|
||||
client.KV,
|
||||
allocator := topology.NewEtcdVolumeIdAllocator(
|
||||
client,
|
||||
v.GetString("master.volume_id_etcd.key_prefix"),
|
||||
timeout,
|
||||
topo.GetMaxVolumeId), nil
|
||||
topo.GetMaxVolumeId)
|
||||
return allocator, allocator, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user