Bugfix updating VCL for self-sharding when Varnish deployments change.

cmd/controller/controller.go

@@ -253,6 +253,10 @@ func NewIngressController(log *logrus.Logger, kubeClient kubernetes.Interface,
 				return
 			}
 			if ok && ingc.isVarnishAdmSvc(svc, namespace) {
+				ingc.log.Infof("Endpoints added for "+
+					"Varnish admin service %s/%s, "+
+					"enqueuing service sync", namespace,
+					svc.Name)
 				ingc.syncQueue.enqueue(svc)
 				return
 			}
@@ -290,6 +294,10 @@ func NewIngressController(log *logrus.Logger, kubeClient kubernetes.Interface,
 				return
 			}
 			if ok && ingc.isVarnishAdmSvc(svc, namespace) {
+				ingc.log.Infof("Endpoints deleted for "+
+					"Varnish admin service %s/%s, "+
+					"enqueuing service sync", namespace,
+					svc.Name)
 				ingc.syncQueue.enqueue(svc)
 				return
 			}
@@ -312,6 +320,10 @@ func NewIngressController(log *logrus.Logger, kubeClient kubernetes.Interface,
 					return
 				}
 				if ok && ingc.isVarnishAdmSvc(svc, namespace) {
+					ingc.log.Infof("Endpoints changed for "+
+						"Varnish admin service %s/%s, "+
+						"enqueuing service sync",
+						namespace, svc.Name)
 					ingc.syncQueue.enqueue(svc)
 					return
 				}
@@ -835,7 +847,31 @@ func (ingc *IngressController) syncSvc(task Task) {
 		return
 	}
 
+	ingc.log.Info("Updating Service:", key)
 	svc := svcObj.(*api_v1.Service)
+
+	// Check if there are Ingresses for which the VCL spec may
+	// change due to changes in Varnish services.
+	updateVCL := false
+	ings, _ := ingc.ingLister.List()
+	for _, ing := range ings.Items {
+		if ing.Namespace != svc.Namespace {
+			continue
+		}
+		if !ingc.isVarnishInVCLSpec(ing) {
+			continue
+		}
+		updateVCL = true
+		ingc.log.Debugf("Requeueing Ingress %s/%s after changed "+
+			"Varnish service %s/%s: %+v", ing.Namespace,
+			ing.Name, svc.Namespace, svc.Name, ing)
+		ingc.syncQueue.enqueue(&ing)
+	}
+	if !updateVCL {
+		ingc.log.Debugf("No change in VCL due to changed Varnish "+
+			"service %s/%s", svc.Namespace, svc.Name)
+	}
+
 	endps, err := ingc.endpLister.GetServiceEndpoints(svc)
 	if err != nil {
 		ingc.syncQueue.requeueAfter(task, err, 5*time.Second)
@@ -876,7 +912,7 @@ func (ingc *IngressController) syncSvc(task Task) {
 			"%v was rejected: %v", key, err)
 		return
 	}
