// CreateOrUpdate attempts to update the current etcd state with the provided
// allocation.
func (e *Etcd) CreateOrUpdate(snapshot *api.RangeAllocation) error {
	e.lock.Lock()
	defer e.lock.Unlock()

	last := ""
	err := e.helper.GuaranteedUpdate(e.baseKey, &api.RangeAllocation{}, true,
		tools.SimpleUpdate(func(input runtime.Object) (output runtime.Object, err error) {
			existing := input.(*api.RangeAllocation)
			switch {
			case len(snapshot.ResourceVersion) != 0 && len(existing.ResourceVersion) != 0:
				if snapshot.ResourceVersion != existing.ResourceVersion {
					return nil, k8serr.NewConflict(e.kind, "", fmt.Errorf("the provided resource version does not match"))
				}
			case len(existing.ResourceVersion) != 0:
				return nil, k8serr.NewConflict(e.kind, "", fmt.Errorf("another caller has already initialized the resource"))
			}
			last = snapshot.ResourceVersion
			return snapshot, nil
		}),
	)
	if err != nil {
		return etcderr.InterpretUpdateError(err, e.kind, "")
	}
	err = e.alloc.Restore(snapshot.Range, snapshot.Data)
	if err == nil {
		e.last = last
	}
	return err
}

// tryUpdate performs a read-update to persist the latest snapshot state of allocation.
func (e *Etcd) tryUpdate(fn func() error) error {
	err := e.helper.GuaranteedUpdate(e.baseKey, &api.RangeAllocation{}, true,
		tools.SimpleUpdate(func(input runtime.Object) (output runtime.Object, err error) {
			existing := input.(*api.RangeAllocation)
			if len(existing.ResourceVersion) == 0 {
				return nil, fmt.Errorf("cannot allocate resources of type %s at this time", e.kind)
			}
			if existing.ResourceVersion != e.last {
				if err := e.alloc.Restore(existing.Range, existing.Data); err != nil {
					return nil, err
				}
				if err := fn(); err != nil {
					return nil, err
				}
			}
			e.last = existing.ResourceVersion
			rangeSpec, data := e.alloc.Snapshot()
			existing.Range = rangeSpec
			existing.Data = data
			return existing, nil
		}),
	)
	if err != nil {
		err = etcderr.InterpretUpdateError(err, e.kind, "")
	}
	return err
}

// UpdateBuildConfig replaces an existing BuildConfig.
func (r *Etcd) UpdateBuildConfig(ctx kapi.Context, config *api.BuildConfig) error {
	key, err := makeBuildConfigKey(ctx, config.Name)
	if err != nil {
		return err
	}
	err = r.SetObj(key, config, nil, 0)
	return etcderr.InterpretUpdateError(err, "buildConfig", config.Name)
}

// UpdateBuild replaces an existing Build.
func (r *Etcd) UpdateBuild(ctx kapi.Context, build *api.Build) error {
	key, err := makeBuildKey(ctx, build.Name)
	if err != nil {
		return err
	}
	err = r.SetObj(key, build, nil, 0)
	return etcderr.InterpretUpdateError(err, "build", build.Name)
}

// UpdateController replaces an existing ReplicationController.
func (r *Registry) UpdateController(ctx api.Context, controller *api.ReplicationController) error {
	key, err := makeControllerKey(ctx, controller.Name)
	if err != nil {
		return err
	}
	err = r.SetObj(key, controller)
	return etcderr.InterpretUpdateError(err, "replicationController", controller.Name)
}

// UpdateRoute replaces an existing Route.
func (registry *Etcd) UpdateRoute(ctx kapi.Context, route *api.Route) error {
	key, err := makeRouteKey(ctx, route.Name)
	if err != nil {
		return err
	}
	err = registry.Set(key, route, nil, 0)
	return etcderr.InterpretUpdateError(err, "route", route.Name)
}

