Initial commit

Dockerfile

+FROM centos:centos7
+
+RUN yum install -y epel-release
+
+RUN yum update -y -q
+
+COPY varnishcache_varnish60.repo /etc/yum.repos.d/
+
+RUN yum -q makecache -y --disablerepo='*' --enablerepo='varnishcache_varnish60'
+
+RUN yum-config-manager --add-repo https://pkg.uplex.de/rpm/7/uplex-varnish/x86_64/
+
+RUN yum install -y -q varnish-6.0.1
+
+RUN yum install -y -q --nogpgcheck vmod-re2
+
+COPY k8s-ingress varnish/vcl/vcl.tmpl /
+
+ENTRYPOINT ["/k8s-ingress"]

controller/controller.go

+/*
+ * Copyright (c) 2018 UPLEX Nils Goroll Systemoptimierung
+ * All rights reserved
+ *
+ * Author: Geoffrey Simmons <geoffrey.simmons@uplex.de>
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+*/
+
+package controller
+
+import (
+	"fmt"
+	"log"
+	"reflect"
+	"time"
+
+	"code.uplex.de/uplex-varnish/k8s-ingress/varnish"
+	"code.uplex.de/uplex-varnish/k8s-ingress/varnish/vcl"
+
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/kubernetes"
+	scheme "k8s.io/client-go/kubernetes/scheme"
+	core_v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+
+	api_v1 "k8s.io/api/core/v1"
+	extensions "k8s.io/api/extensions/v1beta1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// XXX make these configurable
+const (
+	ingressClassKey = "kubernetes.io/ingress.class"
+	resyncPeriod    = 0
+	watchNamespace  = api_v1.NamespaceAll
+//	resyncPeriod    = 30 * time.Second
+)
+
+// IngressController watches Kubernetes API and reconfigures Varnish
+// via VarnishController when needed.
+type IngressController struct {
+	client         kubernetes.Interface
+	vController    *varnish.VarnishController
+	ingController  cache.Controller
+	svcController  cache.Controller
+	endpController cache.Controller
+	ingLister      StoreToIngressLister
+	svcLister      cache.Store
+	endpLister     StoreToEndpointLister
+	syncQueue      *taskQueue
+	stopCh         chan struct{}
+	recorder       record.EventRecorder
+}
+
+var keyFunc = cache.DeletionHandlingMetaNamespaceKeyFunc
+
+// NewIngressController creates a controller
+func NewIngressController(kubeClient kubernetes.Interface,
+	vc *varnish.VarnishController, namespace string) *IngressController {
+
+	ingc := IngressController{
+		client:      kubeClient,
+		stopCh:      make(chan struct{}),
+		vController: vc,
+	}
+
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(log.Printf)
+	eventBroadcaster.StartRecordingToSink(&core_v1.EventSinkImpl{
+		Interface: core_v1.New(ingc.client.Core().RESTClient()).
+			Events(""),
+	})
+	ingc.recorder = eventBroadcaster.NewRecorder(scheme.Scheme,
+		api_v1.EventSource{Component: "varnish-ingress-controller"})
+
+	ingc.syncQueue = NewTaskQueue(ingc.sync)
+
+	log.Print("Varnish Ingress Controller has class: varnish")
+
+	ingHandlers := cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			addIng := obj.(*extensions.Ingress)
+			log.Print("ingHandler.AddFunc:", addIng)
+			if !ingc.isVarnishIngress(addIng) {
+				log.Printf("Ignoring Ingress %v based on "+
+					"Annotation %v", addIng.Name,
+					ingressClassKey)
+				return
+			}
+			log.Printf("Adding Ingress: %v", addIng.Name)
+			ingc.syncQueue.enqueue(obj)
+		},
+		DeleteFunc: func(obj interface{}) {
+			remIng, isIng := obj.(*extensions.Ingress)
+			log.Print("ingHandler.DeleteFunc:", remIng, isIng)
+			if !isIng {
+				deletedState, ok :=
+					obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					log.Print("Error received unexpected "+
+						"object:", obj)
+					return
+				}
+				remIng, ok =
+					deletedState.Obj.(*extensions.Ingress)
+				if !ok {
+					log.Print("Error "+
+						"DeletedFinalStateUnknown "+
+						"contained non-Ingress object:",
+						deletedState.Obj)
+					return
+				}
+			}
+			if !ingc.isVarnishIngress(remIng) {
+				return
+			}
+			ingc.syncQueue.enqueue(obj)
+		},
+		UpdateFunc: func(old, cur interface{}) {
+			curIng := cur.(*extensions.Ingress)
+			oldIng := old.(*extensions.Ingress)
+			log.Print("ingHandler.UpdateFunc:", curIng, oldIng)
+			if !ingc.isVarnishIngress(curIng) {
+				return
+			}
+			if hasChanges(oldIng, curIng) {
+				log.Printf("Ingress %v changed, syncing",
+					curIng.Name)
+				ingc.syncQueue.enqueue(cur)
+			}
+		},
+	}
+	ingc.ingLister.Store, ingc.ingController = cache.NewInformer(
+		cache.NewListWatchFromClient(ingc.client.Extensions().
+			RESTClient(), "ingresses", namespace,
+			fields.Everything()),
+		&extensions.Ingress{}, resyncPeriod, ingHandlers)
+	log.Print("ingc.ingLister.Store:", ingc.ingLister.Store)
+	log.Print("ingc.ingController:", ingc.ingController)
+
+	svcHandlers := cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			addSvc := obj.(*api_v1.Service)
+			log.Print("svcHandler.AddFunc:", addSvc)
+			if ingc.isExternalServiceForStatus(addSvc) {
+				ingc.syncQueue.enqueue(addSvc)
+				return
+			}
+			log.Print("Adding service:", addSvc.Name)
+			ingc.enqueueIngressForService(addSvc)
+		},
+		DeleteFunc: func(obj interface{}) {
+			remSvc, isSvc := obj.(*api_v1.Service)
+			log.Print("svcHandler.DeleteFunc:", remSvc, isSvc)
+			if !isSvc {
+				deletedState, ok :=
+					obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					log.Print("Error received unexpected "+
+						"object:", obj)
+					return
+				}
+				remSvc, ok = deletedState.Obj.(*api_v1.Service)
+				if !ok {
+					log.Print("Error "+
+						"DeletedFinalStateUnknown "+
+						"contained non-Service object:",
+						deletedState.Obj)
+					return
+				}
+			}
+
+			log.Print("Removing service:", remSvc.Name)
+			ingc.enqueueIngressForService(remSvc)
+
+		},
+		UpdateFunc: func(old, cur interface{}) {
+			if !reflect.DeepEqual(old, cur) {
+				curSvc := cur.(*api_v1.Service)
+				log.Print("svcHandler.UpdateFunc:", old, curSvc)
+				if ingc.isExternalServiceForStatus(curSvc) {
+					ingc.syncQueue.enqueue(curSvc)
+					return
+				}
+				log.Printf("Service %v changed, syncing",
+					curSvc.Name)
+				ingc.enqueueIngressForService(curSvc)
+			}
+		},
+	}
+	ingc.svcLister, ingc.svcController = cache.NewInformer(
+		cache.NewListWatchFromClient(ingc.client.Core().RESTClient(),
+			"services", namespace, fields.Everything()),
+		&api_v1.Service{}, resyncPeriod, svcHandlers)
+	log.Print("ingc.svcLister.Store:", ingc.svcLister)
+	log.Print("ingc.svcController:", ingc.svcController)
+
+	endpHandlers := cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			addEndp := obj.(*api_v1.Endpoints)
+			log.Print("endpHandler.UpdateFunc:", addEndp)
+			log.Print("Adding endpoints:", addEndp.Name)
+			ingc.syncQueue.enqueue(obj)
+		},
+		DeleteFunc: func(obj interface{}) {
+			remEndp, isEndp := obj.(*api_v1.Endpoints)
+			log.Print("endpHandler.DeleteFunc:", remEndp, isEndp)
+			if !isEndp {
+				deletedState, ok :=
+					obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					log.Print("Error received unexpected "+
+						"object:", obj)
+					return
+				}
+				remEndp, ok =
+					deletedState.Obj.(*api_v1.Endpoints)
+				if !ok {
+					log.Print("Error "+
+						"DeletedFinalStateUnknown "+
+						"contained non-Endpoints "+
+						"object:", deletedState.Obj)
+					return
+				}
+			}
+			log.Print("Removing endpoints:", remEndp.Name)
+			ingc.syncQueue.enqueue(obj)
+		},
+		UpdateFunc: func(old, cur interface{}) {
+			log.Print("endpHandler.UpdateFunc:", old, cur)
+			if !reflect.DeepEqual(old, cur) {
+				log.Printf("Endpoints %v changed, syncing",
+					cur.(*api_v1.Endpoints).Name)
+				ingc.syncQueue.enqueue(cur)
+			} else {
+				log.Print("Update Endpoints: No change")
+			}
+		},
+	}
+	ingc.endpLister.Store, ingc.endpController = cache.NewInformer(
+		cache.NewListWatchFromClient(ingc.client.Core().RESTClient(),
+			"endpoints", namespace, fields.Everything()),
+		&api_v1.Endpoints{}, resyncPeriod, endpHandlers)
+
+	return &ingc
+}
+
+// hasChanges ignores Status or ResourceVersion changes
+func hasChanges(oldIng *extensions.Ingress, curIng *extensions.Ingress) bool {
+	oldIng.Status.LoadBalancer.Ingress = curIng.Status.LoadBalancer.Ingress
+	oldIng.ResourceVersion = curIng.ResourceVersion
+	return !reflect.DeepEqual(oldIng, curIng)
+}
+
+// Run starts the loadbalancer controller
+func (ingc *IngressController) Run() {
+	go ingc.svcController.Run(ingc.stopCh)
+	go ingc.endpController.Run(ingc.stopCh)
+	go ingc.ingController.Run(ingc.stopCh)
+	go ingc.syncQueue.run(time.Second, ingc.stopCh)
+	<-ingc.stopCh
+}
+
+// Stop shutdowns the load balancer controller
+func (ingc *IngressController) Stop() {
+	close(ingc.stopCh)
+
+	ingc.syncQueue.shutdown()
+}
+
+func (ingc *IngressController) addOrUpdateIng(task Task,
+	ing extensions.Ingress) {
+
+	key := ing.ObjectMeta.Namespace + "/" + ing.ObjectMeta.Name
+	log.Printf("Adding or Updating Ingress: %v\n", key)
+
+	vclSpec, err := ingc.ing2VCLSpec(&ing)
+	if err != nil {
+		// XXX make the requeue interval configurable
+		ingc.syncQueue.requeueAfter(task, err, 5*time.Second)
+		ingc.recorder.Eventf(&ing, api_v1.EventTypeWarning, "Rejected",
+			"%v was rejected: %v", key, err)
+		return
+	}
+
+	err = ingc.vController.Update(key, vclSpec)
+	if err != nil {
+		// XXX as above
+		ingc.syncQueue.requeueAfter(task, err, 5*time.Second)
+		ingc.recorder.Eventf(&ing, api_v1.EventTypeWarning,
+			"AddedOrUpdatedWithError",
+			"Configuration for %v was added or updated, but not "+
+				"applied: %v", key, err)
+	}
+}
+
+func (ingc *IngressController) syncEndp(task Task) {
+	key := task.Key
+	log.Print("Syncing endpoints:", key)
+
+	obj, endpExists, err := ingc.endpLister.GetByKey(key)
+	if err != nil {
+		ingc.syncQueue.requeue(task, err)
+		return
+	}
+
+	if endpExists {
+		ings := ingc.getIngForEndp(obj)
+
+		for _, ing := range ings {
+			if !ingc.isVarnishIngress(&ing) {
+				continue
+			}
+			if !ingc.hasIngress(&ing) {
+				continue
+			}
+			ingc.addOrUpdateIng(task, ing)
+		}
+	}
+}
+
+func (ingc *IngressController) sync(task Task) {
+	log.Printf("Syncing %v", task.Key)
+
+	switch task.Kind {
+	case Ingress:
+		ingc.syncIng(task)
+	case Endpoints:
+		ingc.syncEndp(task)
+		return
+	}
+}
+
+func (ingc *IngressController) syncIng(task Task) {
+	key := task.Key
+	ing, ingExists, err := ingc.ingLister.GetByKeySafe(key)
+	if err != nil {
+		ingc.syncQueue.requeue(task, err)
+		return
+	}
+
+	if !ingExists {
+		log.Print("Deleting Ingress:", key)
+
+		err := ingc.vController.DeleteIngress(key)
+		if err != nil {
+			log.Printf("Error deleting configuration for %v: %v",
+				key, err)
+		}
+		return
+	}
+	ingc.addOrUpdateIng(task, *ing)
+}
+
+func (ingc *IngressController) enqueueIngressForService(svc *api_v1.Service) {
+	ings := ingc.getIngForSvc(svc)
+	for _, ing := range ings {
+		if !ingc.isVarnishIngress(&ing) {
+			continue
+		}
+		if !ingc.hasIngress(&ing) {
+			continue
+		}
+		ingc.syncQueue.enqueue(&ing)
+	}
+}
+
+func (ingc *IngressController) getIngForSvc(svc *api_v1.Service) []extensions.Ingress {
+	ings, err := ingc.ingLister.GetServiceIngress(svc)
+	if err != nil {
+		log.Printf("ignoring service %v: %v", svc.Name, err)
+		return nil
+	}
+	return ings
+}
+
+func (ingc *IngressController) getIngForEndp(obj interface{}) []extensions.Ingress {
+	var ings []extensions.Ingress
+	endp := obj.(*api_v1.Endpoints)
+	svcKey := endp.GetNamespace() + "/" + endp.GetName()
+	svcObj, svcExists, err := ingc.svcLister.GetByKey(svcKey)
+	if err != nil {
+		log.Printf("error getting service %v from the cache: %v",
+			svcKey, err)
+	} else {
+		if svcExists {
+			ings = append(ings,
+				ingc.getIngForSvc(svcObj.(*api_v1.Service))...)
+		}
+	}
+	return ings
+}
+
+func (ingc *IngressController) ing2VCLSpec(ing *extensions.Ingress) (vcl.Spec, error) {
+	vclSpec := vcl.Spec{}
+	vclSpec.AllServices = make(map[string]vcl.Service)
+	if ing.Spec.TLS != nil && len(ing.Spec.TLS) > 0 {
+		log.Printf("TLS config currently ignored in Ingress %s",
+			ing.ObjectMeta.Name)
+	}
+	if ing.Spec.Backend != nil {
+		backend := ing.Spec.Backend
+		addrs, err := ingc.ingBackend2Addrs(*backend, ing.Namespace)
+		if err != nil {
+			return vclSpec, err
+		}
+		vclSvc := vcl.Service{
+			Name:      backend.ServiceName,
+			Addresses: addrs,
+		}
+		vclSpec.DefaultService = vclSvc
+		vclSpec.AllServices[backend.ServiceName] = vclSvc
+	}
+	for _, rule := range ing.Spec.Rules {
+		if rule.Host == "" {
+			return vclSpec,	fmt.Errorf("Ingress rule contains "+
+				"empty Host")
+		}
+		vclRule := vcl.Rule{Host: rule.Host}
+		vclRule.PathMap = make(map[string]vcl.Service)
+		if rule.IngressRuleValue.HTTP == nil {
+			vclSpec.Rules = append(vclSpec.Rules, vclRule)
+			continue
+		}
+		for _, path := range rule.IngressRuleValue.HTTP.Paths {
+			addrs, err := ingc.ingBackend2Addrs(path.Backend,
+				ing.Namespace);
+			if err != nil {
+				return vclSpec, err
+			}
+			vclSvc := vcl.Service{
+				Name:      path.Backend.ServiceName,
+				Addresses: addrs,
+			}
+			vclRule.PathMap[path.Path] = vclSvc
+			vclSpec.AllServices[path.Backend.ServiceName] = vclSvc
+		}
+		vclSpec.Rules = append(vclSpec.Rules, vclRule)
+	}
+	return vclSpec, nil
+}
+
+func (ingc *IngressController) ingBackend2Addrs(backend extensions.IngressBackend,
+	namespace string) ([]vcl.Address, error) {
+
+	var addrs []vcl.Address
+	svcKey := namespace + "/" + backend.ServiceName
+	svcObj, ok, err := ingc.svcLister.GetByKey(svcKey)
+	if err != nil {
+		return addrs, err
+	}
+	if !ok {
+		return addrs, fmt.Errorf("service %s does not exist", svcKey)
+	}
+	svc := svcObj.(*api_v1.Service)
+
+	endps, err := ingc.endpLister.GetServiceEndpoints(svc)
+	if err != nil {
+		return addrs, fmt.Errorf("Error getting endpoints for service "+
+			"%v: %v", svc, err)
+	}
+
+	targetPort := int32(0)
+	ingSvcPort := backend.ServicePort
+	for _, port := range svc.Spec.Ports {
+		if ((ingSvcPort.Type == intstr.Int &&
+			port.Port == int32(ingSvcPort.IntValue())) ||
+			(ingSvcPort.Type == intstr.String &&
+			port.Name == ingSvcPort.String())) {
+
+			targetPort, err = ingc.getTargetPort(&port, svc)
+			if err != nil {
+				return addrs, fmt.Errorf("Error determining "+
+					"target port for port %v in Ingress: "+
+					"%v", ingSvcPort, err)
+			}
+			break
+		}
+	}
+	if targetPort == 0 {
+		return addrs, fmt.Errorf("No port %v in service %s", ingSvcPort,
+			svc.Name)
+	}
+
+	for _, subset := range endps.Subsets {
+		for _, port := range subset.Ports {
+			if port.Port == targetPort {
+				for _, address := range subset.Addresses {
+					addr := vcl.Address{
+						IP:   address.IP,
+						Port: port.Port,
+					}
+					addrs = append(addrs, addr)
+				}
+				return addrs, nil
+			}
+		}
+	}
+	return addrs, fmt.Errorf("No endpoints for target port %v in service "+
+		"%s", targetPort, svc.Name)
+}
+
+func (ingc *IngressController) getTargetPort(svcPort *api_v1.ServicePort, svc *api_v1.Service) (int32, error) {
+	if (svcPort.TargetPort == intstr.IntOrString{}) {
+		return svcPort.Port, nil
+	}
+
+	if svcPort.TargetPort.Type == intstr.Int {
+		return int32(svcPort.TargetPort.IntValue()), nil
+	}
+
+	pods, err := ingc.client.Core().Pods(svc.Namespace).
+		List(meta_v1.ListOptions{
+			LabelSelector: labels.Set(svc.Spec.Selector).String(),
+		})
+	if err != nil {
+		return 0, fmt.Errorf("Error getting pod information: %v", err)
+	}
+
+	if len(pods.Items) == 0 {
+		return 0, fmt.Errorf("No pods of service: %v", svc.Name)
+	}
+
+	pod := &pods.Items[0]
+
+	portNum, err := FindPort(pod, svcPort)
+	if err != nil {
+		return 0, fmt.Errorf("Error finding named port %v in pod %s: "+
+			"%v", svcPort, pod.Name, err)
+	}
+
+	return portNum, nil
+}
+
+// Check if resource ingress class annotation (if exists) matches
+// ingress controller class
+func (ingc *IngressController) isVarnishIngress(ing *extensions.Ingress) bool {
+	if class, exists := ing.Annotations[ingressClassKey]; exists {
+		return class == "varnish" || class == ""
+	}
+	return true
+}
+
+func (ingc *IngressController) hasIngress(ing *extensions.Ingress) bool {
+	name := ing.ObjectMeta.Namespace + "/" + ing.ObjectMeta.Name
+	return ingc.vController.HasIngress(name)
+}
+
+// isExternalServiceForStatus matches the service specified by the
+// external-service arg
+func (ingc *IngressController) isExternalServiceForStatus(svc *api_v1.Service) bool {
+	// return ingc.statusUpdater.namespace == svc.Namespace &&
+	// 	ingc.statusUpdater.externalServiceName == svc.Name
+	return false
+}

