Merge pull request #1628 from aboch/red

Make virtual service programming more robust

libnetwork/ipvs/ipvs.go

@@ -42,6 +42,7 @@ type Destination struct {
 // Handle provides a namespace specific ipvs handle to program ipvs
 // rules.
 type Handle struct {
+	seq  uint32
 	sock *nl.NetlinkSocket
 }
 
@@ -82,6 +83,11 @@ func (i *Handle) NewService(s *Service) error {
 	return i.doCmd(s, nil, ipvsCmdNewService)
 }
 
+// IsServicePresent queries for the ipvs service in the passed handle.
+func (i *Handle) IsServicePresent(s *Service) bool {
+	return nil == i.doCmd(s, nil, ipvsCmdGetService)
+}
+
 // UpdateService updates an already existing service in the passed
 // handle.
 func (i *Handle) UpdateService(s *Service) error {

libnetwork/ipvs/netlink.go

@@ -10,6 +10,7 @@ import (
 	"os/exec"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"syscall"
 	"unsafe"
 
@@ -118,6 +119,7 @@ func fillDestinaton(d *Destination) nl.NetlinkRequestData {
 
 func (i *Handle) doCmd(s *Service, d *Destination, cmd uint8) error {
 	req := newIPVSRequest(cmd)
+	req.Seq = atomic.AddUint32(&i.seq, 1)
 	req.AddData(fillService(s))
 
 	if d != nil {
@@ -206,7 +208,7 @@ done:
 		}
 		for _, m := range msgs {
 			if m.Header.Seq != req.Seq {
-				return nil, fmt.Errorf("Wrong Seq nr %d, expected %d", m.Header.Seq, req.Seq)
+				continue
 			}
 			if m.Header.Pid != pid {
 				return nil, fmt.Errorf("Wrong pid %d, expected %d", m.Header.Pid, pid)

libnetwork/service_common.go

@@ -61,11 +61,6 @@ func (c *controller) cleanupServiceBindings(cleanupNID string) {
 }
 
 func (c *controller) addServiceBinding(name, sid, nid, eid string, vip net.IP, ingressPorts []*PortConfig, aliases []string, ip net.IP) error {
-	var (
-		s          *service
-		addService bool
-	)
-
 	n, err := c.NetworkByID(nid)
 	if err != nil {
 		return err
@@ -123,11 +118,6 @@ func (c *controller) addServiceBinding(name, sid, nid, eid string, vip net.IP, i
 		fwMarkCtrMu.Unlock()
 
 		s.loadBalancers[nid] = lb
-
-		// Since we just created this load balancer make sure
-		// we add a new service service in IPVS rules.
-		addService = true
-
 	}
 
 	lb.backEnds[eid] = ip
@@ -135,7 +125,7 @@ func (c *controller) addServiceBinding(name, sid, nid, eid string, vip net.IP, i
 	// Add loadbalancer service and backend in all sandboxes in
 	// the network only if vip is valid.
 	if len(vip) != 0 {
-		n.(*network).addLBBackend(ip, vip, lb.fwMark, ingressPorts, addService)
+		n.(*network).addLBBackend(ip, vip, lb.fwMark, ingressPorts)
 	}
 
 	return nil

libnetwork/service_linux.go

@@ -68,7 +68,7 @@ func (sb *sandbox) populateLoadbalancers(ep *endpoint) {
 
 	if n.ingress {
 		if err := addRedirectRules(sb.Key(), eIP, ep.ingressPorts); err != nil {
-			logrus.Errorf("Failed to add redirect rules for ep %s: %v", ep.Name(), err)
+			logrus.Errorf("Failed to add redirect rules for ep %s (%s): %v", ep.Name(), ep.ID()[0:7], err)
 		}
 	}
 
@@ -97,20 +97,16 @@ func (sb *sandbox) populateLoadbalancers(ep *endpoint) {
 		}
 
 		lb.service.Lock()
-		addService := true
 		for _, ip := range lb.backEnds {
-			sb.addLBBackend(ip, lb.vip, lb.fwMark, lb.service.ingressPorts,
-				eIP, gwIP, addService, n.ingress)
-			addService = false
+			sb.addLBBackend(ip, lb.vip, lb.fwMark, lb.service.ingressPorts, eIP, gwIP, n.ingress)
 		}
 		lb.service.Unlock()
 	}
 }
 
 // Add loadbalancer backend to all sandboxes which has a connection to
-// this network. If needed add the service as well, as specified by
-// the addService bool.
-func (n *network) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, addService bool) {
+// this network. If needed add the service as well.
+func (n *network) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig) {
 	n.WalkEndpoints(func(e Endpoint) bool {
 		ep := e.(*endpoint)
 		if sb, ok := ep.getSandbox(); ok {
@@ -123,7 +119,7 @@ func (n *network) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*Po
 				gwIP = ep.Iface().Address().IP
 			}
 
-			sb.addLBBackend(ip, vip, fwMark, ingressPorts, ep.Iface().Address(), gwIP, addService, n.ingress)
+			sb.addLBBackend(ip, vip, fwMark, ingressPorts, ep.Iface().Address(), gwIP, n.ingress)
 		}
 
 		return false
@@ -154,7 +150,7 @@ func (n *network) rmLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*Por
 }
 
 // Add loadbalancer backend into one connected sandbox.
-func (sb *sandbox) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, eIP *net.IPNet, gwIP net.IP, addService bool, isIngressNetwork bool) {
+func (sb *sandbox) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, eIP *net.IPNet, gwIP net.IP, isIngressNetwork bool) {
 	if sb.osSbox == nil {
 		return
 	}
@@ -165,7 +161,7 @@ func (sb *sandbox) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*P
 
 	i, err := ipvs.New(sb.Key())
 	if err != nil {
-		logrus.Errorf("Failed to create an ipvs handle for sbox %s: %v", sb.Key(), err)
+		logrus.Errorf("Failed to create an ipvs handle for sbox %s (%s,%s) for lb addition: %v", sb.ID()[0:7], sb.ContainerID()[0:7], sb.Key(), err)
 		return
 	}
 	defer i.Close()
@@ -176,7 +172,7 @@ func (sb *sandbox) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*P
 		SchedName:     ipvs.RoundRobin,
 	}
 
-	if addService {
+	if !i.IsServicePresent(s) {
 		var filteredPorts []*PortConfig
 		if sb.ingress {
 			filteredPorts = filterPortConfigs(ingressPorts, false)
@@ -186,14 +182,14 @@ func (sb *sandbox) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*P
 			}
 		}
 
-		logrus.Debugf("Creating service for vip %s fwMark %d ingressPorts %#v", vip, fwMark, ingressPorts)
+		logrus.Debugf("Creating service for vip %s fwMark %d ingressPorts %#v in sbox %s (%s)", vip, fwMark, ingressPorts, sb.ID()[0:7], sb.ContainerID()[0:7])
 		if err := invokeFWMarker(sb.Key(), vip, fwMark, ingressPorts, eIP, false); err != nil {
-			logrus.Errorf("Failed to add firewall mark rule in sbox %s: %v", sb.Key(), err)
+			logrus.Errorf("Failed to add firewall mark rule in sbox %s (%s): %v", sb.ID()[0:7], sb.ContainerID()[0:7], err)
 			return
 		}
 
 		if err := i.NewService(s); err != nil && err != syscall.EEXIST {
-			logrus.Errorf("Failed to create a new service for vip %s fwmark %d: %v", vip, fwMark, err)
+			logrus.Errorf("Failed to create a new service for vip %s fwmark %d in sbox %s (%s): %v", vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
 			return
 		}
 	}
@@ -208,7 +204,7 @@ func (sb *sandbox) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*P
 	// destination.
 	s.SchedName = ""
 	if err := i.NewDestination(s, d); err != nil && err != syscall.EEXIST {
-		logrus.Errorf("Failed to create real server %s for vip %s fwmark %d in sb %s: %v", ip, vip, fwMark, sb.containerID, err)
+		logrus.Errorf("Failed to create real server %s for vip %s fwmark %d in sbox %s (%s): %v", ip, vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
 	}
 }
 
@@ -224,7 +220,7 @@ func (sb *sandbox) rmLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*Po
 
 	i, err := ipvs.New(sb.Key())
 	if err != nil {
-		logrus.Errorf("Failed to create an ipvs handle for sbox %s: %v", sb.Key(), err)
+		logrus.Errorf("Failed to create an ipvs handle for sbox %s (%s,%s) for lb removal: %v", sb.ID()[0:7], sb.ContainerID()[0:7], sb.Key(), err)
 		return
 	}
 	defer i.Close()
@@ -240,14 +236,14 @@ func (sb *sandbox) rmLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*Po
 		Weight:        1,
 	}
 
-	if err := i.DelDestination(s, d); err != nil {
-		logrus.Infof("Failed to delete real server %s for vip %s fwmark %d: %v", ip, vip, fwMark, err)
+	if err := i.DelDestination(s, d); err != nil && err != syscall.ENOENT {
+		logrus.Errorf("Failed to delete real server %s for vip %s fwmark %d in sbox %s (%s): %v", ip, vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
 	}
 
 	if rmService {
 		s.SchedName = ipvs.RoundRobin
-		if err := i.DelService(s); err != nil {
-			logrus.Errorf("Failed to delete service for vip %s fwmark %d: %v", vip, fwMark, err)
+		if err := i.DelService(s); err != nil && err != syscall.ENOENT {
+			logrus.Errorf("Failed to delete service for vip %s fwmark %d in sbox %s (%s): %v", vip, fwMark, sb.ID()[0:7], sb.ContainerID()[0:7], err)
 		}
 
 		var filteredPorts []*PortConfig
@@ -259,7 +255,7 @@ func (sb *sandbox) rmLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*Po
 		}
 
 		if err := invokeFWMarker(sb.Key(), vip, fwMark, ingressPorts, eIP, true); err != nil {
-			logrus.Errorf("Failed to delete firewall mark rule in sbox %s: %v", sb.Key(), err)
+			logrus.Errorf("Failed to delete firewall mark rule in sbox %s (%s): %v", sb.ID()[0:7], sb.ContainerID()[0:7], err)
 		}
 	}
 }

libnetwork/service_windows.go

@@ -2,7 +2,7 @@ package libnetwork
 
 import "net"
 
-func (n *network) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, addService bool) {
+func (n *network) addLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig) {
 }
 
 func (n *network) rmLBBackend(ip, vip net.IP, fwMark uint32, ingressPorts []*PortConfig, rmService bool) {