func (s consulStore) Delete(id rcf.ID) error {
	key := kp.RollPath(id.String())
	_, err := s.kv.Delete(key, nil)
	if err != nil {
		return consulutil.NewKVError("delete", key, err)
	}
	return nil
}

// test if the farm should work on the given replication controller ID
func (rcf *Farm) shouldWorkOn(rcID fields.ID) (bool, error) {
	if rcf.rcSelector.Empty() {
		return true, nil
	}
	labels, err := rcf.labeler.GetLabels(labels.RC, rcID.String())
	if err != nil {
		return false, err
	}
	return rcf.rcSelector.Matches(labels.Labels), nil
}

// performs a safe (ie check-and-set) mutation of the rc with the given id,
// using the given function
// if the mutator returns an error, it will be propagated out
// if the returned RC has id="", then it will be deleted
func (s *consulStore) mutateRc(id fields.ID, mutator func(fields.RC) (fields.RC, error)) error {
	rcp := kp.RCPath(id.String())
	kvp, meta, err := s.kv.Get(rcp, nil)
	if err != nil {
		return err
	}
	if kvp == nil {
		return fmt.Errorf("replication controller %s does not exist", id)
	}

	rc, err := s.kvpToRC(kvp)
	if err != nil {
		return err
	}
	newKVP := &api.KVPair{
		Key:         rcp,
		ModifyIndex: meta.LastIndex,
	}

	var success bool
	newRC, err := mutator(rc)
	if err != nil {
		return err
	}
	if newRC.ID.String() == "" {
		// TODO: If this fails, then we have some dangling labels.
		// Perhaps they can be cleaned up later.
		// note that if the CAS fails afterwards, we will have still deleted
		// the labels, and then we will retry, which will involve deleting them
		// again
		// really the only way to solve this is a transaction
		err = s.applicator.RemoveAllLabels(labels.RC, id.String())
		if err != nil {
			return err
		}

		success, _, err = s.kv.DeleteCAS(newKVP, nil)
	} else {
		b, err := json.Marshal(newRC)
		if err != nil {
			return err
		}
		newKVP.Value = b
		success, _, err = s.kv.CAS(newKVP, nil)
	}

	if err != nil {
		return err
	}
	if !success {
		return CASError(rcp)
	}
	return nil
}

func (s *consulStore) Get(id fields.ID) (fields.RC, error) {
	kvp, _, err := s.kv.Get(kp.RCPath(id.String()), nil)
	if err != nil {
		return fields.RC{}, err
	}
	if kvp == nil {
		// ID didn't exist
		return fields.RC{}, nil
	}
	return s.kvpToRC(kvp)
}

func (s consulStore) Lock(id rcf.ID, session string) (bool, error) {
	key := kp.LockPath(kp.RollPath(id.String()))
	success, _, err := s.kv.Acquire(&api.KVPair{
		Key:     key,
		Value:   []byte(session),
		Session: session,
	}, nil)
	if err != nil {
		return false, consulutil.NewKVError("acquire", key, err)
	}
	return success, nil
}

// close one child
func (rlf *Farm) releaseChild(id fields.ID) {
	rlf.logger.WithField("ru", id).Infoln("Releasing update")
	close(rlf.children[id].quit)
	delete(rlf.children, id)

	// if our lock is active, attempt to gracefully release it
	if rlf.lock != nil {
		err := rlf.lock.Unlock(kp.LockPath(kp.RollPath(id.String())))
		if err != nil {
			rlf.logger.WithField("ru", id).Warnln("Could not release update lock")
		}
	}
}

func (s consulStore) Get(id rcf.ID) (rollf.Update, error) {
	key := kp.RollPath(id.String())
	kvp, _, err := s.kv.Get(key, nil)
	if err != nil {
		return rollf.Update{}, consulutil.NewKVError("get", key, err)
	}

	var ret rollf.Update
	err = json.Unmarshal(kvp.Value, &ret)
	if err != nil {
		return rollf.Update{}, err
	}
	return ret, nil
}

func (u update) lockPath(id rcf.ID) string {
	// RUs want to lock the RCs they're mutating, but this lock is separate
	// from the RC lock (which is held by the rc.WatchDesires goroutine), so the
	// key being locked is different
	return kp.LockPath(kp.RCPath(id.String(), "update"))
}