-	ingc.vController.AddOrUpdateVarnishSvc(key, addrs)
+	ingc.vController.AddOrUpdateVarnishSvc(key, addrs, !updateVCL)
 }
 
 func (ingc *IngressController) syncSecret(task Task) {
@@ -937,3 +973,10 @@ func (ingc *IngressController) isVarnishAdmSvc(svc *api_v1.Service,
 func (ingc *IngressController) isAdminSecret(secr *api_v1.Secret) bool {
 	return secr.Name == admSecretName
 }
+
+// Return true if changes in Varnish services may lead to changes in
+// the VCL config generated for the Ingress.
+func (ingc *IngressController) isVarnishInVCLSpec(ing extensions.Ingress) bool {
+	_, selfShard := ing.Annotations[selfShardKey]
+	return selfShard
+}

cmd/varnish/varnish.go

@@ -124,6 +124,14 @@ func (vc *VarnishController) Start(errChan chan error) {
 func (vc *VarnishController) updateVarnishInstance(svc *varnishSvc,
 	cfgName string, vclSrc string) error {
 
+	if svc == nil {
+		return VarnishAdmError{
+			addr: "",
+			err:  fmt.Errorf("Service object is nil"),
+		}
+	}
+
+	vc.log.Infof("Update Varnish instance at %s", svc.addr)
 	svc.admMtx.Lock()
 	defer svc.admMtx.Unlock()
 
@@ -232,7 +240,7 @@ func (vc *VarnishController) updateVarnishInstances(svcs []*varnishSvc) error {
 }
 
 func (vc *VarnishController) addVarnishSvc(key string,
-	addrs []vcl.Address) error {
+	addrs []vcl.Address, loadVCL bool) error {
 
 	vc.varnishSvcs[key] = make([]*varnishSvc, len(addrs))
 	for i, addr := range addrs {
@@ -243,6 +251,9 @@ func (vc *VarnishController) addVarnishSvc(key string,
 		}
 		vc.varnishSvcs[key][i] = &svc
 	}
+	if !loadVCL {
+		return nil
+	}
 	return vc.updateVarnishInstances(vc.varnishSvcs[key])
 }
 
@@ -288,6 +299,7 @@ func (vc *VarnishController) removeVarnishInstances(svcs []*varnishSvc) error {
 	var errs VarnishAdmErrors
 
 	for _, svc := range svcs {
+		// XXX health check for sharding config should fail
 		if err := vc.setCfgLabel(svc, notAvailCfg, readinessLabel,
 			true); err != nil {
 
@@ -303,7 +315,7 @@ func (vc *VarnishController) removeVarnishInstances(svcs []*varnishSvc) error {
 }
 
 func (vc *VarnishController) updateVarnishSvc(key string,
-	addrs []vcl.Address) error {
+	addrs []vcl.Address, loadVCL bool) error {
 
 	var errs VarnishAdmErrors
 	var newSvcs, remSvcs, keepSvcs []*varnishSvc
@@ -323,7 +335,7 @@ func (vc *VarnishController) updateVarnishSvc(key string,
 			keepSvcs = append(keepSvcs, svc)
 			continue
 		}
-		newSvc := &varnishSvc{addr: addr}
+		newSvc := &varnishSvc{addr: addr, admMtx: &sync.Mutex{}}
 		newSvcs = append(newSvcs, newSvc)
 	}
 	for addr, svc := range prevAddrs {
@@ -344,13 +356,15 @@ func (vc *VarnishController) updateVarnishSvc(key string,
 		}
 	}
 
-	updateErrs := vc.updateVarnishInstances(vc.varnishSvcs[key])
-	if updateErrs != nil {
-		vadmErrs, ok := updateErrs.(VarnishAdmErrors)
-		if ok {
-			errs = append(errs, vadmErrs...)
-		} else {
-			return updateErrs
+	if loadVCL {
+		updateErrs := vc.updateVarnishInstances(vc.varnishSvcs[key])
+		if updateErrs != nil {
+			vadmErrs, ok := updateErrs.(VarnishAdmErrors)
+			if ok {
+				errs = append(errs, vadmErrs...)
+			} else {
+				return updateErrs
+			}
 		}
 	}
 	if len(errs) == 0 {
@@ -360,7 +374,7 @@ func (vc *VarnishController) updateVarnishSvc(key string,
 }
 
 func (vc *VarnishController) AddOrUpdateVarnishSvc(key string,
-	addrs []vcl.Address) error {
+	addrs []vcl.Address, loadVCL bool) error {
 
 	if vc.admSecret == nil {
 		return fmt.Errorf("Cannot add or update Varnish service %s: "+
@@ -369,9 +383,9 @@ func (vc *VarnishController) AddOrUpdateVarnishSvc(key string,
 
 	_, ok := vc.varnishSvcs[key]
 	if !ok {
-		return vc.addVarnishSvc(key, addrs)
+		return vc.addVarnishSvc(key, addrs, loadVCL)
 	}
-	return vc.updateVarnishSvc(key, addrs)
+	return vc.updateVarnishSvc(key, addrs, loadVCL)
 }
 
 func (vc *VarnishController) DeleteVarnishSvc(key string) error {
@@ -379,8 +393,11 @@ func (vc *VarnishController) DeleteVarnishSvc(key string) error {
 	if !ok {
 		return nil
 	}
-	delete(vc.varnishSvcs, key)
-	return vc.removeVarnishInstances(svcs)
+	err := vc.removeVarnishInstances(svcs)
+	if err != nil {
+		delete(vc.varnishSvcs, key)
+	}
+	return err
 }
 
 func (vc *VarnishController) Update(key, uid string, spec vcl.Spec) error {