controller/utils.go

+/*
+ * Copyright (c) 2018 UPLEX Nils Goroll Systemoptimierung
+ * All rights reserved
+ *
+ * Author: Geoffrey Simmons <geoffrey.simmons@uplex.de>
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+*/
+
+package controller
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+
+	api_v1 "k8s.io/api/core/v1"
+	extensions "k8s.io/api/extensions/v1beta1"
+)
+
+// taskQueue manages a work queue through an independent worker that
+// invokes the given sync function for every work item inserted.
+type taskQueue struct {
+	// queue is the work queue the worker polls
+	queue *workqueue.Type
+	// sync is called for each item in the queue
+	sync func(Task)
+	// workerDone is closed when the worker exits
+	workerDone chan struct{}
+}
+
+func (t *taskQueue) run(period time.Duration, stopCh <-chan struct{}) {
+	wait.Until(t.worker, period, stopCh)
+}
+
+// enqueue enqueues ns/name of the given api object in the task queue.
+func (t *taskQueue) enqueue(obj interface{}) {
+	key, err := keyFunc(obj)
+	if err != nil {
+		log.Printf("Couldn't get key for object %v: %v", obj, err)
+		return
+	}
+
+	task, err := NewTask(key, obj)
+	if err != nil {
+		log.Printf("Couldn't create a task for object %v: %v", obj, err)
+		return
+	}
+
+	log.Print("Adding an element with a key:", task.Key)
+
+	t.queue.Add(task)
+}
+
+func (t *taskQueue) requeue(task Task, err error) {
+	log.Printf("Requeuing %v, err %v", task.Key, err)
+	t.queue.Add(task)
+}
+
+func (t *taskQueue) requeueAfter(task Task, err error, after time.Duration) {
+	log.Printf("Requeuing %v after %s, err %v", task.Key, after.String(),
+		err)
+	go func(task Task, after time.Duration) {
+		time.Sleep(after)
+		t.queue.Add(task)
+	}(task, after)
+}
+
+// worker processes work in the queue through sync.
+func (t *taskQueue) worker() {
+	for {
+		task, quit := t.queue.Get()
+		if quit {
+			close(t.workerDone)
+			return
+		}
+		log.Printf("Syncing %v", task.(Task).Key)
+		t.sync(task.(Task))
+		t.queue.Done(task)
+	}
+}
+
+// shutdown shuts down the work queue and waits for the worker to ACK
+func (t *taskQueue) shutdown() {
+	t.queue.ShutDown()
+	<-t.workerDone
+}
+
+// NewTaskQueue creates a new task queue with the given sync function.
+// The sync function is called for every element inserted into the queue.
+func NewTaskQueue(syncFn func(Task)) *taskQueue {
+	return &taskQueue{
+		queue:      workqueue.New(),
+		sync:       syncFn,
+		workerDone: make(chan struct{}),
+	}
+}
+
+// Kind represents the kind of the Kubernetes resources of a task
+type Kind int
+
+const (
+	// Ingress resource
+	Ingress = iota
+	// Endpoints resource
+	Endpoints
+	// Service resource
+	Service
+)
+
+// Task is an element of a taskQueue
+type Task struct {
+	Kind Kind
+	Key  string
+}
+
+// NewTask creates a new task
+func NewTask(key string, obj interface{}) (Task, error) {
+	var k Kind
+	switch t := obj.(type) {
+	case *extensions.Ingress:
+//		ing := obj.(*extensions.Ingress)
+		k = Ingress
+	case *api_v1.Endpoints:
+		k = Endpoints
+	case *api_v1.Service:
+		k = Service
+	default:
+		return Task{}, fmt.Errorf("Unknown type: %v", t)
+	}
+
+	return Task{k, key}, nil
+}
+
+// compareLinks returns true if the 2 self links are equal.
+// func compareLinks(l1, l2 string) bool {
+// 	// TODO: These can be partial links
+// 	return l1 == l2 && l1 != ""
+// }
+
+// StoreToIngressLister makes a Store that lists Ingress.
+// TODO: Move this to cache/listers post 1.1.
+type StoreToIngressLister struct {
+	cache.Store
+}
+
+// GetByKeySafe calls Store.GetByKeySafe and returns a copy of the ingress so it is
+// safe to modify.
+func (s *StoreToIngressLister) GetByKeySafe(key string) (ing *extensions.Ingress, exists bool, err error) {
+	item, exists, err := s.Store.GetByKey(key)
+	if !exists || err != nil {
+		return nil, exists, err
+	}
+	ing = item.(*extensions.Ingress).DeepCopy()
+	return
+}
+
+// List lists all Ingress' in the store.
+func (s *StoreToIngressLister) List() (ing extensions.IngressList, err error) {
+	for _, m := range s.Store.List() {
+		ing.Items = append(ing.Items, *(m.(*extensions.Ingress)).DeepCopy())
+	}
+	return ing, nil
+}
+
+// GetServiceIngress gets all the Ingress' that have rules pointing to a service.
+// Note that this ignores services without the right nodePorts.
+func (s *StoreToIngressLister) GetServiceIngress(svc *api_v1.Service) (ings []extensions.Ingress, err error) {
+	for _, m := range s.Store.List() {
+		ing := *m.(*extensions.Ingress).DeepCopy()
+		if ing.Namespace != svc.Namespace {
+			continue
+		}
+		if ing.Spec.Backend != nil {
+			if ing.Spec.Backend.ServiceName == svc.Name {
+				ings = append(ings, ing)
+			}
+		}
+		for _, rules := range ing.Spec.Rules {
+			if rules.IngressRuleValue.HTTP == nil {
+				continue
+			}
+			for _, p := range rules.IngressRuleValue.HTTP.Paths {
+				if p.Backend.ServiceName == svc.Name {
+					ings = append(ings, ing)
+				}
+			}
+		}
+	}
+	if len(ings) == 0 {
+		err = fmt.Errorf("No ingress for service %v", svc.Name)
+	}
+	return
+}
+
+// StoreToEndpointLister makes a Store that lists Endpoints
+type StoreToEndpointLister struct {
+	cache.Store
+}
+
+// GetServiceEndpoints returns the endpoints of a service, matched on service name.
+func (s *StoreToEndpointLister) GetServiceEndpoints(svc *api_v1.Service) (ep api_v1.Endpoints, err error) {
+	for _, m := range s.Store.List() {
+		ep = *m.(*api_v1.Endpoints)
+		if svc.Name == ep.Name && svc.Namespace == ep.Namespace {
+			return ep, nil
+		}
+	}
+	err = fmt.Errorf("could not find endpoints for service: %v", svc.Name)
+	return
+}
+
+// FindPort locates the container port for the given pod and portName.  If the
+// targetPort is a number, use that.  If the targetPort is a string, look that
+// string up in all named ports in all containers in the target pod.  If no
+// match is found, fail.
+func FindPort(pod *api_v1.Pod, svcPort *api_v1.ServicePort) (int32, error) {
+	portName := svcPort.TargetPort
+	switch portName.Type {
+	case intstr.String:
+		name := portName.StrVal
+		for _, container := range pod.Spec.Containers {
+			for _, port := range container.Ports {
+				if port.Name == name &&
+					port.Protocol == svcPort.Protocol {
+					return port.ContainerPort, nil
+				}
+			}
+		}
+	case intstr.Int:
+		return int32(portName.IntValue()), nil
+	}
+
+	return 0, fmt.Errorf("no suitable port for manifest: %s", pod.UID)
+}

