Bare metal loadblancers

contrib/service-loadbalancer/Dockerfile

+# Copyright 2015 The Kubernetes Authors. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+FROM ubuntu:14.04
+MAINTAINER Prashanth B <beeps@google.com>
+
+# so apt-get doesn't complain
+ENV DEBIAN_FRONTEND=noninteractive
+RUN sed -i 's/^exit 101/exit 0/' /usr/sbin/policy-rc.d
+
+# TODO: Move to using haproxy:1.5 image instead. Honestly,
+# that image isn't much smaller and the convenience of having
+# an ubuntu container for dev purposes trumps the tiny amounts
+# of disk and bandwidth we'd save in doing so.
+RUN \
+  apt-get update && \
+  apt-get install -y haproxy && \
+  sed -i 's/^ENABLED=.*/ENABLED=1/' /etc/default/haproxy && \
+  rm -rf /var/lib/apt/lists/*
+
+ADD haproxy.cfg /etc/haproxy/haproxy.cfg
+ADD service_loadbalancer service_loadbalancer
+ADD service_loadbalancer.go service_loadbalancer.go
+ADD template.cfg template.cfg
+ADD loadbalancer.json loadbalancer.json
+ADD haproxy_reload haproxy_reload
+ADD README.md README.md
+ENTRYPOINT ["/service_loadbalancer"]

contrib/service-loadbalancer/Makefile

+all: push
+
+# 0.0 shouldn't clobber any released builds
+TAG = 0.0
+PREFIX = bprashanth/servicelb
+
+server: service_loadbalancer.go
+	CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -ldflags '-w' -o service_loadbalancer ./service_loadbalancer.go
+
+container: server
+	docker build -t $(PREFIX):$(TAG) .
+
+push: container
+	docker push $(PREFIX):$(TAG)
+
+clean:
+	rm -f service_loadbalancer

contrib/service-loadbalancer/haproxy.cfg

+# Default haprxy config file. The service-loadbalancer uses
+# go templates (http://golang.org/pkg/text/template/) to
+# generate the config dynamically.
+global
+	daemon
+
+defaults
+	log	global
+	mode	http
+	option	httplog
+	option	dontlognull
+    contimeout 5000
+    clitimeout 50000
+    srvtimeout 50000
+	errorfile 400 /etc/haproxy/errors/400.http
+	errorfile 403 /etc/haproxy/errors/403.http
+	errorfile 408 /etc/haproxy/errors/408.http
+	errorfile 500 /etc/haproxy/errors/500.http
+	errorfile 502 /etc/haproxy/errors/502.http
+	errorfile 503 /etc/haproxy/errors/503.http
+	errorfile 504 /etc/haproxy/errors/504.http
+
+frontend localnodes
+    bind *:80

contrib/service-loadbalancer/haproxy_reload

+#!/bin/bash
+
+# Copyright 2015 The Kubernetes Authors. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# A script to help with haproxy reloads. Needs sudo for :80. Running it for the
+# first time starts haproxy, each subsequent invocation will perform a
+# soft-reload.
+# -f config file
+# -p pid file
+# -D run as daemon
+# -s soft reload, wait for pids to finish handling requests
+# -f send pids a resume signal if reload of new config fails
+
+haproxy -f /etc/haproxy/haproxy.cfg -p /var/run/haproxy.pid -D -sf $(cat /var/run/haproxy.pid)

contrib/service-loadbalancer/loadbalancer.json

+{
+    "name": "haproxy",
+    "reloadCmd": "./haproxy_reload",
+    "config": "/etc/haproxy/haproxy.cfg",
+    "template": "template.cfg",
+    "algorithm": "roundrobin"
+}

contrib/service-loadbalancer/rc.yaml

+apiVersion: v1
+kind: ReplicationController
+metadata:
+  name: service-loadbalancer
+  labels:
+    app: service-loadbalancer
+    version: v1
+spec:
+  replicas: 1
+  selector:
+    app: service-loadbalancer
+    version: v1
+  template:
+    metadata:
+      labels:
+        app: service-loadbalancer
+        version: v1
+    spec:
+      nodeSelector:
+        role: loadbalancer
+      containers:
+      - image: bprashanth/servicelb:0.0
+        imagePullPolicy: Always
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8081
+            scheme: HTTP
+          initialDelaySeconds: 30
+          timeoutSeconds: 5
+        name: haproxy
+        ports:
+        # All http services
+        - containerPort: 80
+          hostPort: 80
+          protocol: TCP
+        # nginx https
+        - containerPort: 443
+          hostPort: 8080
+          protocol: TCP
+        # mysql
+        - containerPort: 3306
+          hostPort: 3306
+          protocol: TCP
+        # haproxy stats
+        - containerPort: 1936
+          hostPort: 1936
+          protocol: TCP
+        resources: {}
+        args:
+        - --tcp-services=mysql:3306,nginxsvc:443

contrib/service-loadbalancer/service_loadbalancer_test.go

+/*
+Copyright 2015 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+	"github.com/golang/glog"
+)
+
+const ns = "default"
+
+// storeEps stores the given endpoints in a store.
+func storeEps(eps []*api.Endpoints) cache.Store {
+	store := cache.NewStore(cache.MetaNamespaceKeyFunc)
+	found := make([]interface{}, 0, len(eps))
+	for i := range eps {
+		found = append(found, eps[i])
+	}
+	if err := store.Replace(found); err != nil {
+		glog.Fatalf("Unable to replace endpoints %v", err)
+	}
+	return store
+}
+
+// storeServices stores the given services in a store.
+func storeServices(svcs []*api.Service) cache.Store {
+	store := cache.NewStore(cache.MetaNamespaceKeyFunc)
+	found := make([]interface{}, 0, len(svcs))
+	for i := range svcs {
+		found = append(found, svcs[i])
+	}
+	if err := store.Replace(found); err != nil {
+		glog.Fatalf("Unable to replace services %v", err)
+	}
+	return store
+}
+
+func getEndpoints(svc *api.Service, endpointAddresses []api.EndpointAddress, endpointPorts []api.EndpointPort) *api.Endpoints {
+	return &api.Endpoints{
+		ObjectMeta: api.ObjectMeta{
+			Name:      svc.Name,
+			Namespace: svc.Namespace,
+		},
+		Subsets: []api.EndpointSubset{{
+			Addresses: endpointAddresses,
+			Ports:     endpointPorts,
+		}},
+	}
+}
+
+func getService(servicePorts []api.ServicePort) *api.Service {
+	return &api.Service{
+		ObjectMeta: api.ObjectMeta{
+			Name: string(util.NewUUID()), Namespace: ns},
+		Spec: api.ServiceSpec{
+			Ports: servicePorts,
+		},
+	}
+}
+
+func newFakeLoadBalancerController(endpoints []*api.Endpoints, services []*api.Service) *loadBalancerController {
+	flb := loadBalancerController{}
+	flb.epLister.Store = storeEps(endpoints)
+	flb.svcLister.Store = storeServices(services)
+	flb.httpPort = 80
+	return &flb
+}
+
+func TestGetEndpoints(t *testing.T) {
+	// 2 pods each of which have 3 targetPorts exposed via a single service
+	endpointAddresses := []api.EndpointAddress{
+		{IP: "1.2.3.4"},
+		{IP: "6.7.8.9"},
+	}
+	ports := []int{80, 443, 3306}
+	endpointPorts := []api.EndpointPort{
+		{Port: ports[0], Protocol: "TCP"},
+		{Port: ports[1], Protocol: "TCP"},
+		{Port: ports[2], Protocol: "TCP", Name: "mysql"},
+	}
+	servicePorts := []api.ServicePort{
+		{Port: 10, TargetPort: util.NewIntOrStringFromInt(ports[0])},
+		{Port: 20, TargetPort: util.NewIntOrStringFromInt(ports[1])},
+		{Port: 30, TargetPort: util.NewIntOrStringFromString("mysql")},
+	}
+
+	svc := getService(servicePorts)
+	endpoints := []*api.Endpoints{getEndpoints(svc, endpointAddresses, endpointPorts)}
+	flb := newFakeLoadBalancerController(endpoints, []*api.Service{svc})
+
+	for i := range ports {
+		eps := flb.getEndpoints(svc, &svc.Spec.Ports[i])
+		expectedEps := util.NewStringSet()
+		for _, address := range endpointAddresses {
+			expectedEps.Insert(fmt.Sprintf("%v:%v", address.IP, ports[i]))
+		}
+
+		receivedEps := util.NewStringSet()
+		for _, ep := range eps {
+			receivedEps.Insert(ep)
+		}
+		if len(receivedEps) != len(expectedEps) || !expectedEps.IsSuperset(receivedEps) {
+			t.Fatalf("Unexpected endpoints, received %+v, expected %+v", receivedEps, expectedEps)
+		}
+		glog.Infof("Got endpoints %+v", receivedEps)
+	}
+}
+
+func TestGetServices(t *testing.T) {
+	endpointAddresses := []api.EndpointAddress{
+		{IP: "1.2.3.4"},
+		{IP: "6.7.8.9"},
+	}
+	ports := []int{80, 443}
+	endpointPorts := []api.EndpointPort{
+		{Port: ports[0], Protocol: "TCP"},
+		{Port: ports[1], Protocol: "TCP"},
+	}
+	servicePorts := []api.ServicePort{
+		{Port: 10, TargetPort: util.NewIntOrStringFromInt(ports[0])},
+		{Port: 20, TargetPort: util.NewIntOrStringFromInt(ports[1])},
+	}
+
+	// 2 services targetting the same endpoints, one of which is declared as a tcp service.
+	svc1 := getService(servicePorts)
+	svc2 := getService(servicePorts)
+	endpoints := []*api.Endpoints{
+		getEndpoints(svc1, endpointAddresses, endpointPorts),
+		getEndpoints(svc2, endpointAddresses, endpointPorts),
+	}
+	flb := newFakeLoadBalancerController(endpoints, []*api.Service{svc1, svc2})
+	flb.tcpServices = map[string]int{
+		svc1.Name: 20,
+	}
+	http, tcp := flb.getServices()
+	serviceURLEp := fmt.Sprintf("%v:%v", svc1.Name, 20)
+	if len(tcp) != 1 || tcp[0].Name != serviceURLEp || tcp[0].FrontendPort != 20 {
+		t.Fatalf("Unexpected tcp service %+v expected %+v", tcp, svc1.Name)
+	}
+
+	// All pods of svc1 exposed under servicePort 20 are tcp
+	expectedTCPEps := util.NewStringSet()
+	for _, address := range endpointAddresses {
+		expectedTCPEps.Insert(fmt.Sprintf("%v:%v", address.IP, 443))
+	}
+	receivedTCPEps := util.NewStringSet()
+	for _, ep := range tcp[0].Ep {
+		receivedTCPEps.Insert(ep)
+	}
+	if len(receivedTCPEps) != len(expectedTCPEps) || !expectedTCPEps.IsSuperset(receivedTCPEps) {
+		t.Fatalf("Unexpected tcp serice %+v", tcp)
+	}
+
+	// All pods of either service not mentioned in the tcpmap are multiplexed on port  :80 as http services.
+	expectedURLMapping := map[string]util.StringSet{
+		fmt.Sprintf("%v:%v", svc1.Name, 10): util.NewStringSet("1.2.3.4:80", "6.7.8.9:80"),
+		fmt.Sprintf("%v:%v", svc2.Name, 10): util.NewStringSet("1.2.3.4:80", "6.7.8.9:80"),
+		fmt.Sprintf("%v:%v", svc2.Name, 20): util.NewStringSet("1.2.3.4:443", "6.7.8.9:443"),
+	}
+	for _, s := range http {
+		if s.FrontendPort != 80 {
+			t.Fatalf("All http services should get multiplexed via the same frontend port: %+v", s)
+		}
+		expectedEps, ok := expectedURLMapping[s.Name]
+		if !ok {
+			t.Fatalf("Expected url endpoint %v, found %+v", s.Name, expectedURLMapping)
+		}
+		receivedEp := util.NewStringSet()
+		for i := range s.Ep {
+			receivedEp.Insert(s.Ep[i])
+		}
+		if len(receivedEp) != len(expectedEps) && !receivedEp.IsSuperset(expectedEps) {
+			t.Fatalf("Expected %+v, got %+v", expectedEps, receivedEp)
+		}
+	}
+}

contrib/service-loadbalancer/template.cfg

+# This file uses golang text templates (http://golang.org/pkg/text/template/) to
+# dynamically configure the haproxy loadbalancer.
+global
+    daemon
+    stats socket /tmp/haproxy
+
+defaults
+    log	global
+    option	dontlognull
+    timeout connect 5000
+    timeout client  50000
+    timeout server  50000
+
+# haproxy stats, required hostport and firewall rules for :1936
+listen stats :1936
+    mode http
+    stats enable
+    stats hide-version
+    stats realm Haproxy\ Statistics
+    stats uri /
+
+frontend httpfrontend
+    # Frontend bound on all network interfaces on port 80
+    bind *:80
+    mode	http
+
+    # inherit default mode, needs changing for tcp
+    # forward everything meant for /foo to the foo backend
+    # default_backend foo
+{{range $i, $svc := .httpServices}}
+    acl url_{{$svc.Name}} path_beg /{{$svc.Name}}
+    use_backend {{$svc.Name}} if url_{{$svc.Name}}
+{{end}}
+
+{{range $i, $svc := .httpServices}}
+{{ $svcName := $svc.Name }}
+backend {{$svc.Name}}
+    mode	http
+    option	httplog
+    errorfile 400 /etc/haproxy/errors/400.http
+    errorfile 403 /etc/haproxy/errors/403.http
+    errorfile 408 /etc/haproxy/errors/408.http
+    errorfile 500 /etc/haproxy/errors/500.http
+    errorfile 502 /etc/haproxy/errors/502.http
+    errorfile 503 /etc/haproxy/errors/503.http
+    errorfile 504 /etc/haproxy/errors/504.http
+
+    balance roundrobin
+    # TODO: Make the path used to access a service customizable.
+    reqrep ^([^\ :]*)\ /{{$svc.Name}}[/]?(.*) \1\ /\2
+    {{range $j, $ep := $svc.Ep}}server {{$svcName}}_{{$j}} {{$ep}} check
+    {{end}}
+{{end}}
+
+
+
+{{range $i, $svc := .tcpServices}}
+{{ $svcName := $svc.Name }}
+frontend {{$svc.Name}}
+    bind *:{{$svc.FrontendPort}}
+    mode tcp
+    default_backend {{$svc.Name}}
+
+backend {{$svc.Name}}
+    balance roundrobin
+    mode tcp
+    {{range $j, $ep := $svc.Ep}}server {{$svcName}}_{{$j}} {{$ep}} check
+    {{end}}
+{{end}}

pkg/client/cache/listers.go

@@ -266,4 +266,27 @@ func (s *StoreToServiceLister) GetPodServices(pod *api.Pod) (services []api.Serv
 	return
 }
 
-// TODO: add StoreToEndpointsLister for use in kube-proxy.
+// StoreToEndpointsLister makes a Store that lists endpoints.
+type StoreToEndpointsLister struct {
+	Store
+}
+
+// List lists all endpoints in the store.
+func (s *StoreToEndpointsLister) List() (services api.EndpointsList, err error) {
+	for _, m := range s.Store.List() {
+		services.Items = append(services.Items, *(m.(*api.Endpoints)))
+	}
+	return services, nil
+}
+
+// GetServiceEndpoints returns the endpoints of a service, matched on service name.
+func (s *StoreToEndpointsLister) GetServiceEndpoints(svc *api.Service) (ep api.Endpoints, err error) {
+	for _, m := range s.Store.List() {
+		ep = *m.(*api.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
+}