Fix headless service issue with SSL passthrough

.travis.yml

@@ -34,6 +34,9 @@ jobs:
     - go get github.com/golang/lint/golint
     - make fmt lint vet
   - stage: Coverage
+    before_script:
+    - make e2e-image
+    - test/e2e/up.sh
     script:
     - go get github.com/mattn/goveralls
     - go get github.com/modocache/gover

cmd/nginx/flags.go

@@ -27,15 +27,12 @@ import (
 
 	apiv1 "k8s.io/api/core/v1"
 
+	"k8s.io/ingress-nginx/internal/ingress"
 	"k8s.io/ingress-nginx/internal/ingress/controller"
 	ngx_config "k8s.io/ingress-nginx/internal/ingress/controller/config"
 	ing_net "k8s.io/ingress-nginx/internal/net"
 )
 
-const (
-	defIngressClass = "nginx"
-)
-
 func parseFlags() (bool, *controller.Configuration, error) {
 	var (
 		flags = pflag.NewFlagSet("", pflag.ExitOnError)
@@ -157,6 +154,8 @@ func parseFlags() (bool, *controller.Configuration, error) {
 		}
 	}
 
+	ingress.IngressClass = *ingressClass
+
 	// check port collisions
 	if !ing_net.IsPortAvailable(*httpPort) {
 		return false, nil, fmt.Errorf("Port %v is already in use. Please check the flag --http-port", *httpPort)
@@ -198,7 +197,6 @@ func parseFlags() (bool, *controller.Configuration, error) {
 		EnableSSLChainCompletion: *enableSSLChainCompletion,
 		ResyncPeriod:             *resyncPeriod,
 		DefaultService:           *defaultSvc,
-		IngressClass:             *ingressClass,
 		Namespace:                *watchNamespace,
 		ConfigMapName:            *configMap,
 		TCPConfigMapName:         *tcpConfigMapName,

cmd/nginx/main.go

@@ -127,6 +127,7 @@ func main() {
 	ngx := controller.NewNGINXController(conf)
 
 	if conf.EnableSSLPassthrough {
+		glog.Info("setting up TLS proxy for SSL passthrough")
 		setupSSLProxy(conf.ListenPorts.HTTPS, conf.ListenPorts.SSLProxy, ngx)
 	}
 

internal/ingress/controller/controller.go

@@ -37,8 +37,6 @@ import (
 	clientset "k8s.io/client-go/kubernetes"
 
 	"k8s.io/ingress-nginx/internal/ingress"
-	"k8s.io/ingress-nginx/internal/ingress/annotations"
-	"k8s.io/ingress-nginx/internal/ingress/annotations/class"
 	"k8s.io/ingress-nginx/internal/ingress/annotations/healthcheck"
 	"k8s.io/ingress-nginx/internal/ingress/annotations/parser"
 	"k8s.io/ingress-nginx/internal/ingress/annotations/proxy"
@@ -76,8 +74,8 @@ type Configuration struct {
 
 	ConfigMapName  string
 	DefaultService string
-	IngressClass   string
-	Namespace      string
+
+	Namespace string
 
 	ForceNamespaceIsolation bool
 
@@ -117,26 +115,6 @@ func (n NGINXController) GetDefaultBackend() defaults.Backend {
 	return n.backendDefaults
 }
 
-// GetPublishService returns the configured service used to set ingress status
-func (n NGINXController) GetPublishService() *apiv1.Service {
-	s, err := n.listers.Service.GetByName(n.cfg.PublishService)
-	if err != nil {
-		return nil
-	}
-
-	return s
-}
-
-// GetSecret searches for a secret in the local secrets Store
-func (n NGINXController) GetSecret(name string) (*apiv1.Secret, error) {
-	return n.listers.Secret.GetByName(name)
-}
-
-// GetService searches for a service in the local secrets Store
-func (n NGINXController) GetService(name string) (*apiv1.Service, error) {
-	return n.listers.Service.GetByName(name)
-}
-
 // GetAnnotationWithPrefix returns the prefix of ingress annotations
 func (n NGINXController) GetAnnotationWithPrefix(suffix string) string {
 	return fmt.Sprintf("%v/%v", n.cfg.AnnotationsPrefix, suffix)
@@ -154,32 +132,20 @@ func (n *NGINXController) syncIngress(item interface{}) error {
 
 	if element, ok := item.(task.Element); ok {
 		if name, ok := element.Key.(string); ok {
-			if obj, exists, _ := n.listers.Ingress.GetByKey(name); exists {
-				ing := obj.(*extensions.Ingress)
+			if ing, err := n.store.GetIngress(name); err == nil {
 				n.readSecrets(ing)
 			}
 		}
 	}
 
 	// Sort ingress rules using the ResourceVersion field
-	ings := n.listers.Ingress.List()
-	sort.SliceStable(ings, func(i, j int) bool {
-		ir := ings[i].(*extensions.Ingress).ResourceVersion
-		jr := ings[j].(*extensions.Ingress).ResourceVersion
+	ingresses := n.store.ListIngresses()
+	sort.SliceStable(ingresses, func(i, j int) bool {
+		ir := ingresses[i].ResourceVersion
+		jr := ingresses[j].ResourceVersion
 		return ir < jr
 	})
 
-	// filter ingress rules
-	var ingresses []*extensions.Ingress
-	for _, ingIf := range ings {
-		ing := ingIf.(*extensions.Ingress)
-		if !class.IsValid(ing, n.cfg.IngressClass, n.cfg.DefaultIngressClass) {
-			continue
-		}
-
-		ingresses = append(ingresses, ing)
-	}
-
 	upstreams, servers := n.getBackendServers(ingresses)
 	var passUpstreams []*ingress.SSLPassthroughBackend
 
@@ -248,7 +214,7 @@ func (n *NGINXController) getStreamServices(configmapName string, proto apiv1.Pr
 		return []ingress.L4Service{}
 	}
 
-	configmap, err := n.listers.ConfigMap.GetByName(configmapName)
+	configmap, err := n.store.GetConfigMap(configmapName)
 	if err != nil {
 		glog.Errorf("unexpected error reading configmap %v: %v", configmapName, err)
 		return []ingress.L4Service{}
@@ -306,19 +272,12 @@ func (n *NGINXController) getStreamServices(configmapName string, proto apiv1.Pr
 			continue
 		}
 
-		svcObj, svcExists, err := n.listers.Service.GetByKey(nsName)
+		svc, err := n.store.GetService(nsName)
 		if err != nil {
 			glog.Warningf("error getting service %v: %v", nsName, err)
 			continue
 		}
 
-		if !svcExists {
-			glog.Warningf("service %v was not found", nsName)
-			continue
-		}
-
-		svc := svcObj.(*apiv1.Service)
-
 		var endps []ingress.Endpoint
 		targetPort, err := strconv.Atoi(svcPort)
 		if err != nil {
@@ -375,20 +334,13 @@ func (n *NGINXController) getDefaultUpstream() *ingress.Backend {
 		Name: defUpstreamName,
 	}
 	svcKey := n.cfg.DefaultService
-	svcObj, svcExists, err := n.listers.Service.GetByKey(svcKey)
+	svc, err := n.store.GetService(svcKey)
 	if err != nil {
 		glog.Warningf("unexpected error searching the default backend %v: %v", n.cfg.DefaultService, err)
 		upstream.Endpoints = append(upstream.Endpoints, n.DefaultEndpoint())
 		return upstream
 	}
 
-	if !svcExists {
-		glog.Warningf("service %v does not exist", svcKey)
-		upstream.Endpoints = append(upstream.Endpoints, n.DefaultEndpoint())
-		return upstream
-	}
-
-	svc := svcObj.(*apiv1.Service)
 	endps := n.getEndpoints(svc, &svc.Spec.Ports[0], apiv1.ProtocolTCP, &healthcheck.Config{})
 	if len(endps) == 0 {
 		glog.Warningf("service %v does not have any active endpoints", svcKey)
@@ -408,7 +360,10 @@ func (n *NGINXController) getBackendServers(ingresses []*extensions.Ingress) ([]
 	servers := n.createServers(ingresses, upstreams, du)
 
 	for _, ing := range ingresses {
-		anns := n.getIngressAnnotations(ing)
+		anns, err := n.store.GetIngressAnnotations(ing)
+		if err != nil {
+			glog.Warningf("%v", err)
+		}
 
 		for _, rule := range ing.Spec.Rules {
 			host := rule.Host
@@ -622,20 +577,19 @@ func (n *NGINXController) getBackendServers(ingresses []*extensions.Ingress) ([]
 
 // GetAuthCertificate is used by the auth-tls annotations to get a cert from a secret
 func (n NGINXController) GetAuthCertificate(name string) (*resolver.AuthSSLCert, error) {
-	if _, exists := n.sslCertTracker.Get(name); !exists {
+	if _, err := n.store.GetLocalSecret(name); err != nil {
 		n.syncSecret(name)
 	}
 
-	_, err := n.listers.Secret.GetByName(name)
+	_, err := n.store.GetLocalSecret(name)
 	if err != nil {
 		return &resolver.AuthSSLCert{}, fmt.Errorf("unexpected error: %v", err)
 	}
 
-	bc, exists := n.sslCertTracker.Get(name)
-	if !exists {
-		return &resolver.AuthSSLCert{}, fmt.Errorf("secret %v does not exist", name)
+	cert, err := n.store.GetLocalSecret(name)
+	if err != nil {
+		return &resolver.AuthSSLCert{}, err
 	}
-	cert := bc.(*ingress.SSLCert)
 	return &resolver.AuthSSLCert{
 		Secret:     name,
 		CAFileName: cert.CAFileName,
@@ -650,7 +604,10 @@ func (n *NGINXController) createUpstreams(data []*extensions.Ingress, du *ingres
 	upstreams[defUpstreamName] = du
 
 	for _, ing := range data {
-		anns := n.getIngressAnnotations(ing)
+		anns, err := n.store.GetIngressAnnotations(ing)
+		if err != nil {
+			glog.Warningf("%v", err)
+		}
 
 		var defBackend string
 		if ing.Spec.Backend != nil {
@@ -737,7 +694,7 @@ func (n *NGINXController) createUpstreams(data []*extensions.Ingress, du *ingres
 					upstreams[name].Endpoints = endp
 				}
 
-				s, err := n.listers.Service.GetByName(svcKey)
+				s, err := n.store.GetService(svcKey)
 				if err != nil {
 					glog.Warningf("error obtaining service: %v", err)
 					continue
@@ -752,13 +709,11 @@ func (n *NGINXController) createUpstreams(data []*extensions.Ingress, du *ingres
 }
 
 func (n *NGINXController) getServiceClusterEndpoint(svcKey string, backend *extensions.IngressBackend) (endpoint ingress.Endpoint, err error) {
-	svcObj, svcExists, err := n.listers.Service.GetByKey(svcKey)
-
-	if !svcExists {
-		return endpoint, fmt.Errorf("service %v does not exist", svcKey)
+	svc, err := n.store.GetService(svcKey)
+	if err != nil {
+		return endpoint, err
 	}
 
-	svc := svcObj.(*apiv1.Service)
 	if svc.Spec.ClusterIP == "" || svc.Spec.ClusterIP == "None" {
 		return endpoint, fmt.Errorf("No ClusterIP found for service %s", svcKey)
 	}
@@ -790,7 +745,7 @@ func (n *NGINXController) getServiceClusterEndpoint(svcKey string, backend *exte
 // to a service.
 func (n *NGINXController) serviceEndpoints(svcKey, backendPort string,
 	hz *healthcheck.Config) ([]ingress.Endpoint, error) {
-	svc, err := n.listers.Service.GetByName(svcKey)
+	svc, err := n.store.GetService(svcKey)
 
 	var upstreams []ingress.Endpoint
 	if err != nil {
@@ -915,7 +870,10 @@ func (n *NGINXController) createServers(data []*extensions.Ingress,
 
 	// initialize all the servers
 	for _, ing := range data {
-		anns := n.getIngressAnnotations(ing)
+		anns, err := n.store.GetIngressAnnotations(ing)
+		if err != nil {
+			glog.Warningf("%v", err)
+		}
 
 		// default upstream server
 		un := du.Name
@@ -966,7 +924,10 @@ func (n *NGINXController) createServers(data []*extensions.Ingress,
 
 	// configure default location, alias, and SSL
 	for _, ing := range data {
-		anns := n.getIngressAnnotations(ing)
+		anns, err := n.store.GetIngressAnnotations(ing)
+		if err != nil {
+			glog.Warningf("%v", err)
+		}
 
 		for _, rule := range ing.Spec.Rules {
 			host := rule.Host
@@ -1031,13 +992,12 @@ func (n *NGINXController) createServers(data []*extensions.Ingress,
 			}
 
 			key := fmt.Sprintf("%v/%v", ing.Namespace, tlsSecretName)
-			bc, exists := n.sslCertTracker.Get(key)
-			if !exists {
-				glog.Warningf("ssl certificate \"%v\" does not exist in local store", key)
+			cert, err := n.store.GetLocalSecret(key)
+			if err != nil {
+				glog.Warningf("%v", err)
 				continue
 			}
 
-			cert := bc.(*ingress.SSLCert)
 			err = cert.Certificate.VerifyHostname(host)
 			if err != nil {
 				glog.Warningf("ssl certificate %v does not contain a Common Name or Subject Alternative Name for host %v", key, host)
@@ -1107,7 +1067,7 @@ func (n *NGINXController) getEndpoints(
 	}
 
 	glog.V(3).Infof("getting endpoints for service %v/%v and port %v", s.Namespace, s.Name, servicePort.String())
-	ep, err := n.listers.Endpoint.GetServiceEndpoints(s)
+	ep, err := n.store.GetServiceEndpoints(s)
 	if err != nil {
 		glog.Warningf("unexpected error obtaining service endpoints: %v", err)
 		return upsServers
@@ -1187,24 +1147,3 @@ func (n *NGINXController) SetForceReload(shouldReload bool) {
 		atomic.StoreInt32(&n.forceReload, 0)
 	}
 }
-
-func (n *NGINXController) extractAnnotations(ing *extensions.Ingress) {
-	anns := n.annotations.Extract(ing)
-	glog.V(3).Infof("updating annotations information for ingres %v/%v", anns.Namespace, anns.Name)
-	n.listers.IngressAnnotation.Update(anns)
-}
-
-// getByIngress returns the parsed annotations from an Ingress
-func (n *NGINXController) getIngressAnnotations(ing *extensions.Ingress) *annotations.Ingress {
-	key := fmt.Sprintf("%v/%v", ing.Namespace, ing.Name)
-	item, exists, err := n.listers.IngressAnnotation.GetByKey(key)
-	if err != nil {
-		glog.Errorf("unexpected error getting ingress annotation %v: %v", key, err)
-		return &annotations.Ingress{}
-	}
-	if !exists {
-		glog.Errorf("ingress annotation %v was not found", key)
-		return &annotations.Ingress{}
-	}
-	return item.(*annotations.Ingress)
-}