rancher/rke
1
0
Fork 0
mirror of https://github.com/rancher/rke.git synced 2025-08-10 19:22:41 +00:00
Code Issues Releases Wiki Activity

reduce rewrite workers, add additional logging around secrets retrieval

Browse Source
This commit is contained in:
Ryan Sanna 2020-12-18 14:01:47 -07:00
parent c3e5f6f768
commit c14c39f8c5
2 changed files with 24 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cluster/cluster.go
View File

@ -85,6 +85,7 @@ const (
FullStateConfigMapName = "full-cluster-state" FullStateConfigMapName = "full-cluster-state"
UpdateStateTimeout = 30 UpdateStateTimeout = 30
GetStateTimeout = 30 GetStateTimeout = 30
RewriteWorkers = 5
SyncWorkers = 10 SyncWorkers = 10
NoneAuthorizationMode = "none" NoneAuthorizationMode = "none"
LocalNodeAddress = "127.0.0.1" LocalNodeAddress = "127.0.0.1"

32
cluster/encryption.go
View File

@ -40,6 +40,7 @@ type encryptionKey struct {
Name string Name string
Secret string Secret string
} }
type keyList struct { type keyList struct {
KeyList []*encryptionKey KeyList []*encryptionKey
} }
@ -142,9 +143,12 @@ func (c *Cluster) RewriteSecrets(ctx context.Context) error {
return true return true
} }
logrus.Debugf("[%v] retrieving cluster secrets with batch size: %v", rewriteSecretsOperation, secretBatchSize)
var continueToken string var continueToken string
var secrets []v1.Secret var secrets []v1.Secret
var restart bool var restart bool
var batchNum uint
for { for {
err := retry.OnError(retry.DefaultRetry, retryErr, func() error { err := retry.OnError(retry.DefaultRetry, retryErr, func() error {
l, err := k8sClient.CoreV1().Secrets("").List(ctx, metav1.ListOptions{ l, err := k8sClient.CoreV1().Secrets("").List(ctx, metav1.ListOptions{
@ -156,11 +160,15 @@ func (c *Cluster) RewriteSecrets(ctx context.Context) error {
logrus.Debugf("[%v] continue token expired, restarting list operation", rewriteSecretsOperation) logrus.Debugf("[%v] continue token expired, restarting list operation", rewriteSecretsOperation)
continueToken = "" continueToken = ""
restart = true restart = true
batchNum = 0
return nil return nil
} }
return err return err
} }
batchNum++
logrus.Debugf("[%v] batch %v, retrieved %v secrets from cluster", rewriteSecretsOperation, batchNum, len(l.Items))
secrets = append(secrets, l.Items...) secrets = append(secrets, l.Items...)
continueToken = l.Continue continueToken = l.Continue
@ -191,14 +199,16 @@ func (c *Cluster) RewriteSecrets(ctx context.Context) error {
// NOTE: since we retrieve secrets in batches, we don't know total number of secrets up front. // NOTE: since we retrieve secrets in batches, we don't know total number of secrets up front.
// Telling the user how many we've rewritten so far is the best we can do // Telling the user how many we've rewritten so far is the best we can do
done := make(chan struct{}, SyncWorkers) done := make(chan struct{}, RewriteWorkers)
defer close(done) var numRewritten int
go func() { var wg sync.WaitGroup
var rewritten int wg.Add(1)
go func() { // track progress of secret rewrites
defer wg.Done()
for range done { for range done {
rewritten++ numRewritten++
if rewritten%50 == 0 { // log a message every 50 secrets if numRewritten%50 == 0 { // log a message every 50 secrets
log.Infof(ctx, "[%s] %v secrets rewritten", rewriteSecretsOperation, rewritten) log.Infof(ctx, "[%s] %v secrets rewritten", rewriteSecretsOperation, numRewritten)
} }
} }
}() }()
@ -214,7 +224,7 @@ func (c *Cluster) RewriteSecrets(ctx context.Context) error {
// spawn workers to perform secret rewrites // spawn workers to perform secret rewrites
var errgrp errgroup.Group var errgrp errgroup.Group
for w := 0; w < SyncWorkers; w++ { for w := 0; w < RewriteWorkers; w++ {
errgrp.Go(func() error { errgrp.Go(func() error {
var errList []error var errList []error
for secret := range rewrites { for secret := range rewrites {
@ -247,10 +257,14 @@ func (c *Cluster) RewriteSecrets(ctx context.Context) error {
return err // worker error from rewrites return err // worker error from rewrites
} }
// All secrets have been sent for rewrite, send exit signal to progress tracking go routine and wait for exit
close(done)
wg.Wait()
if cliErr != nil { if cliErr != nil {
log.Infof(ctx, "[%s] Operation encountered error: %v", rewriteSecretsOperation, cliErr) log.Infof(ctx, "[%s] Operation encountered error: %v", rewriteSecretsOperation, cliErr)
} else { } else {
log.Infof(ctx, "[%s] Operation completed", rewriteSecretsOperation) log.Infof(ctx, "[%s] Operation completed, %v secrets rewritten", rewriteSecretsOperation, numRewritten)
} }
return cliErr return cliErr