examples/cafe-ingress.yaml

+apiVersion: extensions/v1beta1
+kind: Ingress
+metadata:
+  name: cafe-ingress-varnish
+  annotations:
+    kubernetes.io/ingress.class: "varnish"
+  namespace: varnish-ingress
+spec:
+  rules:
+  - host: cafe.example.com
+    http:
+      paths:
+      - path: /tea
+        backend:
+          serviceName: tea-svc
+          servicePort: 80
+      - path: /coffee
+        backend:
+          serviceName: coffee-svc
+          servicePort: 80

examples/cafe.yaml

+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: coffee
+  namespace: varnish-ingress
+spec:
+  replicas: 2
+  selector:
+    matchLabels:
+      app: coffee
+  template:
+    metadata:
+      labels:
+        app: coffee
+    spec:
+      containers:
+      - name: coffee
+        image: nginxdemos/hello:plain-text
+        ports:
+        - containerPort: 80
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: coffee-svc
+  namespace: varnish-ingress
+spec:
+  ports:
+  - port: 80
+    targetPort: 80
+    protocol: TCP
+    name: http
+  selector:
+    app: coffee
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: tea
+  namespace: varnish-ingress
+spec:
+  replicas: 3
+  selector:
+    matchLabels:
+      app: tea 
+  template:
+    metadata:
+      labels:
+        app: tea 
+    spec:
+      containers:
+      - name: tea 
+        image: nginxdemos/hello:plain-text
+        ports:
+        - containerPort: 80
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: tea-svc
+  namespace: varnish-ingress
+  labels:
+spec:
+  ports:
+  - port: 80
+    targetPort: 80
+    protocol: TCP
+    name: http
+  selector:
+    app: tea

