Skip to content

Commit

Permalink
fix master service endpoint system for multiple masters
Browse files Browse the repository at this point in the history
  • Loading branch information
@lavalamp
lavalamp committed May 4, 2015
1 parent bb82fbb commit 966a4fc
Show file tree
Hide file tree
Showing 2 changed files with 206 additions and 47 deletions.
83 changes: 69 additions & 14 deletions pkg/master/publish.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@ package master

import (
"net"
"reflect"
"time"

"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/rest"

Expand All @@ -38,7 +38,7 @@ func (m *Master) serviceWriterLoop(stop chan struct{}) {
glog.Errorf("Can't create master namespace: %v", err)
}
if m.serviceReadWriteIP != nil {
if err := m.createMasterServiceIfNeeded("kubernetes", m.serviceReadWriteIP, m.serviceReadWritePort); err != nil {
if err := m.createMasterServiceIfNeeded("kubernetes", m.serviceReadWriteIP, m.serviceReadWritePort); err != nil && !errors.IsAlreadyExists(err) {
glog.Errorf("Can't create rw service: %v", err)
}
if err := m.setEndpoints("kubernetes", m.clusterIP, m.publicReadWritePort); err != nil {
Expand All @@ -63,7 +63,7 @@ func (m *Master) roServiceWriterLoop(stop chan struct{}) {
glog.Errorf("Can't create master namespace: %v", err)
}
if m.serviceReadOnlyIP != nil {
if err := m.createMasterServiceIfNeeded("kubernetes-ro", m.serviceReadOnlyIP, m.serviceReadOnlyPort); err != nil {
if err := m.createMasterServiceIfNeeded("kubernetes-ro", m.serviceReadOnlyIP, m.serviceReadOnlyPort); err != nil && !errors.IsAlreadyExists(err) {
glog.Errorf("Can't create ro service: %v", err)
}
if err := m.setEndpoints("kubernetes-ro", m.clusterIP, m.publicReadOnlyPort); err != nil {
Expand Down Expand Up @@ -128,15 +128,20 @@ func (m *Master) createMasterServiceIfNeeded(serviceName string, serviceIP net.I
return err
}

// setEndpoints sets the endpoints for the given service.
// TODO: in a multi-master scenario this needs to consider all masters.
// setEndpoints sets the endpoints for the given apiserver service (ro or rw).
// setEndpoints expects that the endpoints objects it manages will all be
// managed only by setEndpoints; therefore, to understand this, you need only
// understand the requirements and the body of this function.
//
// Requirements:
// * All apiservers MUST use the same ports for their {rw, ro} services.
// * All apiservers MUST use setEndpoints and only setEndpoints to manage the
// endpoints for their {rw, ro} services.
// * All apiservers MUST know and agree on the number of apiservers expected
// to be running (m.masterCount).
// * setEndpoints is called periodically from all apiservers.
//
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 {
Expand All @@ -147,11 +152,61 @@ func (m *Master) setEndpoints(serviceName string, ip net.IP, port int) error {
},
}
}
if !reflect.DeepEqual(e.Subsets, want) {
e.Subsets = want
glog.Infof("setting endpoints for master service %q to %v", serviceName, e)

// First, determine if the endpoint is in the format we expect (one
// subset, one port, N IP addresses).
formatCorrect, ipCorrect := m.checkEndpointSubsetFormat(e, ip.String(), port)
if !formatCorrect {
// Something is egregiously wrong, just re-make the endpoints record.
e.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: ip.String()}},
Ports: []api.EndpointPort{{Port: port, Protocol: api.ProtocolTCP}},
}}
glog.Infof("Resetting endpoints for master service %q to %v", serviceName, e)
return m.endpointRegistry.UpdateEndpoints(ctx, e)
} else if !ipCorrect {
// So we *always* add our own IP address; if there are too many
// IP addresses, we remove the one lexicographically after our
// own IP address. Given the requirements stated at the top of
// this function, this should cause the list of IP addresses to
// become eventually correct.
e.Subsets[0].Addresses = append(e.Subsets[0].Addresses, api.EndpointAddress{IP: ip.String()})
e.Subsets = endpoints.RepackSubsets(e.Subsets)
addrs := &e.Subsets[0].Addresses
for i, addr := range *addrs {
if addr.IP == ip.String() {
for len(*addrs) > m.masterCount {
remove := (i + 1) % len(*addrs)
*addrs = append((*addrs)[:remove], (*addrs)[remove+1:]...)
}
break
}
}
return m.endpointRegistry.UpdateEndpoints(ctx, e)
}
// We didn't make any changes, no need to actually call update.
return nil
}

// Determine if the endpoint is in the format setEndpoints expect (one subset,
// one port, N IP addresses); and if the specified IP address is present and
// the correct number of ip addresses are found.
func (m *Master) checkEndpointSubsetFormat(e *api.Endpoints, ip string, port int) (formatCorrect, ipCorrect bool) {
if len(e.Subsets) != 1 {
return false, false
}
sub := &e.Subsets[0]
if len(sub.Ports) != 1 {
return false, false
}
p := &sub.Ports[0]
if p.Port != port || p.Protocol != api.ProtocolTCP {
return false, false
}
for _, addr := range sub.Addresses {
if addr.IP == ip {
return true, len(sub.Addresses) == m.masterCount
}
}
return true, false
}
170 changes: 137 additions & 33 deletions pkg/master/publish_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,21 +26,32 @@ import (
)

func TestSetEndpoints(t *testing.T) {
ns := api.NamespaceDefault
om := func(name string) api.ObjectMeta {
return api.ObjectMeta{Namespace: ns, Name: name}
}
tests := []struct {
testName string
serviceName string
ip string
port int
endpoints *api.EndpointsList
expectUpdate bool
testName string
serviceName string
ip string
port int
additionalMasters int
endpoints *api.EndpointsList
expectUpdate *api.Endpoints // nil means none expected
}{
{
testName: "no existing endpoints",
serviceName: "foo",
ip: "1.2.3.4",
port: 8080,
endpoints: nil,
expectUpdate: true,
testName: "no existing endpoints",
serviceName: "foo",
ip: "1.2.3.4",
port: 8080,
endpoints: nil,
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
{
testName: "existing endpoints satisfy",
Expand All @@ -49,14 +60,13 @@ func TestSetEndpoints(t *testing.T) {
port: 8080,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: api.ObjectMeta{Name: "foo"},
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
}},
},
expectUpdate: false,
},
{
testName: "existing endpoints satisfy but too many",
Expand All @@ -65,14 +75,88 @@ func TestSetEndpoints(t *testing.T) {
port: 8080,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: api.ObjectMeta{Name: "foo"},
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}, {IP: "4.3.2.1"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
}},
},
expectUpdate: true,
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
{
testName: "existing endpoints satisfy but too many + extra masters",
serviceName: "foo",
ip: "1.2.3.4",
port: 8080,
additionalMasters: 3,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{
{IP: "1.2.3.4"},
{IP: "4.3.2.1"},
{IP: "4.3.2.2"},
{IP: "4.3.2.3"},
{IP: "4.3.2.4"},
},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
}},
},
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{
{IP: "1.2.3.4"},
{IP: "4.3.2.2"},
{IP: "4.3.2.3"},
{IP: "4.3.2.4"},
},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
{
testName: "existing endpoints satisfy but too many + extra masters + delete first",
serviceName: "foo",
ip: "4.3.2.4",
port: 8080,
additionalMasters: 3,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{
{IP: "1.2.3.4"},
{IP: "4.3.2.1"},
{IP: "4.3.2.2"},
{IP: "4.3.2.3"},
{IP: "4.3.2.4"},
},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
}},
},
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{
{IP: "4.3.2.1"},
{IP: "4.3.2.2"},
{IP: "4.3.2.3"},
{IP: "4.3.2.4"},
},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
{
testName: "existing endpoints wrong name",
Expand All @@ -81,14 +165,20 @@ func TestSetEndpoints(t *testing.T) {
port: 8080,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: api.ObjectMeta{Name: "bar"},
ObjectMeta: om("bar"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
}},
},
expectUpdate: true,
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
{
testName: "existing endpoints wrong IP",
Expand All @@ -97,14 +187,20 @@ func TestSetEndpoints(t *testing.T) {
port: 8080,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: api.ObjectMeta{Name: "foo"},
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "4.3.2.1"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
}},
},
expectUpdate: true,
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
{
testName: "existing endpoints wrong port",
Expand All @@ -113,14 +209,20 @@ func TestSetEndpoints(t *testing.T) {
port: 8080,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: api.ObjectMeta{Name: "foo"},
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 9090, Protocol: "TCP"}},
}},
}},
},
expectUpdate: true,
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
{
testName: "existing endpoints wrong protocol",
Expand All @@ -129,18 +231,24 @@ func TestSetEndpoints(t *testing.T) {
port: 8080,
endpoints: &api.EndpointsList{
Items: []api.Endpoints{{
ObjectMeta: api.ObjectMeta{Name: "foo"},
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "UDP"}},
}},
}},
},
expectUpdate: true,
expectUpdate: &api.Endpoints{
ObjectMeta: om("foo"),
Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}},
},
},
}
for _, test := range tests {
master := Master{}
master := Master{masterCount: test.additionalMasters + 1}
registry := &registrytest.EndpointRegistry{
Endpoints: test.endpoints,
}
Expand All @@ -149,18 +257,14 @@ func TestSetEndpoints(t *testing.T) {
if err != nil {
t.Errorf("case %q: unexpected error: %v", test.testName, err)
}
if test.expectUpdate {
expectedSubsets := []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "1.2.3.4"}},
Ports: []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
}}
if test.expectUpdate != nil {
if len(registry.Updates) != 1 {
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)
} else if e, a := test.expectUpdate, &registry.Updates[0]; !reflect.DeepEqual(e, a) {
t.Errorf("case %q: expected update:\n%#v\ngot:\n%#v\n", test.testName, e, a)
}
}
if !test.expectUpdate && len(registry.Updates) > 0 {
if test.expectUpdate == nil && len(registry.Updates) > 0 {
t.Errorf("case %q: no update expected, yet saw: %v", test.testName, registry.Updates)
}
}
Expand Down

0 comments on commit 966a4fc

Please sign in to comment.