// UpdateService replaces an existing Service.
func (r *Registry) UpdateService(ctx api.Context, svc *api.Service) error {
	key, err := makeServiceKey(ctx, svc.Name)
	if err != nil {
		return err
	}
	err = r.SetObj(key, svc)
	return etcderr.InterpretUpdateError(err, "service", svc.Name)
}

// Update updates the item.
func (e *Etcd) Update(ctx api.Context, id string, obj runtime.Object) error {
	key, err := e.KeyFunc(ctx, id)
	if err != nil {
		return err
	}
	// TODO: verify that SetObj checks ResourceVersion before succeeding.
	err = e.Helper.SetObj(key, obj)
	return etcderr.InterpretUpdateError(err, e.EndpointName, id)
}

// UpdateEndpoints update Endpoints of a Service.
func (r *Registry) UpdateEndpoints(e *api.Endpoints) error {
	// TODO: this is a really bad misuse of AtomicUpdate, need to compute a diff inside the loop.
	err := r.AtomicUpdate(makeServiceEndpointsKey(e.ID), &api.Endpoints{},
		func(input runtime.Object) (runtime.Object, error) {
			// TODO: racy - label query is returning different results for two simultaneous updaters
			return e, nil
		})
	return etcderr.InterpretUpdateError(err, "endpoints", e.ID)
}

// UpdateService replaces an existing Service.
func (r *Registry) UpdateService(ctx api.Context, svc *api.Service) (*api.Service, error) {
	key, err := makeServiceKey(ctx, svc.Name)
	if err != nil {
		return nil, err
	}
	out := &api.Service{}
	err = r.SetObj(key, svc, out, 0)
	return out, etcderr.InterpretUpdateError(err, "service", svc.Name)
}

// UpdateController replaces an existing ReplicationController.
func (r *Registry) UpdateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error) {
	key, err := makeControllerKey(ctx, controller.Name)
	if err != nil {
		return nil, err
	}
	out := &api.ReplicationController{}
	err = r.SetObj(key, controller, out, 0)
	return out, etcderr.InterpretUpdateError(err, "replicationController", controller.Name)
}

// assignPod assigns the given pod to the given machine.
func (r *BindingREST) assignPod(ctx api.Context, podID string, machine string, annotations map[string]string) (err error) {
	if _, err = r.setPodHostAndAnnotations(ctx, podID, "", machine, annotations); err != nil {
		err = etcderr.InterpretGetError(err, "pod", podID)
		err = etcderr.InterpretUpdateError(err, "pod", podID)
		if _, ok := err.(*errors.StatusError); !ok {
			err = errors.NewConflict("binding", podID, err)
		}
	}
	return
}

// UpdateEndpoints update Endpoints of a Service.
func (r *Registry) UpdateEndpoints(ctx api.Context, endpoints *api.Endpoints) error {
	key, err := makeServiceEndpointsKey(ctx, endpoints.Name)
	if err != nil {
		return err
	}
	// TODO: this is a really bad misuse of AtomicUpdate, need to compute a diff inside the loop.
	err = r.AtomicUpdate(key, &api.Endpoints{}, true,
		func(input runtime.Object) (runtime.Object, uint64, error) {
			// TODO: racy - label query is returning different results for two simultaneous updaters
			return endpoints, 0, nil
		})
	return etcderr.InterpretUpdateError(err, "endpoints", endpoints.Name)
}

// UpdateWithName updates the item with the provided name
// DEPRECATED: use Update instead
func (e *Etcd) UpdateWithName(ctx api.Context, name string, obj runtime.Object) error {
	key, err := e.KeyFunc(ctx, name)
	if err != nil {
		return err
	}
	ttl, err := e.calculateTTL(obj, 0, true)
	if err != nil {
		return err
	}
	err = e.Helper.SetObj(key, obj, nil, ttl)
	err = etcderr.InterpretUpdateError(err, e.EndpointName, name)
	if err == nil && e.Decorator != nil {
		err = e.Decorator(obj)
	}
	return err
}

