Merge pull request #5939 from thockin/plural_endpoints_31_endpt_subsets

Implement multi-port 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,35 +408,104 @@ 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"}},
+					},
+				}},
 		},
 	}
 
@@ -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.