Merge pull request #66012 from Lion-Wei/ipvs-graceful-termination

ipvs support graceful termination

pkg/proxy/ipvs/BUILD

@@ -37,6 +37,7 @@ go_test(
 go_library(
     name = "go_default_library",
     srcs = [
+        "graceful_termination.go",
         "ipset.go",
         "netlink.go",
         "netlink_linux.go",

pkg/proxy/ipvs/graceful_termination.go

@@ -0,0 +1,223 @@
+/*
+Copyright 2015 The Kubernetes Authors.
+
+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 ipvs
+
+import (
+	"sync"
+	"time"
+
+	"fmt"
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/util/wait"
+	utilipvs "k8s.io/kubernetes/pkg/util/ipvs"
+)
+
+const (
+	rsGracefulDeletePeriod = 15 * time.Minute
+	rsCheckDeleteInterval  = 1 * time.Minute
+)
+
+// listItem stores real server information and the process time.
+// If nothing special happened, real server will be delete after process time.
+type listItem struct {
+	VirtualServer *utilipvs.VirtualServer
+	RealServer    *utilipvs.RealServer
+}
+
+// String return the unique real server name(with virtual server information)
+func (g *listItem) String() string {
+	return GetUniqueRSName(g.VirtualServer, g.RealServer)
+}
+
+// GetUniqueRSName return a string type unique rs name with vs information
+func GetUniqueRSName(vs *utilipvs.VirtualServer, rs *utilipvs.RealServer) string {
+	return vs.String() + "/" + rs.String()
+}
+
+type graceTerminateRSList struct {
+	lock sync.Mutex
+	list map[string]*listItem
+}
+
+// add push an new element to the rsList
+func (q *graceTerminateRSList) add(rs *listItem) bool {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	uniqueRS := rs.String()
+	if _, ok := q.list[uniqueRS]; ok {
+		return false
+	}
+
+	glog.V(5).Infof("Adding rs %v to graceful delete rsList", rs)
+	q.list[uniqueRS] = rs
+	return true
+}
+
+// remove remove an element from the rsList
+func (q *graceTerminateRSList) remove(rs *listItem) bool {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	uniqueRS := rs.String()
+	if _, ok := q.list[uniqueRS]; ok {
+		return false
+	}
+	delete(q.list, uniqueRS)
+	return true
+}
+
+func (q *graceTerminateRSList) flushList(handler func(rsToDelete *listItem) (bool, error)) bool {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	success := true
+	for name, rs := range q.list {
+		deleted, err := handler(rs)
+		if err != nil {
+			glog.Errorf("Try delete rs %q err: %v", name, err)
+			success = false
+		}
+		if deleted {
+			glog.Infof("lw: remote out of the list: %s", name)
+			q.remove(rs)
+		}
+	}
+	return success
+}
+
+// exist check whether the specified unique RS is in the rsList
+func (q *graceTerminateRSList) exist(uniqueRS string) (*listItem, bool) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	if _, ok := q.list[uniqueRS]; ok {
+		return nil, false
+	}
+	return nil, false
+}
+
+// GracefulTerminationManager manage rs graceful termination information and do graceful termination work
+// rsList is the rs list to graceful termination, ipvs is the ipvsinterface to do ipvs delete/update work
+type GracefulTerminationManager struct {
+	rsList graceTerminateRSList
+	ipvs   utilipvs.Interface
+}
+
+// NewGracefulTerminationManager create a gracefulTerminationManager to manage ipvs rs graceful termination work
+func NewGracefulTerminationManager(ipvs utilipvs.Interface) *GracefulTerminationManager {
+	l := make(map[string]*listItem)
+	return &GracefulTerminationManager{
+		rsList: graceTerminateRSList{
+			list: l,
+		},
+		ipvs: ipvs,
+	}
+}
+
+// InTerminationList to check whether specified unique rs name is in graceful termination list
+func (m *GracefulTerminationManager) InTerminationList(uniqueRS string) bool {
+	_, exist := m.rsList.exist(uniqueRS)
+	return exist
+}
+
+// GracefulDeleteRS to update rs weight to 0, and add rs to graceful terminate list
+func (m *GracefulTerminationManager) GracefulDeleteRS(vs *utilipvs.VirtualServer, rs *utilipvs.RealServer) error {
+	// Try to delete rs before add it to graceful delete list
+	ele := &listItem{
+		VirtualServer: vs,
+		RealServer:    rs,
+	}
+	deleted, err := m.deleteRsFunc(ele)
+	if err != nil {
+		glog.Errorf("Delete rs %q err: %v", err)
+	}
+	if deleted {
+		return nil
+	}
+	rs.Weight = 0
+	err = m.ipvs.UpdateRealServer(vs, rs)
+	if err != nil {
+		return err
+	}
+	glog.V(5).Infof("Adding an element to graceful delete rsList: %+v", ele)
+	m.rsList.add(ele)
+	return nil
+}
+
+func (m *GracefulTerminationManager) deleteRsFunc(rsToDelete *listItem) (bool, error) {
+	glog.Infof("Trying to delete rs: %s", rsToDelete.String())
+	rss, err := m.ipvs.GetRealServers(rsToDelete.VirtualServer)
+	if err != nil {
+		return false, err
+	}
+	for _, rs := range rss {
+		if rsToDelete.RealServer.Equal(rs) {
+			if rs.ActiveConn != 0 {
+				return false, nil
+			}
+			glog.Infof("Deleting rs: %s", rsToDelete.String())
+			err := m.ipvs.DeleteRealServer(rsToDelete.VirtualServer, rs)
+			if err != nil {
+				return false, fmt.Errorf("Delete destination %q err: %v", rs.String(), err)
+			}
+			return true, nil
+		}
+	}
+	return false, fmt.Errorf("Failed to delete rs %q, can't find the real server", rsToDelete.String())
+}
+
+func (m *GracefulTerminationManager) tryDeleteRs() {
+	if !m.rsList.flushList(m.deleteRsFunc) {
+		glog.Errorf("Try flush graceful termination list err")
+	}
+}
+
+// MoveRSOutofGracefulDeleteList to delete an rs and remove it from the rsList immediately
+func (m *GracefulTerminationManager) MoveRSOutofGracefulDeleteList(uniqueRS string) error {
+	rsToDelete, find := m.rsList.exist(uniqueRS)
+	if !find || rsToDelete == nil {
+		return fmt.Errorf("failed to find rs: %q", uniqueRS)
+	}
+	err := m.ipvs.DeleteRealServer(rsToDelete.VirtualServer, rsToDelete.RealServer)
+	if err != nil {
+		return err
+	}
+	m.rsList.remove(rsToDelete)
+	return nil
+}
+
+// Run start a goroutine to try to delete rs in the graceful delete rsList with an interval 1 minute
+func (m *GracefulTerminationManager) Run() {
+	// before start, add leftover in delete rs to graceful delete rsList
+	vss, err := m.ipvs.GetVirtualServers()
+	if err != nil {
+		glog.Errorf("IPVS graceful delete manager failed to get IPVS virtualserver")
+	}
+	for _, vs := range vss {
+		rss, err := m.ipvs.GetRealServers(vs)
+		if err != nil {
+			glog.Errorf("IPVS graceful delete manager failed to get %v realserver", vs)
+			continue
+		}
+		for _, rs := range rss {
+			m.GracefulDeleteRS(vs, rs)
+		}
+	}
+
+	go wait.Until(m.tryDeleteRs, rsCheckDeleteInterval, wait.NeverStop)
+}

pkg/proxy/ipvs/proxier.go

@@ -231,7 +231,8 @@ type Proxier struct {
 	nodePortAddresses []string
 	// networkInterfacer defines an interface for several net library functions.
 	// Inject for test purpose.
-	networkInterfacer utilproxy.NetworkInterfacer
+	networkInterfacer     utilproxy.NetworkInterfacer
+	gracefuldeleteManager *GracefulTerminationManager
 }
 
 // IPGetter helps get node network interface IP
@@ -353,38 +354,39 @@ func NewProxier(ipt utiliptables.Interface,
 	healthChecker := healthcheck.NewServer(hostname, recorder, nil, nil) // use default implementations of deps
 
 	proxier := &Proxier{
-		portsMap:          make(map[utilproxy.LocalPort]utilproxy.Closeable),
+		portsMap:              make(map[utilproxy.LocalPort]utilproxy.Closeable),
-		serviceMap:        make(proxy.ServiceMap),
+		serviceMap:            make(proxy.ServiceMap),
-		serviceChanges:    proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder),
+		serviceChanges:        proxy.NewServiceChangeTracker(newServiceInfo, &isIPv6, recorder),
-		endpointsMap:      make(proxy.EndpointsMap),
+		endpointsMap:          make(proxy.EndpointsMap),
-		endpointsChanges:  proxy.NewEndpointChangeTracker(hostname, nil, &isIPv6, recorder),
+		endpointsChanges:      proxy.NewEndpointChangeTracker(hostname, nil, &isIPv6, recorder),
-		syncPeriod:        syncPeriod,
+		syncPeriod:            syncPeriod,
-		minSyncPeriod:     minSyncPeriod,
+		minSyncPeriod:         minSyncPeriod,
-		excludeCIDRs:      excludeCIDRs,
+		excludeCIDRs:          excludeCIDRs,
-		iptables:          ipt,
+		iptables:              ipt,
-		masqueradeAll:     masqueradeAll,
+		masqueradeAll:         masqueradeAll,
-		masqueradeMark:    masqueradeMark,
+		masqueradeMark:        masqueradeMark,
-		exec:              exec,
+		exec:                  exec,
-		clusterCIDR:       clusterCIDR,
+		clusterCIDR:           clusterCIDR,
-		hostname:          hostname,
+		hostname:              hostname,
-		nodeIP:            nodeIP,
+		nodeIP:                nodeIP,
-		portMapper:        &listenPortOpener{},
+		portMapper:            &listenPortOpener{},
-		recorder:          recorder,
+		recorder:              recorder,
-		healthChecker:     healthChecker,
+		healthChecker:         healthChecker,
-		healthzServer:     healthzServer,
+		healthzServer:         healthzServer,
-		ipvs:              ipvs,
+		ipvs:                  ipvs,
-		ipvsScheduler:     scheduler,
+		ipvsScheduler:         scheduler,
-		ipGetter:          &realIPGetter{nl: NewNetLinkHandle()},
+		ipGetter:              &realIPGetter{nl: NewNetLinkHandle()},
-		iptablesData:      bytes.NewBuffer(nil),
+		iptablesData:          bytes.NewBuffer(nil),
-		filterChainsData:  bytes.NewBuffer(nil),
+		filterChainsData:      bytes.NewBuffer(nil),
-		natChains:         bytes.NewBuffer(nil),
+		natChains:             bytes.NewBuffer(nil),
-		natRules:          bytes.NewBuffer(nil),
+		natRules:              bytes.NewBuffer(nil),
-		filterChains:      bytes.NewBuffer(nil),
+		filterChains:          bytes.NewBuffer(nil),
-		filterRules:       bytes.NewBuffer(nil),
+		filterRules:           bytes.NewBuffer(nil),
-		netlinkHandle:     NewNetLinkHandle(),
+		netlinkHandle:         NewNetLinkHandle(),
-		ipset:             ipset,
+		ipset:                 ipset,
-		nodePortAddresses: nodePortAddresses,
+		nodePortAddresses:     nodePortAddresses,
-		networkInterfacer: utilproxy.RealNetwork{},
+		networkInterfacer:     utilproxy.RealNetwork{},
+		gracefuldeleteManager: NewGracefulTerminationManager(ipvs),
 	}
 	// initialize ipsetList with all sets we needed
 	proxier.ipsetList = make(map[string]*IPSet)
@@ -397,6 +399,7 @@ func NewProxier(ipt utiliptables.Interface,
 	burstSyncs := 2
 	glog.V(3).Infof("minSyncPeriod: %v, syncPeriod: %v, burstSyncs: %d", minSyncPeriod, syncPeriod, burstSyncs)
 	proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
+	proxier.gracefuldeleteManager.Run()
 	return proxier, nil
 }
 
@@ -1510,53 +1513,72 @@ func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNode
 		newEndpoints.Insert(epInfo.String())
 	}
 
-	if !curEndpoints.Equal(newEndpoints) {
+	// Create new endpoints
-		// Create new endpoints
+	for _, ep := range newEndpoints.List() {
-		for _, ep := range newEndpoints.Difference(curEndpoints).UnsortedList() {
+		ip, port, err := net.SplitHostPort(ep)
-			ip, port, err := net.SplitHostPort(ep)
+		if err != nil {
-			if err != nil {
+			glog.Errorf("Failed to parse endpoint: %v, error: %v", ep, err)
-				glog.Errorf("Failed to parse endpoint: %v, error: %v", ep, err)
+			continue
-				continue
+		}
-			}
+		portNum, err := strconv.Atoi(port)
-			portNum, err := strconv.Atoi(port)
+		if err != nil {
-			if err != nil {
+			glog.Errorf("Failed to parse endpoint port %s, error: %v", port, err)
-				glog.Errorf("Failed to parse endpoint port %s, error: %v", port, err)
+			continue
-				continue
+		}
-			}
 
-			newDest := &utilipvs.RealServer{
+		newDest := &utilipvs.RealServer{
-				Address: net.ParseIP(ip),
+			Address: net.ParseIP(ip),
-				Port:    uint16(portNum),
+			Port:    uint16(portNum),
-				Weight:  1,
+			Weight:  1,
+		}
+
+		if curEndpoints.Has(ep) {
+			// check if newEndpoint is in gracefulDelete list, is true, delete this ep immediately
+			uniqueRS := GetUniqueRSName(vs, newDest)
+			if !proxier.gracefuldeleteManager.InTerminationList(uniqueRS) {
+				continue
 			}
-			err = proxier.ipvs.AddRealServer(appliedVirtualServer, newDest)
+			glog.V(5).Infof("new ep %q is in graceful delete list", uniqueRS)
+			err := proxier.gracefuldeleteManager.MoveRSOutofGracefulDeleteList(uniqueRS)
 			if err != nil {
-				glog.Errorf("Failed to add destination: %v, error: %v", newDest, err)
+				glog.Errorf("Failed to delete endpoint: %v in gracefulDeleteQueue, error: %v", ep, err)
 				continue
 			}
 		}
-		// Delete old endpoints
+		err = proxier.ipvs.AddRealServer(appliedVirtualServer, newDest)
-		for _, ep := range curEndpoints.Difference(newEndpoints).UnsortedList() {
+		if err != nil {
-			ip, port, err := net.SplitHostPort(ep)
+			glog.Errorf("Failed to add destination: %v, error: %v", newDest, err)
-			if err != nil {
+			continue
-				glog.Errorf("Failed to parse endpoint: %v, error: %v", ep, err)
+		}
-				continue
+	}
-			}
+	// Delete old endpoints
-			portNum, err := strconv.Atoi(port)
+	for _, ep := range curEndpoints.Difference(newEndpoints).UnsortedList() {
-			if err != nil {
+		// if curEndpoint is in gracefulDelete, skip
-				glog.Errorf("Failed to parse endpoint port %s, error: %v", port, err)
+		uniqueRS := vs.String() + "/" + ep
-				continue
+		if proxier.gracefuldeleteManager.InTerminationList(uniqueRS) {
-			}
+			continue
+		}
+		ip, port, err := net.SplitHostPort(ep)
+		if err != nil {
+			glog.Errorf("Failed to parse endpoint: %v, error: %v", ep, err)
+			continue
+		}
+		portNum, err := strconv.Atoi(port)
+		if err != nil {
+			glog.Errorf("Failed to parse endpoint port %s, error: %v", port, err)
+			continue
+		}
 
-			delDest := &utilipvs.RealServer{
+		delDest := &utilipvs.RealServer{
-				Address: net.ParseIP(ip),
+			Address: net.ParseIP(ip),
-				Port:    uint16(portNum),
+			Port:    uint16(portNum),
-			}
+		}
-			err = proxier.ipvs.DeleteRealServer(appliedVirtualServer, delDest)
+
-			if err != nil {
+		glog.V(5).Infof("Using graceful delete to delete: %v", delDest)
-				glog.Errorf("Failed to delete destination: %v, error: %v", delDest, err)
+		err = proxier.gracefuldeleteManager.GracefulDeleteRS(appliedVirtualServer, delDest)
-				continue
+		if err != nil {
-			}
+			glog.Errorf("Failed to delete destination: %v, error: %v", delDest, err)
+			continue
 		}
 	}
 	return nil
@@ -1569,6 +1591,11 @@ func (proxier *Proxier) cleanLegacyService(activeServices map[string]bool, curre
 			// This service was not processed in the latest sync loop so before deleting it,
 			// make sure it does not fall within an excluded CIDR range.
 			okayToDelete := true
+			rsList, err := proxier.ipvs.GetRealServers(svc)
+			if len(rsList) != 0 && err == nil {
+				glog.V(5).Infof("Will not delete VS: %v, cause it have RS: %v", svc, rsList)
+				okayToDelete = false
+			}
 			for _, excludedCIDR := range proxier.excludeCIDRs {
 				// Any validation of this CIDR already should have occurred.
 				_, n, _ := net.ParseCIDR(excludedCIDR)

pkg/util/ipvs/ipvs.go

@@ -41,6 +41,8 @@ type Interface interface {
 	GetRealServers(*VirtualServer) ([]*RealServer, error)
 	// DeleteRealServer deletes the specified real server from the specified virtual server.
 	DeleteRealServer(*VirtualServer, *RealServer) error
+	// UpdateRealServer updates the specified real server from the specified virtual server.
+	UpdateRealServer(*VirtualServer, *RealServer) error
 }
 
 // VirtualServer is an user-oriented definition of an IPVS virtual server in its entirety.
@@ -91,9 +93,11 @@ func (svc *VirtualServer) String() string {
 
 // RealServer is an user-oriented definition of an IPVS real server in its entirety.
 type RealServer struct {
-	Address net.IP
+	Address      net.IP
-	Port    uint16
+	Port         uint16
-	Weight  int
+	Weight       int
+	ActiveConn   int
+	InactiveConn int
 }
 
 func (rs *RealServer) String() string {

pkg/util/ipvs/ipvs_linux.go

@@ -144,6 +144,18 @@ func (runner *runner) DeleteRealServer(vs *VirtualServer, rs *RealServer) error
 	return runner.ipvsHandle.DelDestination(svc, dst)
 }
 
+func (runner *runner) UpdateRealServer(vs *VirtualServer, rs *RealServer) error {
+	svc, err := toIPVSService(vs)
+	if err != nil {
+		return err
+	}
+	dst, err := toIPVSDestination(rs)
+	if err != nil {
+		return err
+	}
+	return runner.ipvsHandle.UpdateDestination(svc, dst)
+}
+
 // GetRealServers is part of ipvs.Interface.
 func (runner *runner) GetRealServers(vs *VirtualServer) ([]*RealServer, error) {
 	svc, err := toIPVSService(vs)
@@ -203,9 +215,11 @@ func toRealServer(dst *libipvs.Destination) (*RealServer, error) {
 		return nil, errors.New("ipvs destination should not be empty")
 	}
 	return &RealServer{
-		Address: dst.Address,
+		Address:      dst.Address,
-		Port:    dst.Port,
+		Port:         dst.Port,
-		Weight:  dst.Weight,
+		Weight:       dst.Weight,
+		ActiveConn:   dst.ActiveConnections,
+		InactiveConn: dst.InactiveConnections,
 	}, nil
 }
 

pkg/util/ipvs/ipvs_unsupported.go

@@ -68,4 +68,8 @@ func (runner *runner) DeleteRealServer(*VirtualServer, *RealServer) error {
 	return fmt.Errorf("IPVS not supported for this platform")
 }
 
+func (runner *runner) UpdateRealServer(*VirtualServer, *RealServer) error {
+	return fmt.Errorf("IPVS not supported for this platform")
+}
+
 var _ = Interface(&runner{})

pkg/util/ipvs/testing/fake.go

@@ -193,4 +193,13 @@ func (f *FakeIPVS) DeleteRealServer(serv *utilipvs.VirtualServer, dest *utilipvs
 	return nil
 }
 
+// UpdateRealServer is a fake implementation, it deletes the old real server then add new real server
+func (f *FakeIPVS) UpdateRealServer(serv *utilipvs.VirtualServer, dest *utilipvs.RealServer) error {
+	err := f.DeleteRealServer(serv, dest)
+	if err != nil {
+		return err
+	}
+	return f.AddRealServer(serv, dest)
+}
+
 var _ = utilipvs.Interface(&FakeIPVS{})

pkg/util/ipvs/testing/fake_test.go

@@ -124,9 +124,9 @@ func TestRealServer(t *testing.T) {
 		Protocol: string("TCP"),
 	}
 	rss := []*utilipvs.RealServer{
-		{net.ParseIP("172.16.2.1"), 8080, 1},
+		{Address: net.ParseIP("172.16.2.1"), Port: 8080, Weight: 1},
-		{net.ParseIP("172.16.2.2"), 8080, 2},
+		{Address: net.ParseIP("172.16.2.2"), Port: 8080, Weight: 2},
-		{net.ParseIP("172.16.2.3"), 8080, 3},
+		{Address: net.ParseIP("172.16.2.3"), Port: 8080, Weight: 3},
 	}
 	err := fake.AddVirtualServer(vs)
 	if err != nil {