// Creates a rolling update that may or may not already have an existing old
// RC. If one matches the oldRCSelector, it will be used as the old RC in the
// new update.  If one does not exist, a "dummy" old RC will be created that is
// identical to the specifications for the new RC.
// Returns an error if the old RC exists but is part of another RU, or if
// the label selector returns more than one match.
func (s consulStore) CreateRollingUpdateFromOneMaybeExistingWithLabelSelector(
	oldRCSelector klabels.Selector,
	desiredReplicas int,
	minimumReplicas int,
	leaveOld bool,
	rollDelay time.Duration,
	newRCManifest manifest.Manifest,
	newRCNodeSelector klabels.Selector,
	newRCPodLabels klabels.Set,
	newRCLabels klabels.Set,
	rollLabels klabels.Set,
) (u roll_fields.Update, err error) {
	// This function may or may not create old and new RCs and subsequently
	// fail, so we defer a function that does any cleanup (if applicable)
	var cleanupOldRC func()
	var cleanupNewRC func()

	defer func() {
		if err != nil {
			if cleanupOldRC != nil {
				cleanupOldRC()
			}

			if cleanupNewRC != nil {
				cleanupNewRC()
			}
		}
	}()

	session, renewalErrCh, err := s.newRUCreationSession()
	if err != nil {
		return roll_fields.Update{}, err
	}
	defer session.Destroy()

	// Check if any RCs match the oldRCSelector
	matches, err := s.labeler.GetMatches(oldRCSelector, labels.RC, false)
	if err != nil {
		return roll_fields.Update{}, err
	}

	var oldRCID rc_fields.ID
	if len(matches) > 1 {
		return roll_fields.Update{}, AmbiguousRCSelector
	} else if len(matches) == 1 {
		oldRCID = rc_fields.ID(matches[0].ID)
	} else {
		if leaveOld {
			return roll_fields.Update{}, util.Errorf(
				"Can't create an update with LeaveOld set if there is no old RC (sel=%s)",
				oldRCSelector.String(),
			)
		}

		// Create the old RC using the same info as the new RC, it'll be
		// removed when the update completes anyway
		rc, err := s.rcstore.Create(newRCManifest, newRCNodeSelector, newRCPodLabels)
		if err != nil {
			return roll_fields.Update{}, err
		}

		oldRCID = rc.ID
		cleanupOldRC = func() {
			err = s.rcstore.Delete(oldRCID, false)
			if err != nil {
				s.logger.WithError(err).Errorf("Unable to cleanup newly-created old RC %s after update creation failure:", oldRCID)
			}

			// Any labels we wrote will be deleted by rcstore.Delete()
		}

		// Copy the new RC labels to the old RC as well
		err = s.labeler.SetLabels(labels.RC, oldRCID.String(), newRCLabels)
		if err != nil {
			return roll_fields.Update{}, err
		}
	}

	// Lock the old RC to guarantee that no new updates can use it
	err = s.lockRCs(rc_fields.IDs{oldRCID}, session)
	if err != nil {
		return roll_fields.Update{}, err
	}

	// Check for updates that exist that operate on the old RC
	err = s.checkForConflictingUpdates(rc_fields.IDs{oldRCID})
	if err != nil {
		return roll_fields.Update{}, err
	}

	// Create the new RC
	var newRCID rc_fields.ID
	select {
	case err = <-renewalErrCh:
//.........这里部分代码省略.........

// Like CreateRollingUpdateFromExistingRCs except will create the new RC based
// on passed parameters, using oldRCID for the old RC. The new RC and new RU
// will be created transactionally (all or nothing)
func (s consulStore) CreateRollingUpdateFromOneExistingRCWithID(
	oldRCID rc_fields.ID,
	desiredReplicas int,
	minimumReplicas int,
	leaveOld bool,
	rollDelay time.Duration,
	newRCManifest manifest.Manifest,
	newRCNodeSelector klabels.Selector,
	newRCPodLabels klabels.Set,
	newRCLabels klabels.Set,
	rollLabels klabels.Set,
) (u roll_fields.Update, err error) {
	// There are cases where this function will create the new RC and
	// subsequently fail, in which case we need to do some cleanup.

	// cleans up new RC, might be nil if we didn't create one
	var newRCCleanup func()

	// If we had an error and the rc cleanup function is set, run it
	defer func() {
		if err != nil && newRCCleanup != nil {
			newRCCleanup()
		}
	}()

	var session kp.Session
	var renewalErrCh chan error
	session, renewalErrCh, err = s.newRUCreationSession()
	if err != nil {
		return roll_fields.Update{}, err
	}
	defer session.Destroy()

	rcIDs := rc_fields.IDs{oldRCID}
	err = s.lockRCs(rcIDs, session)
	if err != nil {
		return roll_fields.Update{}, err
	}

	err = s.checkForConflictingUpdates(rcIDs)
	if err != nil {
		return roll_fields.Update{}, err
	}

	// Now create the new RC, first checking if our session is still valid
	var newRCID rc_fields.ID
	select {
	case err = <-renewalErrCh:
		return roll_fields.Update{}, err
	default:
		rc, err := s.rcstore.Create(newRCManifest, newRCNodeSelector, newRCPodLabels)
		if err != nil {
			return roll_fields.Update{}, err
		}

		newRCCleanup = func() {
			err := s.rcstore.Delete(newRCID, false)
			if err != nil {
				s.logger.WithError(err).Errorln("Unable to cleanup RC %s after failed RU creation attempt", newRCID)
			}
		}

		newRCID = rc.ID

		// Get a lock on the new RC we just created so no parallel
		// update creations can use it
		err = s.lockRCs(rc_fields.IDs{newRCID}, session)
		if err != nil {
			return roll_fields.Update{}, err
		}
	}

	rcIDs = append(rcIDs, newRCID)
	// Check for conflicts again in case an update was created on the new
	// RC between when we created it and locked it
	err = s.checkForConflictingUpdates(rcIDs)
	if err != nil {
		return roll_fields.Update{}, err
	}

	err = s.labeler.SetLabels(labels.RC, newRCID.String(), newRCLabels)
	if err != nil {
		return roll_fields.Update{}, err
	}

	u = roll_fields.Update{
		OldRC:           oldRCID,
		NewRC:           newRCID,
		DesiredReplicas: desiredReplicas,
		MinimumReplicas: minimumReplicas,
		LeaveOld:        leaveOld,
		RollDelay:       rollDelay,
	}

	return s.attemptRUCreation(u, rollLabels, renewalErrCh)
}