Kubernetes 并发控制与数据一致性的实现原理

type ObjectMeta struct {      ......
      // An opaque value that represents the internal version of this object   that can      // be used by clients to determine when objects have changed. May be   used for optimistic      // concurrency, change detection, and the watch operation on a   resource or set of resources.      // Clients must treat these values as opaque and passed unmodified   back to the server.      // They may only be valid for a particular resource or set of   resources.      //      // Populated by the system.      // Read-only.      // Value must be treated as opaque by clients and .      // More info:   https://git.k8s.io/community/contributors/devel/api-conventions.md#concurrency-control-and-consistency      // +optional      ResourceVersion string           ......}

Client #1: GET FooClient #2: GET Foo
Client #1: Set Foo.Bar =   "one"Client #1: PUT Foo
Client #2: Set Foo.Baz =   "two"Client #2: PUT Foo

Client #1: GET Foo  //初始Foo.ResourceVersion=1Client #2: GET Foo  //初始Foo.ResourceVersion=1
Client #1: Set Foo.Bar =   "one"Client #1: PUT Foo  //更新Foo.ResourceVersion=2
Client #2: Set Foo.Baz =   "two"Client #2: PUT Foo  //返回409冲突

e.Storage.GuaranteedUpdate(ctx,   key...) (runtime.Object, *uint64, error) {              //   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 storage object.              resourceVersion,   err := e.Storage.Versioner().ObjectResourceVersion(obj)              if   err != nil {                     return   nil, nil, err              }                version, err :=   e.Storage.Versioner().ObjectResourceVersion(existing)              doUnconditionalUpdate   := resourceVersion == 0 && e.UpdateStrategy.AllowUnconditionalUpdate()                             if   doUnconditionalUpdate {                     //   Update the object's resource version to match the latest                     //   storage object's resource version.                     err   = e.Storage.Versioner().UpdateObject(obj, res.ResourceVersion)                     if   err != nil {                            return   nil, nil, err                     }              }   else {                     //   Check if the object's resource version matches the latest                     //   resource version.                     ......                     if   resourceVersion != version {                            return   nil, nil, kubeerr.NewConflict(qualifiedResource, name,   fmt.Errorf(OptimisticLockErrorMsg))                     }              }         ......       return   out, creating, nil}

kubectl patch pod my-pod -p   '{"spec":{"containers":[{"name":"my-container","image":"new-image"}]}}'

func (p *patcher) patchResource(ctx   context.Context) (runtime.Object, error) {       p.namespace   = request.NamespaceValue(ctx)       switch   p.patchType {       case   types.JSONPatchType, types.MergePatchType:              p.mechanism   = &jsonPatcher{patcher: p}       case   types.StrategicMergePatchType:              schemaReferenceObj,   err := p.unsafeConvertor.ConvertToVersion(p.restPatcher.New(),   p.kind.GroupVersion())              if   err != nil {                     return   nil, err              }              p.mechanism   = &smpPatcher{patcher: p, schemaReferenceObj: schemaReferenceObj}       default:              return   nil, fmt.Errorf("%v: unimplemented patch type", p.patchType)       }       p.updatedObjectInfo   = rest.DefaultUpdatedObjectInfo(nil, p.applyPatch, p.applyAdmission)       return   finishRequest(p.timeout, func() (runtime.Object, error) {              updateObject,   _, updateErr := p.restPatcher.Update(ctx, p.name, p.updatedObjectInfo,   p.createValidation, p.updateValidation, false, p.options)              return   updateObject, updateErr       })}