瀏覽代碼

Distributed delete processing

Signed-off-by: Madhu Venugopal <madhu@docker.com>
Madhu Venugopal 10 年之前
父節點
當前提交
2677a461de
共有 6 個文件被更改,包括 100 次插入57 次删除
  1. 3 0
      libnetwork/controller.go
  2. 4 1
      libnetwork/datastore/datastore.go
  3. 1 1
      libnetwork/endpoint.go
  4. 0 9
      libnetwork/libnetwork_test.go
  5. 2 0
      libnetwork/network.go
  6. 90 46
      libnetwork/store.go

+ 3 - 0
libnetwork/controller.go

@@ -263,6 +263,9 @@ func (c *controller) addNetwork(n *network) error {
 	if err := d.CreateNetwork(n.id, n.generic); err != nil {
 	if err := d.CreateNetwork(n.id, n.generic); err != nil {
 		return err
 		return err
 	}
 	}
+	if err := n.watchEndpoints(); err != nil {
+		return err
+	}
 	c.Lock()
 	c.Lock()
 	c.networks[n.id] = n
 	c.networks[n.id] = n
 	c.Unlock()
 	c.Unlock()

+ 4 - 1
libnetwork/datastore/datastore.go

@@ -30,7 +30,10 @@ type DataStore interface {
 }
 }
 
 
 // ErrKeyModified is raised for an atomic update when the update is working on a stale state
 // ErrKeyModified is raised for an atomic update when the update is working on a stale state
-var ErrKeyModified = store.ErrKeyModified
+var (
+	ErrKeyModified = store.ErrKeyModified
+	ErrKeyNotFound = store.ErrKeyNotFound
+)
 
 
 type datastore struct {
 type datastore struct {
 	store store.Store
 	store store.Store

+ 1 - 1
libnetwork/endpoint.go

@@ -554,7 +554,7 @@ func (ep *endpoint) deleteEndpoint() error {
 	_, ok := n.endpoints[epid]
 	_, ok := n.endpoints[epid]
 	if !ok {
 	if !ok {
 		n.Unlock()
 		n.Unlock()
-		return &UnknownEndpointError{name: name, id: string(epid)}
+		return nil
 	}
 	}
 
 
 	nid := n.id
 	nid := n.id

+ 0 - 9
libnetwork/libnetwork_test.go

@@ -547,15 +547,6 @@ func TestUnknownEndpoint(t *testing.T) {
 		t.Fatal(err)
 		t.Fatal(err)
 	}
 	}
 
 
-	err = ep.Delete()
-	if err == nil {
-		t.Fatal("Expected to fail. But instead succeeded")
-	}
-
-	if _, ok := err.(*libnetwork.UnknownEndpointError); !ok {
-		t.Fatalf("Did not fail with expected error. Actual error: %v", err)
-	}
-
 	// Done testing. Now cleanup
 	// Done testing. Now cleanup
 	if err := network.Delete(); err != nil {
 	if err := network.Delete(); err != nil {
 		t.Fatal(err)
 		t.Fatal(err)

+ 2 - 0
libnetwork/network.go

@@ -62,6 +62,7 @@ type network struct {
 	endpoints   endpointTable
 	endpoints   endpointTable
 	generic     options.Generic
 	generic     options.Generic
 	dbIndex     uint64
 	dbIndex     uint64
+	stopWatchCh chan struct{}
 	sync.Mutex
 	sync.Mutex
 }
 }
 
 
@@ -248,6 +249,7 @@ func (n *network) deleteNetwork() error {
 		}
 		}
 		log.Warnf("driver error deleting network %s : %v", n.name, err)
 		log.Warnf("driver error deleting network %s : %v", n.name, err)
 	}
 	}
+	n.stopWatch()
 	return nil
 	return nil
 }
 }
 
 

+ 90 - 46
libnetwork/store.go

