Merge pull request #862 from mavenugo/epcleanup

Force endpoint delete

libnetwork/api/api.go

@@ -529,7 +529,7 @@ func procDeleteEndpoint(c libnetwork.NetworkController, vars map[string]string,
 		return nil, errRsp
 	}
 
-	err := ep.Delete()
+	err := ep.Delete(false)
 	if err != nil {
 		return nil, convertNetworkError(err)
 	}
@@ -641,13 +641,22 @@ func procPublishService(c libnetwork.NetworkController, vars map[string]string,
 }
 
 func procUnpublishService(c libnetwork.NetworkController, vars map[string]string, body []byte) (interface{}, *responseStatus) {
+	var sd serviceDelete
+
+	if body != nil {
+		err := json.Unmarshal(body, &sd)
+		if err != nil {
+			return "", &responseStatus{Status: "Invalid body: " + err.Error(), StatusCode: http.StatusBadRequest}
+		}
+	}
+
 	epT, epBy := detectEndpointTarget(vars)
 	sv, errRsp := findService(c, epT, epBy)
 	if !errRsp.isOK() {
 		return nil, errRsp
 	}
-	err := sv.Delete()
-	if err != nil {
+
+	if err := sv.Delete(sd.Force); err != nil {
 		return nil, endpointToService(convertNetworkError(err))
 	}
 	return nil, &successResponse

libnetwork/api/api_test.go

@@ -690,15 +690,15 @@ func TestProcGetServices(t *testing.T) {
 	}
 
 	delete(vars, urlEpPID)
-	err = ep11.Delete()
+	err = ep11.Delete(false)
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = ep12.Delete()
+	err = ep12.Delete(false)
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = ep21.Delete()
+	err = ep21.Delete(false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -1014,7 +1014,7 @@ func TestAttachDetachBackend(t *testing.T) {
 		t.Fatalf("Did not find expected sandbox. Got %v", sb)
 	}
 
-	err = ep1.Delete()
+	err = ep1.Delete(false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -1495,7 +1495,7 @@ func TestFindEndpointUtil(t *testing.T) {
 		t.Fatalf("Diffenrent queries returned different endpoints")
 	}
 
-	ep.Delete()
+	ep.Delete(false)
 
 	_, errRsp = findEndpoint(c, nid, "secondEp", byID, byName)
 	if errRsp == &successResponse {

libnetwork/api/types.go

@@ -76,6 +76,12 @@ type servicePublish struct {
 	PortMapping  []types.PortBinding   `json:"port_mapping"`
 }
 
+// serviceDelete represents the body of the "unpublish service" http request message
+type serviceDelete struct {
+	Name  string `json:"name"`
+	Force bool   `json:"force"`
+}
+
 // extraHost represents the extra host object
 type extraHost struct {
 	Name    string `json:"name"`

libnetwork/client/service.go

@@ -191,6 +191,7 @@ func (cli *NetworkCli) CmdServicePublish(chain string, args ...string) error {
 // CmdServiceUnpublish handles service delete UI
 func (cli *NetworkCli) CmdServiceUnpublish(chain string, args ...string) error {
 	cmd := cli.Subcmd(chain, "unpublish", "SERVICE[.NETWORK]", "Removes a service", false)
+	force := cmd.Bool([]string{"f", "-force"}, false, "force unpublish service")
 	cmd.Require(flag.Exact, 1)
 	err := cmd.ParseFlags(args, true)
 	if err != nil {
@@ -203,7 +204,8 @@ func (cli *NetworkCli) CmdServiceUnpublish(chain string, args ...string) error {
 		return err
 	}
 
-	_, _, err = readBody(cli.call("DELETE", "/services/"+serviceID, nil, nil))
+	sd := serviceDelete{Name: sn, Force: *force}
+	_, _, err = readBody(cli.call("DELETE", "/services/"+serviceID, sd, nil))
 
 	return err
 }

libnetwork/client/types.go

@@ -49,6 +49,12 @@ type serviceCreate struct {
 	PortMapping  []types.PortBinding   `json:"port_mapping"`
 }
 
+// serviceDelete represents the body of the "unpublish service" http request message
+type serviceDelete struct {
+	Name  string `json:"name"`
+	Force bool   `json:"force"`
+}
+
 // serviceAttach represents the expected body of the "attach/detach sandbox to/from service" http request messages
 type serviceAttach struct {
 	SandboxID string   `json:"sandbox_id"`

libnetwork/controller.go

@@ -216,6 +216,31 @@ func (c *controller) validateHostDiscoveryConfig() bool {
 	return true
 }
 
+func (c *controller) clusterHostID() string {
+	c.Lock()
+	defer c.Unlock()
+	if c.cfg == nil || c.cfg.Cluster.Address == "" {
+		return ""
+	}
+	addr := strings.Split(c.cfg.Cluster.Address, ":")
+	return addr[0]
+}
+
+func (c *controller) isNodeAlive(node string) bool {
+	if c.discovery == nil {
+		return false
+	}
+
+	nodes := c.discovery.Fetch()
+	for _, n := range nodes {
+		if n.String() == node {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (c *controller) initDiscovery(watcher discovery.Watcher) error {
 	if c.cfg == nil {
 		return fmt.Errorf("discovery initialization requires a valid configuration")

libnetwork/default_gateway.go

@@ -87,7 +87,7 @@ func (sb *sandbox) clearDefaultGW() error {
 	if err := ep.sbLeave(sb); err != nil {
 		return fmt.Errorf("container %s: endpoint leaving GW Network failed: %v", sb.containerID, err)
 	}
-	if err := ep.Delete(); err != nil {
+	if err := ep.Delete(false); err != nil {
 		return fmt.Errorf("container %s: deleting endpoint on GW Network failed: %v", sb.containerID, err)
 	}
 	return nil

libnetwork/endpoint.go

@@ -41,7 +41,7 @@ type Endpoint interface {
 	DriverInfo() (map[string]interface{}, error)
 
 	// Delete and detaches this endpoint from the network.
-	Delete() error
+	Delete(force bool) error
 }
 
 // EndpointOption is a option setter function type used to pass varios options to Network
@@ -56,6 +56,7 @@ type endpoint struct {
 	iface             *endpointInterface
 	joinInfo          *endpointJoinInfo
 	sandboxID         string
+	locator           string
 	exposedPorts      []types.TransportPort
 	anonymous         bool
 	disableResolution bool
@@ -84,6 +85,7 @@ func (ep *endpoint) MarshalJSON() ([]byte, error) {
 		epMap["generic"] = ep.generic
 	}
 	epMap["sandbox"] = ep.sandboxID
+	epMap["locator"] = ep.locator
 	epMap["anonymous"] = ep.anonymous
 	epMap["disableResolution"] = ep.disableResolution
 	epMap["myAliases"] = ep.myAliases
@@ -167,6 +169,9 @@ func (ep *endpoint) UnmarshalJSON(b []byte) (err error) {
 	if v, ok := epMap["disableResolution"]; ok {
 		ep.disableResolution = v.(bool)
 	}
+	if l, ok := epMap["locator"]; ok {
+		ep.locator = l.(string)
+	}
 	ma, _ := json.Marshal(epMap["myAliases"])
 	var myAliases []string
 	json.Unmarshal(ma, &myAliases)
@@ -186,6 +191,7 @@ func (ep *endpoint) CopyTo(o datastore.KVObject) error {
 	dstEp.name = ep.name
 	dstEp.id = ep.id
 	dstEp.sandboxID = ep.sandboxID
+	dstEp.locator = ep.locator
 	dstEp.dbIndex = ep.dbIndex
 	dstEp.dbExists = ep.dbExists
 	dstEp.anonymous = ep.anonymous
@@ -600,7 +606,23 @@ func (ep *endpoint) sbLeave(sbox Sandbox, options ...EndpointOption) error {
 	return sb.clearDefaultGW()
 }
 
-func (ep *endpoint) Delete() error {
+func (n *network) validateForceDelete(locator string) error {
+	if n.Scope() == datastore.LocalScope {
+		return nil
+	}
+
+	if locator == "" {
+		return fmt.Errorf("invalid endpoint locator identifier")
+	}
+
+	if n.getController().isNodeAlive(locator) {
+		return fmt.Errorf("the remote host %s hosting the container is alive", locator)
+	}
+
+	return nil
+}
+
+func (ep *endpoint) Delete(force bool) error {
 	var err error
 	n, err := ep.getNetworkFromStore()
 	if err != nil {
@@ -615,18 +637,33 @@ func (ep *endpoint) Delete() error {
 	ep.Lock()
 	epid := ep.id
 	name := ep.name
-	sb, _ := n.getController().SandboxByID(ep.sandboxID)
-	if sb != nil {
-		ep.Unlock()
+	sbid := ep.sandboxID
+	locator := ep.locator
+	ep.Unlock()
+
+	if force {
+		if err = n.validateForceDelete(locator); err != nil {
+			return fmt.Errorf("unable to force delete endpoint %s: %v", name, err)
+		}
+	}
+
+	sb, _ := n.getController().SandboxByID(sbid)
+	if sb != nil && !force {
 		return &ActiveContainerError{name: name, id: epid}
 	}
-	ep.Unlock()
+
+	if sb != nil {
+		if e := ep.sbLeave(sb); e != nil {
+			log.Warnf("failed to leave sandbox for endpoint %s : %v", name, e)
+		}
+	}
 
 	if err = n.getController().deleteFromStore(ep); err != nil {
 		return err
 	}
+
 	defer func() {
-		if err != nil {
+		if err != nil && !force {
 			ep.dbExists = false
 			if e := n.getController().updateToStore(ep); e != nil {
 				log.Warnf("failed to recreate endpoint in store %s : %v", name, e)
@@ -634,11 +671,11 @@ func (ep *endpoint) Delete() error {
 		}
 	}()
 
-	if err = n.getEpCnt().DecEndpointCnt(); err != nil {
+	if err = n.getEpCnt().DecEndpointCnt(); err != nil && !force {
 		return err
 	}
 	defer func() {
-		if err != nil {
+		if err != nil && !force {
 			if e := n.getEpCnt().IncEndpointCnt(); e != nil {
 				log.Warnf("failed to update network %s : %v", n.name, e)
 			}
@@ -648,7 +685,7 @@ func (ep *endpoint) Delete() error {
 	// unwatch for service records
 	n.getController().unWatchSvcRecord(ep)
 
-	if err = ep.deleteEndpoint(); err != nil {
+	if err = ep.deleteEndpoint(); err != nil && !force {
 		return err
 	}
 
@@ -923,7 +960,7 @@ func (c *controller) cleanupLocalEndpoints() {
 		}
 
 		for _, ep := range epl {
-			if err := ep.Delete(); err != nil {
+			if err := ep.Delete(false); err != nil {
 				log.Warnf("Could not delete local endpoint %s during endpoint cleanup: %v", ep.name, err)
 			}
 		}

libnetwork/libnetwork_internal_test.go

@@ -407,7 +407,7 @@ func TestIpamReleaseOnNetDriverFailures(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer ep.Delete()
+	defer ep.Delete(false)
 
 	expectedIP, _ := types.ParseCIDR("10.34.0.1/16")
 	if !types.CompareIPNet(ep.Info().Iface().Address(), expectedIP) {

libnetwork/libnetwork_test.go

@@ -135,7 +135,7 @@ func TestNull(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := ep.Delete(); err != nil {
+	if err := ep.Delete(false); err != nil {
 		t.Fatal(err)
 	}
 
@@ -213,11 +213,11 @@ func TestHost(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := ep1.Delete(); err != nil {
+	if err := ep1.Delete(false); err != nil {
 		t.Fatal(err)
 	}
 
-	if err := ep2.Delete(); err != nil {
+	if err := ep2.Delete(false); err != nil {
 		t.Fatal(err)
 	}
 
@@ -249,7 +249,7 @@ func TestHost(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if err := ep3.Delete(); err != nil {
+	if err := ep3.Delete(false); err != nil {
 		t.Fatal(err)
 	}
 
@@ -305,7 +305,7 @@ func TestBridge(t *testing.T) {
 		t.Fatalf("Incomplete data for port mapping in endpoint operational data: %d", len(pm))
 	}
 
-	if err := ep.Delete(); err != nil {
+	if err := ep.Delete(false); err != nil {
 		t.Fatal(err)
 	}
 
@@ -358,7 +358,7 @@ func TestBridgeIpv6FromMac(t *testing.T) {
 		t.Fatalf("Expected %v. Got: %v", expIP, iface.AddressIPv6())
 	}
 
-	if err := ep.Delete(); err != nil {
+	if err := ep.Delete(false); err != nil {
 		t.Fatal(err)
 	}
 
@@ -514,7 +514,7 @@ func TestDeleteNetworkWithActiveEndpoints(t *testing.T) {
 	}
 
 	// Done testing. Now cleanup.
-	if err := ep.Delete(); err != nil {
+	if err := ep.Delete(false); err != nil {
 		t.Fatal(err)
 	}
 
@@ -586,7 +586,7 @@ func TestUnknownEndpoint(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = ep.Delete()
+	err = ep.Delete(false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -624,7 +624,7 @@ func TestNetworkEndpointsWalkers(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep11.Delete(); err != nil {
+		if err := ep11.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -634,7 +634,7 @@ func TestNetworkEndpointsWalkers(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep12.Delete(); err != nil {
+		if err := ep12.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -752,7 +752,7 @@ func TestDuplicateEndpoint(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep.Delete(); err != nil {
+		if err := ep.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -761,7 +761,7 @@ func TestDuplicateEndpoint(t *testing.T) {
 	defer func() {
 		// Cleanup ep2 as well, else network cleanup might fail for failure cases
 		if ep2 != nil {
-			if err := ep2.Delete(); err != nil {
+			if err := ep2.Delete(false); err != nil {
 				t.Fatal(err)
 			}
 		}
@@ -904,7 +904,7 @@ func TestNetworkQuery(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep11.Delete(); err != nil {
+		if err := ep11.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -914,7 +914,7 @@ func TestNetworkQuery(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep12.Delete(); err != nil {
+		if err := ep12.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -1032,7 +1032,7 @@ func TestEndpointJoin(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep1.Delete(); err != nil {
+		if err := ep1.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -1150,7 +1150,7 @@ func TestEndpointJoin(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep2.Delete(); err != nil {
+		if err := ep2.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -1253,7 +1253,7 @@ func externalKeyTest(t *testing.T, reexec bool) {
 		t.Fatal(err)
 	}
 	defer func() {
-		err = ep.Delete()
+		err = ep.Delete(false)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -1264,7 +1264,7 @@ func externalKeyTest(t *testing.T, reexec bool) {
 		t.Fatal(err)
 	}
 	defer func() {
-		err = ep2.Delete()
+		err = ep2.Delete(false)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -1402,7 +1402,7 @@ func TestEndpointDeleteWithActiveContainer(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		err = ep.Delete()
+		err = ep.Delete(false)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -1431,7 +1431,7 @@ func TestEndpointDeleteWithActiveContainer(t *testing.T) {
 		}
 	}()
 
-	err = ep.Delete()
+	err = ep.Delete(false)
 	if err == nil {
 		t.Fatal("Expected to fail. But instead succeeded")
 	}
@@ -1465,7 +1465,7 @@ func TestEndpointMultipleJoins(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep.Delete(); err != nil {
+		if err := ep.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -1589,7 +1589,7 @@ func TestontainerInvalidLeave(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
-		if err := ep.Delete(); err != nil {
+		if err := ep.Delete(false); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -2334,7 +2334,7 @@ func runParallelTests(t *testing.T, thrNumber int) {
 			t.Fatal(err)
 		}
 	} else {
-		err = ep.Delete()
+		err = ep.Delete(false)
 		if err != nil {
 			t.Fatal(err)
 		}

libnetwork/network.go

@@ -681,6 +681,7 @@ func (n *network) CreateEndpoint(name string, options ...EndpointOption) (Endpoi
 	// 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 possible.
 	ep.network = n
+	ep.locator = n.getController().clusterHostID()
 	ep.network, err = ep.getNetworkFromStore()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get network during CreateEndpoint: %v", err)

libnetwork/sandbox.go

@@ -198,7 +198,7 @@ func (sb *sandbox) Delete() error {
 			log.Warnf("Failed detaching sandbox %s from endpoint %s: %v\n", sb.ID(), ep.ID(), err)
 		}
 
-		if err := ep.Delete(); err != nil {
+		if err := ep.Delete(false); err != nil {
 			log.Warnf("Failed deleting endpoint %s: %v\n", ep.ID(), err)
 		}
 	}

libnetwork/test/integration/dnet/overlay-consul.bats

@@ -25,6 +25,46 @@ load helpers
     test_overlay consul skip_add
 }
 
+@test "Test overlay network with dnet ungraceful shutdown" {
+    skip_for_circleci
+    dnet_cmd $(inst_id2port 1) network create -d overlay multihost
+    start=1
+    end=3
+    for i in `seq ${start} ${end}`;
+    do
+        dnet_cmd $(inst_id2port $i) container create container_${i}
+        net_connect ${i} container_${i} multihost
+    done
+
+    hrun runc $(dnet_container_name 1 consul) $(get_sbox_id 1 container_1) "ifconfig eth0"
+    container_1_ip=$(echo ${output} | grep 'inet addr:' | cut -d: -f2 | awk '{ print $1}')
+
+    # forcefully unpublish the service from dnet2 when dnet1 is alive.
+    # operation must fail
+    set +e
+    dnet_cmd $(inst_id2port 2) service unpublish -f container_1.multihost
+    status="$?"
+    set -e
+    [ "${status}" -ne 0 ]
+
+    # ungracefully kill dnet-1-consul container
+    docker rm -f dnet-1-consul
+    # sleep for 60 seconds to make sure the discovery catches up
+    sleep 60
+
+    # forcefully unpublish the service from dnet2 when dnet1 is dead.
+    dnet_cmd $(inst_id2port 2) service unpublish -f container_1.multihost
+    dnet_cmd $(inst_id2port 2) container create container_1
+    net_connect 2 container_1 multihost
+
+    hrun runc $(dnet_container_name 2 consul) $(get_sbox_id 2 container_1) "ifconfig eth0"
+    container_1_new_ip=$(echo ${output} | grep 'inet addr:' | cut -d: -f2 | awk '{ print $1}')
+
+    if [ "$container_1_ip" != "$container_1_new_ip" ]; then
+        exit 1
+    fi
+}
+
 @test "Test overlay network internal network with consul" {
     skip_for_circleci
     test_overlay consul internal