github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-31 23:37:01 +00:00
Code Issues Projects Releases Wiki Activity

Migrate Create to Kubernetes client

Browse Source
This commit is contained in:
Marek Siarkowicz 2024-05-31 10:57:41 +02:00
parent 092a6d1e0d
commit 53ca81da29
1 changed files with 8 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
View File

@ -217,9 +217,12 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
} }
span.AddEvent("Encode succeeded", attribute.Int("len", len(data))) span.AddEvent("Encode succeeded", attribute.Int("len", len(data)))
opts, err := s.ttlOpts(ctx, int64(ttl)) var lease clientv3.LeaseID
if err != nil { if ttl != 0 {
return err lease, err = s.leaseManager.GetLease(ctx, int64(ttl))
if err != nil {
return err
}
} }
newData, err := s.transformer.TransformToStorage(ctx, data, authenticatedDataString(preparedKey)) newData, err := s.transformer.TransformToStorage(ctx, data, authenticatedDataString(preparedKey))
@ -230,11 +233,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
span.AddEvent("TransformToStorage succeeded") span.AddEvent("TransformToStorage succeeded")
startTime := time.Now() startTime := time.Now()
txnResp, err := s.client.KV.Txn(ctx).If( txnResp, err := s.client.Kubernetes.OptimisticPut(ctx, preparedKey, newData, 0, kubernetes.PutOptions{LeaseID: lease})
notFound(preparedKey),
).Then(
clientv3.OpPut(preparedKey, string(newData), opts...),
).Commit()
metrics.RecordEtcdRequest("create", s.groupResourceString, err, startTime) metrics.RecordEtcdRequest("create", s.groupResourceString, err, startTime)
if err != nil { if err != nil {
span.AddEvent("Txn call failed", attribute.String("err", err.Error())) span.AddEvent("Txn call failed", attribute.String("err", err.Error()))
@ -247,8 +246,7 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
} }
if out != nil { if out != nil {
putResp := txnResp.Responses[0].GetResponsePut() err = s.decoder.Decode(data, out, txnResp.Revision)
err = s.decoder.Decode(data, out, putResp.Header.Revision)
if err != nil { if err != nil {
span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error())) span.AddEvent("decode failed", attribute.Int("len", len(data)), attribute.String("err", err.Error()))
recordDecodeError(s.groupResourceString, preparedKey) recordDecodeError(s.groupResourceString, preparedKey)
@ -1052,10 +1050,6 @@ func recordDecodeError(resource string, key string) {
klog.V(4).Infof("Decoding %s \"%s\" failed", resource, key) klog.V(4).Infof("Decoding %s \"%s\" failed", resource, key)
} }
func notFound(key string) clientv3.Cmp {
return clientv3.Compare(clientv3.ModRevision(key), "=", 0)
}
// getTypeName returns type name of an object for reporting purposes. // getTypeName returns type name of an object for reporting purposes.
func getTypeName(obj interface{}) string { func getTypeName(obj interface{}) string {
return reflect.TypeOf(obj).String() return reflect.TypeOf(obj).String()