Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add support for Headless hostPort services #324

Merged
merged 5 commits into from
Nov 20, 2017
Merged

Add support for Headless hostPort services #324

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
10d7b2c
Added initial support for Headless services
Aug 29, 2017
b1fde0a
service.go: Fixed some formatting
Aug 29, 2017
7eb493a
forgot to run gometalinter
Aug 29, 2017
20e9e11
service: implemented alternative proposal, to reuse existing annotation
Aug 31, 2017
325a0bc
source: some refactoring
Aug 31, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Binary file added source/debug.test
View file
Open in desktop
Binary file not shown.
27 changes: 27 additions & 0 deletions source/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
log "github.com/Sirupsen/logrus"

metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/pkg/api/v1"

Expand Down Expand Up @@ -114,6 +115,28 @@ func (sc *serviceSource) Endpoints() ([]*endpoint.Endpoint, error) {
return endpoints, nil
}

func (sc *serviceSource) extractHeadlessEndpoint(svc *v1.Service, hostname string) []*endpoint.Endpoint {

var endpoints []*endpoint.Endpoint

// Get all the Pods
if pods, err := sc.client.CoreV1().Pods(svc.Namespace).List(metav1.ListOptions{LabelSelector: labels.Set(svc.Spec.Selector).AsSelectorPreValidated().String()}); err != nil {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the spirit of the original request around convention and suggestion. I personally have not see this style of expression used much, where the else is referencing the result of the assignment in the guarded if statement and it took me a few seconds to figure out where that was even coming from.

Personally I find it more clear to simply make the assignment and have a small err != nil check that short circuits which eventually you get used to reading all over the place.

That's a completely arbitrary assessment though and may not be true to those who have worked and looked at larger code bases.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good. I will take a look shortly.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mean something like this?

pods, err := sc.client.CoreV1().Pods(svc.Namespace).List(metav1.ListOptions{LabelSelector: labels.Set(svc.Spec.Selector).AsSelectorPreValidated().String()})
	// Get all the Pods
	if err != nil {
		log.Errorf("List Pods of service[%s] error:%v", svc.GetName(), err)
	} else {}

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can have an empty return to short circuit: https://play.golang.org/p/jpRxJ19fl5

Ex.

pods, err := sc.client.CoreV1().Pods(svc.Namespace).List(metav1.ListOptions{LabelSelector: labels.Set(svc.Spec.Selector).AsSelectorPreValidated().String()}); 

if err != nil {
 	log.Errorf("List Pods of service[%s] error:%v", svc.GetName(), err)
 	return 
} 

for _, v := range pods.Items {
	headlessDomain := v.Spec.Hostname + "." + hostname
	log.Debugf("Generating matching endpoint %s with HostIP %s", headlessDomain, v.Status.HostIP)
	// To reduce traffice on the DNS API only add record for running Pods. Good Idea?
	if v.Status.Phase == v1.PodRunning {
		endpoints = append(endpoints, endpoint.NewEndpoint(headlessDomain, v.Status.HostIP, endpoint.RecordTypeA))
	} else {
		log.Debugf("Pod %s is not in running phase", v.Spec.Hostname)
	}
}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But the function should return the Endpoints no? Or should we redesign that part?

log.Errorf("List Pods of service[%s] error:%v", svc.GetName(), err)
} else {
for _, v := range pods.Items {
headlessDomain := v.Spec.Hostname + "." + hostname
log.Debugf("Generating matching endpoint %s with HostIP %s", headlessDomain, v.Status.HostIP)
// To reduce traffice on the DNS API only add record for running Pods. Good Idea?
if v.Status.Phase == v1.PodRunning {
endpoints = append(endpoints, endpoint.NewEndpoint(headlessDomain, v.Status.HostIP, endpoint.RecordTypeA))
} else {
log.Debugf("Pod %s is not in running phase", v.Spec.Hostname)
}
}
}

return endpoints
}
func (sc *serviceSource) endpointsFromTemplate(svc *v1.Service) ([]*endpoint.Endpoint, error) {
var endpoints []*endpoint.Endpoint

Expand Down Expand Up @@ -159,6 +182,10 @@ func (sc *serviceSource) generateEndpoints(svc *v1.Service, hostname string) []*
if sc.publishInternal {
endpoints = append(endpoints, extractServiceIps(svc, hostname)...)
}
if svc.Spec.ClusterIP == v1.ClusterIPNone {
endpoints = append(endpoints, sc.extractHeadlessEndpoint(svc, hostname)...)
}

}

