mirrors/kube-router
1
0
Fork 0
mirror of https://github.com/cloudnativelabs/kube-router.git synced 2025-10-07 16:01:08 +02:00
Code Issues Packages Projects Releases Wiki Activity

Prevent masquerading pod -> NodeIP traffic (#174)

Browse Source 
* Move getNodeIP logic to utils package

Remove redundant ipset lookups

utils.NewIPSet() does this for us.

* Don't masquerade pod -> nodeAddrsIPSet traffic

Previously with Pod egress enabled, this would get masqueraded.
This change also adds cleanup for said ipset.

* Enhanced cleanup of Pod egress, overlay networking

- Delete old/bad pod egress iptables rule(s) from old versions
- When pod egress or overlay are disabled, cleanup as needed

* Update IPSet.Sets to map type

* ipset enhancements

- Avoid providing method that would delete all ipset sets on a system
- New method DestroyAllWithin() destroys sets tracked by an IPSet
- Create() now handles cases where Sets/System state are not in sync
- Refresh() now handles leftover -temp set gracefully
- Swap() now uses ipset swap
- Delete() improved sync of Sets and system state
- Get() now validates if map element exists before trying
- etc

* Update routes controller to reflect ipset changes
This commit is contained in:
Bryan Zubrod 2017-10-07 04:14:13 -05:00 committed by GitHub
parent 3debbfa4e2
commit 342ea5ac9f
5 changed files with 390 additions and 202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
app/controllers/network_policy_controller.go
View File

@ -7,7 +7,6 @@ import (
"errors"
"fmt"
"net"
"os/exec"
"strconv"
"strings"
"sync"
@ -20,7 +19,6 @@ import (
"github.com/golang/glog"
"k8s.io/client-go/kubernetes"
api "k8s.io/client-go/pkg/api/v1"
apiv1 "k8s.io/client-go/pkg/api/v1"
apiextensions "k8s.io/client-go/pkg/apis/extensions/v1beta1"
networking "k8s.io/client-go/pkg/apis/networking/v1"
)
@ -45,7 +43,7 @@ type NetworkPolicyController struct {
// list of all active network policies expressed as networkPolicyInfo
networkPoliciesInfo *[]networkPolicyInfo
ipset *utils.IPSet
ipSetHandler *utils.IPSet
}
// internal structure to represent a network policy
@ -168,11 +166,6 @@ func (npc *NetworkPolicyController) Sync() error {
npc.mu.Lock()
defer npc.mu.Unlock()
_, err = exec.LookPath("ipset")
if err != nil {
return errors.New("Ensure ipset package is installed: " + err.Error())
}
start := time.Now()
defer func() {
glog.Infof("sync iptables took %v", time.Since(start))
@ -239,7 +232,7 @@ func (npc *NetworkPolicyController) syncNetworkPolicyChains() (map[string]bool,
// create a ipset for all destination pod ip's matched by the policy spec PodSelector
destPodIpSetName := policyDestinationPodIpSetName(policy.namespace, policy.name)
destPodIpSet, err := npc.ipset.Create(destPodIpSetName, utils.TypeHashIP, utils.OptionTimeout, "0")
destPodIpSet, err := npc.ipSetHandler.Create(destPodIpSetName, utils.TypeHashIP, utils.OptionTimeout, "0")
if err != nil {
return nil, nil, fmt.Errorf("failed to create ipset: %s", err.Error())
}
@ -274,7 +267,7 @@ func (npc *NetworkPolicyController) syncNetworkPolicyChains() (map[string]bool,
if len(ingressRule.srcPods) != 0 {
srcPodIpSetName := policySourcePodIpSetName(policy.namespace, policy.name, i)
srcPodIpSet, err := npc.ipset.Create(srcPodIpSetName, utils.TypeHashIP, utils.OptionTimeout, "0")
srcPodIpSet, err := npc.ipSetHandler.Create(srcPodIpSetName, utils.TypeHashIP, utils.OptionTimeout, "0")
if err != nil {
return nil, nil, fmt.Errorf("failed to create ipset: %s", err.Error())
}
@ -869,21 +862,6 @@ func policySourcePodIpSetName(namespace, policyName string, ingressRuleNo int) s
return "KUBE-SRC-" + encoded[:16]
}
func getNodeIP(node *apiv1.Node) (net.IP, error) {
addresses := node.Status.Addresses
addressMap := make(map[apiv1.NodeAddressType][]apiv1.NodeAddress)
for i := range addresses {
addressMap[addresses[i].Type] = append(addressMap[addresses[i].Type], addresses[i])
}
if addresses, ok := addressMap[apiv1.NodeInternalIP]; ok {
return net.ParseIP(addresses[0].Address), nil
}
if addresses, ok := addressMap[apiv1.NodeExternalIP]; ok {
return net.ParseIP(addresses[0].Address), nil
}
return nil, errors.New("host IP unknown")
}
// Cleanup cleanup configurations done
func (npc *NetworkPolicyController) Cleanup() {
@ -961,7 +939,7 @@ func (npc *NetworkPolicyController) Cleanup() {
}
// delete all ipsets
err = npc.ipset.Destroy()
err = npc.ipSetHandler.DestroyAllWithin()
if err != nil {
glog.Errorf("Failed to clean up ipsets: " + err.Error())
}
@ -989,7 +967,7 @@ func NewNetworkPolicyController(clientset *kubernetes.Clientset, config *options
npc.nodeHostName = node.Name
nodeIP, err := getNodeIP(node)
nodeIP, err := utils.GetNodeIP(node)
if err != nil {
return nil, err
}
@ -1003,7 +981,7 @@ func NewNetworkPolicyController(clientset *kubernetes.Clientset, config *options
if err != nil {
return nil, err
}
npc.ipset = ipset
npc.ipSetHandler = ipset
watchers.PodWatcher.RegisterHandler(&npc)
watchers.NetworkPolicyWatcher.RegisterHandler(&npc)

299
app/controllers/network_routes_controller.go
View File

@ -52,20 +52,26 @@ type NetworkRoutingController struct {
globalPeerRouters []*config.NeighborConfig
nodePeerRouters []string
bgpFullMeshMode bool
podSubnetsIpSet *utils.Set
ipSetHandler *utils.IPSet
enableOverlays bool
}
var (
activeNodes = make(map[string]bool)
podEgressArgs = []string{"-m", "set", "--match-set", podSubnetIpSetName, "src",
"-m", "set", "!", "--match-set", podSubnetIpSetName, "dst",
podEgressArgs = []string{"-m", "set", "--match-set", podSubnetsIPSetName, "src",
"-m", "set", "!", "--match-set", podSubnetsIPSetName, "dst",
"-m", "set", "!", "--match-set", nodeAddrsIPSetName, "dst",
"-j", "MASQUERADE"}
podEgressArgsBad = [][]string{{"-m", "set", "--match-set", podSubnetsIPSetName, "src",
"-m", "set", "!", "--match-set", podSubnetsIPSetName, "dst",
"-j", "MASQUERADE"}}
)
const (
clustetNeighborsSet = "clusterneighborsset"
podSubnetIpSetName = "kube-router-pod-subnets"
customRouteTableID = 77
customRouteTableName = "kube-router"
podSubnetsIPSetName = "kube-router-pod-subnets"
nodeAddrsIPSetName = "kube-router-node-ips"
)
// Run runs forever until we are notified on stop channel
@ -89,6 +95,12 @@ func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGr
}
}
glog.Info("Populating ipsets.")
err = nrc.syncNodeIPSets()
if err != nil {
glog.Errorf("Failed initial ipset setup: %s", err)
}
// In case of cluster provisioned on AWS disable source-destination check
nrc.disableSourceDestinationCheck()
@ -98,11 +110,45 @@ func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGr
glog.Errorf("Failed to enable IP forwarding of traffic from pods: %s", err.Error())
}
// enable policy based routing rules
// Handle ipip tunnel overlay
if nrc.enableOverlays {
glog.Info("IPIP Tunnel Overlay enabled in configuration.")
glog.Info("Setting up overlay networking.")
err = nrc.enablePolicyBasedRouting()
if err != nil {
glog.Errorf("Failed to enable required policy based routing: %s", err.Error())
}
} else {
glog.Info("IPIP Tunnel Overlay disabled in configuration.")
glog.Info("Cleaning up old overlay networking if needed.")
err = nrc.disablePolicyBasedRouting()
if err != nil {
glog.Errorf("Failed to disable policy based routing: %s", err.Error())
}
}
glog.Info("Performing cleanup of depreciated rules/ipsets (if needed).")
err = deleteBadPodEgressRules()
if err != nil {
glog.Errorf("Error cleaning up old/bad Pod egress rules: %s", err.Error())
}
// Handle Pod egress masquerading configuration
if nrc.enablePodEgress {
glog.Infoln("Enabling Pod egress.")
err = createPodEgressRule()
if err != nil {
glog.Errorf("Error enabling Pod egress: %s", err.Error())
}
} else {
glog.Infoln("Disabling Pod egress.")
err = deletePodEgressRule()
if err != nil {
glog.Warningf("Error cleaning up Pod Egress related networking: %s", err)
}
}
// enable netfilter for the bridge
if _, err := exec.Command("modprobe", "br_netfilter").CombinedOutput(); err != nil {
@ -118,27 +164,6 @@ func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGr
glog.Infof("Starting network route controller")
// Handle Pod egress masquerading configuration
if nrc.enablePodEgress {
glog.Infoln("Enabling Pod egress.")
err = createPodEgressRule()
if err != nil {
glog.Errorf("Error enabling Pod egress: %s", err.Error())
}
} else {
glog.Infoln("Disabling Pod egress.")
err = deletePodEgressRule()
// TODO: Don't error if removing non-existent Pod egress rules/ipsets.
if err != nil {
glog.Infof("Error disabling Pod egress: %s", err.Error())
}
err = deletePodSubnetIpSet()
if err != nil {
glog.Infof("Error disabling Pod egress: %s", err.Error())
}
}
// Wait till we are ready to launch BGP server
for {
err := nrc.startBgpServer()
@ -168,11 +193,12 @@ func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGr
default:
}
// Update Pod subnet ipset entries
if nrc.enablePodEgress {
err := nrc.syncPodSubnetIpSet()
// Update ipset entries
if nrc.enablePodEgress || nrc.enableOverlays {
glog.Info("Syncing ipsets.")
err := nrc.syncNodeIPSets()
if err != nil {
glog.Errorf("Error synchronizing Pod subnet ipset: %s", err.Error())
glog.Errorf("Error synchronizing ipsets: %s", err.Error())
}
}
@ -228,6 +254,7 @@ func createPodEgressRule() error {
err.Error() + "External connectivity will not work.")
}
glog.Infof("Added iptables rule to masqurade outbound traffic from pods.")
return nil
}
@ -242,6 +269,7 @@ func deletePodEgressRule() error {
if err != nil {
return errors.New("Failed to lookup iptable rule to masqurade outbound traffic from pods: " + err.Error())
}
if exists {
err = iptablesCmdHandler.Delete("nat", "POSTROUTING", podEgressArgs...)
if err != nil {
@ -250,6 +278,34 @@ func deletePodEgressRule() error {
}
glog.Infof("Deleted iptables rule to masqurade outbound traffic from pods.")
}
return nil
}
func deleteBadPodEgressRules() error {
iptablesCmdHandler, err := iptables.New()
if err != nil {
return errors.New("Failed create iptables handler:" + err.Error())
}
for _, args := range podEgressArgsBad {
exists, err := iptablesCmdHandler.Exists("nat", "POSTROUTING", args...)
if err != nil {
return fmt.Errorf("Failed to lookup iptables rule: %s", err.Error())
}
if exists {
err = iptablesCmdHandler.Delete("nat", "POSTROUTING", args...)
if err != nil {
return fmt.Errorf("Failed to delete old/bad iptable rule to "+
"masqurade outbound traffic from pods: %s.\n"+
"Pod egress might still work, or bugs may persist after upgrade...",
err)
}
glog.Infof("Deleted old/bad iptables rule to masqurade outbound traffic from pods.")
}
}
return nil
}
@ -622,12 +678,14 @@ func (nrc *NetworkRoutingController) injectRoute(path *table.Path) error {
glog.Infof("Tunnel interface: " + tunnelName + " for the node " + nexthop.String() + " already exists.")
}
out, err := exec.Command("ip", "route", "list", "table", "kube-router").Output()
out, err := exec.Command("ip", "route", "list", "table", customRouteTableName).Output()
if err != nil {
return errors.New("Failed to verify if route already exists in kube-router table due to " + err.Error())
return fmt.Errorf("Failed to verify if route already exists in %s table: %s",
customRouteTableName, err.Error())
}
if !strings.Contains(string(out), tunnelName) {
if err = exec.Command("ip", "route", "add", nexthop.String(), "dev", tunnelName, "table", "kube-router").Run(); err != nil {
if err = exec.Command("ip", "route", "add", nexthop.String(), "dev", tunnelName, "table",
customRouteTableName).Run(); err != nil {
return errors.New("Failed to add route in custom route table due to: " + err.Error())
}
}
@ -655,36 +713,21 @@ func (nrc *NetworkRoutingController) injectRoute(path *table.Path) error {
// Cleanup performs the cleanup of configurations done
func (nrc *NetworkRoutingController) Cleanup() {
// Pod egress cleanup
err := deletePodEgressRule()
if err != nil {
glog.Errorf("Error deleting Pod egress iptable rule: %s", err.Error())
glog.Warningf("Error deleting Pod egress iptable rule: %s", err.Error())
}
err = deletePodSubnetIpSet()
err = deleteBadPodEgressRules()
if err != nil {
glog.Errorf("Error deleting Pod subnet ipset: %s", err.Error())
}
}
func deletePodSubnetIpSet() error {
_, err := exec.LookPath("ipset")
if err != nil {
return errors.New("Ensure ipset package is installed: " + err.Error())
glog.Warningf("Error deleting Pod egress iptable rule: %s", err.Error())
}
ipset, err := utils.NewIPSet()
err = nrc.ipSetHandler.DestroyAllWithin()
if err != nil {
return err
glog.Warningf("Error deleting ipset: %s", err.Error())
}
ipset.Sets[podSubnetIpSetName] = &utils.Set{
Name: podSubnetIpSetName,
}
err = ipset.Destroy()
if err != nil {
return errors.New("Failure deleting Pod egress ipset: " + err.Error())
}
return nil
}
func (nrc *NetworkRoutingController) disableSourceDestinationCheck() {
@ -728,24 +771,53 @@ func (nrc *NetworkRoutingController) disableSourceDestinationCheck() {
}
}
func (nrc *NetworkRoutingController) syncPodSubnetIpSet() error {
glog.Infof("Syncing Pod subnet ipset entries.")
// get the current list of the nodes from API server
func (nrc *NetworkRoutingController) syncNodeIPSets() error {
// Get the current list of the nodes from API server
nodes, err := nrc.clientset.Core().Nodes().List(metav1.ListOptions{})
if err != nil {
return errors.New("Failed to list nodes from API server: " + err.Error())
}
// Collect active PodCIDR(s) from nodes
// Collect active PodCIDR(s) and NodeIPs from nodes
currentPodCidrs := make([]string, 0)
currentNodeIPs := make([]string, 0)
for _, node := range nodes.Items {
currentPodCidrs = append(currentPodCidrs, node.Spec.PodCIDR)
nodeIP, err := utils.GetNodeIP(&node)
if err != nil {
return fmt.Errorf("Failed to find a node IP: %s", err)
}
currentNodeIPs = append(currentNodeIPs, nodeIP.String())
}
err = nrc.podSubnetsIpSet.Refresh(currentPodCidrs, utils.OptionTimeout, "0")
// Syncing Pod subnet ipset entries
psSet := nrc.ipSetHandler.Get(podSubnetsIPSetName)
if psSet == nil {
glog.Infof("Creating missing ipset \"%s\"", podSubnetsIPSetName)
_, err = nrc.ipSetHandler.Create(podSubnetsIPSetName, utils.OptionTimeout, "0")
if err != nil {
return errors.New("Failed to update Pod subnet ipset: " + err.Error())
return fmt.Errorf("ipset \"%s\" not found in controller instance.",
podSubnetsIPSetName)
}
}
err = psSet.Refresh(currentPodCidrs, psSet.Options...)
if err != nil {
return fmt.Errorf("Failed to sync Pod Subnets ipset: %s", err)
}
// Syncing Node Addresses ipset entries
naSet := nrc.ipSetHandler.Get(nodeAddrsIPSetName)
if naSet == nil {
glog.Infof("Creating missing ipset \"%s\"", nodeAddrsIPSetName)
_, err = nrc.ipSetHandler.Create(nodeAddrsIPSetName, utils.OptionTimeout, "0")
if err != nil {
return fmt.Errorf("ipset \"%s\" not found in controller instance.",
nodeAddrsIPSetName)
}
}
err = naSet.Refresh(currentNodeIPs, naSet.Options...)
if err != nil {
return fmt.Errorf("Failed to sync Node Addresses ipset: %s", err)
}
return nil
@ -769,7 +841,7 @@ func (nrc *NetworkRoutingController) syncPeers() {
// establish peer and add Pod CIDRs with current set of nodes
currentNodes := make([]string, 0)
for _, node := range nodes.Items {
nodeIP, _ := getNodeIP(&node)
nodeIP, _ := utils.GetNodeIP(&node)
// skip self
if nodeIP.String() == nrc.nodeIP.String() {
@ -900,33 +972,23 @@ func (nrc *NetworkRoutingController) enableForwarding() error {
// setup a custom routing table that will be used for policy based routing to ensure traffic originating
// on tunnel interface only leaves through tunnel interface irrespective rp_filter enabled/disabled
func (nrc *NetworkRoutingController) enablePolicyBasedRouting() error {
b, err := ioutil.ReadFile("/etc/iproute2/rt_tables")
err := rtTablesAdd(customRouteTableID, customRouteTableName)
if err != nil {
return fmt.Errorf("Failed to enable required policy based routing due to: %s", err.Error())
}
if !strings.Contains(string(b), "kube-router") {
f, err := os.OpenFile("/etc/iproute2/rt_tables", os.O_APPEND|os.O_WRONLY, 0600)
if err != nil {
return fmt.Errorf("Failed to enable required policy based routing. Failed to create custom route table to route tunneled traffic properly due to: %s", err.Error())
}
if _, err = f.WriteString("77 kube-router"); err != nil {
return fmt.Errorf("Failed to enable required policy based routing.Failed to add custom router table entry in /etc/iproute2/rt_tables due to: %s", err.Error())
}
return fmt.Errorf("Failed to update rt_tables file: %s", err)
}
cidr, err := utils.GetPodCidrFromNodeSpec(nrc.clientset, nrc.hostnameOverride)
if err != nil {
return fmt.Errorf("Failed to enable required policy based routing.Failed to get the pod CIDR allocated for the node due to: %s", err.Error())
return fmt.Errorf("Failed to get the pod CIDR allocated for the node: %s", err.Error())
}
out, err := exec.Command("ip", "rule", "list").Output()
if err != nil {
return fmt.Errorf("Failed to enable required policy based routing as failed to verify if `ip rule` exists due to %s", err.Error())
return fmt.Errorf("Failed to verify if `ip rule` exists: %s", err.Error())
}
if !strings.Contains(string(out), cidr) {
err = exec.Command("ip", "rule", "add", "from", cidr, "table", "kube-router").Run()
err = exec.Command("ip", "rule", "add", "from", cidr, "table", customRouteTableName).Run()
if err != nil {
return fmt.Errorf("Failed to add ip rule due to: %s", err.Error())
}
@ -935,6 +997,54 @@ func (nrc *NetworkRoutingController) enablePolicyBasedRouting() error {
return nil
}
func (nrc *NetworkRoutingController) disablePolicyBasedRouting() error {
err := rtTablesAdd(customRouteTableID, customRouteTableName)
if err != nil {
return fmt.Errorf("Failed to update rt_tables file: %s", err)
}
cidr, err := utils.GetPodCidrFromNodeSpec(nrc.clientset, nrc.hostnameOverride)
if err != nil {
return fmt.Errorf("Failed to get the pod CIDR allocated for the node: %s",
err.Error())
}
out, err := exec.Command("ip", "rule", "list").Output()
if err != nil {
return fmt.Errorf("Failed to verify if `ip rule` exists: %s",
err.Error())
}
if strings.Contains(string(out), cidr) {
err = exec.Command("ip", "rule", "del", "from", cidr, "table", customRouteTableName).Run()
if err != nil {
return fmt.Errorf("Failed to delete ip rule: %s", err.Error())
}
}
return nil
}
func rtTablesAdd(num uint, s string) error {
b, err := ioutil.ReadFile("/etc/iproute2/rt_tables")
if err != nil {
return fmt.Errorf("Failed to read: %s", err.Error())
}
if !strings.Contains(string(b), s) {
f, err := os.OpenFile("/etc/iproute2/rt_tables", os.O_APPEND|os.O_WRONLY, 0600)
if err != nil {
return fmt.Errorf("Failed to open: %s", err.Error())
}
if _, err = f.WriteString(strconv.Itoa(int(num)) + " " + s); err != nil {
return fmt.Errorf("Failed to write: %s",
err.Error())
}
}
return nil
}
// OnNodeUpdate Handle updates from Node watcher. Node watcher calls this method whenever there is
// new node is added or old node is deleted. So peer up with new node and drop peering
// from old node
@ -943,7 +1053,7 @@ func (nrc *NetworkRoutingController) OnNodeUpdate(nodeUpdate *watchers.NodeUpdat
defer nrc.mu.Unlock()
node := nodeUpdate.Node
nodeIP, _ := getNodeIP(node)
nodeIP, _ := utils.GetNodeIP(node)
if nodeUpdate.Op == watchers.ADD {
glog.Infof("Received node %s added update from watch API so peer with new node", nodeIP)
n := &config.Neighbor{
@ -1133,34 +1243,37 @@ func getNodeSubnet(nodeIp net.IP) (net.IPNet, string, error) {
return net.IPNet{}, "", errors.New("Failed to find interface with specified node ip")
}
// func (nrc *NetworkRoutingController) getExternalNodeIPs(
// NewNetworkRoutingController returns new NetworkRoutingController object
func NewNetworkRoutingController(clientset *kubernetes.Clientset,
kubeRouterConfig *options.KubeRouterConfig) (*NetworkRoutingController, error) {
// TODO: Remove lookup, ipset.New already does this.
_, err := exec.LookPath("ipset")
if err != nil {
return nil, errors.New("Ensure ipset package is installed: " + err.Error())
}
var err error
nrc := NetworkRoutingController{}
nrc.bgpFullMeshMode = kubeRouterConfig.FullMeshMode
nrc.enablePodEgress = kubeRouterConfig.EnablePodEgress
nrc.syncPeriod = kubeRouterConfig.RoutesSyncPeriod
nrc.clientset = clientset
ipset, err := utils.NewIPSet()
nrc.ipSetHandler, err = utils.NewIPSet()
if err != nil {
return nil, err
}
if nrc.enablePodEgress || len(nrc.clusterCIDR) != 0 {
nrc.enablePodEgress = true
// TODO: Add bitmap hashtype support to ipset package. It would work well here.
nrc.podSubnetsIpSet, err = ipset.Create(podSubnetIpSetName, utils.TypeHashNet, utils.OptionTimeout, "0")
_, err = nrc.ipSetHandler.Create(podSubnetsIPSetName, utils.TypeHashNet, utils.OptionTimeout, "0")
if err != nil {
return nil, fmt.Errorf("failed to create Pod subnet ipset: %s", err.Error())
return nil, err
}
_, err = nrc.ipSetHandler.Create(nodeAddrsIPSetName, utils.TypeHashIP, utils.OptionTimeout, "0")
if err != nil {
return nil, err
}
if kubeRouterConfig.EnablePodEgress || len(nrc.clusterCIDR) != 0 {
nrc.enablePodEgress = true
}
if kubeRouterConfig.ClusterAsn != 0 {
@ -1205,7 +1318,7 @@ func NewNetworkRoutingController(clientset *kubernetes.Clientset,
nrc.nodeHostName = node.Name
nodeIP, err := getNodeIP(node)
nodeIP, err := utils.GetNodeIP(node)
if err != nil {
return nil, errors.New("Failed getting IP address from node object: " + err.Error())
}

2
app/controllers/network_services_controller.go
View File

@ -928,7 +928,7 @@ func NewNetworkServicesController(clientset *kubernetes.Clientset, config *optio
}
nsc.nodeHostName = node.Name
nodeIP, err := getNodeIP(node)
nodeIP, err := utils.GetNodeIP(node)
if err != nil {
return nil, err
}

220
utils/ipset.go
View File

@ -81,29 +81,21 @@ const (
// IPSet represent ipset sets managed by.
type IPSet struct {
// ipset bianry path.
ipSetPath *string
// Sets maintainted by ipset.
Sets map[string]*Set
}
// Set reprensent a ipset set entry.
type Set struct {
// ipset parent to get ipSetPath.
IPSet *IPSet
// set name.
Parent *IPSet
Name string
// set entries.
Entries []*Entry
// set created options.
Options []string
}
// Entry of ipset Set.
type Entry struct {
// set parent to get ipSetPath.
Set *Set
// entry created options.
Options []string
}
@ -117,12 +109,12 @@ func getIPSetPath() (*string, error) {
}
// Used to run ipset binary with args and return stdout.
func (set *IPSet) run(args ...string) (string, error) {
func (ipset *IPSet) run(args ...string) (string, error) {
var stderr bytes.Buffer
var stdout bytes.Buffer
cmd := exec.Cmd{
Path: *set.ipSetPath,
Args: append([]string{*set.ipSetPath}, args...),
Path: *ipset.ipSetPath,
Args: append([]string{*ipset.ipSetPath}, args...),
Stderr: &stderr,
Stdout: &stdout,
}
@ -135,12 +127,12 @@ func (set *IPSet) run(args ...string) (string, error) {
}
// Used to run ipset binary with arg and inject stdin buffer and return stdout.
func (set *IPSet) runWithStdin(stdin *bytes.Buffer, args ...string) (string, error) {
func (ipset *IPSet) runWithStdin(stdin *bytes.Buffer, args ...string) (string, error) {
var stderr bytes.Buffer
var stdout bytes.Buffer
cmd := exec.Cmd{
Path: *set.ipSetPath,
Args: append([]string{*set.ipSetPath}, args...),
Path: *ipset.ipSetPath,
Args: append([]string{*ipset.ipSetPath}, args...),
Stderr: &stderr,
Stdout: &stdout,
Stdin: stdin,
@ -166,77 +158,150 @@ func NewIPSet() (*IPSet, error) {
return ipSet, nil
}
// Create a set identified with setname and specified type. The type may require type specific options. If the -exist option is specified, ipset ignores the error otherwise raised when the same set (setname and create parameters are identical) already exists.
func (ipSet *IPSet) Create(setName string, createOptions ...string) (*Set, error) {
set := &Set{
// Create a set identified with setname and specified type. The type may
// require type specific options. Does not create set on the system if it
// already exists by the same name.
func (ipset *IPSet) Create(setName string, createOptions ...string) (*Set, error) {
// Populate Set map if needed
if ipset.Get(setName) == nil {
ipset.Sets[setName] = &Set{
Name: setName,
Options: createOptions,
IPSet: ipSet,
Parent: ipset,
}
ipSet.Sets[setName] = set
_, err := ipSet.run(append([]string{"create", "-exist", set.Name}, createOptions...)...)
}
// Determine if set with the same name is already active on the system
setIsActive, err := ipset.Sets[setName].IsActive()
if err != nil {
return nil, err
return nil, fmt.Errorf("Failed to determine if ipset set %s exists: %s",
setName, err)
}
return set, nil
// Create set if missing from the system
if !setIsActive {
_, err := ipset.run(append([]string{"create", "-exist", setName},
createOptions...)...)
if err != nil {
return nil, fmt.Errorf("Failed to create ipset set on system: %s", err)
}
}
return ipset.Sets[setName], nil
}
// Add a given entry to the set. If the -exist option is specified, ipset ignores if the entry already added to the set.
// Adds a given Set to an IPSet
func (ipset *IPSet) Add(set *Set) error {
_, err := ipset.Create(set.Name, set.Options...)
if err != nil {
return err
}
for _, entry := range set.Entries {
_, err := ipset.Get(set.Name).Add(entry.Options...)
if err != nil {
return err
}
}
return nil
}
// Add a given entry to the set. If the -exist option is specified, ipset
// ignores if the entry already added to the set.
func (set *Set) Add(addOptions ...string) (*Entry, error) {
entry := &Entry{
Set: set,
Options: addOptions,
}
set.Entries = append(set.Entries, entry)
_, err := set.IPSet.run(append([]string{"add", "-exist", entry.Set.Name}, addOptions...)...)
_, err := set.Parent.run(append([]string{"add", "-exist", entry.Set.Name}, addOptions...)...)
if err != nil {
return nil, err
}
return entry, nil
}
// Del an entry from a set. If the -exist option is specified and the entry is not in the set (maybe already expired), then the command is ignored.
// Del an entry from a set. If the -exist option is specified and the entry is
// not in the set (maybe already expired), then the command is ignored.
func (entry *Entry) Del() error {
_, err := entry.Set.IPSet.run(append([]string{"del", entry.Set.Name}, entry.Options...)...)
_, err := entry.Set.Parent.run(append([]string{"del", entry.Set.Name}, entry.Options...)...)
if err != nil {
return err
}
entry.Set.IPSet.Save()
entry.Set.Parent.Save()
return nil
}
// Test wether an entry is in a set or not. Exit status number is zero if the tested entry is in the set and nonzero if it is missing from the set.
// Test wether an entry is in a set or not. Exit status number is zero if the
// tested entry is in the set and nonzero if it is missing from the set.
func (set *Set) Test(testOptions ...string) (bool, error) {
_, err := set.IPSet.run(append([]string{"test", set.Name}, testOptions...)...)
_, err := set.Parent.run(append([]string{"test", set.Name}, testOptions...)...)
if err != nil {
return false, err
}
return true, nil
}
// Destroy the specified set or all the sets if none is given. If the set has got reference(s), nothing is done and no set destroyed.
// Destroy the specified set or all the sets if none is given. If the set has
// got reference(s), nothing is done and no set destroyed.
func (set *Set) Destroy() error {
_, err := set.IPSet.run("destroy", set.Name)
_, err := set.Parent.run("destroy", set.Name)
if err != nil {
return err
}
delete(set.Parent.Sets, set.Name)
return nil
}
// Destroy the specified set or all the sets if none is given. If the set has got reference(s), nothing is done and no set destroyed.
func (set *IPSet) Destroy() error {
_, err := set.run("destroy")
// Destroy the specified set by name. If the set has got reference(s), nothing
// is done and no set destroyed. If the IPSet does not contain the named set
// then Destroy is a no-op.
func (ipset *IPSet) Destroy(setName string) error {
set := ipset.Get(setName)
if set == nil {
return nil
}
err := set.Destroy()
if err != nil {
return err
}
return nil
}
// DestroyAllWithin destroys all sets contained within the IPSet's Sets.
func (ipset *IPSet) DestroyAllWithin() error {
for _, v := range ipset.Sets {
err := v.Destroy()
if err != nil {
return err
}
}
return nil
}
// IsActive checks if a set exists on the system with the same name.
func (set *Set) IsActive() (bool, error) {
_, err := set.Parent.run("list", set.Name)
if err != nil {
if strings.Contains(err.Error(), "name does not exist") {
return false, nil
}
return false, err
}
return true, nil
}
// Parse ipset save stdout.
// ex:
// create KUBE-DST-3YNVZWWGX3UQQ4VQ hash:ip family inet hashsize 1024 maxelem 65536 timeout 0
// add KUBE-DST-3YNVZWWGX3UQQ4VQ 100.96.1.6 timeout 0
func parseIPSetSave(ipSet *IPSet, result string) map[string]*Set {
func parseIPSetSave(ipset *IPSet, result string) map[string]*Set {
sets := make(map[string]*Set)
// Save is always in order
lines := strings.Split(result, "\n")
@ -244,7 +309,7 @@ func parseIPSetSave(ipSet *IPSet, result string) map[string]*Set {
content := strings.Split(line, " ")
if content[0] == "create" {
sets[content[1]] = &Set{
IPSet: ipSet,
Parent: ipset,
Name: content[1],
Options: content[2:],
}
@ -264,9 +329,9 @@ func parseIPSetSave(ipSet *IPSet, result string) map[string]*Set {
// ex:
// create KUBE-DST-3YNVZWWGX3UQQ4VQ hash:ip family inet hashsize 1024 maxelem 65536 timeout 0
// add KUBE-DST-3YNVZWWGX3UQQ4VQ 100.96.1.6 timeout 0
func buildIPSetRestore(ipSet *IPSet) string {
func buildIPSetRestore(ipset *IPSet) string {
ipSetRestore := ""
for _, set := range ipSet.Sets {
for _, set := range ipset.Sets {
ipSetRestore += fmt.Sprintf("create %s %s\n", set.Name, strings.Join(set.Options[:], " "))
for _, entry := range set.Entries {
ipSetRestore += fmt.Sprintf("add %s %s\n", set.Name, strings.Join(entry.Options[:], " "))
@ -275,22 +340,28 @@ func buildIPSetRestore(ipSet *IPSet) string {
return ipSetRestore
}
// Save the given set, or all sets if none is given to stdout in a format that restore can read. The option -file can be used to specify a filename instead of stdout.
// Save the given set, or all sets if none is given to stdout in a format that
// restore can read. The option -file can be used to specify a filename instead
// of stdout.
// save "ipset save" command output to ipset.sets.
func (set *IPSet) Save() error {
stdout, err := set.run("save")
func (ipset *IPSet) Save() error {
stdout, err := ipset.run("save")
if err != nil {
return err
}
set.Sets = parseIPSetSave(set, stdout)
ipset.Sets = parseIPSetSave(ipset, stdout)
return nil
}
// Restore a saved session generated by save. The saved session can be fed from stdin or the option -file can be used to specify a filename instead of stdin. Please note, existing sets and elements are not erased by restore unless specified so in the restore file. All commands are allowed in restore mode except list, help, version, interactive mode and restore itself.
// Restore a saved session generated by save. The saved session can be fed from
// stdin or the option -file can be used to specify a filename instead of
// stdin. Please note, existing sets and elements are not erased by restore
// unless specified so in the restore file. All commands are allowed in restore
// mode except list, help, version, interactive mode and restore itself.
// Send formated ipset.sets into stdin of "ipset restore" command.
func (set *IPSet) Restore() error {
stdin := bytes.NewBufferString(buildIPSetRestore(set))
_, err := set.runWithStdin(stdin, "restore", "-exist")
func (ipset *IPSet) Restore() error {
stdin := bytes.NewBufferString(buildIPSetRestore(ipset))
_, err := ipset.runWithStdin(stdin, "restore", "-exist")
if err != nil {
return err
}
@ -299,7 +370,7 @@ func (set *IPSet) Restore() error {
// Flush all entries from the specified set or flush all sets if none is given.
func (set *Set) Flush() error {
_, err := set.IPSet.run("flush", set.Name)
_, err := set.Parent.run("flush", set.Name)
if err != nil {
return err
}
@ -307,8 +378,8 @@ func (set *Set) Flush() error {
}
// Flush all entries from the specified set or flush all sets if none is given.
func (set *IPSet) Flush() error {
_, err := set.run("flush")
func (ipset *IPSet) Flush() error {
_, err := ipset.run("flush")
if err != nil {
return err
}
@ -317,21 +388,28 @@ func (set *IPSet) Flush() error {
// Get Set by Name.
func (ipset *IPSet) Get(setName string) *Set {
return ipset.Sets[setName]
set, ok := ipset.Sets[setName]
if !ok {
return nil
}
return set
}
// Rename a set. Set identified by SETNAME-TO must not exist.
func (set *Set) Rename(newName string) error {
_, err := set.IPSet.run("rename", set.Name, newName)
_, err := set.Parent.run("rename", set.Name, newName)
if err != nil {
return err
}
return nil
}
// Swap the content of two sets, or in another words, exchange the name of two sets. The referred sets must exist and compatible type of sets can be swapped only.
// Swap the content of two sets, or in another words, exchange the name of two
// sets. The referred sets must exist and compatible type of sets can be
// swapped only.
func (set *Set) Swap(setTo *Set) error {
_, err := set.IPSet.run("rename", set.Name, setTo.Name)
_, err := set.Parent.run("swap", set.Name, setTo.Name)
if err != nil {
return err
}
@ -340,38 +418,36 @@ func (set *Set) Swap(setTo *Set) error {
// Refresh a Set with new entries.
func (set *Set) Refresh(entries []string, extraOptions ...string) error {
var err error
tempName := set.Name + "-temp"
s := &Set{
IPSet: set.IPSet,
newSet := &Set{
Parent: set.Parent,
Name: tempName,
Options: set.Options,
}
err = set.Parent.Add(newSet)
if err != nil {
return err
}
for _, entry := range entries {
s.Entries = append(s.Entries, &Entry{
Set: s,
Options: append([]string{entry}, extraOptions...),
})
_, err = newSet.Add(entry)
if err != nil {
return err
}
set.IPSet.Sets[tempName] = s
err := set.IPSet.Restore()
}
err = set.Swap(newSet)
if err != nil {
return err
}
err = set.Swap(s)
err = set.Parent.Destroy(tempName)
if err != nil {
return err
}
err = s.Destroy()
if err != nil {
return err
}
s.Name = set.Name
set.IPSet.Sets[set.Name] = s
delete(set.IPSet.Sets, tempName)
return nil
}

21
utils/node.go
View File

@ -1,7 +1,9 @@
package utils
import (
"errors"
"fmt"
"net"
"os"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@ -38,3 +40,22 @@ func GetNodeObject(clientset *kubernetes.Clientset, hostnameOverride string) (*a
return nil, fmt.Errorf("Failed to identify the node by NODE_NAME, hostname or --hostname-override")
}
// GetNodeIP returns the most valid external facing IP address for a node.
// Order of preference:
// 1. NodeInternalIP
// 2. NodeExternalIP (Only set on cloud providers usually)
func GetNodeIP(node *apiv1.Node) (net.IP, error) {
addresses := node.Status.Addresses
addressMap := make(map[apiv1.NodeAddressType][]apiv1.NodeAddress)
for i := range addresses {
addressMap[addresses[i].Type] = append(addressMap[addresses[i].Type], addresses[i])
}
if addresses, ok := addressMap[apiv1.NodeInternalIP]; ok {
return net.ParseIP(addresses[0].Address), nil
}
if addresses, ok := addressMap[apiv1.NodeExternalIP]; ok {
return net.ParseIP(addresses[0].Address), nil
}
return nil, errors.New("host IP unknown")
}