Ingress Hostname based traffic forwarding

docs/user-guide/component/ingress/README.md

@@ -25,7 +25,8 @@ hosting. This plugin also support configurable application ports with all the fe
   - [Cross namespace routing support](named-virtual-hosting.md),
   - [URL and Request Header Re-writing](header-rewrite.md),
   - [Wildcard Name based virtual hosting](named-virtual-hosting.md),
-  - Persistent sessions, Loadbalancer stats.
+  - Persistent sessions, Loadbalancer stats,
+  - [Route Traffic to StatefulSet Pods Based on Host Name](statefulset-pod.md)
 
 ### Comparison with Kubernetes
 | Feauture | Kube Ingress | AppsCode Ingress |
@@ -38,6 +39,7 @@ hosting. This plugin also support configurable application ports with all the fe
 | URL and Header rewriting | :x: | :white_check_mark: |
 | Wildcard name virtual hosting | :x: | :white_check_mark: |
 | Loadbalancer statistics | :x: | :white_check_mark: |
+| Route Traffic to StatefulSet Pods Based on Host Name | :x: | :white_check_mark: |
 
 ## AppsCode Ingress Flow
 Typically, services and pods have IPs only routable by the cluster network. All traffic that ends up at an
@@ -126,6 +128,7 @@ same ingress resource. Learn more by reading the certificate doc.
 - [URL and Header Rewriting](header-rewrite.md)
 - [TCP Loadbalancing](tcp.md)
 - [TLS Termination](tls.md)
+- [Route Traffic to StatefulSet Pods Based on Host Name](statefulset-pod.md)
 
 
 ## Example

docs/user-guide/component/ingress/statefulset-pod.md

@@ -0,0 +1,99 @@
+### Forward Traffic to StatefulSet
+There is the regular way to forward traffic to StatefulSet. Create a service with the pods label selector as
+selector, and use the service name as Backend ServiceName. By following:
+
+```
+apiVersion: apps/v1beta1
+kind: StatefulSet
+metadata:
+  name: web
+spec:
+  serviceName: "nginx-set"
+  replicas: 2
+  template:
+    metadata:
+      labels:
+        app: nginx
+    spec:
+      containers:
+      - name: nginx
+        image: gcr.io/google_containers/nginx-slim:0.8
+        ports:
+        - containerPort: 80
+          name: web
+----
+apiVersion: v1
+kind: Service
+metadata:
+  name: nginx-set
+  labels:
+    app: nginx
+spec:
+  ports:
+  - port: 80
+    name: web
+  clusterIP: None
+  selector:
+    app: nginx
+```
+
+Create another service for StatefulSets pods with selector.
+```
+apiVersion: v1
+kind: Service
+metadata:
+  name: nginx-service
+  labels:
+    app: nginx
+spec:
+  ports:
+  - port: 80
+    name: web
+  selector:
+    app: nginx
+
+```
+
+And Use the service in the ingress Backend service name, as:
+```
+backend:
+  serviceName: nginx-service
+  servicePort: '80'
+```
+
+That will forward traffic to your StatefulSets Pods.
+
+
+#### Forward Traffic to specific Pods of StatefulSet
+There is a way to send traffic to all or specific pod of a StatefulSet using voyager. You can set
+`hostNames` field in `Backend`, traffic will only forwarded to those pods.
+
+For Example the above StatefulSet will create two pod.
+```
+web-0
+web-1
+```
+Those are the host names.
+
+Now Create a ingress that will only forward traffic to web-0
+```yaml
+apiVersion: appscode.com/v1beta1
+kind: Ingress
+metadata:
+  name: test-ingress
+  namespace: default
+spec:
+  rules:
+  - host: appscode.example.com
+    http:
+      paths:
+      - path: '/testPath'
+        backend:
+          hostNames:
+          - web-0
+          serviceName: nginx-set #! There is no extra service. This
+          servicePort: '80'      # is the Statefulset's Headless Service
+```
+
+Viola. Now all `/testPath` traffic will be sent to pod web-0 only. There is no extra service also.
+The StatefulSet's Headless Service is enough. By using all the hostNames You can forward traffic to all pods.

hack/make.py

@@ -163,7 +163,7 @@ def test(type, *args):
     elif type == 'clean':
         e2e_test_clean()
     else:
-        print '{test unit|minikube|e2e|clean}'
+        print '{test unit|minikube|e2e}'
 
 def unit_test():
     die(call(libbuild.GOC + ' test -v ./cmd/... ./pkg/... -args -v=3 -verbose=true -mode=unit'))
@@ -180,10 +180,6 @@ def integration(args):
     st = ' '.join(args)
     die(call(libbuild.GOC + ' test -v ./test/integration/... -timeout 10h -args -v=3 -verbose=true -mode=e2e -in-cluster=true ' + st))
 