install/nodeport.yaml

+apiVersion: v1
+kind: Service
+metadata:
+  name: varnish-ingress
+  namespace: varnish-ingress
+spec:
+  type: NodePort 
+  ports:
+  - port: 80
+    targetPort: 80
+    protocol: TCP
+    name: http
+  selector:
+    app: varnish-ingress

install/ns-and-sa.yaml

+apiVersion: v1
+kind: Namespace
+metadata:
+  name: varnish-ingress 
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: varnish-ingress 
+  namespace: varnish-ingress
\ No newline at end of file

install/rbac.yaml

+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1beta1
+metadata:
+  name: varnish-ingress
+rules:
+- apiGroups:
+  - ""
+  resources:
+  - services
+  - endpoints
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  verbs:
+  - list
+- apiGroups:
+  - ""
+  resources:
+  - events
+  verbs:
+  - create
+  - patch
+- apiGroups:
+  - extensions
+  resources:
+  - ingresses
+  verbs:
+  - list
+  - watch
+  - get
+- apiGroups:
+  - "extensions"
+  resources:
+  - ingresses/status
+  verbs:
+  - update
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1beta1
+metadata:
+  name: varnish-ingress
+subjects:
+- kind: ServiceAccount
+  name: varnish-ingress
+  namespace: varnish-ingress
+roleRef:
+  kind: ClusterRole
+  name: varnish-ingress
+  apiGroup: rbac.authorization.k8s.io

