Implement multi-port endpoints

cmd/integration/integration.go

@@ -283,11 +283,25 @@ func endpointsSet(c *client.Client, serviceNamespace, serviceID string, endpoint
 			glog.Infof("Error on creating endpoints: %v", err)
 			return false, nil
 		}
-		for _, e := range endpoints.Endpoints {
-			glog.Infof("%s/%s endpoint: %s:%d %#v", serviceNamespace, serviceID, e.IP, e.Port, e.TargetRef)
+		count := 0
+		for _, ss := range endpoints.Subsets {
+			for _, addr := range ss.Addresses {
+				for _, port := range ss.Ports {
+					count++
+					glog.Infof("%s/%s endpoint: %s:%d %#v", serviceNamespace, serviceID, addr.IP, port.Port, addr.TargetRef)
+				}
+			}
 		}
-		return len(endpoints.Endpoints) == endpointCount, nil
+		return count == endpointCount, nil
+	}
+}
+
+func countEndpoints(eps *api.Endpoints) int {
+	count := 0
+	for i := range eps.Subsets {
+		count += len(eps.Subsets[i].Addresses) * len(eps.Subsets[i].Ports)
 	}
+	return count
 }
 
 func podExists(c *client.Client, podNamespace string, podID string) wait.ConditionFunc {
@@ -669,7 +683,7 @@ func runMasterServiceTest(client *client.Client) {
 		if err != nil {
 			glog.Fatalf("unexpected error listing endpoints for kubernetes service: %v", err)
 		}
-		if len(ep.Endpoints) == 0 {
+		if countEndpoints(ep) == 0 {
 			glog.Fatalf("no endpoints for kubernetes service: %v", ep)
 		}
 	} else {
@@ -680,7 +694,7 @@ func runMasterServiceTest(client *client.Client) {
 		if err != nil {
 			glog.Fatalf("unexpected error listing endpoints for kubernetes service: %v", err)
 		}
-		if len(ep.Endpoints) == 0 {
+		if countEndpoints(ep) == 0 {
 			glog.Fatalf("no endpoints for kubernetes service: %v", ep)
 		}
 	} else {

contrib/for-tests/network-tester/webserver.go

@@ -45,6 +45,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 )
 
 var (
@@ -218,9 +219,17 @@ func contactOthers(state *State) {
 			time.Sleep(time.Duration(1+rand.Intn(10)) * time.Second)
 		}
 
-		for i, e := range endpoints.Endpoints {
-			state.Logf("Attempting to contact IP %s at endpoint number %d port %v", e.IP, i, e.Port)
-			contactSingle(fmt.Sprintf("http://%s:%d", e.IP, e.Port), state)
+		eps := util.StringSet{}
+		for _, ss := range endpoints.Subsets {
+			for _, a := range ss.Addresses {
+				for _, p := range ss.Ports {
+					eps.Insert(fmt.Sprintf("http://%s:%d", a.IP, p.Port))
+				}
+			}
+		}
+		for ep := range eps {
+			state.Logf("Attempting to contact %s", ep)
+			contactSingle(ep, state)
 		}
 
 		time.Sleep(5 * time.Second)

pkg/api/endpoints/util.go

+/*
+Copyright 2015 Google Inc. 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 endpoints
+
+import (
+	"bytes"
+	"crypto/md5"
+	"encoding/hex"
+	"hash"
+	"sort"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+)
+
+// RepackSubsets takes a slice of EndpointSubset objects, expands it to the full
+// representation, and then repacks that into the canonical layout.  This
+// ensures that code which operates on these objects can rely on the common
+// form for things like comparison.  The result is a newly allocated slice.
+func RepackSubsets(subsets []api.EndpointSubset) []api.EndpointSubset {
+	// First map each unique port definition to the sets of hosts that
+	// offer it.  The sets of hosts must be de-duped, using IP as the key.
+	type ipString string
+	allAddrs := map[ipString]*api.EndpointAddress{}
+	portsToAddrs := map[api.EndpointPort]addressSet{}
+	for i := range subsets {
+		for j := range subsets[i].Ports {
+			epp := &subsets[i].Ports[j]
+			for k := range subsets[i].Addresses {
+				epa := &subsets[i].Addresses[k]
+				ipstr := ipString(epa.IP)
+				// Accumulate the most "complete" address we can.
+				if allAddrs[ipstr] == nil {
+					// Make a copy so we don't write to the
+					// input args of this function.
+					p := &api.EndpointAddress{}
+					*p = *epa
+					allAddrs[ipstr] = p
+				} else if allAddrs[ipstr].TargetRef == nil {
+					allAddrs[ipstr].TargetRef = epa.TargetRef
+				}
+				// Remember that this port maps to this address.
+				if _, found := portsToAddrs[*epp]; !found {
+					portsToAddrs[*epp] = addressSet{}
+				}
+				portsToAddrs[*epp].Insert(allAddrs[ipstr])
+			}
+		}
+	}
+
+	// Next, map the sets of hosts to the sets of ports they offer.
+	// Go does not allow maps or slices as keys to maps, so we have
+	// to synthesize and artificial key and do a sort of 2-part
+	// associative entity.
+	type keyString string
+	addrSets := map[keyString]addressSet{}
+	addrSetsToPorts := map[keyString][]api.EndpointPort{}
+	for epp, addrs := range portsToAddrs {
+		key := keyString(hashAddresses(addrs))
+		addrSets[key] = addrs
+		addrSetsToPorts[key] = append(addrSetsToPorts[key], epp)
+	}
+
+	// Next, build the N-to-M association the API wants.
+	final := []api.EndpointSubset{}
+	for key, ports := range addrSetsToPorts {
+		addrs := []api.EndpointAddress{}
+		for k := range addrSets[key] {
+			addrs = append(addrs, *k)
+		}
+		final = append(final, api.EndpointSubset{Addresses: addrs, Ports: ports})
+	}
+
+	// Finally, sort it.
+	return SortSubsets(final)
+}
+
+type addressSet map[*api.EndpointAddress]struct{}
+
+func (set addressSet) Insert(addr *api.EndpointAddress) {
+	set[addr] = struct{}{}
+}
+
+func hashAddresses(addrs addressSet) string {
+	// Flatten the list of addresses into a string so it can be used as a
+	// map key.
+	hasher := md5.New()
+	util.DeepHashObject(hasher, addrs)
+	return hex.EncodeToString(hasher.Sum(nil)[0:])
+}
+
+// SortSubsets sorts an array of EndpointSubset objects in place.  For ease of
+// use it returns the input slice.
+func SortSubsets(subsets []api.EndpointSubset) []api.EndpointSubset {
+	for i := range subsets {
+		ss := &subsets[i]
+		sort.Sort(addrsByIP(ss.Addresses))
+		sort.Sort(portsByHash(ss.Ports))
+	}
+	sort.Sort(subsetsByHash(subsets))
+	return subsets
+}
+
+func hashObject(hasher hash.Hash, obj interface{}) []byte {
+	util.DeepHashObject(hasher, obj)
+	return hasher.Sum(nil)
+}
+
+type subsetsByHash []api.EndpointSubset
+
+func (sl subsetsByHash) Len() int      { return len(sl) }
+func (sl subsetsByHash) Swap(i, j int) { sl[i], sl[j] = sl[j], sl[i] }
+func (sl subsetsByHash) Less(i, j int) bool {
+	hasher := md5.New()
+	h1 := hashObject(hasher, sl[i])
+	h2 := hashObject(hasher, sl[j])
+	return bytes.Compare(h1, h2) < 0
+}
+
+type addrsByIP []api.EndpointAddress
+
+func (sl addrsByIP) Len() int      { return len(sl) }
+func (sl addrsByIP) Swap(i, j int) { sl[i], sl[j] = sl[j], sl[i] }
+func (sl addrsByIP) Less(i, j int) bool {
+	return bytes.Compare([]byte(sl[i].IP), []byte(sl[j].IP)) < 0
+}
+
+type portsByHash []api.EndpointPort
+
+func (sl portsByHash) Len() int      { return len(sl) }
+func (sl portsByHash) Swap(i, j int) { sl[i], sl[j] = sl[j], sl[i] }
+func (sl portsByHash) Less(i, j int) bool {
+	hasher := md5.New()
+	h1 := hashObject(hasher, sl[i])
+	h2 := hashObject(hasher, sl[j])
+	return bytes.Compare(h1, h2) < 0
+}

pkg/api/endpoints/util_test.go

+/*
+Copyright 2015 Google Inc. 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 endpoints
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/davecgh/go-spew/spew"
+)
+
+func TestPackSubsets(t *testing.T) {
+	// The downside of table-driven tests is that some things have to live outside the table.
+	fooObjRef := api.ObjectReference{Name: "foo"}
+	barObjRef := api.ObjectReference{Name: "bar"}
+
+	testCases := []struct {
+		name   string
+		given  []api.EndpointSubset
+		expect []api.EndpointSubset
+	}{
+		{
+			name:   "empty everything",
+			given:  []api.EndpointSubset{{Addresses: []api.EndpointAddress{}, Ports: []api.EndpointPort{}}},
+			expect: []api.EndpointSubset{},
+		}, {
+			name:   "empty addresses",
+			given:  []api.EndpointSubset{{Addresses: []api.EndpointAddress{}, Ports: []api.EndpointPort{{Port: 111}}}},
+			expect: []api.EndpointSubset{},
+		}, {
+			name:   "empty ports",
+			given:  []api.EndpointSubset{{Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}}, Ports: []api.EndpointPort{}}},
+			expect: []api.EndpointSubset{},
+		}, {
+			name: "one set, one ip, one port",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+		}, {
+			name: "one set, two ips, one port",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+		}, {
+			name: "one set, one ip, two ports",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}, {Port: 222}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}, {Port: 222}},
+			}},
+		}, {
+			name: "one set, dup ips, one port",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+		}, {
+			name: "one set, dup ips with target-refs, one port",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{
+					{IP: "1.2.3.4", TargetRef: &fooObjRef},
+					{IP: "1.2.3.4", TargetRef: &barObjRef},
+				},
+				Ports: []api.EndpointPort{{Port: 111}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4", TargetRef: &fooObjRef}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+		}, {
+			name: "one set, one ip, dup ports",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}, {Port: 111}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+		}, {
+			name: "two sets, dup ip, dup port",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+		}, {
+			name: "two sets, dup ip, two ports",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 222}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}, {Port: 222}},
+			}},
+		}, {
+			name: "two sets, two ips, dup port",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "5.6.7.8"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}},
+		}, {
+			name: "two sets, two ips, two ports",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "5.6.7.8"}},
+				Ports:     []api.EndpointPort{{Port: 222}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "5.6.7.8"}},
+				Ports:     []api.EndpointPort{{Port: 222}},
+			}},
+		}, {
+			name: "four sets, three ips, three ports, jumbled",
+			given: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.5"}},
+				Ports:     []api.EndpointPort{{Port: 222}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.6"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.5"}},
+				Ports:     []api.EndpointPort{{Port: 333}},
+			}},
+			expect: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "1.2.3.6"}},
+				Ports:     []api.EndpointPort{{Port: 111}},
+			}, {
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.5"}},
+				Ports:     []api.EndpointPort{{Port: 222}, {Port: 333}},
+			}},
+		},
+	}
+
+	for _, tc := range testCases {
+		result := RepackSubsets(tc.given)
+		if !reflect.DeepEqual(result, SortSubsets(tc.expect)) {
+			t.Errorf("case %q: expected %s, got %s", tc.name, spew.Sprintf("%v", SortSubsets(tc.expect)), spew.Sprintf("%v", result))
+		}
+	}
+}

pkg/api/testing/fuzzer.go

@@ -17,7 +17,6 @@ limitations under the License.
 package testing
 
 import (
-	"fmt"
 	"math/rand"
 	"strconv"
 	"testing"
@@ -204,11 +203,6 @@ func FuzzerFor(t *testing.T, version string, src rand.Source) *fuzz.Fuzzer {
 		func(s *api.NamespaceStatus, c fuzz.Continue) {
 			s.Phase = api.NamespaceActive
 		},
-		func(ep *api.Endpoint, c fuzz.Continue) {
-			// TODO: If our API used a particular type for IP fields we could just catch that here.
-			ep.IP = fmt.Sprintf("%d.%d.%d.%d", c.Rand.Intn(256), c.Rand.Intn(256), c.Rand.Intn(256), c.Rand.Intn(256))
-			ep.Port = c.Rand.Intn(65536)
-		},
 		func(http *api.HTTPGetAction, c fuzz.Continue) {
 			c.FuzzNoCustom(http)        // fuzz self without calling this function again
 			http.Path = "/" + http.Path // can't be blank

pkg/api/types.go

@@ -914,29 +914,62 @@ type Service struct {
 	Status ServiceStatus `json:"status,omitempty"`
 }
 
-// Endpoints is a collection of endpoints that implement the actual service, for example:
-// Name: "mysql", Endpoints: [{"ip": "10.10.1.1", "port": 1909}, {"ip": "10.10.2.2", "port": 8834}]
+// Endpoints is a collection of endpoints that implement the actual service.  Example:
+//   Name: "mysvc",
+//   Subsets: [
+//     {
+//       Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//       Ports: [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//     },
+//     {
+//       Addresses: [{"ip": "10.10.3.3"}],
+//       Ports: [{"name": "a", "port": 93}, {"name": "b", "port": 76}]
+//     },
+//  ]
 type Endpoints struct {
 	TypeMeta   `json:",inline"`
 	ObjectMeta `json:"metadata,omitempty"`
 
-	// Optional: The IP protocol for these endpoints. Supports "TCP" and
-	// "UDP".  Defaults to "TCP".
-	Protocol  Protocol   `json:"protocol,omitempty"`
-	Endpoints []Endpoint `json:"endpoints,omitempty"`
+	// The set of all endpoints is the union of all subsets.
+	Subsets []EndpointSubset
 }
 
-// Endpoint is a single IP endpoint of a service.
-type Endpoint struct {
-	// Required: The IP of this endpoint.
-	// TODO: This should allow hostname or IP, see #4447.
-	IP string `json:"ip"`
+// EndpointSubset is a group of addresses with a common set of ports.  The
+// expanded set of endpoints is the Cartesian product of Addresses x Ports.
+// For example, given:
+//   {
+//     Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//     Ports:     [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//   }
+// The resulting set of endpoints can be viewed as:
+//     a: [ 10.10.1.1:8675, 10.10.2.2:8675 ],
+//     b: [ 10.10.1.1:309, 10.10.2.2:309 ]
+type EndpointSubset struct {
+	Addresses []EndpointAddress
+	Ports     []EndpointPort
+}
 
-	// Required: The destination port to access.
-	Port int `json:"port"`
+// EndpointAddress is a tuple that describes single IP address.
+type EndpointAddress struct {
+	// The IP of this endpoint.
+	// TODO: This should allow hostname or IP, see #4447.
+	IP string
 
 	// Optional: The kubernetes object related to the entry point.
-	TargetRef *ObjectReference `json:"targetRef,omitempty"`
+	TargetRef *ObjectReference
+}
+
+// EndpointPort is a tuple that describes a single port.
+type EndpointPort struct {
+	// The name of this port (corresponds to ServicePort.Name).  Optional
+	// if only one port is defined.  Must be a DNS_LABEL.
+	Name string
+
+	// The port number.
+	Port int
+
+	// The IP protocol for this port.
+	Protocol Protocol
 }
 
 // EndpointsList is a list of endpoints.

pkg/api/v1beta1/conversion.go

@@ -1253,21 +1253,42 @@ func init() {
 			if err := s.Convert(&in.ObjectMeta, &out.TypeMeta, 0); err != nil {
 				return err
 			}
-			if err := s.Convert(&in.Protocol, &out.Protocol, 0); err != nil {
+			if err := s.Convert(&in.Subsets, &out.Subsets, 0); err != nil {
 				return err
 			}
-			for i := range in.Endpoints {
-				ep := &in.Endpoints[i]
-				hostPort := net.JoinHostPort(ep.IP, strconv.Itoa(ep.Port))
-				out.Endpoints = append(out.Endpoints, hostPort)
-				if ep.TargetRef != nil {
-					target := EndpointObjectReference{
-						Endpoint: hostPort,
-					}
-					if err := s.Convert(ep.TargetRef, &target.ObjectReference, 0); err != nil {
+			// Produce back-compat fields.
+			firstPortName := ""
+			if len(in.Subsets) > 0 {
+				if len(in.Subsets[0].Ports) > 0 {
+					if err := s.Convert(&in.Subsets[0].Ports[0].Protocol, &out.Protocol, 0); err != nil {
 						return err
 					}
-					out.TargetRefs = append(out.TargetRefs, target)
+					firstPortName = in.Subsets[0].Ports[0].Name
+				}
+			} else {
+				out.Protocol = ProtocolTCP
+			}
+			for i := range in.Subsets {
+				ss := &in.Subsets[i]
+				for j := range ss.Ports {
+					ssp := &ss.Ports[j]
+					if ssp.Name != firstPortName {
+						continue
+					}
+					for k := range ss.Addresses {
+						ssa := &ss.Addresses[k]
+						hostPort := net.JoinHostPort(ssa.IP, strconv.Itoa(ssp.Port))
+						out.Endpoints = append(out.Endpoints, hostPort)
+						if ssa.TargetRef != nil {
+							target := EndpointObjectReference{
+								Endpoint: hostPort,
+							}
+							if err := s.Convert(ssa.TargetRef, &target.ObjectReference, 0); err != nil {
+								return err
+							}
+							out.TargetRefs = append(out.TargetRefs, target)
+						}
+					}
 				}
 			}
 			return nil
@@ -1279,32 +1300,10 @@ func init() {
 			if err := s.Convert(&in.TypeMeta, &out.ObjectMeta, 0); err != nil {
 				return err
 			}
-			if err := s.Convert(&in.Protocol, &out.Protocol, 0); err != nil {
+			if err := s.Convert(&in.Subsets, &out.Subsets, 0); err != nil {
 				return err
 			}
-			for i := range in.Endpoints {
-				out.Endpoints = append(out.Endpoints, newer.Endpoint{})
-				ep := &out.Endpoints[i]
-				host, port, err := net.SplitHostPort(in.Endpoints[i])
-				if err != nil {
-					return err
-				}
-				ep.IP = host
-				pn, err := strconv.Atoi(port)
-				if err != nil {
-					return err
-				}
-				ep.Port = pn
-				for j := range in.TargetRefs {
-					if in.TargetRefs[j].Endpoint != in.Endpoints[i] {
-						continue
-					}
-					ep.TargetRef = &newer.ObjectReference{}
-					if err := s.Convert(&in.TargetRefs[j].ObjectReference, ep.TargetRef, 0); err != nil {
-						return err
-					}
-				}
-			}
+			// Back-compat fields are handled in the defaulting phase.
 			return nil
 		},
 

pkg/api/v1beta1/conversion_test.go

@@ -408,36 +408,105 @@ func TestEndpointsConversion(t *testing.T) {
 				Endpoints: []string{},
 			},
 			expected: newer.Endpoints{
-				Protocol:  newer.ProtocolTCP,
-				Endpoints: []newer.Endpoint{},
+				Subsets: []newer.EndpointSubset{},
 			},
 		},
 		{
 			given: current.Endpoints{
 				TypeMeta: current.TypeMeta{
-					ID: "one",
+					ID: "one legacy",
 				},
 				Protocol:  current.ProtocolTCP,
 				Endpoints: []string{"1.2.3.4:88"},
 			},
 			expected: newer.Endpoints{
-				Protocol:  newer.ProtocolTCP,
-				Endpoints: []newer.Endpoint{{IP: "1.2.3.4", Port: 88}},
+				Subsets: []newer.EndpointSubset{{
+					Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolTCP}},
+					Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+				}},
 			},
 		},
 		{
 			given: current.Endpoints{
 				TypeMeta: current.TypeMeta{
-					ID: "several",
+					ID: "several legacy",
 				},
 				Protocol:  current.ProtocolUDP,
 				Endpoints: []string{"1.2.3.4:88", "1.2.3.4:89", "1.2.3.4:90"},
 			},
 			expected: newer.Endpoints{
-				Protocol:  newer.ProtocolUDP,
-				Endpoints: []newer.Endpoint{{IP: "1.2.3.4", Port: 88}, {IP: "1.2.3.4", Port: 89}, {IP: "1.2.3.4", Port: 90}},
+				Subsets: []newer.EndpointSubset{
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 89, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 90, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+					},
+				}},
+		},
+		{
+			given: current.Endpoints{
+				TypeMeta: current.TypeMeta{
+					ID: "one subset",
+				},
+				Protocol:  current.ProtocolTCP,
+				Endpoints: []string{"1.2.3.4:88"},
+				Subsets: []current.EndpointSubset{{
+					Ports:     []current.EndpointPort{{Name: "", Port: 88, Protocol: current.ProtocolTCP}},
+					Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}},
+				}},
+			},
+			expected: newer.Endpoints{
+				Subsets: []newer.EndpointSubset{{
+					Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolTCP}},
+					Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+				}},
 			},
 		},
+		{
+			given: current.Endpoints{
+				TypeMeta: current.TypeMeta{
+					ID: "several subset",
+				},
+				Protocol:  current.ProtocolUDP,
+				Endpoints: []string{"1.2.3.4:88", "5.6.7.8:88", "1.2.3.4:89", "5.6.7.8:89"},
+				Subsets: []current.EndpointSubset{
+					{
+						Ports:     []current.EndpointPort{{Name: "", Port: 88, Protocol: current.ProtocolUDP}},
+						Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []current.EndpointPort{{Name: "", Port: 89, Protocol: current.ProtocolUDP}},
+						Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []current.EndpointPort{{Name: "named", Port: 90, Protocol: current.ProtocolUDP}},
+						Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+				},
+			},
+			expected: newer.Endpoints{
+				Subsets: []newer.EndpointSubset{
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 89, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "named", Port: 90, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+				}},
+		},
 	}
 
 	for i, tc := range testCases {
@@ -447,8 +516,8 @@ func TestEndpointsConversion(t *testing.T) {
 			t.Errorf("[Case: %d] Unexpected error: %v", i, err)
 			continue
 		}
-		if got.Protocol != tc.expected.Protocol || !newer.Semantic.DeepEqual(got.Endpoints, tc.expected.Endpoints) {
-			t.Errorf("[Case: %d] Expected %v, got %v", i, tc.expected, got)
+		if !newer.Semantic.DeepEqual(got.Subsets, tc.expected.Subsets) {
+			t.Errorf("[Case: %d] Expected %#v, got %#v", i, tc.expected.Subsets, got.Subsets)
 		}
 
 		// Convert internal -> versioned.
@@ -458,7 +527,7 @@ func TestEndpointsConversion(t *testing.T) {
 			continue
 		}
 		if got2.Protocol != tc.given.Protocol || !newer.Semantic.DeepEqual(got2.Endpoints, tc.given.Endpoints) {
-			t.Errorf("[Case: %d] Expected %v, got %v", i, tc.given, got2)
+			t.Errorf("[Case: %d] Expected %#v, got %#v", i, tc.given.Endpoints, got2.Endpoints)
 		}
 	}
 }

pkg/api/v1beta1/defaults.go

@@ -17,10 +17,13 @@ limitations under the License.
 package v1beta1
 
 import (
+	"net"
+	"strconv"
 	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+	"github.com/golang/glog"
 )
 
 func init() {
@@ -93,7 +96,42 @@ func init() {
 		},
 		func(obj *Endpoints) {
 			if obj.Protocol == "" {
-				obj.Protocol = "TCP"
+				obj.Protocol = ProtocolTCP
+			}
+			if len(obj.Subsets) == 0 && len(obj.Endpoints) > 0 {
+				// Must be a legacy-style object - populate
+				// Subsets from the older fields.  Do this the
+				// simplest way, which is dumb (but valid).
+				for i := range obj.Endpoints {
+					host, portStr, err := net.SplitHostPort(obj.Endpoints[i])
+					if err != nil {
+						glog.Errorf("failed to SplitHostPort(%q)", obj.Endpoints[i])
+					}
+					var tgtRef *ObjectReference
+					for j := range obj.TargetRefs {
+						if obj.TargetRefs[j].Endpoint == obj.Endpoints[i] {
+							tgtRef = &ObjectReference{}
+							*tgtRef = obj.TargetRefs[j].ObjectReference
+						}
+					}
+					port, err := strconv.Atoi(portStr)
+					if err != nil {
+						glog.Errorf("failed to Atoi(%q)", portStr)
+					}
+					obj.Subsets = append(obj.Subsets, EndpointSubset{
+						Addresses: []EndpointAddress{{IP: host, TargetRef: tgtRef}},
+						Ports:     []EndpointPort{{Protocol: obj.Protocol, Port: port}},
+					})
+				}
+			}
+			for i := range obj.Subsets {
+				ss := &obj.Subsets[i]
+				for i := range ss.Ports {
+					ep := &ss.Ports[i]
+					if ep.Protocol == "" {
+						ep.Protocol = ProtocolTCP
+					}
+				}
 			}
 		},
 		func(obj *HTTPGetAction) {

pkg/api/v1beta1/defaults_test.go

@@ -61,14 +61,57 @@ func TestSetDefaultSecret(t *testing.T) {
 	}
 }
 
+// Test that we use "legacy" fields if "modern" fields are not provided.
+func TestSetDefaulEndpointsLegacy(t *testing.T) {
+	in := &current.Endpoints{
+		Protocol:   "UDP",
+		Endpoints:  []string{"1.2.3.4:93", "5.6.7.8:76"},
+		TargetRefs: []current.EndpointObjectReference{{Endpoint: "1.2.3.4:93", ObjectReference: current.ObjectReference{ID: "foo"}}},
+	}
+	obj := roundTrip(t, runtime.Object(in))
+	out := obj.(*current.Endpoints)
+
+	if len(out.Subsets) != 2 {
+		t.Errorf("Expected 2 EndpointSubsets, got %d (%#v)", len(out.Subsets), out.Subsets)
+	}
+	expected := []current.EndpointSubset{
+		{
+			Addresses: []current.EndpointAddress{{IP: "1.2.3.4", TargetRef: &current.ObjectReference{ID: "foo"}}},
+			Ports:     []current.EndpointPort{{Protocol: current.ProtocolUDP, Port: 93}},
+		},
+		{
+			Addresses: []current.EndpointAddress{{IP: "5.6.7.8"}},
+			Ports:     []current.EndpointPort{{Protocol: current.ProtocolUDP, Port: 76}},
+		},
+	}
+	if !reflect.DeepEqual(out.Subsets, expected) {
+		t.Errorf("Expected %#v, got %#v", expected, out.Subsets)
+	}
+}
+
 func TestSetDefaulEndpointsProtocol(t *testing.T) {
-	in := &current.Endpoints{}
+	in := &current.Endpoints{Subsets: []current.EndpointSubset{
+		{Ports: []current.EndpointPort{{}, {Protocol: "UDP"}, {}}},
+	}}
 	obj := roundTrip(t, runtime.Object(in))
 	out := obj.(*current.Endpoints)
 
 	if out.Protocol != current.ProtocolTCP {
 		t.Errorf("Expected protocol %s, got %s", current.ProtocolTCP, out.Protocol)
 	}
+	for i := range out.Subsets {
+		for j := range out.Subsets[i].Ports {
+			if in.Subsets[i].Ports[j].Protocol == "" {
+				if out.Subsets[i].Ports[j].Protocol != current.ProtocolTCP {
+					t.Errorf("Expected protocol %s, got %s", current.ProtocolTCP, out.Subsets[i].Ports[j].Protocol)
+				}
+			} else {
+				if out.Subsets[i].Ports[j].Protocol != in.Subsets[i].Ports[j].Protocol {
+					t.Errorf("Expected protocol %s, got %s", in.Subsets[i].Ports[j].Protocol, out.Subsets[i].Ports[j].Protocol)
+				}
+			}
+		}
+	}
 }
 
 func TestSetDefaultNamespace(t *testing.T) {

pkg/api/v1beta1/types.go

@@ -762,12 +762,61 @@ type EndpointObjectReference struct {
 // Name: "mysql", Endpoints: ["10.10.1.1:1909", "10.10.2.2:8834"]
 type Endpoints struct {
 	TypeMeta `json:",inline"`
-	// Optional: The IP protocol for these endpoints. Supports "TCP" and
-	// "UDP".  Defaults to "TCP".
-	Protocol  Protocol `json:"protocol,omitempty" description:"IP protocol for endpoint ports; must be UDP or TCP; TCP if unspecified"`
-	Endpoints []string `json:"endpoints" description:"list of endpoints corresponding to a service, of the form address:port, such as 10.10.1.1:1909"`
-	// Optional: The kubernetes object related to the entry point.
+
+	// These fields are retained for backwards compatibility.  For
+	// multi-port services, use the Subsets field instead.  Upon a create or
+	// update operation, the following logic applies:
+	//   * If Subsets is specified, Protocol, Endpoints, and TargetRefs will
+	//     be overwritten by data from Subsets.
+	//   * If Subsets is not specified, Protocol, Endpoints, and TargetRefs
+	//     will be used to generate Subsets.
+	Protocol  Protocol `json:"protocol,omitempty" description:"IP protocol for the first set of endpoint ports; must be UDP or TCP; TCP if unspecified"`
+	Endpoints []string `json:"endpoints" description:"first set of endpoints corresponding to a service, of the form address:port, such as 10.10.1.1:1909"`
+	// Optional: The kubernetes objects related to the first set of entry points.
 	TargetRefs []EndpointObjectReference `json:"targetRefs,omitempty" description:"list of references to objects providing the endpoints"`
+
+	// The set of all endpoints is the union of all subsets.  If this field
+	// is not empty it must include the backwards-compatible protocol and
+	// endpoints.
+	Subsets []EndpointSubset `json:"subsets" description:"sets of addresses and ports that comprise a service"`
+}
+
+// EndpointSubset is a group of addresses with a common set of ports.  The
+// expanded set of endpoints is the Cartesian product of Addresses x Ports.
+// For example, given:
+//   {
+//     Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//     Ports:     [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//   }
+// The resulting set of endpoints can be viewed as:
+//     a: [ 10.10.1.1:8675, 10.10.2.2:8675 ],
+//     b: [ 10.10.1.1:309, 10.10.2.2:309 ]
+type EndpointSubset struct {
+	Addresses []EndpointAddress `json:"addresses,omitempty" description:"IP addresses which offer the related ports"`
+	Ports     []EndpointPort    `json:"ports,omitempty" description:"port numbers available on the related IP addresses"`
+}
+
+// EndpointAddress is a tuple that describes single IP address.
+type EndpointAddress struct {
+	// The IP of this endpoint.
+	// TODO: This should allow hostname or IP, see #4447.
+	IP string `json:"IP" description:"IP address of the endpoint"`
+
+	// Optional: The kubernetes object related to the entry point.
+	TargetRef *ObjectReference `json:"targetRef,omitempty" description:"reference to object providing the endpoint"`
+}
+
+// EndpointPort is a tuple that describes a single port.
+type EndpointPort struct {
+	// The name of this port (corresponds to ServicePort.Name).  Optional
+	// if only one port is defined.  Must be a DNS_LABEL.
+	Name string `json:"name,omitempty" description:"name of this port"`
+
+	// The port number.
+	Port int `json:"port" description:"port number of the endpoint"`
+
+	// The IP protocol for this port.
+	Protocol Protocol `json:"protocol,omitempty" description:"protocol for this port; must be UDP or TCP; TCP if unspecified"`
 }
 
 // EndpointsList is a list of endpoints.

pkg/api/v1beta2/conversion.go

@@ -1181,21 +1181,42 @@ func init() {
 			if err := s.Convert(&in.ObjectMeta, &out.TypeMeta, 0); err != nil {
 				return err
 			}
-			if err := s.Convert(&in.Protocol, &out.Protocol, 0); err != nil {
+			if err := s.Convert(&in.Subsets, &out.Subsets, 0); err != nil {
 				return err
 			}
-			for i := range in.Endpoints {
-				ep := &in.Endpoints[i]
-				hostPort := net.JoinHostPort(ep.IP, strconv.Itoa(ep.Port))
-				out.Endpoints = append(out.Endpoints, hostPort)
-				if ep.TargetRef != nil {
-					target := EndpointObjectReference{
-						Endpoint: hostPort,
-					}
-					if err := s.Convert(ep.TargetRef, &target.ObjectReference, 0); err != nil {
+			// Produce back-compat fields.
+			firstPortName := ""
+			if len(in.Subsets) > 0 {
+				if len(in.Subsets[0].Ports) > 0 {
+					if err := s.Convert(&in.Subsets[0].Ports[0].Protocol, &out.Protocol, 0); err != nil {
 						return err
 					}
-					out.TargetRefs = append(out.TargetRefs, target)
+					firstPortName = in.Subsets[0].Ports[0].Name
+				}
+			} else {
+				out.Protocol = ProtocolTCP
+			}
+			for i := range in.Subsets {
+				ss := &in.Subsets[i]
+				for j := range ss.Ports {
+					ssp := &ss.Ports[j]
+					if ssp.Name != firstPortName {
+						continue
+					}
+					for k := range ss.Addresses {
+						ssa := &ss.Addresses[k]
+						hostPort := net.JoinHostPort(ssa.IP, strconv.Itoa(ssp.Port))
+						out.Endpoints = append(out.Endpoints, hostPort)
+						if ssa.TargetRef != nil {
+							target := EndpointObjectReference{
+								Endpoint: hostPort,
+							}
+							if err := s.Convert(ssa.TargetRef, &target.ObjectReference, 0); err != nil {
+								return err
+							}
+							out.TargetRefs = append(out.TargetRefs, target)
+						}
+					}
 				}
 			}
 			return nil
@@ -1207,32 +1228,10 @@ func init() {
 			if err := s.Convert(&in.TypeMeta, &out.ObjectMeta, 0); err != nil {
 				return err
 			}
-			if err := s.Convert(&in.Protocol, &out.Protocol, 0); err != nil {
+			if err := s.Convert(&in.Subsets, &out.Subsets, 0); err != nil {
 				return err
 			}
-			for i := range in.Endpoints {
-				out.Endpoints = append(out.Endpoints, newer.Endpoint{})
-				ep := &out.Endpoints[i]
-				host, port, err := net.SplitHostPort(in.Endpoints[i])
-				if err != nil {
-					return err
-				}
-				ep.IP = host
-				pn, err := strconv.Atoi(port)
-				if err != nil {
-					return err
-				}
-				ep.Port = pn
-				for j := range in.TargetRefs {
-					if in.TargetRefs[j].Endpoint != in.Endpoints[i] {
-						continue
-					}
-					ep.TargetRef = &newer.ObjectReference{}
-					if err := s.Convert(&in.TargetRefs[j], ep.TargetRef, 0); err != nil {
-						return err
-					}
-				}
-			}
+			// Back-compat fields are handled in the defaulting phase.
 			return nil
 		},
 

pkg/api/v1beta2/conversion_test.go

@@ -227,36 +227,105 @@ func TestEndpointsConversion(t *testing.T) {
 				Endpoints: []string{},
 			},
 			expected: newer.Endpoints{
-				Protocol:  newer.ProtocolTCP,
-				Endpoints: []newer.Endpoint{},
+				Subsets: []newer.EndpointSubset{},
 			},
 		},
 		{
 			given: current.Endpoints{
 				TypeMeta: current.TypeMeta{
-					ID: "one",
+					ID: "one legacy",
 				},
 				Protocol:  current.ProtocolTCP,
 				Endpoints: []string{"1.2.3.4:88"},
 			},
 			expected: newer.Endpoints{
-				Protocol:  newer.ProtocolTCP,
-				Endpoints: []newer.Endpoint{{IP: "1.2.3.4", Port: 88}},
+				Subsets: []newer.EndpointSubset{{
+					Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolTCP}},
+					Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+				}},
 			},
 		},
 		{
 			given: current.Endpoints{
 				TypeMeta: current.TypeMeta{
-					ID: "several",
+					ID: "several legacy",
 				},
 				Protocol:  current.ProtocolUDP,
 				Endpoints: []string{"1.2.3.4:88", "1.2.3.4:89", "1.2.3.4:90"},
 			},
 			expected: newer.Endpoints{
-				Protocol:  newer.ProtocolUDP,
-				Endpoints: []newer.Endpoint{{IP: "1.2.3.4", Port: 88}, {IP: "1.2.3.4", Port: 89}, {IP: "1.2.3.4", Port: 90}},
+				Subsets: []newer.EndpointSubset{
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 89, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 90, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+					},
+				}},
+		},
+		{
+			given: current.Endpoints{
+				TypeMeta: current.TypeMeta{
+					ID: "one subset",
+				},
+				Protocol:  current.ProtocolTCP,
+				Endpoints: []string{"1.2.3.4:88"},
+				Subsets: []current.EndpointSubset{{
+					Ports:     []current.EndpointPort{{Name: "", Port: 88, Protocol: current.ProtocolTCP}},
+					Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}},
+				}},
+			},
+			expected: newer.Endpoints{
+				Subsets: []newer.EndpointSubset{{
+					Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolTCP}},
+					Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}},
+				}},
 			},
 		},
+		{
+			given: current.Endpoints{
+				TypeMeta: current.TypeMeta{
+					ID: "several subset",
+				},
+				Protocol:  current.ProtocolUDP,
+				Endpoints: []string{"1.2.3.4:88", "5.6.7.8:88", "1.2.3.4:89", "5.6.7.8:89"},
+				Subsets: []current.EndpointSubset{
+					{
+						Ports:     []current.EndpointPort{{Name: "", Port: 88, Protocol: current.ProtocolUDP}},
+						Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []current.EndpointPort{{Name: "", Port: 89, Protocol: current.ProtocolUDP}},
+						Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []current.EndpointPort{{Name: "named", Port: 90, Protocol: current.ProtocolUDP}},
+						Addresses: []current.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+				},
+			},
+			expected: newer.Endpoints{
+				Subsets: []newer.EndpointSubset{
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 88, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "", Port: 89, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+					{
+						Ports:     []newer.EndpointPort{{Name: "named", Port: 90, Protocol: newer.ProtocolUDP}},
+						Addresses: []newer.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+					},
+				}},
+		},
 	}
 
 	for i, tc := range testCases {
@@ -266,8 +335,8 @@ func TestEndpointsConversion(t *testing.T) {
 			t.Errorf("[Case: %d] Unexpected error: %v", i, err)
 			continue
 		}
-		if got.Protocol != tc.expected.Protocol || !newer.Semantic.DeepEqual(got.Endpoints, tc.expected.Endpoints) {
-			t.Errorf("[Case: %d] Expected %v, got %v", i, tc.expected, got)
+		if !newer.Semantic.DeepEqual(got.Subsets, tc.expected.Subsets) {
+			t.Errorf("[Case: %d] Expected %#v, got %#v", i, tc.expected.Subsets, got.Subsets)
 		}
 
 		// Convert internal -> versioned.
@@ -277,7 +346,7 @@ func TestEndpointsConversion(t *testing.T) {
 			continue
 		}
 		if got2.Protocol != tc.given.Protocol || !newer.Semantic.DeepEqual(got2.Endpoints, tc.given.Endpoints) {
-			t.Errorf("[Case: %d] Expected %v, got %v", i, tc.given, got2)
+			t.Errorf("[Case: %d] Expected %#v, got %#v", i, tc.given.Endpoints, got2.Endpoints)
 		}
 	}
 }

pkg/api/v1beta2/defaults.go

@@ -17,6 +17,8 @@ limitations under the License.
 package v1beta2
 
 import (
+	"net"
+	"strconv"
 	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
@@ -95,7 +97,42 @@ func init() {
 		},
 		func(obj *Endpoints) {
 			if obj.Protocol == "" {
-				obj.Protocol = "TCP"
+				obj.Protocol = ProtocolTCP
+			}
+			if len(obj.Subsets) == 0 && len(obj.Endpoints) > 0 {
+				// Must be a legacy-style object - populate
+				// Subsets from the older fields.  Do this the
+				// simplest way, which is dumb (but valid).
+				for i := range obj.Endpoints {
+					host, portStr, err := net.SplitHostPort(obj.Endpoints[i])
+					if err != nil {
+						glog.Errorf("failed to SplitHostPort(%q)", obj.Endpoints[i])
+					}
+					var tgtRef *ObjectReference
+					for j := range obj.TargetRefs {
+						if obj.TargetRefs[j].Endpoint == obj.Endpoints[i] {
+							tgtRef = &ObjectReference{}
+							*tgtRef = obj.TargetRefs[j].ObjectReference
+						}
+					}
+					port, err := strconv.Atoi(portStr)
+					if err != nil {
+						glog.Errorf("failed to Atoi(%q)", portStr)
+					}
+					obj.Subsets = append(obj.Subsets, EndpointSubset{
+						Addresses: []EndpointAddress{{IP: host, TargetRef: tgtRef}},
+						Ports:     []EndpointPort{{Protocol: obj.Protocol, Port: port}},
+					})
+				}
+			}
+			for i := range obj.Subsets {
+				ss := &obj.Subsets[i]
+				for i := range ss.Ports {
+					ep := &ss.Ports[i]
+					if ep.Protocol == "" {
+						ep.Protocol = ProtocolTCP
+					}
+				}
 			}
 		},
 		func(obj *HTTPGetAction) {

pkg/api/v1beta2/defaults_test.go

@@ -61,14 +61,56 @@ func TestSetDefaultSecret(t *testing.T) {
 	}
 }
 
+func TestSetDefaulEndpointsLegacy(t *testing.T) {
+	in := &current.Endpoints{
+		Protocol:   "UDP",
+		Endpoints:  []string{"1.2.3.4:93", "5.6.7.8:76"},
+		TargetRefs: []current.EndpointObjectReference{{Endpoint: "1.2.3.4:93", ObjectReference: current.ObjectReference{ID: "foo"}}},
+	}
+	obj := roundTrip(t, runtime.Object(in))
+	out := obj.(*current.Endpoints)
+
+	if len(out.Subsets) != 2 {
+		t.Errorf("Expected 2 EndpointSubsets, got %d (%#v)", len(out.Subsets), out.Subsets)
+	}
+	expected := []current.EndpointSubset{
+		{
+			Addresses: []current.EndpointAddress{{IP: "1.2.3.4", TargetRef: &current.ObjectReference{ID: "foo"}}},
+			Ports:     []current.EndpointPort{{Protocol: current.ProtocolUDP, Port: 93}},
+		},
+		{
+			Addresses: []current.EndpointAddress{{IP: "5.6.7.8"}},
+			Ports:     []current.EndpointPort{{Protocol: current.ProtocolUDP, Port: 76}},
+		},
+	}
+	if !reflect.DeepEqual(out.Subsets, expected) {
+		t.Errorf("Expected %#v, got %#v", expected, out.Subsets)
+	}
+}
+
 func TestSetDefaulEndpointsProtocol(t *testing.T) {
-	in := &current.Endpoints{}
+	in := &current.Endpoints{Subsets: []current.EndpointSubset{
+		{Ports: []current.EndpointPort{{}, {Protocol: "UDP"}, {}}},
+	}}
 	obj := roundTrip(t, runtime.Object(in))
 	out := obj.(*current.Endpoints)
 
 	if out.Protocol != current.ProtocolTCP {
 		t.Errorf("Expected protocol %s, got %s", current.ProtocolTCP, out.Protocol)
 	}
+	for i := range out.Subsets {
+		for j := range out.Subsets[i].Ports {
+			if in.Subsets[i].Ports[j].Protocol == "" {
+				if out.Subsets[i].Ports[j].Protocol != current.ProtocolTCP {
+					t.Errorf("Expected protocol %s, got %s", current.ProtocolTCP, out.Subsets[i].Ports[j].Protocol)
+				}
+			} else {
+				if out.Subsets[i].Ports[j].Protocol != in.Subsets[i].Ports[j].Protocol {
+					t.Errorf("Expected protocol %s, got %s", in.Subsets[i].Ports[j].Protocol, out.Subsets[i].Ports[j].Protocol)
+				}
+			}
+		}
+	}
 }
 
 func TestSetDefaultNamespace(t *testing.T) {

pkg/api/v1beta2/types.go

@@ -763,12 +763,61 @@ type EndpointObjectReference struct {
 // Name: "mysql", Endpoints: ["10.10.1.1:1909", "10.10.2.2:8834"]
 type Endpoints struct {
 	TypeMeta `json:",inline"`
-	// Optional: The IP protocol for these endpoints. Supports "TCP" and
-	// "UDP".  Defaults to "TCP".
-	Protocol  Protocol `json:"protocol,omitempty" description:"IP protocol for endpoint ports; must be UDP or TCP; TCP if unspecified"`
-	Endpoints []string `json:"endpoints" description:"list of endpoints corresponding to a service, of the form address:port, such as 10.10.1.1:1909"`
-	// Optional: The kubernetes object related to the entry point.
+
+	// These fields are retained for backwards compatibility.  For
+	// multi-port services, use the Subsets field instead.  Upon a create or
+	// update operation, the following logic applies:
+	//   * If Subsets is specified, Protocol, Endpoints, and TargetRefs will
+	//     be overwritten by data from Subsets.
+	//   * If Subsets is not specified, Protocol, Endpoints, and TargetRefs
+	//     will be used to generate Subsets.
+	Protocol  Protocol `json:"protocol,omitempty" description:"IP protocol for the first set of endpoint ports; must be UDP or TCP; TCP if unspecified"`
+	Endpoints []string `json:"endpoints" description:"first set of endpoints corresponding to a service, of the form address:port, such as 10.10.1.1:1909"`
+	// Optional: The kubernetes objects related to the first set of entry points.
 	TargetRefs []EndpointObjectReference `json:"targetRefs,omitempty" description:"list of references to objects providing the endpoints"`
+
+	// The set of all endpoints is the union of all subsets.  If this field
+	// is not empty it must include the backwards-compatible protocol and
+	// endpoints.
+	Subsets []EndpointSubset `json:"subsets" description:"sets of addresses and ports that comprise a service"`
+}
+
+// EndpointSubset is a group of addresses with a common set of ports.  The
+// expanded set of endpoints is the Cartesian product of Addresses x Ports.
+// For example, given:
+//   {
+//     Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//     Ports:     [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//   }
+// The resulting set of endpoints can be viewed as:
+//     a: [ 10.10.1.1:8675, 10.10.2.2:8675 ],
+//     b: [ 10.10.1.1:309, 10.10.2.2:309 ]
+type EndpointSubset struct {
+	Addresses []EndpointAddress `json:"addresses,omitempty" description:"IP addresses which offer the related ports"`
+	Ports     []EndpointPort    `json:"ports,omitempty" description:"port numbers available on the related IP addresses"`
+}
+
+// EndpointAddress is a tuple that describes single IP address.
+type EndpointAddress struct {
+	// The IP of this endpoint.
+	// TODO: This should allow hostname or IP, see #4447.
+	IP string `json:"IP" description:"IP address of the endpoint"`
+
+	// Optional: The kubernetes object related to the entry point.
+	TargetRef *ObjectReference `json:"targetRef,omitempty" description:"reference to object providing the endpoint"`
+}
+
+// EndpointPort is a tuple that describes a single port.
+type EndpointPort struct {
+	// The name of this port (corresponds to ServicePort.Name).  Optional
+	// if only one port is defined.  Must be a DNS_LABEL.
+	Name string `json:"name,omitempty" description:"name of this port"`
+
+	// The port number.
+	Port int `json:"port" description:"port number of the endpoint"`
+
+	// The IP protocol for this port.
+	Protocol Protocol `json:"protocol,omitempty" description:"protocol for this port; must be UDP or TCP; TCP if unspecified"`
 }
 
 // EndpointsList is a list of endpoints.

pkg/api/v1beta3/defaults.go

@@ -82,8 +82,14 @@ func init() {
 			}
 		},
 		func(obj *Endpoints) {
-			if obj.Protocol == "" {
-				obj.Protocol = "TCP"
+			for i := range obj.Subsets {
+				ss := &obj.Subsets[i]
+				for i := range ss.Ports {
+					ep := &ss.Ports[i]
+					if ep.Protocol == "" {
+						ep.Protocol = ProtocolTCP
+					}
+				}
 			}
 		},
 		func(obj *HTTPGetAction) {

pkg/api/v1beta3/defaults_test.go

@@ -63,12 +63,24 @@ func TestSetDefaultSecret(t *testing.T) {
 }
 
 func TestSetDefaulEndpointsProtocol(t *testing.T) {
-	in := &current.Endpoints{}
+	in := &current.Endpoints{Subsets: []current.EndpointSubset{
+		{Ports: []current.EndpointPort{{}, {Protocol: "UDP"}, {}}},
+	}}
 	obj := roundTrip(t, runtime.Object(in))
 	out := obj.(*current.Endpoints)
 
-	if out.Protocol != current.ProtocolTCP {
-		t.Errorf("Expected protocol %s, got %s", current.ProtocolTCP, out.Protocol)
+	for i := range out.Subsets {
+		for j := range out.Subsets[i].Ports {
+			if in.Subsets[i].Ports[j].Protocol == "" {
+				if out.Subsets[i].Ports[j].Protocol != current.ProtocolTCP {
+					t.Errorf("Expected protocol %s, got %s", current.ProtocolTCP, out.Subsets[i].Ports[j].Protocol)
+				}
+			} else {
+				if out.Subsets[i].Ports[j].Protocol != in.Subsets[i].Ports[j].Protocol {
+					t.Errorf("Expected protocol %s, got %s", in.Subsets[i].Ports[j].Protocol, out.Subsets[i].Ports[j].Protocol)
+				}
+			}
+		}
 	}
 }
 

pkg/api/v1beta3/types.go

@@ -914,32 +914,64 @@ type ServiceList struct {
 	Items []Service `json:"items" description:"list of services"`
 }
 
-// Endpoints is a collection of endpoints that implement the actual service, for example:
-// Name: "mysql", Endpoints: [{"ip": "10.10.1.1", "port": 1909}, {"ip": "10.10.2.2", "port": 8834}]
+// Endpoints is a collection of endpoints that implement the actual service.  Example:
+//   Name: "mysvc",
+//   Subsets: [
+//     {
+//       Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//       Ports: [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//     },
+//     {
+//       Addresses: [{"ip": "10.10.3.3"}],
+//       Ports: [{"name": "a", "port": 93}, {"name": "b", "port": 76}]
+//     },
+//  ]
 type Endpoints struct {
 	TypeMeta   `json:",inline"`
 	ObjectMeta `json:"metadata,omitempty" description:"standard object metadata; see https://github.com/GoogleCloudPlatform/kubernetes/blob/master/docs/api-conventions.md#metadata"`
 
-	// Optional: The IP protocol for these endpoints. Supports "TCP" and
-	// "UDP".  Defaults to "TCP".
-	Protocol Protocol `json:"protocol,omitempty" description:"IP protocol for endpoint ports; must be UDP or TCP; TCP if unspecified"`
+	// The set of all endpoints is the union of all subsets.
+	Subsets []EndpointSubset `json:"subsets" description:"sets of addresses and ports that comprise a service"`
+}
 
-	Endpoints []Endpoint `json:"endpoints,omitempty" description:"list of endpoints corresponding to a service"`
+// EndpointSubset is a group of addresses with a common set of ports.  The
+// expanded set of endpoints is the Cartesian product of Addresses x Ports.
+// For example, given:
+//   {
+//     Addresses: [{"ip": "10.10.1.1"}, {"ip": "10.10.2.2"}],
+//     Ports:     [{"name": "a", "port": 8675}, {"name": "b", "port": 309}]
+//   }
+// The resulting set of endpoints can be viewed as:
+//     a: [ 10.10.1.1:8675, 10.10.2.2:8675 ],
+//     b: [ 10.10.1.1:309, 10.10.2.2:309 ]
+type EndpointSubset struct {
+	Addresses []EndpointAddress `json:"addresses,omitempty" description:"IP addresses which offer the related ports"`
+	Ports     []EndpointPort    `json:"ports,omitempty" description:"port numbers available on the related IP addresses"`
 }
 
-// Endpoint is a single IP endpoint of a service.
-type Endpoint struct {
-	// Required: The IP of this endpoint.
+// EndpointAddress is a tuple that describes single IP address.
+type EndpointAddress struct {
+	// The IP of this endpoint.
 	// TODO: This should allow hostname or IP, see #4447.
-	IP string `json:"ip" description:"IP of this endpoint"`
-
-	// Required: The destination port to access.
-	Port int `json:"port" description:"destination port of this endpoint"`
+	IP string `json:"IP" description:"IP address of the endpoint"`
 
 	// Optional: The kubernetes object related to the entry point.
 	TargetRef *ObjectReference `json:"targetRef,omitempty" description:"reference to object providing the endpoint"`
 }
 
+// EndpointPort is a tuple that describes a single port.
+type EndpointPort struct {
+	// The name of this port (corresponds to ServicePort.Name).  Optional
+	// if only one port is defined.  Must be a DNS_LABEL.
+	Name string `json:"name,omitempty" description:"name of this port"`
+
+	// The port number.
+	Port int `json:"port" description:"port number of the endpoint"`
+
+	// The IP protocol for this port.
+	Protocol Protocol `json:"protocol,omitempty" description:"protocol for this port; must be UDP or TCP; TCP if unspecified"`
+}
+
 // EndpointsList is a list of endpoints.
 type EndpointsList struct {
 	TypeMeta `json:",inline"`

pkg/api/validation/validation.go

@@ -1158,7 +1158,6 @@ func ValidateNamespaceFinalizeUpdate(newNamespace, oldNamespace *api.Namespace)
 	}
 	newNamespace.ObjectMeta = oldNamespace.ObjectMeta
 	newNamespace.Status = oldNamespace.Status
-	fmt.Printf("NEW NAMESPACE FINALIZERS : %v\n", newNamespace.Spec.Finalizers)
 	return allErrs
 }
 
@@ -1166,6 +1165,28 @@ func ValidateNamespaceFinalizeUpdate(newNamespace, oldNamespace *api.Namespace)
 func ValidateEndpoints(endpoints *api.Endpoints) errs.ValidationErrorList {
 	allErrs := errs.ValidationErrorList{}
 	allErrs = append(allErrs, ValidateObjectMeta(&endpoints.ObjectMeta, true, ValidateEndpointsName).Prefix("metadata")...)
+	allErrs = append(allErrs, validateEndpointSubsets(endpoints.Subsets).Prefix("subsets")...)
+	return allErrs
+}
+
+func validateEndpointSubsets(subsets []api.EndpointSubset) errs.ValidationErrorList {
+	allErrs := errs.ValidationErrorList{}
+
+	for i := range subsets {
+		ss := &subsets[i]
+
+		ssErrs := errs.ValidationErrorList{}
+
+		if len(ss.Addresses) == 0 {
+			ssErrs = append(ssErrs, errs.NewFieldRequired("addresses"))
+		}
+		if len(ss.Ports) == 0 {
+			ssErrs = append(ssErrs, errs.NewFieldRequired("ports"))
+		}
+		// TODO: validate each address and port
+		allErrs = append(allErrs, ssErrs.PrefixIndex(i)...)
+	}
+
 	return allErrs
 }
 
@@ -1173,5 +1194,6 @@ func ValidateEndpoints(endpoints *api.Endpoints) errs.ValidationErrorList {
 func ValidateEndpointsUpdate(oldEndpoints *api.Endpoints, endpoints *api.Endpoints) errs.ValidationErrorList {
 	allErrs := errs.ValidationErrorList{}
 	allErrs = append(allErrs, ValidateObjectMetaUpdate(&oldEndpoints.ObjectMeta, &endpoints.ObjectMeta).Prefix("metadata")...)
+	allErrs = append(allErrs, validateEndpointSubsets(endpoints.Subsets).Prefix("subsets")...)
 	return allErrs
 }

pkg/api/validation/validation_test.go

@@ -2746,3 +2746,7 @@ func TestValidateSecret(t *testing.T) {
 		}
 	}
 }
+
+func TestValidateEndpoints(t *testing.T) {
+	// TODO: implement this
+}

pkg/client/endpoints_test.go

@@ -24,7 +24,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 )
 
-func TestListEndpooints(t *testing.T) {
+func TestListEndpoints(t *testing.T) {
 	ns := api.NamespaceDefault
 	c := &testClient{
 		Request: testRequest{Method: "GET", Path: testapi.ResourcePath("endpoints", ns, ""), Query: buildQueryValues(ns, nil)},
@@ -33,8 +33,10 @@ func TestListEndpooints(t *testing.T) {
 				Items: []api.Endpoints{
 					{
 						ObjectMeta: api.ObjectMeta{Name: "endpoint-1"},
-						Endpoints: []api.Endpoint{
-							{IP: "10.245.1.2", Port: 8080}, {IP: "10.245.1.3", Port: 8080}},
+						Subsets: []api.EndpointSubset{{
+							Addresses: []api.EndpointAddress{{IP: "10.245.1.2"}, {IP: "10.245.1.3"}},
+							Ports:     []api.EndpointPort{{Port: 8080}},
+						}},
 					},
 				},
 			},

pkg/kubectl/describe.go

@@ -334,7 +334,7 @@ func describeService(service *api.Service, endpoints *api.Endpoints, events *api
 			fmt.Fprintf(out, "Public IPs:\t%s\n", list)
 		}
 		fmt.Fprintf(out, "Port:\t%d\n", service.Spec.Port)
-		fmt.Fprintf(out, "Endpoints:\t%s\n", formatEndpoints(endpoints.Endpoints))
+		fmt.Fprintf(out, "Endpoints:\t%s\n", formatEndpoints(endpoints))
 		fmt.Fprintf(out, "Session Affinity:\t%s\n", service.Spec.SessionAffinity)
 		if events != nil {
 			describeEvents(events, out)

pkg/kubectl/resource_printer.go

@@ -22,10 +22,8 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"net"
 	"reflect"
 	"sort"
-	"strconv"
 	"strings"
 	"text/tabwriter"
 	"text/template"
@@ -277,16 +275,35 @@ func (h *HumanReadablePrinter) printHeader(columnNames []string, w io.Writer) er
 	return nil
 }
 
-func formatEndpoints(endpoints []api.Endpoint) string {
-	if len(endpoints) == 0 {
+func formatEndpoints(endpoints *api.Endpoints) string {
+	if len(endpoints.Subsets) == 0 {
 		return "<none>"
 	}
 	list := []string{}
-	for i := range endpoints {
-		ep := &endpoints[i]
-		list = append(list, net.JoinHostPort(ep.IP, strconv.Itoa(ep.Port)))
+	max := 3
+	more := false
+Loop:
+	for i := range endpoints.Subsets {
+		ss := &endpoints.Subsets[i]
+		for i := range ss.Ports {
+			port := &ss.Ports[i]
+			if port.Name == "" { // TODO: add multi-port support.
+				for i := range ss.Addresses {
+					if len(list) == max {
+						more = true
+						break Loop
+					}
+					addr := &ss.Addresses[i]
+					list = append(list, fmt.Sprintf("%s:%d", addr.IP, port.Port))
+				}
+			}
+		}
+	}
+	ret := strings.Join(list, ",")
+	if more {
+		ret += "..."
 	}
-	return strings.Join(list, ",")
+	return ret
 }
 
 func podHostString(host, ip string) string {
@@ -387,8 +404,8 @@ func printServiceList(list *api.ServiceList, w io.Writer) error {
 	return nil
 }
 
-func printEndpoints(endpoint *api.Endpoints, w io.Writer) error {
-	_, err := fmt.Fprintf(w, "%s\t%s\n", endpoint.Name, formatEndpoints(endpoint.Endpoints))
+func printEndpoints(endpoints *api.Endpoints, w io.Writer) error {
+	_, err := fmt.Fprintf(w, "%s\t%s\n", endpoints.Name, formatEndpoints(endpoints))
 	return err
 }
 

pkg/kubectl/resource_printer_test.go

@@ -469,7 +469,11 @@ func TestPrinters(t *testing.T) {
 		"pod":             &api.Pod{ObjectMeta: om("pod")},
 		"emptyPodList":    &api.PodList{},
 		"nonEmptyPodList": &api.PodList{Items: []api.Pod{{}}},
-		"endpoints":       &api.Endpoints{Endpoints: []api.Endpoint{{IP: "127.0.0.1"}, {IP: "localhost", Port: 8080}}},
+		"endpoints": &api.Endpoints{
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}, {IP: "localhost"}},
+				Ports:     []api.EndpointPort{{Port: 8080}},
+			}}},
 	}
 	// map of printer name to set of objects it should fail on.
 	expectedErrors := map[string]util.StringSet{

pkg/master/publish.go

@@ -18,6 +18,7 @@ package master
 
 import (
 	"net"
+	"reflect"
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
@@ -40,7 +41,7 @@ func (m *Master) serviceWriterLoop(stop chan struct{}) {
 			if err := m.createMasterServiceIfNeeded("kubernetes", m.serviceReadWriteIP, m.serviceReadWritePort); err != nil {
 				glog.Errorf("Can't create rw service: %v", err)
 			}
-			if err := m.ensureEndpointsContain("kubernetes", m.clusterIP, m.publicReadWritePort); err != nil {
+			if err := m.setEndpoints("kubernetes", m.clusterIP, m.publicReadWritePort); err != nil {
 				glog.Errorf("Can't create rw endpoints: %v", err)
 			}
 		}
@@ -65,7 +66,7 @@ func (m *Master) roServiceWriterLoop(stop chan struct{}) {
 			if err := m.createMasterServiceIfNeeded("kubernetes-ro", m.serviceReadOnlyIP, m.serviceReadOnlyPort); err != nil {
 				glog.Errorf("Can't create ro service: %v", err)
 			}
-			if err := m.ensureEndpointsContain("kubernetes-ro", m.clusterIP, m.publicReadOnlyPort); err != nil {
+			if err := m.setEndpoints("kubernetes-ro", m.clusterIP, m.publicReadOnlyPort); err != nil {
 				glog.Errorf("Can't create ro endpoints: %v", err)
 			}
 		}
@@ -128,37 +129,28 @@ func (m *Master) createMasterServiceIfNeeded(serviceName string, serviceIP net.I
 	return err
 }
 
-// ensureEndpointsContain sets the endpoints for the given service. Also removes
-// excess endpoints (as determined by m.masterCount). Extra endpoints could appear
-// in the list if, for example, the master starts running on a different machine,
-// changing IP addresses.
-func (m *Master) ensureEndpointsContain(serviceName string, ip net.IP, port int) error {
+// setEndpoints sets the endpoints for the given service.
+// TODO: in a multi-master scenario this needs to consider all masters.
+func (m *Master) setEndpoints(serviceName string, ip net.IP, port int) error {
+	// The setting we want to find.
+	want := []api.EndpointSubset{{
+		Addresses: []api.EndpointAddress{{IP: ip.String()}},
+		Ports:     []api.EndpointPort{{Port: port, Protocol: api.ProtocolTCP}},
+	}}
+
 	ctx := api.NewDefaultContext()
 	e, err := m.endpointRegistry.GetEndpoints(ctx, serviceName)
-	if err != nil || e.Protocol != api.ProtocolTCP {
+	if err != nil {
 		e = &api.Endpoints{
 			ObjectMeta: api.ObjectMeta{
 				Name:      serviceName,
 				Namespace: api.NamespaceDefault,
 			},
-			Protocol: api.ProtocolTCP,
-		}
-	}
-	found := false
-	for i := range e.Endpoints {
-		ep := &e.Endpoints[i]
-		if ep.IP == ip.String() && ep.Port == port {
-			found = true
-			break
 		}
 	}
-	if !found {
-		e.Endpoints = append(e.Endpoints, api.Endpoint{IP: ip.String(), Port: port})
-		if len(e.Endpoints) > m.masterCount {
-			// We append to the end and remove from the beginning, so this should
-			// converge rapidly with all masters performing this operation.
-			e.Endpoints = e.Endpoints[len(e.Endpoints)-m.masterCount:]
-		}
+	if !reflect.DeepEqual(e.Subsets, want) {
+		e.Subsets = want
+		glog.Infof("setting endpoints for master service %q to %v", serviceName, e)
 		return m.endpointRegistry.UpdateEndpoints(ctx, e)
 	}
 	// We didn't make any changes, no need to actually call update.

pkg/master/publish_test.go

@@ -19,245 +19,149 @@ package master
 import (
 	"net"
 	"reflect"
-	"sync"
 	"testing"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/registrytest"
 )
 
-func TestEnsureEndpointsContain(t *testing.T) {
+func TestSetEndpoints(t *testing.T) {
 	tests := []struct {
-		serviceName       string
-		ip                string
-		port              int
-		expectError       bool
-		expectUpdate      bool
-		endpoints         *api.EndpointsList
-		expectedEndpoints []api.Endpoint
-		err               error
-		masterCount       int
+		testName     string
+		serviceName  string
+		ip           string
+		port         int
+		endpoints    *api.EndpointsList
+		expectUpdate bool
 	}{
 		{
+			testName:     "no existing endpoints",
 			serviceName:  "foo",
 			ip:           "1.2.3.4",
 			port:         8080,
-			expectError:  false,
+			endpoints:    nil,
 			expectUpdate: true,
-			masterCount:  1,
 		},
 		{
-			serviceName:  "foo",
-			ip:           "1.2.3.4",
-			port:         8080,
-			expectError:  false,
-			expectUpdate: false,
+			testName:    "existing endpoints satisfy",
+			serviceName: "foo",
+			ip:          "1.2.3.4",
+			port:        8080,
 			endpoints: &api.EndpointsList{
-				Items: []api.Endpoints{
-					{
-						ObjectMeta: api.ObjectMeta{
-							Name: "foo",
-						},
-						Endpoints: []api.Endpoint{
-							{
-								IP:   "1.2.3.4",
-								Port: 8080,
-							},
-						},
-						Protocol: api.ProtocolTCP,
-					},
-				},
+				Items: []api.Endpoints{{
+					ObjectMeta: api.ObjectMeta{Name: "foo"},
+					Subsets: []api.EndpointSubset{{
+						Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+						Ports:     []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
+					}},
+				}},
 			},
-			masterCount:       1,
-			expectedEndpoints: []api.Endpoint{{"1.2.3.4", 8080, nil}},
+			expectUpdate: false,
 		},
 		{
-			serviceName:  "foo",
-			ip:           "1.2.3.4",
-			port:         8080,
-			expectError:  false,
-			expectUpdate: true,
+			testName:    "existing endpoints satisfy but too many",
+			serviceName: "foo",
+			ip:          "1.2.3.4",
+			port:        8080,
 			endpoints: &api.EndpointsList{
-				Items: []api.Endpoints{
-					{
-						ObjectMeta: api.ObjectMeta{
-							Name:      "foo",
-							Namespace: api.NamespaceDefault,
-						},
-						Endpoints: []api.Endpoint{
-							{
-								IP:   "4.3.2.1",
-								Port: 8080,
-							},
-						},
-						Protocol: api.ProtocolTCP,
-					},
-				},
+				Items: []api.Endpoints{{
+					ObjectMeta: api.ObjectMeta{Name: "foo"},
+					Subsets: []api.EndpointSubset{{
+						Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "4.3.2.1"}},
+						Ports:     []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
+					}},
+				}},
 			},
-			masterCount: 1,
+			expectUpdate: true,
 		},
 		{
-			serviceName:  "foo",
-			ip:           "1.2.3.4",
-			port:         8080,
-			expectError:  false,
+			testName:    "existing endpoints wrong name",
+			serviceName: "foo",
+			ip:          "1.2.3.4",
+			port:        8080,
+			endpoints: &api.EndpointsList{
+				Items: []api.Endpoints{{
+					ObjectMeta: api.ObjectMeta{Name: "bar"},
+					Subsets: []api.EndpointSubset{{
+						Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+						Ports:     []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
+					}},
+				}},
+			},
 			expectUpdate: true,
+		},
+		{
+			testName:    "existing endpoints wrong IP",
+			serviceName: "foo",
+			ip:          "1.2.3.4",
+			port:        8080,
 			endpoints: &api.EndpointsList{
-				Items: []api.Endpoints{
-					{
-						ObjectMeta: api.ObjectMeta{
-							Name:      "foo",
-							Namespace: api.NamespaceDefault,
-						},
-						Endpoints: []api.Endpoint{
-							{
-								IP:   "4.3.2.1",
-								Port: 9090,
-							},
-						},
-						Protocol: api.ProtocolTCP,
-					},
-				},
+				Items: []api.Endpoints{{
+					ObjectMeta: api.ObjectMeta{Name: "foo"},
+					Subsets: []api.EndpointSubset{{
+						Addresses: []api.EndpointAddress{{IP: "4.3.2.1"}},
+						Ports:     []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
+					}},
+				}},
 			},
-			masterCount:       2,
-			expectedEndpoints: []api.Endpoint{{"4.3.2.1", 9090, nil}, {"1.2.3.4", 8080, nil}},
+			expectUpdate: true,
 		},
 		{
-			serviceName:  "foo",
-			ip:           "1.2.3.4",
-			port:         8080,
-			expectError:  false,
+			testName:    "existing endpoints wrong port",
+			serviceName: "foo",
+			ip:          "1.2.3.4",
+			port:        8080,
+			endpoints: &api.EndpointsList{
+				Items: []api.Endpoints{{
+					ObjectMeta: api.ObjectMeta{Name: "foo"},
+					Subsets: []api.EndpointSubset{{
+						Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+						Ports:     []api.EndpointPort{{Port: 9090, Protocol: "TCP"}},
+					}},
+				}},
+			},
 			expectUpdate: true,
+		},
+		{
+			testName:    "existing endpoints wrong protocol",
+			serviceName: "foo",
+			ip:          "1.2.3.4",
+			port:        8080,
 			endpoints: &api.EndpointsList{
-				Items: []api.Endpoints{
-					{
-						ObjectMeta: api.ObjectMeta{
-							Name:      "foo",
-							Namespace: api.NamespaceDefault,
-						},
-						Endpoints: []api.Endpoint{
-							{
-								IP:   "4.3.2.1",
-								Port: 9090,
-							},
-							{
-								IP:   "1.2.3.4",
-								Port: 8000,
-							},
-						},
-						Protocol: api.ProtocolTCP,
-					},
-				},
+				Items: []api.Endpoints{{
+					ObjectMeta: api.ObjectMeta{Name: "foo"},
+					Subsets: []api.EndpointSubset{{
+						Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+						Ports:     []api.EndpointPort{{Port: 8080, Protocol: "UDP"}},
+					}},
+				}},
 			},
-			masterCount:       2,
-			expectedEndpoints: []api.Endpoint{{"1.2.3.4", 8000, nil}, {"1.2.3.4", 8080, nil}},
+			expectUpdate: true,
 		},
 	}
 	for _, test := range tests {
 		master := Master{}
 		registry := &registrytest.EndpointRegistry{
 			Endpoints: test.endpoints,
-			Err:       test.err,
 		}
 		master.endpointRegistry = registry
-		master.masterCount = test.masterCount
-		err := master.ensureEndpointsContain(test.serviceName, net.ParseIP(test.ip), test.port)
-		if test.expectError && err == nil {
-			t.Errorf("unexpected non-error")
-		}
-		if !test.expectError && err != nil {
-			t.Errorf("unexpected error: %v", err)
+		err := master.setEndpoints(test.serviceName, net.ParseIP(test.ip), test.port)
+		if err != nil {
+			t.Errorf("case %q: unexpected error: %v", test.testName, err)
 		}
 		if test.expectUpdate {
-			if test.expectedEndpoints == nil {
-				test.expectedEndpoints = []api.Endpoint{{test.ip, test.port, nil}}
-			}
-			expectedUpdate := api.Endpoints{
-				ObjectMeta: api.ObjectMeta{
-					Name:      test.serviceName,
-					Namespace: "default",
-				},
-				Endpoints: test.expectedEndpoints,
-				Protocol:  "TCP",
-			}
+			expectedSubsets := []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+				Ports:     []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
+			}}
 			if len(registry.Updates) != 1 {
-				t.Errorf("unexpected updates: %v", registry.Updates)
-			} else if !reflect.DeepEqual(expectedUpdate, registry.Updates[0]) {
-				t.Errorf("expected update:\n%#v\ngot:\n%#v\n", expectedUpdate, registry.Updates[0])
+				t.Errorf("case %q: unexpected updates: %v", test.testName, registry.Updates)
+			} else if !reflect.DeepEqual(expectedSubsets, registry.Updates[0].Subsets) {
+				t.Errorf("case %q: expected update:\n%#v\ngot:\n%#v\n", test.testName, expectedSubsets, registry.Updates[0].Subsets)
 			}
 		}
 		if !test.expectUpdate && len(registry.Updates) > 0 {
-			t.Errorf("no update expected, yet saw: %v", registry.Updates)
-		}
-	}
-}
-
-func TestEnsureEndpointsContainConverges(t *testing.T) {
-	master := Master{}
-	registry := &registrytest.EndpointRegistry{
-		Endpoints: &api.EndpointsList{
-			Items: []api.Endpoints{
-				{
-					ObjectMeta: api.ObjectMeta{
-						Name:      "foo",
-						Namespace: api.NamespaceDefault,
-					},
-					Endpoints: []api.Endpoint{
-						{
-							IP:   "4.3.2.1",
-							Port: 9000,
-						},
-						{
-							IP:   "1.2.3.4",
-							Port: 8000,
-						},
-					},
-					Protocol: api.ProtocolTCP,
-				},
-			},
-		},
-	}
-	master.endpointRegistry = registry
-	master.masterCount = 2
-	// This is purposefully racy, it shouldn't matter the order that these things arrive,
-	// we should still converge on the right answer.
-	wg := sync.WaitGroup{}
-	wg.Add(2)
-	go func() {
-		for i := 0; i < 10; i++ {
-			if err := master.ensureEndpointsContain("foo", net.ParseIP("4.3.2.1"), 9090); err != nil {
-				t.Errorf("unexpected error: %v", err)
-				t.Fail()
-			}
-		}
-		wg.Done()
-	}()
-	go func() {
-		for i := 0; i < 10; i++ {
-			if err := master.ensureEndpointsContain("foo", net.ParseIP("1.2.3.4"), 8080); err != nil {
-				t.Errorf("unexpected error: %v", err)
-				t.Fail()
-			}
-		}
-		wg.Done()
-	}()
-	wg.Wait()
-
-	// We should see at least two updates.
-	if len(registry.Updates) > 2 {
-		t.Errorf("unexpected updates: %v", registry.Updates)
-	}
-	// Pick up the last update and validate.
-	endpoints := registry.Updates[len(registry.Updates)-1]
-	if len(endpoints.Endpoints) != 2 {
-		t.Errorf("unexpected update: %v", endpoints)
-	}
-	for _, endpoint := range endpoints.Endpoints {
-		if endpoint.IP == "4.3.2.1" && endpoint.Port != 9090 {
-			t.Errorf("unexpected endpoint state: %v", endpoint)
-		}
-		if endpoint.IP == "1.2.3.4" && endpoint.Port != 8080 {
-			t.Errorf("unexpected endpoint state: %v", endpoint)
+			t.Errorf("case %q: no update expected, yet saw: %v", test.testName, registry.Updates)
 		}
 	}
 }

pkg/proxy/config/api_test.go

@@ -185,7 +185,10 @@ func TestServicesFromZeroError(t *testing.T) {
 func TestEndpoints(t *testing.T) {
 	endpoint := api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Name: "bar", ResourceVersion: "2"},
-		Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: 9000}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+			Ports:     []api.EndpointPort{{Port: 9000}},
+		}},
 	}
 
 	fakeWatch := watch.NewFake()
@@ -235,7 +238,10 @@ func TestEndpoints(t *testing.T) {
 func TestEndpointsFromZero(t *testing.T) {
 	endpoint := api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Name: "bar", ResourceVersion: "2"},
-		Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: 9000}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+			Ports:     []api.EndpointPort{{Port: 9000}},
+		}},
 	}
 
 	fakeWatch := watch.NewFake()

pkg/proxy/config/config_test.go

@@ -218,11 +218,17 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddedAndNotified(t *testing.
 	config.RegisterHandler(handler2)
 	endpointsUpdate1 := CreateEndpointsUpdate(ADD, api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Namespace: "testnamespace", Name: "foo"},
-		Endpoints:  []api.Endpoint{{IP: "endpoint1"}, {IP: "endpoint2"}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "1.1.1.1"}, {IP: "2.2.2.2"}},
+			Ports:     []api.EndpointPort{{Port: 80}},
+		}},
 	})
 	endpointsUpdate2 := CreateEndpointsUpdate(ADD, api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Namespace: "testnamespace", Name: "bar"},
-		Endpoints:  []api.Endpoint{{IP: "endpoint3"}, {IP: "endpoint4"}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "3.3.3.3"}, {IP: "4.4.4.4"}},
+			Ports:     []api.EndpointPort{{Port: 80}},
+		}},
 	})
 	handler.Wait(2)
 	handler2.Wait(2)
@@ -244,11 +250,17 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddRemoveSetAndNotified(t *t
 	config.RegisterHandler(handler2)
 	endpointsUpdate1 := CreateEndpointsUpdate(ADD, api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Namespace: "testnamespace", Name: "foo"},
-		Endpoints:  []api.Endpoint{{IP: "endpoint1"}, {IP: "endpoint2"}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "1.1.1.1"}, {IP: "2.2.2.2"}},
+			Ports:     []api.EndpointPort{{Port: 80}},
+		}},
 	})
 	endpointsUpdate2 := CreateEndpointsUpdate(ADD, api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Namespace: "testnamespace", Name: "bar"},
-		Endpoints:  []api.Endpoint{{IP: "endpoint3"}, {IP: "endpoint4"}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "3.3.3.3"}, {IP: "4.4.4.4"}},
+			Ports:     []api.EndpointPort{{Port: 80}},
+		}},
 	})
 	handler.Wait(2)
 	handler2.Wait(2)
@@ -262,7 +274,10 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddRemoveSetAndNotified(t *t
 	// Add one more
 	endpointsUpdate3 := CreateEndpointsUpdate(ADD, api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Namespace: "testnamespace", Name: "foobar"},
-		Endpoints:  []api.Endpoint{{IP: "endpoint5"}, {IP: "endpoint6"}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "5.5.5.5"}, {IP: "6.6.6.6"}},
+			Ports:     []api.EndpointPort{{Port: 80}},
+		}},
 	})
 	handler.Wait(1)
 	handler2.Wait(1)
@@ -274,7 +289,10 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddRemoveSetAndNotified(t *t
 	// Update the "foo" service with new endpoints
 	endpointsUpdate1 = CreateEndpointsUpdate(ADD, api.Endpoints{
 		ObjectMeta: api.ObjectMeta{Namespace: "testnamespace", Name: "foo"},
-		Endpoints:  []api.Endpoint{{IP: "endpoint7"}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "7.7.7.7"}},
+			Ports:     []api.EndpointPort{{Port: 80}},
+		}},
 	})
 	handler.Wait(1)
 	handler2.Wait(1)

pkg/proxy/loadbalancer.go

@@ -26,8 +26,8 @@ import (
 // LoadBalancer is an interface for distributing incoming requests to service endpoints.
 type LoadBalancer interface {
 	// NextEndpoint returns the endpoint to handle a request for the given
-	// service and source address.
-	NextEndpoint(service types.NamespacedName, srcAddr net.Addr) (string, error)
-	NewService(service types.NamespacedName, sessionAffinityType api.AffinityType, stickyMaxAgeMinutes int) error
-	CleanupStaleStickySessions(service types.NamespacedName)
+	// service-port and source address.
+	NextEndpoint(service types.NamespacedName, port string, srcAddr net.Addr) (string, error)
+	NewService(service types.NamespacedName, port string, sessionAffinityType api.AffinityType, stickyMaxAgeMinutes int) error
+	CleanupStaleStickySessions(service types.NamespacedName, port string)
 }

pkg/proxy/proxier.go

@@ -70,7 +70,8 @@ type tcpProxySocket struct {
 
 func tryConnect(service types.NamespacedName, srcAddr net.Addr, protocol string, proxier *Proxier) (out net.Conn, err error) {
 	for _, retryTimeout := range endpointDialTimeout {
-		endpoint, err := proxier.loadBalancer.NextEndpoint(service, srcAddr)
+		// TODO: support multiple service ports
+		endpoint, err := proxier.loadBalancer.NextEndpoint(service, "", srcAddr)
 		if err != nil {
 			glog.Errorf("Couldn't find an endpoint for %s: %v", service, err)
 			return nil, err
@@ -388,7 +389,8 @@ func (proxier *Proxier) ensurePortals() {
 func (proxier *Proxier) cleanupStaleStickySessions() {
 	for name, info := range proxier.serviceMap {
 		if info.sessionAffinityType != api.AffinityTypeNone {
-			proxier.loadBalancer.CleanupStaleStickySessions(name)
+			// TODO: support multiple service ports
+			proxier.loadBalancer.CleanupStaleStickySessions(name, "")
 		}
 	}
 }
@@ -509,7 +511,8 @@ func (proxier *Proxier) OnUpdate(services []api.Service) {
 		if err != nil {
 			glog.Errorf("Failed to open portal for %q: %v", serviceName, err)
 		}
-		proxier.loadBalancer.NewService(serviceName, info.sessionAffinityType, info.stickyMaxAgeMinutes)
+		// TODO: support multiple service ports
+		proxier.loadBalancer.NewService(serviceName, "", info.sessionAffinityType, info.stickyMaxAgeMinutes)
 	}
 	proxier.mu.Lock()
 	defer proxier.mu.Unlock()

pkg/proxy/proxier_test.go

@@ -199,7 +199,10 @@ func TestTCPProxy(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: tcpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: tcpServerPort}},
+			}},
 		},
 	})
 
@@ -220,7 +223,10 @@ func TestUDPProxy(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: udpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: udpServerPort}},
+			}},
 		},
 	})
 
@@ -250,7 +256,10 @@ func TestTCPProxyStop(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: tcpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: tcpServerPort}},
+			}},
 		},
 	})
 
@@ -282,7 +291,10 @@ func TestUDPProxyStop(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: udpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: udpServerPort}},
+			}},
 		},
 	})
 
@@ -314,7 +326,10 @@ func TestTCPProxyUpdateDelete(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: tcpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: tcpServerPort}},
+			}},
 		},
 	})
 
@@ -345,7 +360,10 @@ func TestUDPProxyUpdateDelete(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: udpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: udpServerPort}},
+			}},
 		},
 	})
 
@@ -376,7 +394,10 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: tcpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: tcpServerPort}},
+			}},
 		},
 	})
 
@@ -416,7 +437,10 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: udpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: udpServerPort}},
+			}},
 		},
 	})
 
@@ -456,7 +480,10 @@ func TestTCPProxyUpdatePort(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: tcpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: tcpServerPort}},
+			}},
 		},
 	})
 
@@ -493,7 +520,10 @@ func TestUDPProxyUpdatePort(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: udpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: udpServerPort}},
+			}},
 		},
 	})
 
@@ -527,7 +557,10 @@ func TestProxyUpdatePortal(t *testing.T) {
 	lb.OnUpdate([]api.Endpoints{
 		{
 			ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
-			Endpoints:  []api.Endpoint{{IP: "127.0.0.1", Port: tcpServerPort}},
+			Subsets: []api.EndpointSubset{{
+				Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+				Ports:     []api.EndpointPort{{Port: tcpServerPort}},
+			}},
 		},
 	})
 

pkg/registry/endpoint/etcd/etcd_test.go

@@ -50,15 +50,20 @@ func validNewEndpoints() *api.Endpoints {
 			Name:      "foo",
 			Namespace: api.NamespaceDefault,
 		},
-		Protocol:  "TCP",
-		Endpoints: []api.Endpoint{{IP: "baz"}},
+		Subsets: []api.EndpointSubset{{
+			Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+			Ports:     []api.EndpointPort{{Port: 80, Protocol: "TCP"}},
+		}},
 	}
 }
 
 func validChangedEndpoints() *api.Endpoints {
 	endpoints := validNewEndpoints()
 	endpoints.ResourceVersion = "1"
-	endpoints.Endpoints = []api.Endpoint{{IP: "baz"}, {IP: "bar"}}
+	endpoints.Subsets = []api.EndpointSubset{{
+		Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "5.6.7.8"}},
+		Ports:     []api.EndpointPort{{Port: 80, Protocol: "TCP"}},
+	}}
 	return endpoints
 }
 
@@ -113,10 +118,18 @@ func TestEtcdListEndpoints(t *testing.T) {
 			Node: &etcd.Node{
 				Nodes: []*etcd.Node{
 					{
-						Value: runtime.EncodeOrDie(latest.Codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Protocol: "TCP", Endpoints: []api.Endpoint{{IP: "127.0.0.1", Port: 8345}}}),
+						Value: runtime.EncodeOrDie(latest.Codec, &api.Endpoints{
+							ObjectMeta: api.ObjectMeta{Name: "foo"},
+							Subsets: []api.EndpointSubset{{
+								Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
+								Ports:     []api.EndpointPort{{Port: 8345, Protocol: "TCP"}},
+							}},
+						}),
 					},
 					{
-						Value: runtime.EncodeOrDie(latest.Codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "bar"}, Protocol: "TCP"}),
+						Value: runtime.EncodeOrDie(latest.Codec, &api.Endpoints{
+							ObjectMeta: api.ObjectMeta{Name: "bar"},
+						}),
 					},
 				},
 			},

pkg/registry/endpoint/rest.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	endptspkg "github.com/GoogleCloudPlatform/kubernetes/pkg/api/endpoints"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
@@ -45,13 +46,15 @@ func (endpointsStrategy) NamespaceScoped() bool {
 
 // PrepareForCreate clears fields that are not allowed to be set by end users on creation.
 func (endpointsStrategy) PrepareForCreate(obj runtime.Object) {
-	_ = obj.(*api.Endpoints)
+	endpoints := obj.(*api.Endpoints)
+	endpoints.Subsets = endptspkg.RepackSubsets(endpoints.Subsets)
 }
 
 // PrepareForUpdate clears fields that are not allowed to be set by end users on update.
 func (endpointsStrategy) PrepareForUpdate(obj, old runtime.Object) {
-	_ = obj.(*api.Endpoints)
+	newEndpoints := obj.(*api.Endpoints)
 	_ = old.(*api.Endpoints)
+	newEndpoints.Subsets = endptspkg.RepackSubsets(newEndpoints.Subsets)
 }
 
 // Validate validates a new endpoints.

pkg/registry/etcd/etcd_test.go

@@ -539,7 +539,7 @@ func TestEtcdDeleteService(t *testing.T) {
 	key, _ := makeServiceKey(ctx, "foo")
 	fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &api.Service{ObjectMeta: api.ObjectMeta{Name: "foo"}}), 0)
 	endpointsKey, _ := etcdgeneric.NamespaceKeyFunc(ctx, "/registry/services/endpoints", "foo")
-	fakeClient.Set(endpointsKey, runtime.EncodeOrDie(latest.Codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Protocol: "TCP"}), 0)
+	fakeClient.Set(endpointsKey, runtime.EncodeOrDie(latest.Codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}}), 0)
 
 	err := registry.DeleteService(ctx, "foo")
 	if err != nil {

pkg/registry/service/rest.go

@@ -23,6 +23,7 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
@@ -233,19 +234,43 @@ var _ = rest.Redirector(&REST{})
 
 // ResourceLocation returns a URL to which one can send traffic for the specified service.
 func (rs *REST) ResourceLocation(ctx api.Context, id string) (*url.URL, http.RoundTripper, error) {
-	eps, err := rs.endpoints.GetEndpoints(ctx, id)
+	// Allow ID as "svcname" or "svcname:port".
+	parts := strings.Split(id, ":")
+	if len(parts) > 2 {
+		return nil, nil, errors.NewBadRequest(fmt.Sprintf("invalid service request %q", id))
+	}
+	svcName := parts[0]
+	portStr := ""
+	if len(parts) == 2 {
+		portStr = parts[1]
+	}
+
+	eps, err := rs.endpoints.GetEndpoints(ctx, svcName)
 	if err != nil {
 		return nil, nil, err
 	}
-	if len(eps.Endpoints) == 0 {
-		return nil, nil, fmt.Errorf("no endpoints available for %v", id)
+	if len(eps.Subsets) == 0 {
+		return nil, nil, fmt.Errorf("no endpoints available for %q", svcName)
+	}
+	// Pick a random Subset to start searching from.
+	ssSeed := rand.Intn(len(eps.Subsets))
+	// Find a Subset that has the port.
+	for ssi := 0; ssi < len(eps.Subsets); ssi++ {
+		ss := &eps.Subsets[(ssSeed+ssi)%len(eps.Subsets)]
+		for i := range ss.Ports {
+			if ss.Ports[i].Name == portStr {
+				// Pick a random address.
+				ip := ss.Addresses[rand.Intn(len(ss.Addresses))].IP
+				port := ss.Ports[i].Port
+				// We leave off the scheme ('http://') because we have no idea what sort of server
+				// is listening at this endpoint.
+				return &url.URL{
+					Host: net.JoinHostPort(ip, strconv.Itoa(port)),
+				}, nil, nil
+			}
+		}
 	}
-	// We leave off the scheme ('http://') because we have no idea what sort of server
-	// is listening at this endpoint.
-	ep := &eps.Endpoints[rand.Intn(len(eps.Endpoints))]
-	return &url.URL{
-		Host: net.JoinHostPort(ep.IP, strconv.Itoa(ep.Port)),
-	}, nil, nil
+	return nil, nil, fmt.Errorf("no endpoints available for %q", id)
 }
 
 func (rs *REST) getLoadbalancerName(ctx api.Context, service *api.Service) string {

pkg/registry/service/rest_test.go

@@ -373,13 +373,10 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
 					Name:      "foo",
 					Namespace: api.NamespaceDefault,
 				},
-				Endpoints: []api.Endpoint{
-					{
-						IP:   "100.100.100.100",
-						Port: 80,
-					},
-				},
-				Protocol: api.ProtocolTCP,
+				Subsets: []api.EndpointSubset{{
+					Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
+					Ports:     []api.EndpointPort{{Name: "", Port: 80}, {Name: "p", Port: 93}},
+				}},
 			},
 		},
 	}
@@ -391,6 +388,8 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
 		},
 	})
 	redirector := rest.Redirector(storage)
+
+	// Test a simple id.
 	location, _, err := redirector.ResourceLocation(ctx, "foo")
 	if err != nil {
 		t.Errorf("Unexpected error: %v", err)
@@ -398,10 +397,28 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
 	if location == nil {
 		t.Errorf("Unexpected nil: %v", location)
 	}
-	if e, a := "//100.100.100.100:80", location.String(); e != a {
+	if e, a := "//1.2.3.4:80", location.String(); e != a {
+		t.Errorf("Expected %v, but got %v", e, a)
+	}
+
+	// Test a name + port.
+	location, _, err = redirector.ResourceLocation(ctx, "foo:p")
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+	if location == nil {
+		t.Errorf("Unexpected nil: %v", location)
+	}
+	if e, a := "//1.2.3.4:93", location.String(); e != a {
 		t.Errorf("Expected %v, but got %v", e, a)
 	}
 
+	// Test a non-existent name + port.
+	location, _, err = redirector.ResourceLocation(ctx, "foo:q")
+	if err == nil {
+		t.Errorf("Unexpected nil error")
+	}
+
 	// Test error path
 	if _, _, err = redirector.ResourceLocation(ctx, "bar"); err == nil {
 		t.Errorf("unexpected nil error")

pkg/service/endpoints_controller.go

@@ -18,8 +18,10 @@ package service
 
 import (
 	"fmt"
+	"reflect"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/endpoints"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/v1beta1"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/v1beta2"
@@ -50,7 +52,9 @@ func (e *EndpointController) SyncServiceEndpoints() error {
 		return err
 	}
 	var resultErr error
-	for _, service := range services.Items {
+	for i := range services.Items {
+		service := &services.Items[i]
+
 		if service.Spec.Selector == nil {
 			// services without a selector receive no endpoints from this controller;
 			// these services will receive the endpoints that are created out-of-band via the REST API.
@@ -64,14 +68,19 @@ func (e *EndpointController) SyncServiceEndpoints() error {
 			resultErr = err
 			continue
 		}
-		endpoints := []api.Endpoint{}
 
-		for _, pod := range pods.Items {
+		subsets := []api.EndpointSubset{}
+		for i := range pods.Items {
+			pod := &pods.Items[i]
+
 			// TODO: Once v1beta1 and v1beta2 are EOL'ed, this can
 			// assume that service.Spec.TargetPort is populated.
 			_ = v1beta1.Dependency
 			_ = v1beta2.Dependency
-			port, err := findPort(&pod, &service)
+			// TODO: Add multiple-ports to Service and expose them here.
+			portName := ""
+			portProto := service.Spec.Protocol
+			portNum, err := findPort(pod, service)
 			if err != nil {
 				glog.Errorf("Failed to find port for service %s/%s: %v", service.Namespace, service.Name, err)
 				continue
@@ -93,18 +102,19 @@ func (e *EndpointController) SyncServiceEndpoints() error {
 				continue
 			}
 
-			endpoints = append(endpoints, api.Endpoint{
-				IP:   pod.Status.PodIP,
-				Port: port,
-				TargetRef: &api.ObjectReference{
-					Kind:            "Pod",
-					Namespace:       pod.ObjectMeta.Namespace,
-					Name:            pod.ObjectMeta.Name,
-					UID:             pod.ObjectMeta.UID,
-					ResourceVersion: pod.ObjectMeta.ResourceVersion,
-				},
-			})
+			epp := api.EndpointPort{Name: portName, Port: portNum, Protocol: portProto}
+			epa := api.EndpointAddress{IP: pod.Status.PodIP, TargetRef: &api.ObjectReference{
+				Kind:            "Pod",
+				Namespace:       pod.ObjectMeta.Namespace,
+				Name:            pod.ObjectMeta.Name,
+				UID:             pod.ObjectMeta.UID,
+				ResourceVersion: pod.ObjectMeta.ResourceVersion,
+			}}
+			subsets = append(subsets, api.EndpointSubset{Addresses: []api.EndpointAddress{epa}, Ports: []api.EndpointPort{epp}})
 		}
+		subsets = endpoints.RepackSubsets(subsets)
+
+		// See if there's actually an update here.
 		currentEndpoints, err := e.client.Endpoints(service.Namespace).Get(service.Name)
 		if err != nil {
 			if errors.IsNotFound(err) {
@@ -112,26 +122,24 @@ func (e *EndpointController) SyncServiceEndpoints() error {
 					ObjectMeta: api.ObjectMeta{
 						Name: service.Name,
 					},
-					Protocol: service.Spec.Protocol,
 				}
 			} else {
 				glog.Errorf("Error getting endpoints: %v", err)
 				continue
 			}
 		}
-		newEndpoints := &api.Endpoints{}
-		*newEndpoints = *currentEndpoints
-		newEndpoints.Endpoints = endpoints
+		if reflect.DeepEqual(currentEndpoints.Subsets, subsets) {
+			glog.V(5).Infof("endpoints are equal for %s/%s, skipping update", service.Namespace, service.Name)
+			continue
+		}
+		newEndpoints := currentEndpoints
+		newEndpoints.Subsets = subsets
 
 		if len(currentEndpoints.ResourceVersion) == 0 {
 			// No previous endpoints, create them
 			_, err = e.client.Endpoints(service.Namespace).Create(newEndpoints)
 		} else {
 			// Pre-existing
-			if currentEndpoints.Protocol == service.Spec.Protocol && endpointsListEqual(currentEndpoints, endpoints) {
-				glog.V(5).Infof("protocol and endpoints are equal for %s/%s, skipping update", service.Namespace, service.Name)
-				continue
-			}
 			_, err = e.client.Endpoints(service.Namespace).Update(newEndpoints)
 		}
 		if err != nil {
@@ -142,83 +150,43 @@ func (e *EndpointController) SyncServiceEndpoints() error {
 	return resultErr
 }
 
-func endpointEqual(this, that *api.Endpoint) bool {
-	if this.IP != that.IP || this.Port != that.Port {
-		return false
-	}
-
-	if this.TargetRef == nil || that.TargetRef == nil {
-		return this.TargetRef == that.TargetRef
-	}
-
-	return *this.TargetRef == *that.TargetRef
-}
-
-func containsEndpoint(haystack *api.Endpoints, needle *api.Endpoint) bool {
-	if haystack == nil || needle == nil {
-		return false
-	}
-	for ix := range haystack.Endpoints {
-		if endpointEqual(&haystack.Endpoints[ix], needle) {
-			return true
-		}
-	}
-	return false
-}
-
-func endpointsListEqual(eps *api.Endpoints, endpoints []api.Endpoint) bool {
-	if len(eps.Endpoints) != len(endpoints) {
-		return false
-	}
-	for i := range endpoints {
-		if !containsEndpoint(eps, &endpoints[i]) {
-			return false
-		}
-	}
-	return true
-}
-
-func findDefaultPort(pod *api.Pod, servicePort int) (int, bool) {
-	foundPorts := []int{}
+func findDefaultPort(pod *api.Pod, servicePort int, proto api.Protocol) int {
 	for _, container := range pod.Spec.Containers {
 		for _, port := range container.Ports {
-			foundPorts = append(foundPorts, port.ContainerPort)
+			if port.Protocol == proto {
+				return port.ContainerPort
+			}
 		}
 	}
-	if len(foundPorts) == 0 {
-		return servicePort, true
-	}
-	if len(foundPorts) == 1 {
-		return foundPorts[0], true
-	}
-	return 0, false
+	return servicePort
 }
 
 // findPort locates the container port for the given manifest and portName.
+// If the targetPort is a non-zero number, use that.  If the targetPort is 0 or
+// not specified, use the first defined port with the same protocol.  If no port
+// is defined, use the service's port.  If the targetPort is an empty string use
+// the first defined port with the same protocol.  If no port is defined, use
+// the service's port.  If the targetPort is a non-empty 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.Pod, service *api.Service) (int, error) {
 	portName := service.Spec.TargetPort
 	switch portName.Kind {
 	case util.IntstrString:
 		if len(portName.StrVal) == 0 {
-			if port, found := findDefaultPort(pod, service.Spec.Port); found {
-				return port, nil
-			}
-			break
+			return findDefaultPort(pod, service.Spec.Port, service.Spec.Protocol), nil
 		}
 		name := portName.StrVal
 		for _, container := range pod.Spec.Containers {
 			for _, port := range container.Ports {
-				if port.Name == name {
+				if port.Name == name && port.Protocol == service.Spec.Protocol {
 					return port.ContainerPort, nil
 				}
 			}
 		}
 	case util.IntstrInt:
 		if portName.IntVal == 0 {
-			if port, found := findDefaultPort(pod, service.Spec.Port); found {
-				return port, nil
-			}
-			break
+			return findDefaultPort(pod, service.Spec.Port, service.Spec.Protocol), nil
 		}
 		return portName.IntVal, nil
 	}

pkg/tools/etcd_helper_watch_test.go

@@ -278,11 +278,22 @@ func TestWatch(t *testing.T) {
 	}
 }
 
+func emptySubsets() []api.EndpointSubset {
+	return []api.EndpointSubset{}
+}
+
+func makeSubsets(ip string, port int) []api.EndpointSubset {
+	return []api.EndpointSubset{{
+		Addresses: []api.EndpointAddress{{IP: ip}},
+		Ports:     []api.EndpointPort{{Port: port}},
+	}}
+}
+
 func TestWatchEtcdState(t *testing.T) {
 	codec := latest.Codec
 	type T struct {
 		Type      watch.EventType
-		Endpoints []api.Endpoint
+		Endpoints []api.EndpointSubset
 	}
 	testCases := map[string]struct {
 		Initial   map[string]EtcdResponseWithError
@@ -296,7 +307,10 @@ func TestWatchEtcdState(t *testing.T) {
 				{
 					Action: "create",
 					Node: &etcd.Node{
-						Value: string(runtime.EncodeOrDie(codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Endpoints: []api.Endpoint{}})),
+						Value: string(runtime.EncodeOrDie(codec, &api.Endpoints{
+							ObjectMeta: api.ObjectMeta{Name: "foo"},
+							Subsets:    emptySubsets(),
+						})),
 					},
 				},
 			},
@@ -310,12 +324,18 @@ func TestWatchEtcdState(t *testing.T) {
 				{
 					Action: "compareAndSwap",
 					Node: &etcd.Node{
-						Value:         string(runtime.EncodeOrDie(codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Endpoints: []api.Endpoint{{IP: "127.0.0.1", Port: 9000}}})),
+						Value: string(runtime.EncodeOrDie(codec, &api.Endpoints{
+							ObjectMeta: api.ObjectMeta{Name: "foo"},
+							Subsets:    makeSubsets("127.0.0.1", 9000),
+						})),
 						CreatedIndex:  1,
 						ModifiedIndex: 2,
 					},
 					PrevNode: &etcd.Node{
-						Value:         string(runtime.EncodeOrDie(codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Endpoints: []api.Endpoint{}})),
+						Value: string(runtime.EncodeOrDie(codec, &api.Endpoints{
+							ObjectMeta: api.ObjectMeta{Name: "foo"},
+							Subsets:    emptySubsets(),
+						})),
 						CreatedIndex:  1,
 						ModifiedIndex: 1,
 					},
@@ -323,7 +343,7 @@ func TestWatchEtcdState(t *testing.T) {
 			},
 			From: 1,
 			Expected: []*T{
-				{watch.Modified, []api.Endpoint{{IP: "127.0.0.1", Port: 9000}}},
+				{watch.Modified, makeSubsets("127.0.0.1", 9000)},
 			},
 		},
 		"from initial state": {
@@ -332,7 +352,10 @@ func TestWatchEtcdState(t *testing.T) {
 					R: &etcd.Response{
 						Action: "get",
 						Node: &etcd.Node{
-							Value:         string(runtime.EncodeOrDie(codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Endpoints: []api.Endpoint{}})),
+							Value: string(runtime.EncodeOrDie(codec, &api.Endpoints{
+								ObjectMeta: api.ObjectMeta{Name: "foo"},
+								Subsets:    emptySubsets(),
+							})),
 							CreatedIndex:  1,
 							ModifiedIndex: 1,
 						},
@@ -345,12 +368,18 @@ func TestWatchEtcdState(t *testing.T) {
 				{
 					Action: "compareAndSwap",
 					Node: &etcd.Node{
-						Value:         string(runtime.EncodeOrDie(codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Endpoints: []api.Endpoint{{IP: "127.0.0.1", Port: 9000}}})),
+						Value: string(runtime.EncodeOrDie(codec, &api.Endpoints{
+							ObjectMeta: api.ObjectMeta{Name: "foo"},
+							Subsets:    makeSubsets("127.0.0.1", 9000),
+						})),
 						CreatedIndex:  1,
 						ModifiedIndex: 2,
 					},
 					PrevNode: &etcd.Node{
-						Value:         string(runtime.EncodeOrDie(codec, &api.Endpoints{ObjectMeta: api.ObjectMeta{Name: "foo"}, Endpoints: []api.Endpoint{}})),
+						Value: string(runtime.EncodeOrDie(codec, &api.Endpoints{
+							ObjectMeta: api.ObjectMeta{Name: "foo"},
+							Subsets:    emptySubsets(),
+						})),
 						CreatedIndex:  1,
 						ModifiedIndex: 1,
 					},
@@ -358,7 +387,7 @@ func TestWatchEtcdState(t *testing.T) {
 			},
 			Expected: []*T{
 				{watch.Added, nil},
-				{watch.Modified, []api.Endpoint{{IP: "127.0.0.1", Port: 9000}}},
+				{watch.Modified, makeSubsets("127.0.0.1", 9000)},
 			},
 		},
 	}
@@ -382,7 +411,7 @@ func TestWatchEtcdState(t *testing.T) {
 				t.Errorf("%s: expected type %v, got %v", k, e, a)
 				break
 			}
-			if e, a := testCase.Expected[i].Endpoints, event.Object.(*api.Endpoints).Endpoints; !api.Semantic.DeepDerivative(e, a) {
+			if e, a := testCase.Expected[i].Endpoints, event.Object.(*api.Endpoints).Subsets; !api.Semantic.DeepDerivative(e, a) {
 				t.Errorf("%s: expected type %v, got %v", k, e, a)
 				break
 			}

test/e2e/service.go

@@ -394,16 +394,22 @@ func validateUniqueOrFail(s []string) {
 	}
 }
 
-func validateIPsOrFail(c *client.Client, ns string, expectedPort int, expectedEndpoints []string, endpoints *api.Endpoints) {
+func flattenSubsets(subsets []api.EndpointSubset, expectedPort int) util.StringSet {
 	ips := util.StringSet{}
-	for _, ep := range endpoints.Endpoints {
-		if ep.Port != expectedPort {
-			Failf("invalid port, expected %d, got %d", expectedPort, ep.Port)
+	for _, ss := range subsets {
+		for _, port := range ss.Ports {
+			if port.Port == expectedPort {
+				for _, addr := range ss.Addresses {
+					ips.Insert(addr.IP)
+				}
+			}
 		}
-		ips.Insert(ep.IP)
 	}
+	return ips
+}
 
-	for _, name := range expectedEndpoints {
+func validateIPsOrFail(c *client.Client, ns string, expectedPods []string, ips util.StringSet) {
+	for _, name := range expectedPods {
 		pod, err := c.Pods(ns).Get(name)
 		if err != nil {
 			Failf("failed to get pod %s, that's pretty weird. validation failed: %s", name, err)
@@ -413,26 +419,26 @@ func validateIPsOrFail(c *client.Client, ns string, expectedPort int, expectedEn
 		}
 		By(fmt.Sprintf(""))
 	}
-	By(fmt.Sprintf("successfully validated IPs %v against expected endpoints %v port %d on namespace %s", ips, expectedEndpoints, expectedPort, ns))
-
+	By(fmt.Sprintf("successfully validated IPs %v against expected endpoints %v on namespace %s", ips, expectedPods, ns))
 }
 
-func validateEndpointsOrFail(c *client.Client, ns, serviceName string, expectedPort int, expectedEndpoints []string) {
+func validateEndpointsOrFail(c *client.Client, ns, serviceName string, expectedPort int, expectedPods []string) {
 	for {
 		endpoints, err := c.Endpoints(ns).Get(serviceName)
 		if err == nil {
-			if len(endpoints.Endpoints) == len(expectedEndpoints) {
-				validateIPsOrFail(c, ns, expectedPort, expectedEndpoints, endpoints)
-				return
+			ips := flattenSubsets(endpoints.Subsets, expectedPort)
+			if len(ips) == len(expectedPods) {
+				validateIPsOrFail(c, ns, expectedPods, ips)
+				break
 			} else {
-				By(fmt.Sprintf("Unexpected number of endpoints: found %v, expected %v (ignoring for 1 second)", endpoints.Endpoints, expectedEndpoints))
+				By(fmt.Sprintf("Unexpected number of endpoints: found %v, expected %v (ignoring for 1 second)", ips, expectedPods))
 			}
 		} else {
 			By(fmt.Sprintf("Failed to get endpoints: %v (ignoring for 1 second)", err))
 		}
 		time.Sleep(time.Second)
 	}
-	By(fmt.Sprintf("successfully validated endpoints %v port %d on service %s/%s", expectedEndpoints, expectedPort, ns, serviceName))
+	By(fmt.Sprintf("successfully validated endpoints %v port %d on service %s/%s", expectedPods, expectedPort, ns, serviceName))
 }
 
 func addEndpointPodOrFail(c *client.Client, ns, name string, labels map[string]string) {