Skip to content

Commit

Permalink
Remove secret sync loop
Browse files Browse the repository at this point in the history
  • Loading branch information
aledbf committed Jul 19, 2017
1 parent 05ef427 commit 5a1f845
Show file tree
Hide file tree
Showing 3 changed files with 64 additions and 63 deletions.
78 changes: 45 additions & 33 deletions core/pkg/ingress/controller/backend_ssl.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,47 +24,46 @@ import (
"github.com/golang/glog"

api "k8s.io/client-go/pkg/api/v1"
extensions "k8s.io/client-go/pkg/apis/extensions/v1beta1"
"k8s.io/client-go/tools/cache"

"k8s.io/ingress/core/pkg/ingress"
"k8s.io/ingress/core/pkg/ingress/annotations/parser"
"k8s.io/ingress/core/pkg/net/ssl"
)

// syncSecret keeps in sync Secrets used by Ingress rules with the files on
// disk to allow copy of the content of the secret to disk to be used
// by external processes.
func (ic *GenericController) syncSecret() {
glog.V(3).Infof("starting syncing of secrets")
func (ic *GenericController) syncSecret(key string) {
glog.V(3).Infof("starting syncing of secret %v", key)

var cert *ingress.SSLCert
var err error

for _, k := range ic.secretTracker.List() {
key := k.(string)
cert, err = ic.getPemCertificate(key)
if err != nil {
glog.Warningf("error obtaining PEM from secret %v: %v", key, err)
continue
}
cert, err = ic.getPemCertificate(key)
if err != nil {
glog.Warningf("error obtaining PEM from secret %v: %v", key, err)
return
}

// create certificates and add or update the item in the store
cur, exists := ic.sslCertTracker.Get(key)
if exists {
s := cur.(*ingress.SSLCert)
if reflect.DeepEqual(s, cert) {
// no need to update
continue
}
glog.Infof("updating secret %v in the local store", key)
ic.sslCertTracker.Update(key, cert)
ic.reloadRequired = true
continue
// create certificates and add or update the item in the store
cur, exists := ic.sslCertTracker.Get(key)
if exists {
s := cur.(*ingress.SSLCert)
if reflect.DeepEqual(s, cert) {
// no need to update
return
}

glog.Infof("adding secret %v to the local store", key)
ic.sslCertTracker.Add(key, cert)
glog.Infof("updating secret %v in the local store", key)
ic.sslCertTracker.Update(key, cert)
ic.reloadRequired = true
return
}

glog.Infof("adding secret %v to the local store", key)
ic.sslCertTracker.Add(key, cert)
ic.reloadRequired = true
}

// getPemCertificate receives a secret, and creates a ingress.SSLCert as return.
Expand Down Expand Up @@ -106,6 +105,26 @@ func (ic *GenericController) getPemCertificate(secretName string) (*ingress.SSLC
return s, nil
}

// secrReferenced checks if a secret is referenced or not by one or more Ingress rules
func (ic *GenericController) secrReferenced(name, namespace string) bool {
for _, ingIf := range ic.ingLister.Store.List() {
ing := ingIf.(*extensions.Ingress)
str, err := parser.GetStringAnnotation("ingress.kubernetes.io/auth-tls-secret", ing)
if err == nil && str == fmt.Sprintf("%v/%v", namespace, name) {
return true
}
if ing.Namespace != namespace {
continue
}
for _, tls := range ing.Spec.TLS {
if tls.SecretName == name {
return true
}
}
}
return false
}

// sslCertTracker holds a store of referenced Secrets in Ingress rules
type sslCertTracker struct {
cache.ThreadSafeStore
Expand All @@ -117,13 +136,6 @@ func newSSLCertTracker() *sslCertTracker {
}
}

// secretTracker holds a store of Secrets
type secretTracker struct {
cache.ThreadSafeStore
}

func newSecretTracker() *secretTracker {
return &secretTracker{
cache.NewThreadSafeStore(cache.Indexers{}, cache.Indices{}),
}
func (s *sslCertTracker) DeleteAll(key string) {
s.Delete(key)
}
9 changes: 4 additions & 5 deletions core/pkg/ingress/controller/backend_ssl_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,6 @@ func buildGenericControllerForBackendSSL() *GenericController {
mapController: buildControllerForBackendSSL(),

sslCertTracker: newSSLCertTracker(),
secretTracker: newSecretTracker(),
}
}

Expand Down Expand Up @@ -157,7 +156,6 @@ func TestSyncSecret(t *testing.T) {
for _, foo := range foos {
t.Run(foo.tn, func(t *testing.T) {
ic := buildGenericControllerForBackendSSL()
ic.secretTracker.Add(foo.secretName, foo.secretName)

// init secret for getPemCertificate
secret := buildSecretForBackendSSL()
Expand All @@ -166,16 +164,17 @@ func TestSyncSecret(t *testing.T) {
secret.Data = foo.Data
ic.secrLister.Add(secret)

key := "default/foo_secret"
// for add
ic.syncSecret()
ic.syncSecret(key)
if foo.expectSuccess {
// validate
_, exist := ic.sslCertTracker.Get(foo.secretName)
_, exist := ic.sslCertTracker.Get(key)
if !exist {
t.Errorf("Failed to sync secret: %s", foo.secretName)
} else {
// for update
ic.syncSecret()
ic.syncSecret(key)
}
}
})
Expand Down
40 changes: 15 additions & 25 deletions core/pkg/ingress/controller/controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@ import (
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
clientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme"
unversionedcore "k8s.io/client-go/kubernetes/typed/core/v1"
Expand All @@ -51,7 +50,6 @@ import (
"k8s.io/ingress/core/pkg/ingress/defaults"
"k8s.io/ingress/core/pkg/ingress/resolver"
"k8s.io/ingress/core/pkg/ingress/status"
"k8s.io/ingress/core/pkg/ingress/status/leaderelection/resourcelock"
"k8s.io/ingress/core/pkg/ingress/store"
"k8s.io/ingress/core/pkg/k8s"
"k8s.io/ingress/core/pkg/net/ssl"
Expand Down Expand Up @@ -99,8 +97,6 @@ type GenericController struct {
// local store of SSL certificates
// (only certificates used in ingress)
sslCertTracker *sslCertTracker
// store of secret names referenced from Ingress
secretTracker *secretTracker

syncRateLimiter flowcontrol.RateLimiter

Expand Down Expand Up @@ -167,7 +163,6 @@ func newIngressController(config *Configuration) *GenericController {
Component: "ingress-controller",
}),
sslCertTracker: newSSLCertTracker(),
secretTracker: newSecretTracker(),
}

ic.syncQueue = task.NewTaskQueue(ic.syncIngress)
Expand Down Expand Up @@ -219,39 +214,36 @@ func newIngressController(config *Configuration) *GenericController {
}

secrEventHandler := cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
sec := obj.(*api.Secret)
key := fmt.Sprintf("%v/%v", sec.Namespace, sec.Name)
if ic.secrReferenced(sec.Namespace, sec.Name) {
ic.syncSecret(key)
}
},
UpdateFunc: func(old, cur interface{}) {
if !reflect.DeepEqual(old, cur) {
ic.syncSecret()
sec := cur.(*api.Secret)
key := fmt.Sprintf("%v/%v", sec.Namespace, sec.Name)
ic.syncSecret(key)
}
},
DeleteFunc: func(obj interface{}) {
sec := obj.(*api.Secret)
key := fmt.Sprintf("%v/%v", sec.Namespace, sec.Name)
ic.sslCertTracker.Delete(key)
ic.secretTracker.Delete(key)
ic.sslCertTracker.DeleteAll(key)
},
}

eventHandler := cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
ep := obj.(*api.Endpoints)
_, found := ep.Annotations[resourcelock.LeaderElectionRecordAnnotationKey]
if found {
return
}
ic.syncQueue.Enqueue(obj)
},
DeleteFunc: func(obj interface{}) {
ic.syncQueue.Enqueue(obj)
},
UpdateFunc: func(old, cur interface{}) {
if !reflect.DeepEqual(old, cur) {
ep := cur.(*api.Endpoints)
_, found := ep.Annotations[resourcelock.LeaderElectionRecordAnnotationKey]
if found {
return
}

ic.syncQueue.Enqueue(cur)
}
},
Expand Down Expand Up @@ -754,8 +746,8 @@ func (ic *GenericController) getBackendServers() ([]*ingress.Backend, []*ingress

// GetAuthCertificate ...
func (ic GenericController) GetAuthCertificate(secretName string) (*resolver.AuthSSLCert, error) {
if _, exists := ic.secretTracker.Get(secretName); !exists {
ic.secretTracker.Add(secretName, secretName)
if _, exists := ic.sslCertTracker.Get(secretName); !exists {
ic.syncSecret(secretName)
}

_, err := ic.GetSecret(secretName)
Expand Down Expand Up @@ -1168,9 +1160,9 @@ func (ic GenericController) extractSecretNames(ing *extensions.Ingress) {
}

key := fmt.Sprintf("%v/%v", ing.Namespace, tls.SecretName)
_, exists := ic.secretTracker.Get(key)
_, exists := ic.sslCertTracker.Get(key)
if !exists {
ic.secretTracker.Add(key, key)
ic.syncSecret(key)
}
}
}
Expand Down Expand Up @@ -1219,8 +1211,6 @@ func (ic GenericController) Start() {

go ic.syncQueue.Run(10*time.Second, ic.stopCh)

go wait.Forever(ic.syncSecret, 10*time.Second)

if ic.syncStatus != nil {
go ic.syncStatus.Run(ic.stopCh)
}
Expand Down

0 comments on commit 5a1f845

Please sign in to comment.