install/varnish-ingress.yaml

+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: varnish-ingress
+  namespace: varnish-ingress
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: varnish-ingress
+  template:
+    metadata:
+      labels:
+        app: varnish-ingress
+    spec:
+      serviceAccountName: varnish-ingress
+      containers:
+      - image: varnish-ingress
+        imagePullPolicy: IfNotPresent
+        name: varnish-ingress
+        ports:
+        - name: http
+          containerPort: 80
+        env:
+        - name: POD_NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        - name: POD_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.name
+        # args:
+        #         # - -varnish-configmaps=$(POD_NAMESPACE)/varnish-config
+        #   - -default-server-tls-secret=$(POD_NAMESPACE)/default-server-secret
+        #   - -v=3 # Enables extensive logging. Useful for trooublshooting.
+        #   - -report-ingress-status
+        #   - -external-service=varnish-ingress
+        #  #- -enable-leader-election

main.go

+package main
+
+//go:generate gogitversion -p main
+
+import (
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"code.uplex.de/uplex-varnish/k8s-ingress/varnish"
+	"code.uplex.de/uplex-varnish/k8s-ingress/controller"
+
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+var versionF = flag.Bool("version", false, "print version and exit")
+
+func main() {
+	flag.Parse()
+
+	if *versionF {
+		fmt.Printf("%s version %s", os.Args[0], version)
+		os.Exit(0)
+	}
+
+	log.Print("Starting Varnish Ingress controller version:", version)
+
+	var err error
+	var config *rest.Config
+
+	config, err = rest.InClusterConfig()
+	if err != nil {
+		log.Fatal("error creating client configuration: %v", err)
+	}
+	kubeClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Fatal("Failed to create client:", err)
+	}
+
+	vController := varnish.NewVarnishController()
+
+	varnishDone := make(chan error, 1)
+	vController.Start(varnishDone)
+
+	namespace := os.Getenv("POD_NAMESPACE")
+	ingController := controller.NewIngressController(kubeClient,
+		vController, namespace)
+	go handleTermination(ingController, vController, varnishDone)
+	ingController.Run()
+}
+
+func handleTermination(ingc *controller.IngressController,
+	vc *varnish.VarnishController, varnishDone chan error) {
+
+	signalChan := make(chan os.Signal, 1)
+	signal.Notify(signalChan, syscall.SIGTERM)
+
+	exitStatus := 0
+	exited := false
+
+	select {
+	case err := <-varnishDone:
+		if err != nil {
+			log.Print("varnish controller exited with an error:",
+				err)
+			exitStatus = 1
+		} else {
+			log.Print("varnish controller exited successfully")
+		}
+		exited = true
+	case <-signalChan:
+		log.Print("Received SIGTERM, shutting down")
+	}
+
+	log.Print("Shutting down the ingress controller")
+	ingc.Stop()
+
+	if !exited {
+		log.Print("Shutting down Varnish")
+		vc.Quit()
+		<-varnishDone
+	}
+
+	log.Print("Exiting with a status:", exitStatus)
+	os.Exit(exitStatus)
+}