return endpoints
Expand Down
139 changes: 139 additions & 0 deletions source/service_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -674,6 +674,145 @@ func TestClusterIpServices(t *testing.T) {
}
}

// TestHeadlessServices tests that headless services generate the correct endpoints.
func TestHeadlessServices(t *testing.T) {
for _, tc := range []struct {
title string
targetNamespace string
svcNamespace string
svcName string
svcType v1.ServiceType
compatibility string
fqdnTemplate string
labels map[string]string
annotations map[string]string
clusterIP string
hostIP string
selector map[string]string
lbs []string
hostnames []string
phases []v1.PodPhase
expected []*endpoint.Endpoint
expectError bool
}{
{
"annotated Headless services return endpoints for each selected Pod",
"",
"testing",
"foo",
v1.ServiceTypeClusterIP,
"",
"",
map[string]string{"component": "foo"},
map[string]string{
hostnameAnnotationKey: "service.example.org",
},
v1.ClusterIPNone,
"1.1.1.1",
map[string]string{
"component": "foo",
},
[]string{},
[]string{"foo-0", "foo-1"},
[]v1.PodPhase{v1.PodRunning, v1.PodRunning},
[]*endpoint.Endpoint{
{DNSName: "foo-0.service.example.org", Target: "1.1.1.1"},
{DNSName: "foo-1.service.example.org", Target: "1.1.1.1"},
},
false,
},
{
"annotated Headless services return endpoints for each selected Pod, which are in running state",
"",
"testing",
"foo",
v1.ServiceTypeClusterIP,
"",
"",
map[string]string{"component": "foo"},
map[string]string{
hostnameAnnotationKey: "service.example.org",
},
v1.ClusterIPNone,
"1.1.1.1",
map[string]string{
"component": "foo",
},
[]string{},
[]string{"foo-0", "foo-1"},
[]v1.PodPhase{v1.PodRunning, v1.PodFailed},
[]*endpoint.Endpoint{
{DNSName: "foo-0.service.example.org", Target: "1.1.1.1"},
},
false,
},
} {
t.Run(tc.title, func(t *testing.T) {
// Create a Kubernetes testing client
kubernetes := fake.NewSimpleClientset()

service := &v1.Service{
Spec: v1.ServiceSpec{
Type: tc.svcType,
ClusterIP: tc.clusterIP,
Selector: tc.selector,
},
ObjectMeta: metav1.ObjectMeta{
Namespace: tc.svcNamespace,
Name: tc.svcName,
Labels: tc.labels,
Annotations: tc.annotations,
},
Status: v1.ServiceStatus{},
}
_, err := kubernetes.CoreV1().Services(service.Namespace).Create(service)
require.NoError(t, err)

for i, hostname := range tc.hostnames {
pod := &v1.Pod{
Spec: v1.PodSpec{
Containers: []v1.Container{},
Hostname: hostname,
},
ObjectMeta: metav1.ObjectMeta{
Namespace: tc.svcNamespace,
Name: hostname,
Labels: tc.labels,
Annotations: tc.annotations,
},
Status: v1.PodStatus{
HostIP: tc.hostIP,
Phase: tc.phases[i],
},
}

_, err = kubernetes.CoreV1().Pods(tc.svcNamespace).Create(pod)
require.NoError(t, err)
}

// Create our object under test and get the endpoints.
client, _ := NewServiceSource(
kubernetes,
tc.targetNamespace,
tc.fqdnTemplate,
tc.compatibility,
true,
)
require.NoError(t, err)

endpoints, err := client.Endpoints()
if tc.expectError {
require.Error(t, err)
} else {
require.NoError(t, err)
}

// Validate returned endpoints against desired endpoints.
validateEndpoints(t, endpoints, tc.expected)
})
}
}

func BenchmarkServiceEndpoints(b *testing.B) {
kubernetes := fake.NewSimpleClientset()

Expand Down