// Delete removes the item from etcd.
func (e *Etcd) Delete(ctx api.Context, name string, options *api.DeleteOptions) (runtime.Object, error) {
	key, err := e.KeyFunc(ctx, name)
	if err != nil {
		return nil, err
	}

	obj := e.NewFunc()
	trace := util.NewTrace("Delete " + reflect.TypeOf(obj).String())
	defer trace.LogIfLong(time.Second)
	trace.Step("About to read object")
	if err := e.Helper.ExtractObj(key, obj, false); err != nil {
		return nil, etcderr.InterpretDeleteError(err, e.EndpointName, name)
	}

	// support older consumers of delete by treating "nil" as delete immediately
	if options == nil {
		options = api.NewDeleteOptions(0)
	}
	graceful, pendingGraceful, err := rest.BeforeDelete(e.DeleteStrategy, ctx, obj, options)
	if err != nil {
		return nil, err
	}
	if pendingGraceful {
		return e.finalizeDelete(obj, false)
	}
	if graceful && *options.GracePeriodSeconds != 0 {
		trace.Step("Graceful deletion")
		out := e.NewFunc()
		if err := e.Helper.SetObj(key, obj, out, uint64(*options.GracePeriodSeconds)); err != nil {
			return nil, etcderr.InterpretUpdateError(err, e.EndpointName, name)
		}
		return e.finalizeDelete(out, true)
	}

	// delete immediately, or no graceful deletion supported
	out := e.NewFunc()
	trace.Step("About to delete object")
	if err := e.Helper.DeleteObj(key, out); err != nil {
		return nil, etcderr.InterpretDeleteError(err, e.EndpointName, name)
	}
	return e.finalizeDelete(out, true)
}

// UpdateService replaces an existing Service.
func (r *Registry) UpdateService(svc *api.Service) error {
	err := r.SetObj(makeServiceKey(svc.ID), svc)
	return etcderr.InterpretUpdateError(err, "service", svc.ID)
}

func (r *Registry) UpdateMinion(ctx api.Context, minion *api.Node) error {
	// TODO: Add some validations.
	err := r.SetObj(makeNodeKey(minion.Name), minion, nil, 0)
	return etcderr.InterpretUpdateError(err, "minion", minion.Name)
}