varnish/varnish.go

+/*
+ * Copyright (c) 2018 UPLEX Nils Goroll Systemoptimierung
+ * All rights reserved
+ *
+ * Author: Geoffrey Simmons <geoffrey.simmons@uplex.de>
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+*/
+
+/*
+// TODO
+* VCL housekeeping
+  * either discard the previously active VCL immediately on new vcl.use
+  * or periodically clean up
+
+* monitoring
+  * periodically call ping, status, panic.show when otherwise idle
+*/
+
+package varnish
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"os"
+	"os/exec"
+	"os/user"
+	"path/filepath"
+	"strconv"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"code.uplex.de/uplex-varnish/k8s-ingress/varnish/vcl"
+
+	"code.uplex.de/uplex-varnish/varnishapi/pkg/admin"
+	"code.uplex.de/uplex-varnish/varnishapi/pkg/vsm"
+)
+
+// XXX timeout for getting Admin connection (waiting for varnishd start)
+// timeout waiting for child process to stop
+const (
+	vclDir       = "/etc/varnish"
+	vclFile      = "ingress.vcl"
+	varnishLsn   = ":80"
+	varnishdPath = "/usr/sbin/varnishd"
+	notFoundVCL  = `vcl 4.0;
+
+backend default { .host = "192.0.2.255"; .port = "80"; }
+
+sub vcl_recv {
+	return (synth(404));
+}
+`
+)
+
+var (
+	vclPath     = filepath.Join(vclDir, vclFile)
+	tmpPath     = filepath.Join(os.TempDir(), vclFile)
+	varnishArgs = []string{"-a", varnishLsn, "-f", vclPath, "-F"}
+	vcacheUID   int
+	varnishGID  int
+	currentIng  string
+	configCtr   = uint64(0)
+)
+
+type VarnishController struct {
+	varnishdCmd *exec.Cmd
+	adm         *admin.Admin
+	errChan     chan error
+}
+
+func NewVarnishController() *VarnishController {
+	return &VarnishController{}
+}
+
+func (vc *VarnishController) Start(errChan chan error) {
+	vc.errChan = errChan
+
+	log.Print("Starting Varnish controller")
+	vcacheUser, err := user.Lookup("varnish")
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+	varnishGrp, err := user.LookupGroup("varnish")
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+	vcacheUID, err = strconv.Atoi(vcacheUser.Uid)
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+	varnishGID, err = strconv.Atoi(varnishGrp.Gid)
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+
+	notFoundBytes := []byte(notFoundVCL)
+	if err := ioutil.WriteFile(vclPath, notFoundBytes, 0644); err != nil {
+		vc.errChan <- err
+		return
+	}
+	if err := os.Chown(vclPath, vcacheUID, varnishGID); err != nil {
+		vc.errChan <- err
+		return
+	}
+	log.Print("Wrote initial VCL file")
+
+	vc.varnishdCmd = exec.Command(varnishdPath, varnishArgs...)
+	if err := vc.varnishdCmd.Start(); err != nil {
+		vc.errChan <- err
+		return
+	}
+	log.Print("Launched varnishd")
+
+	// XXX config the timeout
+	vsm := vsm.New()
+	if vsm == nil {
+		vc.errChan <- errors.New("Cannot initiate attachment to "+
+			"Varnish shared memory")
+		return
+	}
+	defer vsm.Destroy()
+	if err := vsm.Attach(""); err != nil {
+		vc.errChan <- err
+		return
+	}
+	addr, err := vsm.GetMgmtAddr()
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+	spath, err := vsm.GetSecretPath()
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+	sfile, err := os.Open(spath)
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+	secret, err := ioutil.ReadAll(sfile)
+	if err != nil {
+		vc.errChan <- err
+		return
+	}
+	if vc.adm, err = admin.Dial(addr, secret, 10*time.Second); err != nil {
+		vc.errChan <- err
+		return
+	}
+	log.Print("Got varnish admin connection")
+}
+
+func (vc *VarnishController) Update(key string, spec vcl.Spec) error {
+	if currentIng != "" && currentIng != key {
+		return fmt.Errorf("Multiple Ingress definitions currently not "+
+			"supported: current=%s new=%s", currentIng, key)
+	}
+	currentIng = key
+
+	f, err := os.Create(tmpPath)
+	if err != nil {
+		return err
+	}
+	wr := bufio.NewWriter(f)
+	if err := vcl.Tmpl.Execute(wr, spec); err != nil {
+		return err
+	}
+	wr.Flush()
+	f.Close()
+	log.Printf("Wrote new VCL config to %s", tmpPath)
+	
+	ctr := atomic.AddUint64(&configCtr, 1)
+	configName := fmt.Sprintf("ingress-%d", ctr)
+	if err := vc.adm.VCLLoad(configName, tmpPath); err != nil {
+		log.Print("Failed to load VCL: ", err)
+		return err
+	}
+	log.Printf("Loaded VCL config %s", configName)
+
+	newVCL, err := ioutil.ReadFile(tmpPath)
+	if err != nil {
+		log.Print(err)
+		return err
+	}
+	if err = ioutil.WriteFile(vclPath, newVCL, 0644); err != nil {
+		log.Print(err)
+		return err
+	}
+	log.Printf("Wrote VCL config to %s", vclPath)
+
+	if err = vc.adm.VCLUse(configName); err != nil {
+		log.Print("Failed to activate VCL: ", err)
+		return err
+	}
+	log.Printf("Activated VCL config %s", configName)
+
+	// XXX discard previously active VCL
+	return nil
+}
+
+// We currently only support one Ingress definition at a time, so
+// deleting the Ingress means that we revert to the "boot" config,
+// which returns synthetic 404 Not Found for all requests.
+func (vc *VarnishController) DeleteIngress(key string) error {
+	if currentIng != "" && currentIng != key {
+		return fmt.Errorf("Unknown Ingress %s", key)
+	}
+
+	if err := vc.adm.VCLUse("boot"); err != nil {
+		log.Print("Failed to activate VCL: ", err)
+		return err
+	}
+	log.Printf("Activated VCL config boot")
+
+	currentIng = ""
+	// XXX discard previously active VCL
+	return nil
+}
+
+// Currently only one Ingress at a time
+func (vc *VarnishController) HasIngress(key string) bool {
+	if currentIng == "" {
+		return false
+	}
+	return key == currentIng
+}
+
+func (vc *VarnishController) Quit() {
+	if err := vc.adm.Stop(); err != nil {
+		log.Print("Failed to stop Varnish child process:", err)
+	} else {
+		for {
+			tmoChan := time.After(time.Minute)
+			select {
+			case <-tmoChan:
+				// XXX config the timeout
+				log.Print("timeout waiting for Varnish child " +
+					"process to finish")
+				return
+			default:
+				state, err := vc.adm.Status()
+				if err != nil {
+					log.Print("Can't get Varnish child "+
+						"process status:", err)
+					return
+				}
+				if state != admin.Stopped {
+					continue
+				}
+			}
+		}
+	}
+	vc.adm.Close()
+
+	if err := vc.varnishdCmd.Process.Signal(syscall.SIGTERM); err != nil {
+		log.Print("Failed to stop Varnish:", err)
+		return
+	}
+	log.Print("Stopped Varnish")
+}

