Merge pull request #1416 from sanimej/health

Add sandbox method for task insertion to service LB and service disovery

libnetwork/endpoint.go

@@ -74,6 +74,7 @@ type endpoint struct {
 	ingressPorts      []*PortConfig
 	dbIndex           uint64
 	dbExists          bool
+	serviceEnabled    bool
 	sync.Mutex
 }
 
@@ -303,6 +304,24 @@ func (ep *endpoint) isAnonymous() bool {
 	return ep.anonymous
 }
 
+// CompareAndSwap ep's serviceEnabled. If its in oldState, set it to newState
+// and return true.  If its not in oldState return false
+func (ep *endpoint) casServiceEnabled(oldState, newState bool) bool {
+	ep.Lock()
+	defer ep.Unlock()
+	if ep.serviceEnabled == oldState {
+		ep.serviceEnabled = newState
+		return true
+	}
+	return false
+}
+
+func (ep *endpoint) setServiceEnabled(state bool) {
+	ep.Lock()
+	defer ep.Unlock()
+	ep.serviceEnabled = state
+}
+
 func (ep *endpoint) needResolver() bool {
 	ep.Lock()
 	defer ep.Unlock()
@@ -500,10 +519,6 @@ func (ep *endpoint) sbJoin(sb *sandbox, options ...EndpointOption) error {
 		return err
 	}
 
-	if e := ep.addToCluster(); e != nil {
-		log.Errorf("Could not update state for endpoint %s into cluster: %v", ep.Name(), e)
-	}
-
 	if sb.needDefaultGW() && sb.getEndpointInGWNetwork() == nil {
 		return sb.setupDefaultGW()
 	}

libnetwork/libnetwork_test.go

@@ -1216,6 +1216,14 @@ func (f *fakeSandbox) Endpoints() []libnetwork.Endpoint {
 	return nil
 }
 
+func (f *fakeSandbox) EnableService() error {
+	return nil
+}
+
+func (f *fakeSandbox) DisableService() error {
+	return nil
+}
+
 func TestExternalKey(t *testing.T) {
 	externalKeyTest(t, false)
 }

libnetwork/networkdb/networkdb.go

@@ -203,9 +203,10 @@ func (nDB *NetworkDB) getEntry(tname, nid, key string) (*entry, error) {
 // table, key) tuple and if the NetworkDB is part of the cluster
 // propogates this event to the cluster. It is an error to create an
 // entry for the same tuple for which there is already an existing
-// entry.
+// entry unless the current entry is deleting state.
 func (nDB *NetworkDB) CreateEntry(tname, nid, key string, value []byte) error {
-	if _, err := nDB.GetEntry(tname, nid, key); err == nil {
+	e, _ := nDB.getEntry(tname, nid, key)
+	if e != nil && !e.deleting {
 		return fmt.Errorf("cannot create entry as the entry in table %s with network id %s and key %s already exists", tname, nid, key)
 	}
 

libnetwork/sandbox.go

@@ -50,6 +50,12 @@ type Sandbox interface {
 	ResolveService(name string) ([]*net.SRV, []net.IP, error)
 	// Endpoints returns all the endpoints connected to the sandbox
 	Endpoints() []Endpoint
+	// EnableService  makes a managed container's service available by adding the
+	// endpoint to the service load balancer and service discovery
+	EnableService() error
+	// DisableService removes a managed contianer's endpoints from the load balancer
+	// and service discovery
+	DisableService() error
 }
 
 // SandboxOption is an option setter function type used to pass various options to
@@ -721,6 +727,30 @@ func (sb *sandbox) SetKey(basePath string) error {
 	return nil
 }
 
+func (sb *sandbox) EnableService() error {
+	for _, ep := range sb.getConnectedEndpoints() {
+		if ep.casServiceEnabled(false, true) {
+			if e := ep.addToCluster(); e != nil {
+				ep.setServiceEnabled(false)
+				return fmt.Errorf("could not update state for endpoint %s into cluster: %v", ep.Name(), e)
+			}
+		}
+	}
+	return nil
+}
+
+func (sb *sandbox) DisableService() error {
+	for _, ep := range sb.getConnectedEndpoints() {
+		if ep.casServiceEnabled(true, false) {
+			if e := ep.deleteFromCluster(); e != nil {
+				ep.setServiceEnabled(true)
+				return fmt.Errorf("could not delete state for endpoint %s from cluster: %v", ep.Name(), e)
+			}
+		}
+	}
+	return nil
+}
+
 func releaseOSSboxResources(osSbox osl.Sandbox, ep *endpoint) {
 	for _, i := range osSbox.Info().Interfaces() {
 		// Only remove the interfaces owned by this endpoint from the sandbox.