kubelet/kubenet: split hostport handling into separate module

pkg/kubelet/network/hostport/fake_iptables.go

+/*
+Copyright 2016 The Kubernetes Authors 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 hostport
+
+import (
+	"bytes"
+	"fmt"
+	"net"
+	"strings"
+
+	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
+)
+
+type fakeChain struct {
+	name  utiliptables.Chain
+	rules []string
+}
+
+type fakeTable struct {
+	name   utiliptables.Table
+	chains map[string]*fakeChain
+}
+
+type fakeIptables struct {
+	tables map[string]*fakeTable
+}
+
+func NewFakeIptables() *fakeIptables {
+	return &fakeIptables{
+		tables: make(map[string]*fakeTable, 0),
+	}
+}
+
+func (f *fakeIptables) GetVersion() (string, error) {
+	return "1.4.21", nil
+}
+
+func (f *fakeIptables) getTable(tableName utiliptables.Table) (*fakeTable, error) {
+	table, ok := f.tables[string(tableName)]
+	if !ok {
+		return nil, fmt.Errorf("Table %s does not exist", tableName)
+	}
+	return table, nil
+}
+
+func (f *fakeIptables) getChain(tableName utiliptables.Table, chainName utiliptables.Chain) (*fakeTable, *fakeChain, error) {
+	table, err := f.getTable(tableName)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	chain, ok := table.chains[string(chainName)]
+	if !ok {
+		return table, nil, fmt.Errorf("Chain %s/%s does not exist", tableName, chainName)
+	}
+
+	return table, chain, nil
+}
+
+func (f *fakeIptables) ensureChain(tableName utiliptables.Table, chainName utiliptables.Chain) (bool, *fakeChain) {
+	table, chain, err := f.getChain(tableName, chainName)
+	if err != nil {
+		// either table or table+chain don't exist yet
+		if table == nil {
+			table = &fakeTable{
+				name:   tableName,
+				chains: make(map[string]*fakeChain),
+			}
+			f.tables[string(tableName)] = table
+		}
+		chain := &fakeChain{
+			name:  chainName,
+			rules: make([]string, 0),
+		}
+		table.chains[string(chainName)] = chain
+		return false, chain
+	}
+	return true, chain
+}
+
+func (f *fakeIptables) EnsureChain(tableName utiliptables.Table, chainName utiliptables.Chain) (bool, error) {
+	existed, _ := f.ensureChain(tableName, chainName)
+	return existed, nil
+}
+
+func (f *fakeIptables) FlushChain(tableName utiliptables.Table, chainName utiliptables.Chain) error {
+	_, chain, err := f.getChain(tableName, chainName)
+	if err != nil {
+		return err
+	}
+	chain.rules = make([]string, 0)
+	return nil
+}
+
+func (f *fakeIptables) DeleteChain(tableName utiliptables.Table, chainName utiliptables.Chain) error {
+	table, _, err := f.getChain(tableName, chainName)
+	if err != nil {
+		return err
+	}
+	delete(table.chains, string(chainName))
+	return nil
+}
+
+// Returns index of rule in array; < 0 if rule is not found
+func findRule(chain *fakeChain, rule string) int {
+	for i, candidate := range chain.rules {
+		if rule == candidate {
+			return i
+		}
+	}
+	return -1
+}
+
+func (f *fakeIptables) ensureRule(position utiliptables.RulePosition, tableName utiliptables.Table, chainName utiliptables.Chain, rule string) (bool, error) {
+	_, chain, err := f.getChain(tableName, chainName)
+	if err != nil {
+		_, chain = f.ensureChain(tableName, chainName)
+	}
+
+	rule, err = normalizeRule(rule)
+	if err != nil {
+		return false, err
+	}
+	ruleIdx := findRule(chain, rule)
+	if ruleIdx >= 0 {
+		return true, nil
+	}
+
+	if position == utiliptables.Prepend {
+		chain.rules = append([]string{rule}, chain.rules...)
+	} else if position == utiliptables.Append {
+		chain.rules = append(chain.rules, rule)
+	} else {
+		return false, fmt.Errorf("Unknown position argument %q", position)
+	}
+
+	return false, nil
+}
+
+func normalizeRule(rule string) (string, error) {
+	normalized := ""
+	remaining := strings.TrimSpace(rule)
+	for {
+		var end int
+
+		if strings.HasPrefix(remaining, "--to-destination=") {
+			remaining = strings.Replace(remaining, "=", " ", 1)
+		}
+
+		if remaining[0] == '"' {
+			end = strings.Index(remaining[1:], "\"")
+			if end < 0 {
+				return "", fmt.Errorf("Invalid rule syntax: mismatched quotes")
+			}
+			end += 2
+		} else {
+			end = strings.Index(remaining, " ")
+			if end < 0 {
+				end = len(remaining)
+			}
+		}
+		arg := remaining[:end]
+
+		// Normalize un-prefixed IP addresses like iptables does
+		if net.ParseIP(arg) != nil {
+			arg = arg + "/32"
+		}
+
+		if len(normalized) > 0 {
+			normalized += " "
+		}
+		normalized += strings.TrimSpace(arg)
+		if len(remaining) == end {
+			break
+		}
+		remaining = remaining[end+1:]
+	}
+	return normalized, nil
+}
+
+func (f *fakeIptables) EnsureRule(position utiliptables.RulePosition, tableName utiliptables.Table, chainName utiliptables.Chain, args ...string) (bool, error) {
+	ruleArgs := make([]string, 0)
+	for _, arg := range args {
+		// quote args with internal spaces (like comments)
+		if strings.Index(arg, " ") >= 0 {
+			arg = fmt.Sprintf("\"%s\"", arg)
+		}
+		ruleArgs = append(ruleArgs, arg)
+	}
+	return f.ensureRule(position, tableName, chainName, strings.Join(ruleArgs, " "))
+}
+
+func (f *fakeIptables) DeleteRule(tableName utiliptables.Table, chainName utiliptables.Chain, args ...string) error {
+	_, chain, err := f.getChain(tableName, chainName)
+	if err == nil {
+		rule := strings.Join(args, " ")
+		ruleIdx := findRule(chain, rule)
+		if ruleIdx < 0 {
+			return nil
+		}
+		chain.rules = append(chain.rules[:ruleIdx], chain.rules[ruleIdx+1:]...)
+	}
+	return nil
+}
+
+func (f *fakeIptables) IsIpv6() bool {
+	return false
+}
+
+func saveChain(chain *fakeChain, data *bytes.Buffer) {
+	for _, rule := range chain.rules {
+		data.WriteString(fmt.Sprintf("-A %s %s\n", chain.name, rule))
+	}
+}
+
+func (f *fakeIptables) Save(tableName utiliptables.Table) ([]byte, error) {
+	table, err := f.getTable(tableName)
+	if err != nil {
+		return nil, err
+	}
+
+	data := bytes.NewBuffer(nil)
+	data.WriteString(fmt.Sprintf("*%s\n", table.name))
+
+	rules := bytes.NewBuffer(nil)
+	for _, chain := range table.chains {
+		data.WriteString(fmt.Sprintf(":%s - [0:0]\n", string(chain.name)))
+		saveChain(chain, rules)
+	}
+	data.Write(rules.Bytes())
+	data.WriteString("COMMIT\n")
+	return data.Bytes(), nil
+}
+
+func (f *fakeIptables) SaveAll() ([]byte, error) {
+	data := bytes.NewBuffer(nil)
+	for _, table := range f.tables {
+		tableData, err := f.Save(table.name)
+		if err != nil {
+			return nil, err
+		}
+		if _, err = data.Write(tableData); err != nil {
+			return nil, err
+		}
+	}
+	return data.Bytes(), nil
+}
+
+func (f *fakeIptables) restore(restoreTableName utiliptables.Table, data []byte, flush utiliptables.FlushFlag) error {
+	buf := bytes.NewBuffer(data)
+	var tableName utiliptables.Table
+	for {
+		line, err := buf.ReadString('\n')
+		if err != nil {
+			break
+		}
+		if line[0] == '#' {
+			continue
+		}
+
+		line = strings.TrimSuffix(line, "\n")
+		if strings.HasPrefix(line, "*") {
+			tableName = utiliptables.Table(line[1:])
+		}
+		if tableName != "" {
+			if restoreTableName != "" && restoreTableName != tableName {
+				continue
+			}
+			if strings.HasPrefix(line, ":") {
+				chainName := utiliptables.Chain(strings.Split(line[1:], " ")[0])
+				if flush == utiliptables.FlushTables {
+					table, chain, _ := f.getChain(tableName, chainName)
+					if chain != nil {
+						delete(table.chains, string(chainName))
+					}
+				}
+				_, _ = f.ensureChain(tableName, chainName)
+			} else if strings.HasPrefix(line, "-A") {
+				parts := strings.Split(line, " ")
+				if len(parts) < 3 {
+					return fmt.Errorf("Invalid iptables rule '%s'", line)
+				}
+				chainName := utiliptables.Chain(parts[1])
+				rule := strings.TrimPrefix(line, fmt.Sprintf("-A %s ", chainName))
+				_, err := f.ensureRule(utiliptables.Append, tableName, chainName, rule)
+				if err != nil {
+					return err
+				}
+			} else if strings.HasPrefix(line, "-X") {
+				parts := strings.Split(line, " ")
+				if len(parts) < 3 {
+					return fmt.Errorf("Invalid iptables rule '%s'", line)
+				}
+				if err := f.DeleteChain(tableName, utiliptables.Chain(parts[1])); err != nil {
+					return err
+				}
+			} else if line == "COMMIT" {
+				if restoreTableName == tableName {
+					return nil
+				}
+				tableName = ""
+			}
+		}
+	}
+
+	return nil
+}
+
+func (f *fakeIptables) Restore(tableName utiliptables.Table, data []byte, flush utiliptables.FlushFlag, counters utiliptables.RestoreCountersFlag) error {
+	return f.restore(tableName, data, flush)
+}
+
+func (f *fakeIptables) RestoreAll(data []byte, flush utiliptables.FlushFlag, counters utiliptables.RestoreCountersFlag) error {
+	return f.restore("", data, flush)
+}
+
+func (f *fakeIptables) AddReloadFunc(reloadFunc func()) {
+}
+
+func (f *fakeIptables) Destroy() {
+}

pkg/kubelet/network/hostport/hostport_test.go

+/*
+Copyright 2016 The Kubernetes Authors 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 hostport
+
+import (
+	"fmt"
+	"net"
+	"strings"
+	"testing"
+
+	"k8s.io/kubernetes/pkg/api"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
+)
+
+type fakeSocket struct {
+	port     int32
+	protocol string
+	closed   bool
+}
+
+func (f *fakeSocket) Close() error {
+	if f.closed {
+		return fmt.Errorf("Socket %q.%s already closed!", f.port, f.protocol)
+	}
+	f.closed = true
+	return nil
+}
+
+func openFakeSocket(hp *hostport) (closeable, error) {
+	return &fakeSocket{hp.port, hp.protocol, false}, nil
+}
+
+type ruleMatch struct {
+	hostport int
+	chain    string
+	match    string
+}
+
+func TestOpenPodHostports(t *testing.T) {
+	fakeIptables := NewFakeIptables()
+
+	h := &handler{
+		hostPortMap: make(map[hostport]closeable),
+		iptables:    fakeIptables,
+		portOpener:  openFakeSocket,
+	}
+
+	tests := []struct {
+		pod     *api.Pod
+		ip      string
+		matches []*ruleMatch
+	}{
+		// New pod that we are going to add
+		{
+			&api.Pod{
+				ObjectMeta: api.ObjectMeta{
+					Name:      "test-pod",
+					Namespace: api.NamespaceDefault,
+				},
+				Spec: api.PodSpec{
+					Containers: []api.Container{{
+						Ports: []api.ContainerPort{{
+							HostPort:      4567,
+							ContainerPort: 80,
+							Protocol:      api.ProtocolTCP,
+						}, {
+							HostPort:      5678,
+							ContainerPort: 81,
+							Protocol:      api.ProtocolUDP,
+						}},
+					}},
+				},
+			},
+			"10.1.1.2",
+			[]*ruleMatch{
+				{
+					-1,
+					"KUBE-HOSTPORTS",
+					"-m comment --comment \"test-pod_default hostport 4567\" -m tcp -p tcp --dport 4567",
+				},
+				{
+					4567,
+					"",
+					"-m comment --comment \"test-pod_default hostport 4567\" -s 10.1.1.2/32 -j KUBE-MARK-MASQ",
+				},
+				{
+					4567,
+					"",
+					"-m comment --comment \"test-pod_default hostport 4567\" -m tcp -p tcp -j DNAT --to-destination 10.1.1.2:80",
+				},
+				{
+					-1,
+					"KUBE-HOSTPORTS",
+					"-m comment --comment \"test-pod_default hostport 5678\" -m udp -p udp --dport 5678",
+				},
+				{
+					5678,
+					"",
+					"-m comment --comment \"test-pod_default hostport 5678\" -s 10.1.1.2/32 -j KUBE-MARK-MASQ",
+				},
+				{
+					5678,
+					"",
+					"-m comment --comment \"test-pod_default hostport 5678\" -m udp -p udp -j DNAT --to-destination 10.1.1.2:81",
+				},
+			},
+		},
+		// Already running pod
+		{
+			&api.Pod{
+				ObjectMeta: api.ObjectMeta{
+					Name:      "another-test-pod",
+					Namespace: api.NamespaceDefault,
+				},
+				Spec: api.PodSpec{
+					Containers: []api.Container{{
+						Ports: []api.ContainerPort{{
+							HostPort:      123,
+							ContainerPort: 654,
+							Protocol:      api.ProtocolTCP,
+						}},
+					}},
+				},
+			},
+			"10.1.1.5",
+			[]*ruleMatch{
+				{
+					-1,
+					"KUBE-HOSTPORTS",
+					"-m comment --comment \"another-test-pod_default hostport 123\" -m tcp -p tcp --dport 123",
+				},
+				{
+					123,
+					"",
+					"-m comment --comment \"another-test-pod_default hostport 123\" -s 10.1.1.5/32 -j KUBE-MARK-MASQ",
+				},
+				{
+					123,
+					"",
+					"-m comment --comment \"another-test-pod_default hostport 123\" -m tcp -p tcp -j DNAT --to-destination 10.1.1.5:654",
+				},
+			},
+		},
+	}
+
+	runningPods := make([]*RunningPod, 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
+						}
+					}
+				}
+				if !found {
+					t.Fatalf("Failed to find ContainerPort for match %d/'%s'", match.hostport, match.match)
+				}
+			}
+		}
+		runningPods = append(runningPods, &RunningPod{
+			Pod: test.pod,
+			IP:  net.ParseIP(test.ip),
+		})
+	}
+
+	err := h.OpenPodHostportsAndSync(tests[0].pod, "br0", runningPods)
+	if err != nil {
+		t.Fatalf("Failed to OpenPodHostportsAndSync: %v", err)
+	}
+
+	// Generic rules
+	genericRules := []*ruleMatch{
+		{-1, "POSTROUTING", "-m comment --comment \"SNAT for localhost access to hostports\" -o br0 -s 127.0.0.0/8 -j MASQUERADE"},
+		{-1, "PREROUTING", "-m comment --comment \"kube hostport portals\" -m addrtype --dst-type LOCAL -j KUBE-HOSTPORTS"},
+		{-1, "OUTPUT", "-m comment --comment \"kube hostport portals\" -m addrtype --dst-type LOCAL -j KUBE-HOSTPORTS"},
+	}
+
+	for _, rule := range genericRules {
+		_, chain, err := fakeIptables.getChain(utiliptables.TableNAT, utiliptables.Chain(rule.chain))
+		if err != nil {
+			t.Fatalf("Expected NAT chain %s did not exist", rule.chain)
+		}
+		if !matchRule(chain, rule.match) {
+			t.Fatalf("Expected %s chain rule match '%s' not found", rule.chain, rule.match)
+		}
+	}
+
+	// Pod rules
+	for _, test := range tests {
+		for _, match := range test.matches {
+			// Ensure chain exists
+			_, chain, err := fakeIptables.getChain(utiliptables.TableNAT, utiliptables.Chain(match.chain))
+			if err != nil {
+				t.Fatalf("Expected NAT chain %s did not exist", match.chain)
+			}
+			if !matchRule(chain, match.match) {
+				t.Fatalf("Expected NAT chain %s rule containing '%s' not found", match.chain, match.match)
+			}
+		}
+	}
+}
+
+func matchRule(chain *fakeChain, match string) bool {
+	for _, rule := range chain.rules {
+		if strings.Contains(rule, match) {
+			return true
+		}
+	}
+	return false
+}

pkg/kubelet/network/hostport/testing/fake.go

+/*
+Copyright 2016 The Kubernetes Authors 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 testing
+
+import (
+	"fmt"
+
+	"k8s.io/kubernetes/pkg/api"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	"k8s.io/kubernetes/pkg/kubelet/network/hostport"
+)
+
+type fakeHandler struct{}
+
+func NewFakeHostportHandler() hostport.HostportHandler {
+	return &fakeHandler{}
+}
+
+func (h *fakeHandler) OpenPodHostportsAndSync(newPod *api.Pod, natInterfaceName string, runningPods []*hostport.RunningPod) error {
+	return h.SyncHostports(natInterfaceName, runningPods)
+}
+
+func (h *fakeHandler) SyncHostports(natInterfaceName string, runningPods []*hostport.RunningPod) error {
+	for _, r := range runningPods {
+		if r.IP.To4() == nil {
+			return fmt.Errorf("Invalid or missing pod %s IP", kubecontainer.GetPodFullName(r.Pod))
+		}
+	}
+
+	return nil
+}

pkg/kubelet/network/kubenet/kubenet_linux_test.go

@@ -27,6 +27,7 @@ import (
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/network"
 	"k8s.io/kubernetes/pkg/kubelet/network/cni/testing"
+	hostporttest "k8s.io/kubernetes/pkg/kubelet/network/hostport/testing"
 	nettest "k8s.io/kubernetes/pkg/kubelet/network/testing"
 	"k8s.io/kubernetes/pkg/util/bandwidth"
 	"k8s.io/kubernetes/pkg/util/exec"
@@ -137,6 +138,7 @@ func TestTeardownCallsShaper(t *testing.T) {
 	kubenet.cniConfig = mockcni
 	kubenet.iptables = ipttest.NewFake()
 	kubenet.bandwidthShaper = fshaper
+	kubenet.hostportHandler = hostporttest.NewFakeHostportHandler()
 
 	mockcni.On("DelNetwork", mock.AnythingOfType("*libcni.NetworkConfig"), mock.AnythingOfType("*libcni.RuntimeConf")).Return(nil)