mirror of https://github.com/k3s-io/k3s
Replace 1-weight semaphore on snapshots with simple mutex
Fixes an issue where the semaphore wasn't permanently initialized until a scheduled snapshot was taken, allowing multiple on-demand snapshots to be taken until the first scheduled snapshot was triggered. Signed-off-by: Brad Davidson <brad.davidson@rancher.com>pull/10427/head v1.29.6+k3s1
parent
4a5f69fae1
commit
83ae095ab9
|
@ -16,6 +16,7 @@ import (
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
|
@ -43,7 +44,6 @@ import (
|
||||||
clientv3 "go.etcd.io/etcd/client/v3"
|
clientv3 "go.etcd.io/etcd/client/v3"
|
||||||
"go.etcd.io/etcd/etcdutl/v3/snapshot"
|
"go.etcd.io/etcd/etcdutl/v3/snapshot"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
"golang.org/x/sync/semaphore"
|
|
||||||
v1 "k8s.io/api/core/v1"
|
v1 "k8s.io/api/core/v1"
|
||||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
"k8s.io/apimachinery/pkg/labels"
|
"k8s.io/apimachinery/pkg/labels"
|
||||||
|
@ -105,14 +105,14 @@ var _ managed.Driver = &ETCD{}
|
||||||
type MemberStatus string
|
type MemberStatus string
|
||||||
|
|
||||||
type ETCD struct {
|
type ETCD struct {
|
||||||
client *clientv3.Client
|
client *clientv3.Client
|
||||||
config *config.Control
|
config *config.Control
|
||||||
name string
|
name string
|
||||||
address string
|
address string
|
||||||
cron *cron.Cron
|
cron *cron.Cron
|
||||||
s3 *S3
|
s3 *S3
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
snapshotSem *semaphore.Weighted
|
snapshotMu *sync.Mutex
|
||||||
}
|
}
|
||||||
|
|
||||||
type learnerProgress struct {
|
type learnerProgress struct {
|
||||||
|
@ -166,10 +166,11 @@ func (e *MemberListError) Is(target error) bool {
|
||||||
func errMemberListFailed() error { return &MemberListError{} }
|
func errMemberListFailed() error { return &MemberListError{} }
|
||||||
|
|
||||||
// NewETCD creates a new value of type
|
// NewETCD creates a new value of type
|
||||||
// ETCD with an initialized cron value.
|
// ETCD with initialized cron and snapshot mutex values.
|
||||||
func NewETCD() *ETCD {
|
func NewETCD() *ETCD {
|
||||||
return &ETCD{
|
return &ETCD{
|
||||||
cron: cron.New(cron.WithLogger(cronLogger)),
|
cron: cron.New(cron.WithLogger(cronLogger)),
|
||||||
|
snapshotMu: &sync.Mutex{},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -29,7 +29,6 @@ import (
|
||||||
"github.com/robfig/cron/v3"
|
"github.com/robfig/cron/v3"
|
||||||
"github.com/sirupsen/logrus"
|
"github.com/sirupsen/logrus"
|
||||||
"go.etcd.io/etcd/etcdutl/v3/snapshot"
|
"go.etcd.io/etcd/etcdutl/v3/snapshot"
|
||||||
"golang.org/x/sync/semaphore"
|
|
||||||
v1 "k8s.io/api/core/v1"
|
v1 "k8s.io/api/core/v1"
|
||||||
"k8s.io/apimachinery/pkg/api/equality"
|
"k8s.io/apimachinery/pkg/api/equality"
|
||||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||||
|
@ -44,10 +43,9 @@ import (
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
maxConcurrentSnapshots = 1
|
compressedExtension = ".zip"
|
||||||
compressedExtension = ".zip"
|
metadataDir = ".metadata"
|
||||||
metadataDir = ".metadata"
|
errorTTL = 24 * time.Hour
|
||||||
errorTTL = 24 * time.Hour
|
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
@ -106,16 +104,6 @@ func snapshotDir(config *config.Control, create bool) (string, error) {
|
||||||
return snapshotDir, nil
|
return snapshotDir, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// preSnapshotSetup checks to see if the necessary components are in place
|
|
||||||
// to perform an Etcd snapshot. This is necessary primarily for on-demand
|
|
||||||
// snapshots since they're performed before normal Etcd setup is completed.
|
|
||||||
func (e *ETCD) preSnapshotSetup(ctx context.Context) error {
|
|
||||||
if e.snapshotSem == nil {
|
|
||||||
e.snapshotSem = semaphore.NewWeighted(maxConcurrentSnapshots)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// compressSnapshot compresses the given snapshot and provides the
|
// compressSnapshot compresses the given snapshot and provides the
|
||||||
// caller with the path to the file.
|
// caller with the path to the file.
|
||||||
func (e *ETCD) compressSnapshot(snapshotDir, snapshotName, snapshotPath string, now time.Time) (string, error) {
|
func (e *ETCD) compressSnapshot(snapshotDir, snapshotName, snapshotPath string, now time.Time) (string, error) {
|
||||||
|
@ -208,14 +196,10 @@ func (e *ETCD) decompressSnapshot(snapshotDir, snapshotFile string) (string, err
|
||||||
// subcommand for prune that can be run manually if the user wants to remove old snapshots.
|
// subcommand for prune that can be run manually if the user wants to remove old snapshots.
|
||||||
// Returns metadata about the new and pruned snapshots.
|
// Returns metadata about the new and pruned snapshots.
|
||||||
func (e *ETCD) Snapshot(ctx context.Context) (*managed.SnapshotResult, error) {
|
func (e *ETCD) Snapshot(ctx context.Context) (*managed.SnapshotResult, error) {
|
||||||
if err := e.preSnapshotSetup(ctx); err != nil {
|
if !e.snapshotMu.TryLock() {
|
||||||
return nil, err
|
return nil, errors.New("snapshot save already in progress")
|
||||||
}
|
}
|
||||||
if !e.snapshotSem.TryAcquire(maxConcurrentSnapshots) {
|
defer e.snapshotMu.Unlock()
|
||||||
return nil, fmt.Errorf("%d snapshots already in progress", maxConcurrentSnapshots)
|
|
||||||
}
|
|
||||||
defer e.snapshotSem.Release(maxConcurrentSnapshots)
|
|
||||||
|
|
||||||
// make sure the core.Factory is initialized before attempting to add snapshot metadata
|
// make sure the core.Factory is initialized before attempting to add snapshot metadata
|
||||||
var extraMetadata *v1.ConfigMap
|
var extraMetadata *v1.ConfigMap
|
||||||
if e.config.Runtime.Core == nil {
|
if e.config.Runtime.Core == nil {
|
||||||
|
|
|
@ -150,11 +150,11 @@ func (e *ETCD) withRequest(sr *SnapshotRequest) *ETCD {
|
||||||
EtcdSnapshotName: e.config.EtcdSnapshotName,
|
EtcdSnapshotName: e.config.EtcdSnapshotName,
|
||||||
EtcdSnapshotRetention: e.config.EtcdSnapshotRetention,
|
EtcdSnapshotRetention: e.config.EtcdSnapshotRetention,
|
||||||
},
|
},
|
||||||
name: e.name,
|
name: e.name,
|
||||||
address: e.address,
|
address: e.address,
|
||||||
cron: e.cron,
|
cron: e.cron,
|
||||||
cancel: e.cancel,
|
cancel: e.cancel,
|
||||||
snapshotSem: e.snapshotSem,
|
snapshotMu: e.snapshotMu,
|
||||||
}
|
}
|
||||||
if len(sr.Name) > 0 {
|
if len(sr.Name) > 0 {
|
||||||
re.config.EtcdSnapshotName = sr.Name[0]
|
re.config.EtcdSnapshotName = sr.Name[0]
|
||||||
|
|
Loading…
Reference in New Issue