@@ -31,7 +31,7 @@ func (c *controller) initDataStore() error {
 	c.Lock()
 	c.Lock()
 	c.store = store
 	c.store = store
 	c.Unlock()
 	c.Unlock()
-	return c.watchStore()
+	return c.watchNetworks()
 }
 }
 
 
 func (c *controller) newNetworkFromStore(n *network) error {
 func (c *controller) newNetworkFromStore(n *network) error {
@@ -92,22 +92,6 @@ func (c *controller) newEndpointFromStore(key string, ep *endpoint) error {
 	n := ep.network
 	n := ep.network
 	id := ep.id
 	id := ep.id
 	ep.Unlock()
 	ep.Unlock()
-	if n == nil {
-		// Possibly the watch event for the network has not shown up yet
-		// Try to get network from the store
-		nid, err := networkIDFromEndpointKey(key, ep)
-		if err != nil {
-			return err
-		}
-		n, err = c.getNetworkFromStore(nid)
-		if err != nil {
-			return err
-		}
-		if err := c.newNetworkFromStore(n); err != nil {
-			return err
-		}
-		n = c.networks[nid]
-	}
 
 
 	_, err := n.EndpointByID(string(id))
 	_, err := n.EndpointByID(string(id))
 	if err != nil {
 	if err != nil {
@@ -170,7 +154,11 @@ func (c *controller) deleteEndpointFromStore(ep *endpoint) error {
 	return nil
 	return nil
 }
 }
 
 
-func (c *controller) watchStore() error {
+func (c *controller) watchNetworks() error {
+	if !c.validateDatastoreConfig() {
+		return nil
+	}
+
 	c.Lock()
 	c.Lock()
 	cs := c.store
 	cs := c.store
 	c.Unlock()
 	c.Unlock()
@@ -179,14 +167,17 @@ func (c *controller) watchStore() error {
 	if err != nil {
 	if err != nil {
 		return err
 		return err
 	}
 	}
-	epPairs, err := cs.KVStore().WatchTree(datastore.Key(datastore.EndpointKeyPrefix), nil)
-	if err != nil {
-		return err
-	}
 	go func() {
 	go func() {
 		for {
 		for {
 			select {
 			select {
 			case nws := <-nwPairs:
 			case nws := <-nwPairs:
+				c.Lock()
+				tmpview := networkTable{}
+				lview := c.networks
+				c.Unlock()
+				for k, v := range lview {
+					tmpview[k] = v
+				}
 				for _, kve := range nws {
 				for _, kve := range nws {
 					var n network
 					var n network
 					err := json.Unmarshal(kve.Value, &n)
 					err := json.Unmarshal(kve.Value, &n)
@@ -194,6 +185,7 @@ func (c *controller) watchStore() error {
 						log.Error(err)
 						log.Error(err)
 						continue
 						continue
 					}
 					}
+					delete(tmpview, n.id)
 					n.dbIndex = kve.LastIndex
 					n.dbIndex = kve.LastIndex
 					c.Lock()
 					c.Lock()
 					existing, ok := c.networks[n.id]
 					existing, ok := c.networks[n.id]
@@ -212,8 +204,59 @@ func (c *controller) watchStore() error {
 					if err = c.newNetworkFromStore(&n); err != nil {
 					if err = c.newNetworkFromStore(&n); err != nil {
 						log.Error(err)
 						log.Error(err)
 					}
 					}
+
 				}
 				}
+				// Delete processing
+				for k := range tmpview {
+					c.Lock()
+					existing, ok := c.networks[k]
+					c.Unlock()
+					if !ok {
+						continue
+					}
+					tmp := network{}
+					if err := c.store.GetObject(datastore.Key(existing.Key()...), &tmp); err != datastore.ErrKeyNotFound {
+						continue
+					}
+					if err := existing.deleteNetwork(); err != nil {
+						log.Debugf("Delete failed %s: %s", existing.name, err)
+					}
+				}
+			}
+		}
+	}()
+	return nil
+}
+
+func (n *network) watchEndpoints() error {
+	if !n.ctrlr.validateDatastoreConfig() {
+		return nil
+	}
+
+	n.Lock()
+	cs := n.ctrlr.store
+	tmp := endpoint{network: n}
+	n.stopWatchCh = make(chan struct{})
+	stopCh := n.stopWatchCh
+	n.Unlock()
+
+	epPairs, err := cs.KVStore().WatchTree(datastore.Key(tmp.KeyPrefix()...), stopCh)
+	if err != nil {
+		return err
+	}
+	go func() {
+		for {
+			select {
+			case <-stopCh:
+				return
 			case eps := <-epPairs:
 			case eps := <-epPairs:
+				n.Lock()
+				tmpview := endpointTable{}
+				lview := n.endpoints
+				n.Unlock()
+				for k, v := range lview {
+					tmpview[k] = v
+				}
 				for _, epe := range eps {
 				for _, epe := range eps {
 					var ep endpoint
 					var ep endpoint
 					err := json.Unmarshal(epe.Value, &ep)
 					err := json.Unmarshal(epe.Value, &ep)
@@ -221,22 +264,30 @@ func (c *controller) watchStore() error {
 						log.Error(err)
 						log.Error(err)
 						continue
 						continue
 					}
 					}
+					delete(tmpview, ep.id)
 					ep.dbIndex = epe.LastIndex
 					ep.dbIndex = epe.LastIndex
-					n, err := c.networkFromEndpointKey(epe.Key, &ep)
-					if err != nil {
-						if _, ok := err.(ErrNoSuchNetwork); !ok {
+					ep.network = n
+					if n.ctrlr.processEndpointUpdate(&ep) {
+						err = n.ctrlr.newEndpointFromStore(epe.Key, &ep)
+						if err != nil {
 							log.Error(err)
 							log.Error(err)
-							continue
 						}
 						}
 					}
 					}
-					if n != nil {
-						ep.network = n.(*network)
+				}
+				// Delete processing
+				for k := range tmpview {
+					n.Lock()
+					existing, ok := n.endpoints[k]
+					n.Unlock()
+					if !ok {
+						continue
+					}
+					tmp := endpoint{}
+					if err := cs.GetObject(datastore.Key(existing.Key()...), &tmp); err != datastore.ErrKeyNotFound {
+						continue
 					}
 					}
-					if c.processEndpointUpdate(&ep) {
-						err = c.newEndpointFromStore(epe.Key, &ep)
-						if err != nil {
-							log.Error(err)
-						}
+					if err := existing.deleteEndpoint(); err != nil {
+						log.Debugf("Delete failed %s: %s", existing.name, err)
 					}
 					}
 				}
 				}
 			}
 			}
@@ -245,20 +296,13 @@ func (c *controller) watchStore() error {
 	return nil
 	return nil
 }
 }
 
 
-func (c *controller) networkFromEndpointKey(key string, ep *endpoint) (Network, error) {
-	nid, err := networkIDFromEndpointKey(key, ep)
-	if err != nil {
-		return nil, err
-	}
-	return c.NetworkByID(string(nid))
-}
-
-func networkIDFromEndpointKey(key string, ep *endpoint) (types.UUID, error) {
-	eKey, err := datastore.ParseKey(key)
-	if err != nil {
-		return types.UUID(""), err
+func (n *network) stopWatch() {
+	n.Lock()
+	if n.stopWatchCh != nil {
+		close(n.stopWatchCh)
+		n.stopWatchCh = nil
 	}
 	}
-	return ep.networkIDFromKey(eKey)
+	n.Unlock()
 }
 }
 
 
 func (c *controller) processEndpointUpdate(ep *endpoint) bool {
 func (c *controller) processEndpointUpdate(ep *endpoint) bool {