// Update performs an atomic update and set of the object. Returns the result of the update
// or an error. If the registry allows create-on-update, the create flow will be executed.
// A bool is returned along with the object and any errors, to indicate object creation.
func (e *Etcd) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	trace := util.NewTrace("Update " + reflect.TypeOf(obj).String())
	defer trace.LogIfLong(time.Second)
	name, err := e.ObjectNameFunc(obj)
	if err != nil {
		return nil, false, err
	}
	key, err := e.KeyFunc(ctx, name)
	if err != nil {
		return nil, false, err
	}
	// If AllowUnconditionalUpdate() is true and the object specified by the user does not have a resource version,
	// then we populate it with the latest version.
	// Else, we check that the version specified by the user matches the version of latest etcd object.
	resourceVersion, err := e.Helper.Versioner.ObjectResourceVersion(obj)
	if err != nil {
		return nil, false, err
	}
	doUnconditionalUpdate := resourceVersion == 0 && e.UpdateStrategy.AllowUnconditionalUpdate()
	// TODO: expose TTL
	creating := false
	out := e.NewFunc()
	err = e.Helper.GuaranteedUpdate(key, out, true, func(existing runtime.Object, res tools.ResponseMeta) (runtime.Object, *uint64, error) {
		version, err := e.Helper.Versioner.ObjectResourceVersion(existing)
		if err != nil {
			return nil, nil, err
		}
		if version == 0 {
			if !e.UpdateStrategy.AllowCreateOnUpdate() {
				return nil, nil, kubeerr.NewNotFound(e.EndpointName, name)
			}
			creating = true
			if err := rest.BeforeCreate(e.CreateStrategy, ctx, obj); err != nil {
				return nil, nil, err
			}
			ttl, err := e.calculateTTL(obj, 0, false)
			if err != nil {
				return nil, nil, err
			}
			return obj, &ttl, nil
		}

		creating = false
		if doUnconditionalUpdate {
			// Update the object's resource version to match the latest etcd object's resource version.
			err = e.Helper.Versioner.UpdateObject(obj, res.Expiration, res.ResourceVersion)
			if err != nil {
				return nil, nil, err
			}
		} else {
			// Check if the object's resource version matches the latest resource version.
			newVersion, err := e.Helper.Versioner.ObjectResourceVersion(obj)
			if err != nil {
				return nil, nil, err
			}
			if newVersion != version {
				return nil, nil, kubeerr.NewConflict(e.EndpointName, name, fmt.Errorf("the object has been modified; please apply your changes to the latest version and try again"))
			}
		}
		if err := rest.BeforeUpdate(e.UpdateStrategy, ctx, obj, existing); err != nil {
			return nil, nil, err
		}
		ttl, err := e.calculateTTL(obj, res.TTL, true)
		if err != nil {
			return nil, nil, err
		}
		if int64(ttl) != res.TTL {
			return obj, &ttl, nil
		}
		return obj, nil, nil
	})

	if err != nil {
		if creating {
			err = etcderr.InterpretCreateError(err, e.EndpointName, name)
			err = rest.CheckGeneratedNameError(e.CreateStrategy, err, obj)
		} else {
			err = etcderr.InterpretUpdateError(err, e.EndpointName, name)
		}
		return nil, false, err
	}
	if creating {
		if e.AfterCreate != nil {
			if err := e.AfterCreate(out); err != nil {
				return nil, false, err
			}
		}
	} else {
		if e.AfterUpdate != nil {
			if err := e.AfterUpdate(out); err != nil {
				return nil, false, err
			}
		}
	}
	if e.Decorator != nil {
		if err := e.Decorator(obj); err != nil {
			return nil, false, err
//.........这里部分代码省略.........

// Update performs an atomic update and set of the object. Returns the result of the update
// or an error. If the registry allows create-on-update, the create flow will be executed.
// A bool is returned along with the object and any errors, to indicate object creation.
func (e *Etcd) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	name, err := e.ObjectNameFunc(obj)
	if err != nil {
		return nil, false, err
	}
	key, err := e.KeyFunc(ctx, name)
	if err != nil {
		return nil, false, err
	}
	// TODO: expose TTL
	creating := false
	out := e.NewFunc()
	err = e.Helper.GuaranteedUpdate(key, out, true, func(existing runtime.Object) (runtime.Object, uint64, error) {
		version, err := e.Helper.Versioner.ObjectResourceVersion(existing)
		if err != nil {
			return nil, 0, err
		}
		if version == 0 {
			if !e.UpdateStrategy.AllowCreateOnUpdate() {
				return nil, 0, kubeerr.NewNotFound(e.EndpointName, name)
			}
			creating = true
			if err := rest.BeforeCreate(e.CreateStrategy, ctx, obj); err != nil {
				return nil, 0, err
			}
			ttl := uint64(0)
			if e.TTLFunc != nil {
				ttl, err = e.TTLFunc(obj, true)
				if err != nil {
					return nil, 0, err
				}
			}
			return obj, ttl, nil
		}

		creating = false
		newVersion, err := e.Helper.Versioner.ObjectResourceVersion(obj)
		if err != nil {
			return nil, 0, err
		}
		if newVersion != version {
			// TODO: return the most recent version to a client?
			return nil, 0, kubeerr.NewConflict(e.EndpointName, name, fmt.Errorf("the resource was updated to %d", version))
		}
		if err := rest.BeforeUpdate(e.UpdateStrategy, ctx, obj, existing); err != nil {
			return nil, 0, err
		}
		ttl := uint64(0)
		if e.TTLFunc != nil {
			ttl, err = e.TTLFunc(obj, false)
			if err != nil {
				return nil, 0, err
			}
		}
		return obj, ttl, nil
	})

	if err != nil {
		if creating {
			err = etcderr.InterpretCreateError(err, e.EndpointName, name)
			err = rest.CheckGeneratedNameError(e.CreateStrategy, err, obj)
		} else {
			err = etcderr.InterpretUpdateError(err, e.EndpointName, name)
		}
		return nil, false, err
	}
	if creating {
		if e.AfterCreate != nil {
			if err := e.AfterCreate(out); err != nil {
				return nil, false, err
			}
		}
	} else {
		if e.AfterUpdate != nil {
			if err := e.AfterUpdate(out); err != nil {
				return nil, false, err
			}
		}
	}
	if e.Decorator != nil {
		if err := e.Decorator(obj); err != nil {
			return nil, false, err
		}
	}
	return out, creating, nil
}

// Update performs an atomic update and set of the object. Returns the result of the update
// or an error. If the registry allows create-on-update, the create flow will be executed.
// A bool is returned along with the object and any errors, to indicate object creation.
func (e *Etcd) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	trace := util.NewTrace("Update " + reflect.TypeOf(obj).String())
	defer trace.LogIfLong(time.Second)
	name, err := e.ObjectNameFunc(obj)
	if err != nil {
		return nil, false, err
	}
	key, err := e.KeyFunc(ctx, name)
	if err != nil {
		return nil, false, err
	}
	// TODO: expose TTL
	creating := false
	out := e.NewFunc()
	err = e.Helper.GuaranteedUpdate(key, out, true, func(existing runtime.Object, res tools.ResponseMeta) (runtime.Object, *uint64, error) {
		version, err := e.Helper.Versioner.ObjectResourceVersion(existing)
		if err != nil {
			return nil, nil, err
		}
		if version == 0 {
			if !e.UpdateStrategy.AllowCreateOnUpdate() {
				return nil, nil, kubeerr.NewNotFound(e.EndpointName, name)
			}
			creating = true
			if err := rest.BeforeCreate(e.CreateStrategy, ctx, obj); err != nil {
				return nil, nil, err
			}
			ttl, err := e.calculateTTL(obj, 0, false)
			if err != nil {
				return nil, nil, err
			}
			return obj, &ttl, nil
		}

		creating = false
		newVersion, err := e.Helper.Versioner.ObjectResourceVersion(obj)
		if err != nil {
			return nil, nil, err
		}
		if newVersion != version {
			// TODO: return the most recent version to a client?
			return nil, nil, kubeerr.NewConflict(e.EndpointName, name, fmt.Errorf("the object has been modified; please apply your changes to the latest version and try again"))
		}
		if err := rest.BeforeUpdate(e.UpdateStrategy, ctx, obj, existing); err != nil {
			return nil, nil, err
		}
		ttl, err := e.calculateTTL(obj, res.TTL, true)
		if err != nil {
			return nil, nil, err
		}
		if int64(ttl) != res.TTL {
			return obj, &ttl, nil
		}
		return obj, nil, nil
	})

	if err != nil {
		if creating {
			err = etcderr.InterpretCreateError(err, e.EndpointName, name)
			err = rest.CheckGeneratedNameError(e.CreateStrategy, err, obj)
		} else {
			err = etcderr.InterpretUpdateError(err, e.EndpointName, name)
		}
		return nil, false, err
	}
	if creating {
		if e.AfterCreate != nil {
			if err := e.AfterCreate(out); err != nil {
				return nil, false, err
			}
		}
	} else {
		if e.AfterUpdate != nil {
			if err := e.AfterUpdate(out); err != nil {
				return nil, false, err
			}
		}
	}
	if e.Decorator != nil {
		if err := e.Decorator(obj); err != nil {
			return nil, false, err
		}
	}
	return out, creating, nil
}