-def e2e_test_clean():
-    die(call('./test/hack/cleanup.sh'))
-
-
 def default():
     gen()
     fmt()

pkg/controller/ingress/parser.go

@@ -27,7 +27,7 @@ func (lbc *EngressController) parse() error {
 	return nil
 }
 
-func (lbc *EngressController) serviceEndpoints(name string, port intstr.IntOrString) ([]*Endpoint, error) {
+func (lbc *EngressController) serviceEndpoints(name string, port intstr.IntOrString, hostNames []string) ([]*Endpoint, error) {
 	log.Infoln("getting endpoints for ", lbc.Config.Namespace, name, "port", port)
 
 	// the following lines giving support to
@@ -48,10 +48,10 @@ func (lbc *EngressController) serviceEndpoints(name string, port intstr.IntOrStr
 	if !ok {
 		return nil, errors.New().WithMessage("service port unavaiable").NotFound()
 	}
-	return lbc.getEndpoints(service, p)
+	return lbc.getEndpoints(service, p, hostNames)
 }
 
-func (lbc *EngressController) getEndpoints(s *kapi.Service, servicePort *kapi.ServicePort) (eps []*Endpoint, err error) {
+func (lbc *EngressController) getEndpoints(s *kapi.Service, servicePort *kapi.ServicePort, hostNames []string) (eps []*Endpoint, err error) {
 	ep, err := lbc.EndpointStore.GetServiceEndpoints(s)
 	if err != nil {
 		return nil, errors.New().WithCause(err).Internal()
@@ -86,17 +86,32 @@ func (lbc *EngressController) getEndpoints(s *kapi.Service, servicePort *kapi.Se
 
 			log.Infoln("targert port", targetPort)
 			for _, epAddress := range ss.Addresses {
-				eps = append(eps, &Endpoint{
-					Name: "server-" + epAddress.IP,
-					IP:   epAddress.IP,
-					Port: targetPort,
-				})
+				if isForwardable(hostNames, epAddress.Hostname) {
+					eps = append(eps, &Endpoint{
+						Name: "server-" + epAddress.IP,
+						IP:   epAddress.IP,
+						Port: targetPort,
+					})
+				}
 			}
 		}
 	}
 	return
 }
 
+func isForwardable(hostNames []string, hostName string) bool {
+	if len(hostNames) <= 0 {
+		return true
+	}
+
+	for _, name := range hostNames {
+		if strings.EqualFold(name, hostName) {
+			return true
+		}
+	}
+	return false
+}
+
 func (lbc *EngressController) generateTemplate() error {
 	log.Infoln("Generating Ingress template.")
 	ctx, err := Context(lbc.Parsed)
@@ -154,7 +169,7 @@ func (lbc *EngressController) parseSpec() {
 	lbc.Options.Ports = make([]int, 0)
 	if lbc.Config.Spec.Backend != nil {
 		log.Debugln("generating defulat backend", lbc.Config.Spec.Backend.RewriteRule, lbc.Config.Spec.Backend.HeaderRule)
-		eps, _ := lbc.serviceEndpoints(lbc.Config.Spec.Backend.ServiceName, lbc.Config.Spec.Backend.ServicePort)
+		eps, _ := lbc.serviceEndpoints(lbc.Config.Spec.Backend.ServiceName, lbc.Config.Spec.Backend.ServicePort, lbc.Config.Spec.Backend.HostNames)
 		lbc.Parsed.DefaultBackend = &Backend{
 			Name:      "default-backend",
 			Endpoints: eps,
@@ -185,7 +200,7 @@ func (lbc *EngressController) parseSpec() {
 					AclMatch: svc.Path,
 				}
 
-				eps, err := lbc.serviceEndpoints(svc.Backend.ServiceName, svc.Backend.ServicePort)
+				eps, err := lbc.serviceEndpoints(svc.Backend.ServiceName, svc.Backend.ServicePort, svc.Backend.HostNames)
 				def.Backends = &Backend{
 					Name:      "backend-" + rand.Characters(5),
 					Endpoints: eps,
@@ -216,7 +231,7 @@ func (lbc *EngressController) parseSpec() {
 				ALPNOptions: parseALPNOptions(tcpSvc.ALPN),
 			}
 			log.Infoln(tcpSvc.Backend.ServiceName, tcpSvc.Backend.ServicePort)
-			eps, err := lbc.serviceEndpoints(tcpSvc.Backend.ServiceName, tcpSvc.Backend.ServicePort)
+			eps, err := lbc.serviceEndpoints(tcpSvc.Backend.ServiceName, tcpSvc.Backend.ServicePort, tcpSvc.Backend.HostNames)
 			def.Backends = &Backend{
 				Name:      "backend-" + rand.Characters(5),
 				Endpoints: eps,

pkg/controller/ingress/parser_test.go

@@ -43,3 +43,31 @@ func TestALPNOptions(t *testing.T) {
 		assert.Equal(t, k, parseALPNOptions(v))
 	}
 }
+
+func TestIsIncludeAbleOptions(t *testing.T) {
+	dataTable := map[string]map[bool][]string{
+		"web-0": {
+			true: {
+				"web",
+				"web-0",
+			},
+		},
+
+		"web-1": {
+			true: {},
+		},
+
+		"web-2": {
+			false: {
+				"web",
+				"web-0",
+			},
+		},
+	}
+
+	for k, v := range dataTable {
+		for key, val := range v {
+			assert.Equal(t, key, isIncludeAbleAddress(val, k))
+		}
+	}
+}

test/e2e/ingress.go

@@ -888,3 +888,109 @@ func (ing *IngressTestSuit) TestIngressCoreIngress() error {
 	}
 	return nil
 }
+
+func (ing *IngressTestSuit) TestIngressHostNames() error {
+	headlessSvc, err := ing.t.KubeClient.Core().Services("default").Create(testStatefulSetSvc)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if ing.t.Config.Cleanup {
+			ing.t.KubeClient.Core().Services("default").Delete(headlessSvc.Name, nil)
+		}
+	}()
+
+	ss, err := ing.t.KubeClient.Apps().StatefulSets("default").Create(testServerStatefulSet)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if ing.t.Config.Cleanup {
+			ing.t.KubeClient.Apps().StatefulSets("default").Delete(ss.Name, nil)
+		}
+	}()
+
+	baseIngress := &aci.Ingress{
+		ObjectMeta: api.ObjectMeta{
+			Name:      testIngressName(),
+			Namespace: TestNamespace,
+		},
+		Spec: aci.ExtendedIngressSpec{
+			Rules: []aci.ExtendedIngressRule{
+				{
+					ExtendedIngressRuleValue: aci.ExtendedIngressRuleValue{
+						HTTP: &aci.HTTPExtendedIngressRuleValue{
+							Paths: []aci.HTTPExtendedIngressPath{
+								{
+									Path: "/testpath",
+									Backend: aci.ExtendedIngressBackend{
+										HostNames:   []string{testServerStatefulSet.Name + "-0"},
+										ServiceName: headlessSvc.Name,
+										ServicePort: intstr.FromInt(80),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	_, err = ing.t.ExtensionClient.Ingress(baseIngress.Namespace).Create(baseIngress)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if ing.t.Config.Cleanup {
+			ing.t.ExtensionClient.Ingress(baseIngress.Namespace).Delete(baseIngress.Name)
+		}
+	}()
+
+	// Wait sometime to loadbalancer be opened up.
+	time.Sleep(time.Second * 10)
+	var svc *api.Service
+	for i := 0; i < maxRetries; i++ {
+		svc, err = ing.t.KubeClient.Core().Services(baseIngress.Namespace).Get(ingress.VoyagerPrefix + baseIngress.Name)
+		if err == nil {
+			break
+		}
+		time.Sleep(time.Second * 5)
+		log.Infoln("Waiting for service to be created")
+	}
+	if err != nil {
+		return err
+	}
+	log.Infoln("Service Created for loadbalancer, Checking for service endpoints")
+	for i := 0; i < maxRetries; i++ {
+		_, err = ing.t.KubeClient.Core().Endpoints(svc.Namespace).Get(svc.Name)
+		if err == nil {
+			break
+		}
+		time.Sleep(time.Second * 5)
+		log.Infoln("Waiting for endpoints to be created")
+	}
+	if err != nil {
+		return err
+	}
+
+	serverAddr, err := ing.getURLs(baseIngress)
+	if err != nil {
+		return err
+	}
+	time.Sleep(time.Second * 30)
+	log.Infoln("Loadbalancer created, calling http endpoints, Total", len(serverAddr))
+	for _, url := range serverAddr {
+		resp, err := testserverclient.NewTestHTTPClient(url).Method("GET").Path("/testpath").DoWithRetry(50)
+		if err != nil {
+			return errors.New().WithCause(err).WithMessage("Failed to connect with server").Internal()
+		}
+		log.Infoln("Response", *resp)
+		if resp.Method != http.MethodGet {
+			return errors.New().WithMessage("Method did not matched").Internal()
+		}
+		if resp.PodName != ss.Name+"-0" {
+			return errors.New().WithMessage("PodName did not matched").Internal()
+		}
+	}
+	return nil
+}