varnish/vcl/template_test.go

+package vcl
+
+import (
+	"bytes"
+	"testing"
+)
+
+var teaSvc = Service{
+	Name: "tea-svc",
+	Addresses: []Address{
+		{
+			IP:   "192.0.2.1",
+			Port: 80,
+		},
+		{
+			IP:   "192.0.2.2",
+			Port: 80,
+		},
+		{
+			IP:   "192.0.2.3",
+			Port: 80,
+		},
+	},
+}
+
+var coffeeSvc = Service{
+	Name: "coffee-svc",
+	Addresses: []Address{
+		{
+			IP:   "192.0.2.4",
+			Port: 80,
+		},
+		{
+			IP:   "192.0.2.5",
+			Port: 80,
+		},
+	},
+}
+
+var cafeSpec = Spec{
+	DefaultService: &Service{},
+	Rules: []Rule{{
+		Host: "cafe.example.com",
+		PathMap: map[string]*Service{
+			"/tea":    &teaSvc,
+			"/coffee": &coffeeSvc,
+		},
+	}},
+	AllServices: map[string]*Service{
+		"tea-svc":    &teaSvc,
+		"coffee-svc": &coffeeSvc,
+	},
+}
+
+func TestTemplate(t *testing.T) {
+	var buf bytes.Buffer
+	if err := Tmpl.Execute(&buf, cafeSpec); err != nil {
+		t.Error("Execute():", err)
+	}
+	t.Log(string(buf.Bytes()))
+}

varnish/vcl/vcl.go

