github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor hostport logic

Browse Source
This commit is contained in:
Minhan Xia 2016-12-29 13:30:34 -08:00
parent 560f52390a
commit 8e318b8d9b
6 changed files with 151 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
pkg/kubelet/network/hostport/BUILD
View File

@ -17,7 +17,6 @@ go_library(
tags = ["automanaged"],
deps = [
"//pkg/api/v1:go_default_library",
"//pkg/kubelet/container:go_default_library",
"//pkg/proxy/iptables:go_default_library",
"//pkg/util/dbus:go_default_library",
"//pkg/util/exec:go_default_library",
@ -33,9 +32,7 @@ go_test(
tags = ["automanaged"],
deps = [
"//pkg/api/v1:go_default_library",
"//pkg/kubelet/container:go_default_library",
"//pkg/util/iptables:go_default_library",
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
],
)

133
pkg/kubelet/network/hostport/hostport.go
View File

@ -26,8 +26,8 @@ import (
"time"
"github.com/golang/glog"
"k8s.io/kubernetes/pkg/api/v1"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
iptablesproxy "k8s.io/kubernetes/pkg/proxy/iptables"
utildbus "k8s.io/kubernetes/pkg/util/dbus"
utilexec "k8s.io/kubernetes/pkg/util/exec"
@ -42,13 +42,29 @@ const (
)
type HostportHandler interface {
OpenPodHostportsAndSync(newPod *ActivePod, natInterfaceName string, activePods []*ActivePod) error
SyncHostports(natInterfaceName string, activePods []*ActivePod) error
// SyncHostports gathers all hostports on node and setup iptables rules enable them. And finally clean up stale hostports
SyncHostports(natInterfaceName string, activePodPortMappings []*PodPortMapping) error
// OpenPodHostportsAndSync opens hostports for a new PodPortMapping, gathers all hostports on
// node, sets up iptables rules enable them. And finally clean up stale hostports.
// 'newPortMapping' must also be present in 'activePodPortMappings'.
OpenPodHostportsAndSync(newPortMapping *PodPortMapping, natInterfaceName string, activePodPortMappings []*PodPortMapping) error
}
type ActivePod struct {
Pod *v1.Pod
IP net.IP
// PortMapping represents a network port in a single container
type PortMapping struct {
Name string
HostPort int32
ContainerPort int32
Protocol v1.Protocol
HostIP string
}
type PodPortMapping struct {
Namespace string
Name string
PortMappings []*PortMapping
HostNetwork bool
IP net.IP
}
type hostportOpener func(*hostport) (closeable, error)
@ -87,35 +103,31 @@ func (hp *hostport) String() string {
}
//openPodHostports opens all hostport for pod and returns the map of hostport and socket
func (h *handler) openHostports(pod *v1.Pod) error {
func (h *handler) openHostports(podHostportMapping *PodPortMapping) error {
var retErr error
ports := make(map[hostport]closeable)
for _, container := range pod.Spec.Containers {
for _, port := range container.Ports {
if port.HostPort <= 0 {
// Ignore
continue
}
hp := hostport{
port: port.HostPort,
protocol: strings.ToLower(string(port.Protocol)),
}
socket, err := h.portOpener(&hp)
if err != nil {
retErr = fmt.Errorf("Cannot open hostport %d for pod %s: %v", port.HostPort, kubecontainer.GetPodFullName(pod), err)
break
}
ports[hp] = socket
for _, port := range podHostportMapping.PortMappings {
if port.HostPort <= 0 {
// Ignore
continue
}
if retErr != nil {
hp := hostport{
port: port.HostPort,
protocol: strings.ToLower(string(port.Protocol)),
}
socket, err := h.portOpener(&hp)
if err != nil {
retErr = fmt.Errorf("Cannot open hostport %d for pod %s: %v", port.HostPort, getPodFullName(podHostportMapping), err)
break
}
ports[hp] = socket
}
// If encounter any error, close all hostports that just got opened.
if retErr != nil {
for hp, socket := range ports {
if err := socket.Close(); err != nil {
glog.Errorf("Cannot clean up hostport %d for pod %s: %v", hp.port, kubecontainer.GetPodFullName(pod), err)
glog.Errorf("Cannot clean up hostport %d for pod %s: %v", hp.port, getPodFullName(podHostportMapping), err)
}
}
return retErr
@ -128,27 +140,28 @@ func (h *handler) openHostports(pod *v1.Pod) error {
return nil
}
func getPodFullName(pod *PodPortMapping) string {
// Use underscore as the delimiter because it is not allowed in pod name
// (DNS subdomain format), while allowed in the container name format.
return pod.Name + "_" + pod.Namespace
}
// gatherAllHostports returns all hostports that should be presented on node,
// given the list of pods running on that node and ignoring host network
// pods (which don't need hostport <-> container port mapping).
func gatherAllHostports(activePods []*ActivePod) (map[v1.ContainerPort]targetPod, error) {
podHostportMap := make(map[v1.ContainerPort]targetPod)
for _, r := range activePods {
if r.IP.To4() == nil {
return nil, fmt.Errorf("Invalid or missing pod %s IP", kubecontainer.GetPodFullName(r.Pod))
func gatherAllHostports(activePodPortMapping []*PodPortMapping) (map[*PortMapping]targetPod, error) {
podHostportMap := make(map[*PortMapping]targetPod)
for _, pm := range activePodPortMapping {
if pm.IP.To4() == nil {
return nil, fmt.Errorf("Invalid or missing pod %s IP", getPodFullName(pm))
}
// should not handle hostports for hostnetwork pods
if r.Pod.Spec.HostNetwork {
if pm.HostNetwork {
continue
}
for _, container := range r.Pod.Spec.Containers {
for _, port := range container.Ports {
if port.HostPort != 0 {
podHostportMap[port] = targetPod{podFullName: kubecontainer.GetPodFullName(r.Pod), podIP: r.IP.String()}
}
}
for _, port := range pm.PortMappings {
podHostportMap[port] = targetPod{podFullName: getPodFullName(pm), podIP: pm.IP.String()}
}
}
return podHostportMap, nil
@ -164,44 +177,44 @@ func writeLine(buf *bytes.Buffer, words ...string) {
// then encoding to base32 and truncating with the prefix "KUBE-SVC-". We do
// this because IPTables Chain Names must be <= 28 chars long, and the longer
// they are the harder they are to read.
func hostportChainName(cp v1.ContainerPort, podFullName string) utiliptables.Chain {
hash := sha256.Sum256([]byte(string(cp.HostPort) + string(cp.Protocol) + podFullName))
func hostportChainName(pm *PortMapping, podFullName string) utiliptables.Chain {
hash := sha256.Sum256([]byte(string(pm.HostPort) + string(pm.Protocol) + podFullName))
encoded := base32.StdEncoding.EncodeToString(hash[:])
return utiliptables.Chain(kubeHostportChainPrefix + encoded[:16])
}
// OpenPodHostportsAndSync opens hostports for a new pod, gathers all hostports on
// OpenPodHostportsAndSync opens hostports for a new PodPortMapping, gathers all hostports on
// node, sets up iptables rules enable them. And finally clean up stale hostports.
// 'newPod' must also be present in 'activePods'.
func (h *handler) OpenPodHostportsAndSync(newPod *ActivePod, natInterfaceName string, activePods []*ActivePod) error {
// 'newPortMapping' must also be present in 'activePodPortMappings'.
func (h *handler) OpenPodHostportsAndSync(newPortMapping *PodPortMapping, natInterfaceName string, activePodPortMappings []*PodPortMapping) error {
// try to open pod host port if specified
if err := h.openHostports(newPod.Pod); err != nil {
if err := h.openHostports(newPortMapping); err != nil {
return err
}
// Add the new pod to active pods if it's not present.
var found bool
for _, p := range activePods {
if p.Pod.UID == newPod.Pod.UID {
for _, pm := range activePodPortMappings {
if pm.Namespace == newPortMapping.Namespace && pm.Name == newPortMapping.Name {
found = true
break
}
}
if !found {
activePods = append(activePods, newPod)
activePodPortMappings = append(activePodPortMappings, newPortMapping)
}
return h.SyncHostports(natInterfaceName, activePods)
return h.SyncHostports(natInterfaceName, activePodPortMappings)
}
// SyncHostports gathers all hostports on node and setup iptables rules enable them. And finally clean up stale hostports
func (h *handler) SyncHostports(natInterfaceName string, activePods []*ActivePod) error {
func (h *handler) SyncHostports(natInterfaceName string, activePodPortMappings []*PodPortMapping) error {
start := time.Now()
defer func() {
glog.V(4).Infof("syncHostportsRules took %v", time.Since(start))
}()
containerPortMap, err := gatherAllHostports(activePods)
hostportPodMap, err := gatherAllHostports(activePodPortMappings)
if err != nil {
return err
}
@ -256,9 +269,9 @@ func (h *handler) SyncHostports(natInterfaceName string, activePods []*ActivePod
// Accumulate NAT chains to keep.
activeNATChains := map[utiliptables.Chain]bool{} // use a map as a set
for containerPort, target := range containerPortMap {
protocol := strings.ToLower(string(containerPort.Protocol))
hostportChain := hostportChainName(containerPort, target.podFullName)
for port, target := range hostportPodMap {
protocol := strings.ToLower(string(port.Protocol))
hostportChain := hostportChainName(port, target.podFullName)
if chain, ok := existingNATChains[hostportChain]; ok {
writeLine(natChains, chain)
} else {
@ -270,9 +283,9 @@ func (h *handler) SyncHostports(natInterfaceName string, activePods []*ActivePod
// Redirect to hostport chain
args := []string{
"-A", string(kubeHostportsChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s hostport %d"`, target.podFullName, containerPort.HostPort),
"-m", "comment", "--comment", fmt.Sprintf(`"%s hostport %d"`, target.podFullName, port.HostPort),
"-m", protocol, "-p", protocol,
"--dport", fmt.Sprintf("%d", containerPort.HostPort),
"--dport", fmt.Sprintf("%d", port.HostPort),
"-j", string(hostportChain),
}
writeLine(natRules, args...)
@ -281,7 +294,7 @@ func (h *handler) SyncHostports(natInterfaceName string, activePods []*ActivePod
// If the request comes from the pod that is serving the hostport, then SNAT
args = []string{
"-A", string(hostportChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s hostport %d"`, target.podFullName, containerPort.HostPort),
"-m", "comment", "--comment", fmt.Sprintf(`"%s hostport %d"`, target.podFullName, port.HostPort),
"-s", target.podIP, "-j", string(iptablesproxy.KubeMarkMasqChain),
}
writeLine(natRules, args...)
@ -290,9 +303,9 @@ func (h *handler) SyncHostports(natInterfaceName string, activePods []*ActivePod
// IPTables will maintained the stats for this chain
args = []string{
"-A", string(hostportChain),
"-m", "comment", "--comment", fmt.Sprintf(`"%s hostport %d"`, target.podFullName, containerPort.HostPort),
"-m", "comment", "--comment", fmt.Sprintf(`"%s hostport %d"`, target.podFullName, port.HostPort),
"-m", protocol, "-p", protocol,
"-j", "DNAT", fmt.Sprintf("--to-destination=%s:%d", target.podIP, containerPort.ContainerPort),
"-j", "DNAT", fmt.Sprintf("--to-destination=%s:%d", target.podIP, port.ContainerPort),
}
writeLine(natRules, args...)
}
@ -321,7 +334,7 @@ func (h *handler) SyncHostports(natInterfaceName string, activePods []*ActivePod
return fmt.Errorf("Failed to execute iptables-restore: %v", err)
}
h.cleanupHostportMap(containerPortMap)
h.cleanupHostportMap(hostportPodMap)
return nil
}
@ -364,7 +377,7 @@ func openLocalPort(hp *hostport) (closeable, error) {
}
// cleanupHostportMap closes obsolete hostports
func (h *handler) cleanupHostportMap(containerPortMap map[v1.ContainerPort]targetPod) {
func (h *handler) cleanupHostportMap(containerPortMap map[*PortMapping]targetPod) {
// compute hostports that are supposed to be open
currentHostports := make(map[hostport]bool)
for containerPort := range containerPortMap {

94
pkg/kubelet/network/hostport/hostport_test.go
View File

@ -23,9 +23,7 @@ import (
"strings"
"testing"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/kubernetes/pkg/api/v1"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
utiliptables "k8s.io/kubernetes/pkg/util/iptables"
)
@ -63,32 +61,29 @@ func TestOpenPodHostports(t *testing.T) {
}
tests := []struct {
pod *v1.Pod
ip string
mapping *PodPortMapping
matches []*ruleMatch
}{
// New pod that we are going to add
{
&v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test-pod",
Namespace: metav1.NamespaceDefault,
},
Spec: v1.PodSpec{
Containers: []v1.Container{{
Ports: []v1.ContainerPort{{
HostPort: 4567,
ContainerPort: 80,
Protocol: v1.ProtocolTCP,
}, {
HostPort: 5678,
ContainerPort: 81,
Protocol: v1.ProtocolUDP,
}},
}},
&PodPortMapping{
Name: "test-pod",
Namespace: v1.NamespaceDefault,
IP: net.ParseIP("10.1.1.2"),
HostNetwork: false,
PortMappings: []*PortMapping{
{
HostPort: 4567,
ContainerPort: 80,
Protocol: v1.ProtocolTCP,
},
{
HostPort: 5678,
ContainerPort: 81,
Protocol: v1.ProtocolUDP,
},
},
},
"10.1.1.2",
[]*ruleMatch{
{
-1,
@ -124,22 +119,19 @@ func TestOpenPodHostports(t *testing.T) {
},
// Already running pod
{
&v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "another-test-pod",
Namespace: metav1.NamespaceDefault,
},
Spec: v1.PodSpec{
Containers: []v1.Container{{
Ports: []v1.ContainerPort{{
HostPort: 123,
ContainerPort: 654,
Protocol: v1.ProtocolTCP,
}},
}},
&PodPortMapping{
Name: "another-test-pod",
Namespace: v1.NamespaceDefault,
IP: net.ParseIP("10.1.1.5"),
HostNetwork: false,
PortMappings: []*PortMapping{
{
HostPort: 123,
ContainerPort: 654,
Protocol: v1.ProtocolTCP,
},
},
},
"10.1.1.5",
[]*ruleMatch{
{
-1,
@ -160,20 +152,18 @@ func TestOpenPodHostports(t *testing.T) {
},
}
activePods := make([]*ActivePod, 0)
activePodPortMapping := make([]*PodPortMapping, 0)
// Fill in any match rules missing chain names
for _, test := range tests {
for _, match := range test.matches {
if match.hostport >= 0 {
found := false
for _, c := range test.pod.Spec.Containers {
for _, cp := range c.Ports {
if int(cp.HostPort) == match.hostport {
match.chain = string(hostportChainName(cp, kubecontainer.GetPodFullName(test.pod)))
found = true
break
}
for _, pm := range test.mapping.PortMappings {
if int(pm.HostPort) == match.hostport {
match.chain = string(hostportChainName(pm, getPodFullName(test.mapping)))
found = true
break
}
}
if !found {
@ -181,24 +171,22 @@ func TestOpenPodHostports(t *testing.T) {
}
}
}
activePods = append(activePods, &ActivePod{
Pod: test.pod,
IP: net.ParseIP(test.ip),
})
activePodPortMapping = append(activePodPortMapping, test.mapping)
}
// Already running pod's host port
hp := hostport{
tests[1].pod.Spec.Containers[0].Ports[0].HostPort,
strings.ToLower(string(tests[1].pod.Spec.Containers[0].Ports[0].Protocol)),
tests[1].mapping.PortMappings[0].HostPort,
strings.ToLower(string(tests[1].mapping.PortMappings[0].Protocol)),
}
h.hostPortMap[hp] = &fakeSocket{
tests[1].pod.Spec.Containers[0].Ports[0].HostPort,
strings.ToLower(string(tests[1].pod.Spec.Containers[0].Ports[0].Protocol)),
tests[1].mapping.PortMappings[0].HostPort,
strings.ToLower(string(tests[1].mapping.PortMappings[0].Protocol)),
false,
}
err := h.OpenPodHostportsAndSync(&ActivePod{Pod: tests[0].pod, IP: net.ParseIP(tests[0].ip)}, "br0", activePods)
err := h.OpenPodHostportsAndSync(tests[0].mapping, "br0", activePodPortMapping)
if err != nil {
t.Fatalf("Failed to OpenPodHostportsAndSync: %v", err)
}

5
pkg/kubelet/network/hostport/testing/BUILD
View File

@ -11,10 +11,7 @@ go_library(
name = "go_default_library",
srcs = ["fake.go"],
tags = ["automanaged"],
deps = [
"//pkg/kubelet/container:go_default_library",
"//pkg/kubelet/network/hostport:go_default_library",
],
deps = ["//pkg/kubelet/network/hostport:go_default_library"],
)
filegroup(

11
pkg/kubelet/network/hostport/testing/fake.go
View File

@ -19,7 +19,6 @@ package testing
import (
"fmt"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/network/hostport"
)
@ -29,14 +28,14 @@ func NewFakeHostportHandler() hostport.HostportHandler {
return &fakeHandler{}
}
func (h *fakeHandler) OpenPodHostportsAndSync(newPod *hostport.ActivePod, natInterfaceName string, activePods []*hostport.ActivePod) error {
return h.SyncHostports(natInterfaceName, activePods)
func (h *fakeHandler) OpenPodHostportsAndSync(newPortMapping *hostport.PodPortMapping, natInterfaceName string, activePortMapping []*hostport.PodPortMapping) error {
return h.SyncHostports(natInterfaceName, activePortMapping)
}
func (h *fakeHandler) SyncHostports(natInterfaceName string, activePods []*hostport.ActivePod) error {
for _, r := range activePods {
func (h *fakeHandler) SyncHostports(natInterfaceName string, activePortMapping []*hostport.PodPortMapping) error {
for _, r := range activePortMapping {
if r.IP.To4() == nil {
return fmt.Errorf("Invalid or missing pod %s IP", kubecontainer.GetPodFullName(r.Pod))
return fmt.Errorf("Invalid or missing pod %s/%s IP", r.Namespace, r.Name)
}
}

45
pkg/kubelet/network/kubenet/kubenet_linux.go
View File

@ -375,13 +375,13 @@ func (plugin *kubenetNetworkPlugin) setup(namespace string, name string, id kube
}
// Open any hostports the pod's containers want
activePods, err := plugin.getActivePods()
activePodPortMapping, err := plugin.getPodPortMapping()
if err != nil {
return err
}
newPod := &hostport.ActivePod{Pod: pod, IP: ip4}
if err := plugin.hostportHandler.OpenPodHostportsAndSync(newPod, BridgeName, activePods); err != nil {
newPodPortMapping := constructPodPortMapping(pod, ip4)
if err := plugin.hostportHandler.OpenPodHostportsAndSync(newPodPortMapping, BridgeName, activePodPortMapping); err != nil {
return err
}
@ -471,9 +471,9 @@ func (plugin *kubenetNetworkPlugin) teardown(namespace string, name string, id k
return utilerrors.NewAggregate(errList)
}
activePods, err := plugin.getActivePods()
activePodPortMapping, err := plugin.getPodPortMapping()
if err == nil {
err = plugin.hostportHandler.SyncHostports(BridgeName, activePods)
err = plugin.hostportHandler.SyncHostports(BridgeName, activePodPortMapping)
}
if err != nil {
errList = append(errList, err)
@ -589,15 +589,12 @@ func (plugin *kubenetNetworkPlugin) getNonExitedPods() ([]*kubecontainer.Pod, er
return ret, nil
}
// Returns a list of pods running or ready to run on this node and each pod's IP address.
// Assumes PodSpecs retrieved from the runtime include the name and ID of containers in
// each pod.
func (plugin *kubenetNetworkPlugin) getActivePods() ([]*hostport.ActivePod, error) {
func (plugin *kubenetNetworkPlugin) getPodPortMapping() ([]*hostport.PodPortMapping, error) {
pods, err := plugin.getNonExitedPods()
if err != nil {
return nil, err
}
activePods := make([]*hostport.ActivePod, 0)
activePodPortMappings := make([]*hostport.PodPortMapping, 0)
for _, p := range pods {
containerID, err := plugin.host.GetRuntime().GetPodContainerID(p)
if err != nil {
@ -612,13 +609,33 @@ func (plugin *kubenetNetworkPlugin) getActivePods() ([]*hostport.ActivePod, erro
continue
}
if pod, ok := plugin.host.GetPodByName(p.Namespace, p.Name); ok {
activePods = append(activePods, &hostport.ActivePod{
Pod: pod,
IP: podIP,
activePodPortMappings = append(activePodPortMappings, constructPodPortMapping(pod, podIP))
}
}
return activePodPortMappings, nil
}
func constructPodPortMapping(pod *v1.Pod, podIP net.IP) *hostport.PodPortMapping {
portMappings := make([]*hostport.PortMapping, 0)
for _, c := range pod.Spec.Containers {
for _, port := range c.Ports {
portMappings = append(portMappings, &hostport.PortMapping{
Name: port.Name,
HostPort: port.HostPort,
ContainerPort: port.ContainerPort,
Protocol: port.Protocol,
HostIP: port.HostIP,
})
}
}
return activePods, nil
return &hostport.PodPortMapping{
Namespace: pod.Namespace,
Name: pod.Name,
PortMappings: portMappings,
HostNetwork: pod.Spec.HostNetwork,
IP: podIP,
}
}
// ipamGarbageCollection will release unused IP.