0ct0pu5/moby
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #35422 from pradipd/lbfix

Browse source 
Move load balancer sandbox creation/deletion into libnetwork
This commit is contained in:
Madhu Venugopal 2017-12-01 14:08:48 -08:00 • committed by GitHub
commit 4bb2c2402d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 220 additions and 136 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

105
daemon/network.go
View file

@ -3,6 +3,7 @@ package daemon
import (
"fmt"
"net"
"runtime"
"sort"
"strings"
"sync"
@ -183,21 +184,14 @@ func (daemon *Daemon) setupIngress(create *clustertypes.NetworkCreateRequest, ip
// Otherwise continue down the call to create or recreate sandbox.
}
n, err := daemon.GetNetworkByID(create.ID)
_, err := daemon.GetNetworkByID(create.ID)
if err != nil {
logrus.Errorf("Failed getting ingress network by id after creating: %v", err)
}
if err = daemon.createLoadBalancerSandbox("ingress", create.ID, ip, n, libnetwork.OptionIngress()); err != nil {
logrus.Errorf("Failed creating load balancer sandbox for ingress network: %v", err)
}
}
func (daemon *Daemon) releaseIngress(id string) {
controller := daemon.netController
if err := controller.SandboxDestroy("ingress-sbox"); err != nil {
logrus.Errorf("Failed to delete ingress sandbox: %v", err)
}
if id == "" {
return
@ -209,13 +203,6 @@ func (daemon *Daemon) releaseIngress(id string) {
return
}
for _, ep := range n.Endpoints() {
if err := ep.Delete(true); err != nil {
logrus.Errorf("Failed to delete endpoint %s (%s): %v", ep.Name(), ep.ID(), err)
return
}
}
if err := n.Delete(); err != nil {
logrus.Errorf("Failed to delete ingress network %s: %v", n.ID(), err)
return
@ -270,34 +257,6 @@ func (daemon *Daemon) CreateNetwork(create types.NetworkCreateRequest) (*types.N
return resp, err
}
func (daemon *Daemon) createLoadBalancerSandbox(prefix, id string, ip net.IP, n libnetwork.Network, options ...libnetwork.SandboxOption) error {
c := daemon.netController
sandboxName := prefix + "-sbox"
sb, err := c.NewSandbox(sandboxName, options...)
if err != nil {
if _, ok := err.(networktypes.ForbiddenError); !ok {
return errors.Wrapf(err, "Failed creating %s sandbox", sandboxName)
}
return nil
}
endpointName := prefix + "-endpoint"
ep, err := n.CreateEndpoint(endpointName, libnetwork.CreateOptionIpam(ip, nil, nil, nil), libnetwork.CreateOptionLoadBalancer())
if err != nil {
return errors.Wrapf(err, "Failed creating %s in sandbox %s", endpointName, sandboxName)
}
if err := ep.Join(sb, nil); err != nil {
return errors.Wrapf(err, "Failed joining %s to sandbox %s", endpointName, sandboxName)
}
if err := sb.EnableService(); err != nil {
return errors.Wrapf(err, "Failed enabling service in %s sandbox", sandboxName)
}
return nil
}
func (daemon *Daemon) createNetwork(create types.NetworkCreateRequest, id string, agent bool) (*types.NetworkCreateResponse, error) {
if runconfig.IsPreDefinedNetwork(create.Name) && !agent {
err := fmt.Errorf("%s is a pre-defined network and cannot be created", create.Name)
@ -360,6 +319,15 @@ func (daemon *Daemon) createNetwork(create types.NetworkCreateRequest, id string
nwOptions = append(nwOptions, libnetwork.NetworkOptionConfigFrom(create.ConfigFrom.Network))
}
if agent && driver == "overlay" && (create.Ingress || runtime.GOOS == "windows") {
nodeIP, exists := daemon.GetAttachmentStore().GetIPForNetwork(id)
if !exists {
return nil, fmt.Errorf("Failed to find a load balancer IP to use for network: %v", id)
}
nwOptions = append(nwOptions, libnetwork.NetworkOptionLBEndpoint(nodeIP))
}
n, err := c.NewNetwork(driver, create.Name, id, nwOptions...)
if err != nil {
if _, ok := err.(libnetwork.ErrDataStoreNotInitialized); ok {
@ -375,18 +343,6 @@ func (daemon *Daemon) createNetwork(create types.NetworkCreateRequest, id string
}
daemon.LogNetworkEvent(n, "create")
if agent && !n.Info().Ingress() && n.Type() == "overlay" {
nodeIP, exists := daemon.GetAttachmentStore().GetIPForNetwork(id)
if !exists {
return nil, fmt.Errorf("Failed to find a load balancer IP to use for network: %v", id)
}
if err := daemon.createLoadBalancerSandbox(create.Name, id, nodeIP, n); err != nil {
return nil, err
}
}
return &types.NetworkCreateResponse{
ID: n.ID(),
Warning: warning,
@ -517,43 +473,16 @@ func (daemon *Daemon) DeleteNetwork(networkID string) error {
return daemon.deleteNetwork(networkID, false)
}
func (daemon *Daemon) deleteLoadBalancerSandbox(n libnetwork.Network) {
controller := daemon.netController
//The only endpoint left should be the LB endpoint (nw.Name() + "-endpoint")
endpoints := n.Endpoints()
if len(endpoints) == 1 {
sandboxName := n.Name() + "-sbox"
info := endpoints[0].Info()
if info != nil {
sb := info.Sandbox()
if sb != nil {
if err := sb.DisableService(); err != nil {
logrus.Warnf("Failed to disable service on sandbox %s: %v", sandboxName, err)
//Ignore error and attempt to delete the load balancer endpoint
}
}
}
if err := endpoints[0].Delete(true); err != nil {
logrus.Warnf("Failed to delete endpoint %s (%s) in %s: %v", endpoints[0].Name(), endpoints[0].ID(), sandboxName, err)
//Ignore error and attempt to delete the sandbox.
}
if err := controller.SandboxDestroy(sandboxName); err != nil {
logrus.Warnf("Failed to delete %s sandbox: %v", sandboxName, err)
//Ignore error and attempt to delete the network.
}
}
}
func (daemon *Daemon) deleteNetwork(networkID string, dynamic bool) error {
nw, err := daemon.FindNetwork(networkID)
if err != nil {
return err
}
if nw.Info().Ingress() {
return nil
}
if runconfig.IsPreDefinedNetwork(nw.Name()) && !dynamic {
err := fmt.Errorf("%s is a pre-defined network and cannot be removed", nw.Name())
return notAllowedError{err}
@ -569,10 +498,6 @@ func (daemon *Daemon) deleteNetwork(networkID string, dynamic bool) error {
return notAllowedError{err}
}
if !nw.Info().Ingress() && nw.Type() == "overlay" {
daemon.deleteLoadBalancerSandbox(nw)
}
if err := nw.Delete(); err != nil {
return err
}

43
integration/service/create_test.go
View file

@ -11,12 +11,11 @@ import (
"github.com/docker/docker/client"
"github.com/docker/docker/integration-cli/request"
"github.com/gotestyourself/gotestyourself/poll"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/net/context"
)
func TestCreateWithLBSandbox(t *testing.T) {
func TestCreateServiceMultipleTimes(t *testing.T) {
defer setupTest(t)()
d := newSwarm(t)
defer d.Stop(t)
@ -33,9 +32,8 @@ func TestCreateWithLBSandbox(t *testing.T) {
require.NoError(t, err)
overlayID := netResp.ID
var instances uint64 = 1
var instances uint64 = 4
serviceSpec := swarmServiceSpec("TestService", instances)
serviceSpec.TaskTemplate.Networks = append(serviceSpec.TaskTemplate.Networks, swarm.NetworkAttachmentConfig{Target: overlayName})
serviceResp, err := client.ServiceCreate(context.Background(), serviceSpec, types.ServiceCreateOptions{
@ -56,14 +54,26 @@ func TestCreateWithLBSandbox(t *testing.T) {
_, _, err = client.ServiceInspectWithRaw(context.Background(), serviceID, types.ServiceInspectOptions{})
require.NoError(t, err)
network, err := client.NetworkInspect(context.Background(), overlayID, types.NetworkInspectOptions{})
require.NoError(t, err)
assert.Contains(t, network.Containers, overlayName+"-sbox")
err = client.ServiceRemove(context.Background(), serviceID)
require.NoError(t, err)
poll.WaitOn(t, serviceIsRemoved(client, serviceID), pollSettings)
poll.WaitOn(t, noTasks(client), pollSettings)
serviceResp, err = client.ServiceCreate(context.Background(), serviceSpec, types.ServiceCreateOptions{
QueryRegistry: false,
})
require.NoError(t, err)
serviceID2 := serviceResp.ID
poll.WaitOn(t, serviceRunningTasksCount(client, serviceID2, instances), pollSettings)
err = client.ServiceRemove(context.Background(), serviceID2)
require.NoError(t, err)
poll.WaitOn(t, serviceIsRemoved(client, serviceID2), pollSettings)
poll.WaitOn(t, noTasks(client), pollSettings)
err = client.NetworkRemove(context.Background(), overlayID)
require.NoError(t, err)
@ -112,6 +122,23 @@ func serviceRunningTasksCount(client client.ServiceAPIClient, serviceID string,
}
}
func noTasks(client client.ServiceAPIClient) func(log poll.LogT) poll.Result {
return func(log poll.LogT) poll.Result {
filter := filters.NewArgs()
tasks, err := client.TaskList(context.Background(), types.TaskListOptions{
Filters: filter,
})
switch {
case err != nil:
return poll.Error(err)
case len(tasks) == 0:
return poll.Success()
default:
return poll.Continue("task count at %d waiting for 0", len(tasks))
}
}
}
func serviceIsRemoved(client client.ServiceAPIClient, serviceID string) func(log poll.LogT) poll.Result {
return func(log poll.LogT) poll.Result {
filter := filters.NewArgs()

2
vendor.conf
View file

@ -30,7 +30,7 @@ github.com/moby/buildkit aaff9d591ef128560018433fe61beb802e149de8
github.com/tonistiigi/fsutil dea3a0da73aee887fc02142d995be764106ac5e2
#get libnetwork packages
github.com/docker/libnetwork f7d21337cf1eb628ad54eecac0881fa23ec266df
github.com/docker/libnetwork 64ae58878fc8f95e4a167499d654e13fa36abdc7
github.com/docker/go-events 9461782956ad83b30282bf90e31fa6a70c255ba9
github.com/armon/go-radix e39d623f12e8e41c7b5529e9a9dd67a1e2261f80
github.com/armon/go-metrics eb0af217e5e9747e41dd5303755356b62d28e3ec

23
vendor/github.com/docker/libnetwork/controller.go generated vendored
View file

@ -837,11 +837,34 @@ addToStore:
if err = c.updateToStore(network); err != nil {
return nil, err
}
defer func() {
if err != nil {
if e := c.deleteFromStore(network); e != nil {
logrus.Warnf("could not rollback from store, network %v on failure (%v): %v", network, err, e)
}
}
}()
if network.configOnly {
return network, nil
}
joinCluster(network)
defer func() {
if err != nil {
network.cancelDriverWatches()
if e := network.leaveCluster(); e != nil {
logrus.Warnf("Failed to leave agent cluster on network %s on failure (%v): %v", network.name, err, e)
}
}
}()
if len(network.loadBalancerIP) != 0 {
if err = network.createLoadBalancerSandbox(); err != nil {
return nil, err
}
}
if !c.isDistributedControl() {
c.Lock()
arrangeIngressFilterRule()

181
vendor/github.com/docker/libnetwork/network.go generated vendored
View file

@ -199,39 +199,40 @@ func (i *IpamInfo) UnmarshalJSON(data []byte) error {
}
type network struct {
ctrlr *controller
name string
networkType string
id string
created time.Time
scope string // network data scope
labels map[string]string
ipamType string
ipamOptions map[string]string
addrSpace string
ipamV4Config []*IpamConf
ipamV6Config []*IpamConf
ipamV4Info []*IpamInfo
ipamV6Info []*IpamInfo
enableIPv6 bool
postIPv6 bool
epCnt *endpointCnt
generic options.Generic
dbIndex uint64
dbExists bool
persist bool
stopWatchCh chan struct{}
drvOnce *sync.Once
resolverOnce sync.Once
resolver []Resolver
internal bool
attachable bool
inDelete bool
ingress bool
driverTables []networkDBTable
dynamic bool
configOnly bool
configFrom string
ctrlr *controller
name string
networkType string
id string
created time.Time
scope string // network data scope
labels map[string]string
ipamType string
ipamOptions map[string]string
addrSpace string
ipamV4Config []*IpamConf
ipamV6Config []*IpamConf
ipamV4Info []*IpamInfo
ipamV6Info []*IpamInfo
enableIPv6 bool
postIPv6 bool
epCnt *endpointCnt
generic options.Generic
dbIndex uint64
dbExists bool
persist bool
stopWatchCh chan struct{}
drvOnce *sync.Once
resolverOnce sync.Once
resolver []Resolver
internal bool
attachable bool
inDelete bool
ingress bool
driverTables []networkDBTable
dynamic bool
configOnly bool
configFrom string
loadBalancerIP net.IP
sync.Mutex
}
@ -473,6 +474,7 @@ func (n *network) CopyTo(o datastore.KVObject) error {
dstN.ingress = n.ingress
dstN.configOnly = n.configOnly
dstN.configFrom = n.configFrom
dstN.loadBalancerIP = n.loadBalancerIP
// copy labels
if dstN.labels == nil {
@ -589,6 +591,7 @@ func (n *network) MarshalJSON() ([]byte, error) {
netMap["ingress"] = n.ingress
netMap["configOnly"] = n.configOnly
netMap["configFrom"] = n.configFrom
netMap["loadBalancerIP"] = n.loadBalancerIP
return json.Marshal(netMap)
}
@ -699,6 +702,9 @@ func (n *network) UnmarshalJSON(b []byte) (err error) {
if v, ok := netMap["configFrom"]; ok {
n.configFrom = v.(string)
}
if v, ok := netMap["loadBalancerIP"]; ok {
n.loadBalancerIP = net.ParseIP(v.(string))
}
// Reconcile old networks with the recently added `--ipv6` flag
if !n.enableIPv6 {
n.enableIPv6 = len(n.ipamV6Info) > 0
@ -799,6 +805,13 @@ func NetworkOptionIpam(ipamDriver string, addrSpace string, ipV4 []*IpamConf, ip
}
}
// NetworkOptionLBEndpoint function returns an option setter for the configuration of the load balancer endpoint for this network
func NetworkOptionLBEndpoint(ip net.IP) NetworkOption {
return func(n *network) {
n.loadBalancerIP = ip
}
}
// NetworkOptionDriverOpts function returns an option setter for any driver parameter described by a map
func NetworkOptionDriverOpts(opts map[string]string) NetworkOption {
return func(n *network) {
@ -944,6 +957,18 @@ func (n *network) delete(force bool) error {
return &UnknownNetworkError{name: name, id: id}
}
if len(n.loadBalancerIP) != 0 {
endpoints := n.Endpoints()
if force || len(endpoints) == 1 {
n.deleteLoadBalancerSandbox()
}
//Reload the network from the store to update the epcnt.
n, err = c.getNetworkFromStore(id)
if err != nil {
return &UnknownNetworkError{name: name, id: id}
}
}
if !force && n.getEpCnt().EndpointCnt() != 0 {
if n.configOnly {
return types.ForbiddenErrorf("configuration network %q is in use", n.Name())
@ -1071,12 +1096,19 @@ func (n *network) CreateEndpoint(name string, options ...EndpointOption) (Endpoi
return nil, types.ForbiddenErrorf("endpoint with name %s already exists in network %s", name, n.Name())
}
ep := &endpoint{name: name, generic: make(map[string]interface{}), iface: &endpointInterface{}}
ep.id = stringid.GenerateRandomID()
n.ctrlr.networkLocker.Lock(n.id)
defer n.ctrlr.networkLocker.Unlock(n.id)
return n.createEndpoint(name, options...)
}
func (n *network) createEndpoint(name string, options ...EndpointOption) (Endpoint, error) {
var err error
ep := &endpoint{name: name, generic: make(map[string]interface{}), iface: &endpointInterface{}}
ep.id = stringid.GenerateRandomID()
// Initialize ep.network with a possibly stale copy of n. We need this to get network from
// store. But once we get it from store we will have the most uptodate copy possibly.
ep.network = n
@ -2021,3 +2053,80 @@ func (c *controller) getConfigNetwork(name string) (*network, error) {
return n.(*network), nil
}
func (n *network) createLoadBalancerSandbox() error {
sandboxName := n.name + "-sbox"
sbOptions := []SandboxOption{}
if n.ingress {
sbOptions = append(sbOptions, OptionIngress())
}
sb, err := n.ctrlr.NewSandbox(sandboxName, sbOptions...)
if err != nil {
return err
}
defer func() {
if err != nil {
if e := n.ctrlr.SandboxDestroy(sandboxName); e != nil {
logrus.Warnf("could not delete sandbox %s on failure on failure (%v): %v", sandboxName, err, e)
}
}
}()
endpointName := n.name + "-endpoint"
epOptions := []EndpointOption{
CreateOptionIpam(n.loadBalancerIP, nil, nil, nil),
CreateOptionLoadBalancer(),
}
ep, err := n.createEndpoint(endpointName, epOptions...)
if err != nil {
return err
}
defer func() {
if err != nil {
if e := ep.Delete(true); e != nil {
logrus.Warnf("could not delete endpoint %s on failure on failure (%v): %v", endpointName, err, e)
}
}
}()
if err := ep.Join(sb, nil); err != nil {
return err
}
return sb.EnableService()
}
func (n *network) deleteLoadBalancerSandbox() {
n.Lock()
c := n.ctrlr
name := n.name
n.Unlock()
endpointName := name + "-endpoint"
sandboxName := name + "-sbox"
endpoint, err := n.EndpointByName(endpointName)
if err != nil {
logrus.Warnf("Failed to find load balancer endpoint %s on network %s: %v", endpointName, name, err)
} else {
info := endpoint.Info()
if info != nil {
sb := info.Sandbox()
if sb != nil {
if err := sb.DisableService(); err != nil {
logrus.Warnf("Failed to disable service on sandbox %s: %v", sandboxName, err)
//Ignore error and attempt to delete the load balancer endpoint
}
}
}
if err := endpoint.Delete(true); err != nil {
logrus.Warnf("Failed to delete endpoint %s (%s) in %s: %v", endpoint.Name(), endpoint.ID(), sandboxName, err)
//Ignore error and attempt to delete the sandbox.
}
}
if err := c.SandboxDestroy(sandboxName); err != nil {
logrus.Warnf("Failed to delete %s sandbox: %v", sandboxName, err)
}
}

2
vendor/github.com/docker/libnetwork/store.go generated vendored
View file

@ -256,7 +256,7 @@ retry:
if err := cs.GetObject(datastore.Key(kvObject.Key()...), kvObject); err != nil {
return fmt.Errorf("could not update the kvobject to latest when trying to delete: %v", err)
}
logrus.Errorf("Error (%v) deleting object %v, retrying....", err, kvObject.Key())
logrus.Warnf("Error (%v) deleting object %v, retrying....", err, kvObject.Key())
goto retry
}
return err