+/*
+ * Copyright (c) 2018 UPLEX Nils Goroll Systemoptimierung
+ * All rights reserved
+ *
+ * Author: Geoffrey Simmons <geoffrey.simmons@uplex.de>
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+*/
+
+package vcl
+
+import (
+	"fmt"
+	"regexp"
+	"text/template"
+)
+
+type Address struct {
+	IP   string
+	Port int32
+}
+
+type Service struct {
+	Name      string
+	Addresses []Address
+}
+
+type Rule struct {
+	Host    string
+	PathMap map[string]Service
+}
+
+type Spec struct {
+	DefaultService Service
+	Rules          []Rule
+	AllServices    map[string]Service
+}
+
+var fMap = template.FuncMap{
+	"plusOne":   func(i int) int { return i + 1 },
+	"vclMangle": func(s string) string { return Mangle(s) },
+	"backendName": func(svc Service, addr string) string {
+		return BackendName(svc, addr)
+	},
+	"dirName": func(svc Service) string {
+		return DirectorName(svc)
+	},
+	"urlMatcher": func(rule Rule) string {
+		return URLMatcher(rule)
+	},
+}
+
+const tmplSrc = "vcl.tmpl"
+
+var (
+	Tmpl        = template.Must(template.New(tmplSrc).Funcs(fMap).ParseFiles(tmplSrc))
+	symPattern  = regexp.MustCompile("^[[:alpha:]][[:word:]-]*$")
+	first       = regexp.MustCompile("[[:alpha:]]")
+	restIllegal = regexp.MustCompile("[^[:word:]-]+")
+)
+
+func replIllegal(ill []byte) []byte {
+	repl := []byte("_")
+	for _, b := range ill {
+		repl = append(repl, []byte(fmt.Sprintf("%02x", b))...)
+	}
+	repl = append(repl, []byte("_")...)
+	return repl
+}
+
+func Mangle(s string) string {
+	var mangled string
+	bytes := []byte(s)
+	if s == "" || symPattern.Match(bytes) {
+		return s
+	}
+	mangled = string(bytes[0])
+	if !first.Match(bytes[0:1]) {
+		mangled = "V" + mangled
+	}
+	rest := restIllegal.ReplaceAllFunc(bytes[1:], replIllegal)
+	mangled = mangled + string(rest)
+	return mangled
+}
+
+func BackendName(svc Service, addr string) string {
+	return Mangle(svc.Name + "_" + addr)
+}
+
+func DirectorName(svc Service) string {
+	return Mangle(svc.Name + "_director")
+}
+
+func URLMatcher(rule Rule) string {
+	return Mangle(rule.Host + "_url")
+}

varnish/vcl/vcl.tmpl

+vcl 4.0;
+
+import std;
+import directors;
+import re2;
+
+backend notfound {
+	# 192.0.2.0/24 reserved for docs & examples (RFC5737).
+	.host = "192.0.2.255";
+	.port = "80";
+}
+
+{{range $name, $svc := .AllServices -}}
+{{range $addr := $svc.Addresses -}}
+backend {{backendName $svc $addr.IP}} {
+	.host = "{{$addr.IP}}";
+	.port = "{{$addr.Port}}";
+}
+{{end -}}
+{{end}}
+
+sub vcl_init {
+{{- if .Rules}}
+	new hosts = re2.set(posix_syntax=true, literal=true, anchor=both);
+	{{- range $rule := .Rules}}
+	hosts.add("{{$rule.Host}}");
+	{{- end}}
+	hosts.compile();
+{{end}}
+
+{{- range $name, $svc := .AllServices}}
+	new {{dirName $svc}} = directors.round_robin();
+	{{- range $addr := $svc.Addresses}}
+	{{dirName $svc}}.add_backend({{backendName $svc $addr.IP}});
+	{{- end}}
+{{end}}
+{{- range $rule := .Rules}}
+	new {{urlMatcher $rule}} = re2.set(posix_syntax=true, anchor=start);
+	{{- range $path, $svc := $rule.PathMap}}
+	{{urlMatcher $rule}}.add("{{$path}}",
+				backend={{dirName $svc}}.backend());
+	{{- end}}
+	{{urlMatcher $rule}}.compile();
+{{end -}}
+}
+
+sub set_backend {
+	set req.backend_hint = notfound;
+{{- if .Rules}}
+	if (hosts.match(req.http.Host)) {
+		if (hosts.nmatches() != 1) {
+			# Fail fast when the match was not unique.
+			return (fail);
+		}
+		if (0 != 0) {
+			#
+		}
+		{{- range $i, $rule := .Rules}}
+		elsif (hosts.which() == {{plusOne $i}}) {
+			if ({{urlMatcher $rule}}.match(req.url)) {
+				set req.backend_hint = {{urlMatcher $rule}}.backend(select=FIRST);
+			}
+		}
+		{{- end}}
+	}
+{{- end}}
+
+	if (req.backend_hint == notfound) {
+{{- if ne .DefaultService.Name ""}}
+		set req.backend_hint = {{dirName .DefaultService}}.backend();
+{{- else}}
+		return (synth(404));
+{{- end}}
+	}
+}
+
+sub vcl_miss {
+	call set_backend;
+}
+
+sub vcl_pass {
+	call set_backend;
+}

varnishcache_varnish60.repo

+[varnishcache_varnish60]
+name=varnishcache_varnish60
+baseurl=https://packagecloud.io/varnishcache/varnish60/el/7/$basearch
+repo_gpgcheck=1
+gpgcheck=0
+enabled=1
+gpgkey=https://packagecloud.io/varnishcache/varnish60/gpgkey
+sslverify=1
+sslcacert=/etc/pki/tls/certs/ca-bundle.crt
+metadata_expire=300
+
+[varnishcache_varnish60-source]
+name=varnishcache_varnish60-source
+baseurl=https://packagecloud.io/varnishcache/varnish60/el/7/SRPMS
+repo_gpgcheck=1
+gpgcheck=0
+enabled=1
+gpgkey=https://packagecloud.io/varnishcache/varnish60/gpgkey
+sslverify=1
+sslcacert=/etc/pki/tls/certs/ca-bundle.crt
+metadata_expire=300