Introduce Kubernetes KV interface to etcd client

Signed-off-by: Marek Siarkowicz <[email protected]>
serathius · Jul 2, 2024 · 98ebb5d · 98ebb5d
1 parent a043da5
commit 98ebb5d
Show file tree

Hide file tree

Showing 11 changed files with 314 additions and 79 deletions.
diff --git a/api/v3rpc/rpctypes/error.go b/api/v3rpc/rpctypes/error.go
@@ -15,6 +15,8 @@
 package rpctypes
 
 import (
+	"context"
+
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -277,3 +279,25 @@ func ErrorDesc(err error) string {
 	}
 	return err.Error()
 }
+
+func ContextError(ctx context.Context, err error) error {
+	if err == nil {
+		return nil
+	}
+	err = Error(err)
+	if _, ok := err.(EtcdError); ok {
+		return err
+	}
+	if ev, ok := status.FromError(err); ok {
+		code := ev.Code()
+		switch code {
+		case codes.DeadlineExceeded:
+			fallthrough
+		case codes.Canceled:
+			if ctx.Err() != nil {
+				err = ctx.Err()
+			}
+		}
+	}
+	return err
+}
diff --git a/client/v3/auth.go b/client/v3/auth.go
@@ -23,6 +23,7 @@ import (
 
 	"go.etcd.io/etcd/api/v3/authpb"
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 )
 
 type (
@@ -135,67 +136,67 @@ func NewAuthFromAuthClient(remote pb.AuthClient, c *Client) Auth {
 
 func (auth *authClient) Authenticate(ctx context.Context, name string, password string) (*AuthenticateResponse, error) {
 	resp, err := auth.remote.Authenticate(ctx, &pb.AuthenticateRequest{Name: name, Password: password}, auth.callOpts...)
-	return (*AuthenticateResponse)(resp), toErr(ctx, err)
+	return (*AuthenticateResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) AuthEnable(ctx context.Context) (*AuthEnableResponse, error) {
 	resp, err := auth.remote.AuthEnable(ctx, &pb.AuthEnableRequest{}, auth.callOpts...)
-	return (*AuthEnableResponse)(resp), toErr(ctx, err)
+	return (*AuthEnableResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) AuthDisable(ctx context.Context) (*AuthDisableResponse, error) {
 	resp, err := auth.remote.AuthDisable(ctx, &pb.AuthDisableRequest{}, auth.callOpts...)
-	return (*AuthDisableResponse)(resp), toErr(ctx, err)
+	return (*AuthDisableResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) AuthStatus(ctx context.Context) (*AuthStatusResponse, error) {
 	resp, err := auth.remote.AuthStatus(ctx, &pb.AuthStatusRequest{}, auth.callOpts...)
-	return (*AuthStatusResponse)(resp), toErr(ctx, err)
+	return (*AuthStatusResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserAdd(ctx context.Context, name string, password string) (*AuthUserAddResponse, error) {
 	resp, err := auth.remote.UserAdd(ctx, &pb.AuthUserAddRequest{Name: name, Password: password, Options: &authpb.UserAddOptions{NoPassword: false}}, auth.callOpts...)
-	return (*AuthUserAddResponse)(resp), toErr(ctx, err)
+	return (*AuthUserAddResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserAddWithOptions(ctx context.Context, name string, password string, options *UserAddOptions) (*AuthUserAddResponse, error) {
 	resp, err := auth.remote.UserAdd(ctx, &pb.AuthUserAddRequest{Name: name, Password: password, Options: (*authpb.UserAddOptions)(options)}, auth.callOpts...)
-	return (*AuthUserAddResponse)(resp), toErr(ctx, err)
+	return (*AuthUserAddResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserDelete(ctx context.Context, name string) (*AuthUserDeleteResponse, error) {
 	resp, err := auth.remote.UserDelete(ctx, &pb.AuthUserDeleteRequest{Name: name}, auth.callOpts...)
-	return (*AuthUserDeleteResponse)(resp), toErr(ctx, err)
+	return (*AuthUserDeleteResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserChangePassword(ctx context.Context, name string, password string) (*AuthUserChangePasswordResponse, error) {
 	resp, err := auth.remote.UserChangePassword(ctx, &pb.AuthUserChangePasswordRequest{Name: name, Password: password}, auth.callOpts...)
-	return (*AuthUserChangePasswordResponse)(resp), toErr(ctx, err)
+	return (*AuthUserChangePasswordResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserGrantRole(ctx context.Context, user string, role string) (*AuthUserGrantRoleResponse, error) {
 	resp, err := auth.remote.UserGrantRole(ctx, &pb.AuthUserGrantRoleRequest{User: user, Role: role}, auth.callOpts...)
-	return (*AuthUserGrantRoleResponse)(resp), toErr(ctx, err)
+	return (*AuthUserGrantRoleResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserGet(ctx context.Context, name string) (*AuthUserGetResponse, error) {
 	resp, err := auth.remote.UserGet(ctx, &pb.AuthUserGetRequest{Name: name}, auth.callOpts...)
-	return (*AuthUserGetResponse)(resp), toErr(ctx, err)
+	return (*AuthUserGetResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserList(ctx context.Context) (*AuthUserListResponse, error) {
 	resp, err := auth.remote.UserList(ctx, &pb.AuthUserListRequest{}, auth.callOpts...)
-	return (*AuthUserListResponse)(resp), toErr(ctx, err)
+	return (*AuthUserListResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) UserRevokeRole(ctx context.Context, name string, role string) (*AuthUserRevokeRoleResponse, error) {
 	resp, err := auth.remote.UserRevokeRole(ctx, &pb.AuthUserRevokeRoleRequest{Name: name, Role: role}, auth.callOpts...)
-	return (*AuthUserRevokeRoleResponse)(resp), toErr(ctx, err)
+	return (*AuthUserRevokeRoleResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) RoleAdd(ctx context.Context, name string) (*AuthRoleAddResponse, error) {
 	resp, err := auth.remote.RoleAdd(ctx, &pb.AuthRoleAddRequest{Name: name}, auth.callOpts...)
-	return (*AuthRoleAddResponse)(resp), toErr(ctx, err)
+	return (*AuthRoleAddResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) RoleGrantPermission(ctx context.Context, name string, key, rangeEnd string, permType PermissionType) (*AuthRoleGrantPermissionResponse, error) {
@@ -205,27 +206,27 @@ func (auth *authClient) RoleGrantPermission(ctx context.Context, name string, ke
 		PermType: authpb.Permission_Type(permType),
 	}
 	resp, err := auth.remote.RoleGrantPermission(ctx, &pb.AuthRoleGrantPermissionRequest{Name: name, Perm: perm}, auth.callOpts...)
-	return (*AuthRoleGrantPermissionResponse)(resp), toErr(ctx, err)
+	return (*AuthRoleGrantPermissionResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) RoleGet(ctx context.Context, role string) (*AuthRoleGetResponse, error) {
 	resp, err := auth.remote.RoleGet(ctx, &pb.AuthRoleGetRequest{Role: role}, auth.callOpts...)
-	return (*AuthRoleGetResponse)(resp), toErr(ctx, err)
+	return (*AuthRoleGetResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) RoleList(ctx context.Context) (*AuthRoleListResponse, error) {
 	resp, err := auth.remote.RoleList(ctx, &pb.AuthRoleListRequest{}, auth.callOpts...)
-	return (*AuthRoleListResponse)(resp), toErr(ctx, err)
+	return (*AuthRoleListResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) RoleRevokePermission(ctx context.Context, role string, key, rangeEnd string) (*AuthRoleRevokePermissionResponse, error) {
 	resp, err := auth.remote.RoleRevokePermission(ctx, &pb.AuthRoleRevokePermissionRequest{Role: role, Key: []byte(key), RangeEnd: []byte(rangeEnd)}, auth.callOpts...)
-	return (*AuthRoleRevokePermissionResponse)(resp), toErr(ctx, err)
+	return (*AuthRoleRevokePermissionResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func (auth *authClient) RoleDelete(ctx context.Context, role string) (*AuthRoleDeleteResponse, error) {
 	resp, err := auth.remote.RoleDelete(ctx, &pb.AuthRoleDeleteRequest{Role: role}, auth.callOpts...)
-	return (*AuthRoleDeleteResponse)(resp), toErr(ctx, err)
+	return (*AuthRoleDeleteResponse)(resp), rpctypes.ContextError(ctx, err)
 }
 
 func StrToPermissionType(s string) (PermissionType, error) {

diff --git a/client/v3/client.go b/client/v3/client.go
@@ -153,7 +153,7 @@ func (c *Client) Close() error {
 		c.Lease.Close()
 	}
 	if c.conn != nil {
-		return toErr(c.ctx, c.conn.Close())
+		return rpctypes.ContextError(c.ctx, c.conn.Close())
 	}
 	return c.ctx.Err()
 }
@@ -590,28 +590,6 @@ func isUnavailableErr(ctx context.Context, err error) bool {
 	return false
 }
 
-func toErr(ctx context.Context, err error) error {
-	if err == nil {
-		return nil
-	}
-	err = rpctypes.Error(err)
-	if _, ok := err.(rpctypes.EtcdError); ok {
-		return err
-	}
-	if ev, ok := status.FromError(err); ok {
-		code := ev.Code()
-		switch code {
-		case codes.DeadlineExceeded:
-			fallthrough
-		case codes.Canceled:
-			if ctx.Err() != nil {
-				err = ctx.Err()
-			}
-		}
-	}
-	return err
-}
-
 func canceledByCaller(stopCtx context.Context, err error) bool {
 	if stopCtx.Err() == nil || err == nil {
 		return false

diff --git a/client/v3/cluster.go b/client/v3/cluster.go
@@ -20,6 +20,7 @@ import (
 	"google.golang.org/grpc"
 
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
+	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	"go.etcd.io/etcd/client/pkg/v3/types"
 )
 
@@ -93,7 +94,7 @@ func (c *cluster) memberAdd(ctx context.Context, peerAddrs []string, isLearner b
 	}
 	resp, err := c.remote.MemberAdd(ctx, r, c.callOpts...)
 	if err != nil {
-		return nil, toErr(ctx, err)
+		return nil, rpctypes.ContextError(ctx, err)
 	}
 	return (*MemberAddResponse)(resp), nil
 }
@@ -102,7 +103,7 @@ func (c *cluster) MemberRemove(ctx context.Context, id uint64) (*MemberRemoveRes
 	r := &pb.MemberRemoveRequest{ID: id}
 	resp, err := c.remote.MemberRemove(ctx, r, c.callOpts...)
 	if err != nil {
-		return nil, toErr(ctx, err)
+		return nil, rpctypes.ContextError(ctx, err)
 	}
 	return (*MemberRemoveResponse)(resp), nil
 }
@@ -119,7 +120,7 @@ func (c *cluster) MemberUpdate(ctx context.Context, id uint64, peerAddrs []strin
 	if err == nil {
 		return (*MemberUpdateResponse)(resp), nil
 	}
-	return nil, toErr(ctx, err)
+	return nil, rpctypes.ContextError(ctx, err)
 }
 
 func (c *cluster) MemberList(ctx context.Context, opts ...OpOption) (*MemberListResponse, error) {
@@ -128,14 +129,14 @@ func (c *cluster) MemberList(ctx context.Context, opts ...OpOption) (*MemberList
 	if err == nil {
 		return (*MemberListResponse)(resp), nil
 	}
-	return nil, toErr(ctx, err)
+	return nil, rpctypes.ContextError(ctx, err)
 }
 
 func (c *cluster) MemberPromote(ctx context.Context, id uint64) (*MemberPromoteResponse, error) {
 	r := &pb.MemberPromoteRequest{ID: id}
 	resp, err := c.remote.MemberPromote(ctx, r, c.callOpts...)
 	if err != nil {
-		return nil, toErr(ctx, err)
+		return nil, rpctypes.ContextError(ctx, err)
 	}
 	return (*MemberPromoteResponse)(resp), nil
 }
diff --git a/client/v3/kubernetes/interface.go b/client/v3/kubernetes/interface.go
@@ -0,0 +1,62 @@
+package kubernetes
+
+import (
+	"context"
+
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
+)
+
+type Interface interface {
+	Get(ctx context.Context, key string, opts GetOptions) (GetResponse, error)
+	List(ctx context.Context, prefix string, opts ListOptions) (ListResponse, error)
+	Count(ctx context.Context, prefix string) (int64, error)
+	OptimisticPut(ctx context.Context, key string, value []byte, opts PutOptions) (PutResponse, error)
+	OptimisticDelete(ctx context.Context, key string, opts DeleteOptions) (DeleteResponse, error)
+}
+
+type GetOptions struct {
+	Revision int64
+}
+
+type ListOptions struct {
+	Revision int64
+	Limit    int64
+	Continue string
+}
+
+type PutOptions struct {
+	ExpectedRevision int64
+	GetOnFailure     bool
+	// LeaseID
+	// Deprecated: Should be replaced with TTL when Interface starts using one lease per object.
+	LeaseID clientv3.LeaseID
+}
+
+type DeleteOptions struct {
+	ExpectedRevision int64
+	GetOnFailure     bool
+}
+
+type GetResponse struct {
+	KV       *mvccpb.KeyValue
+	Revision int64
+}
+
+type ListResponse struct {
+	KVs      []*mvccpb.KeyValue
+	Count    int64
+	Revision int64
+}
+
+type PutResponse struct {
+	KV        *mvccpb.KeyValue
+	Succeeded bool
+	Revision  int64
+}
+
+type DeleteResponse struct {
+	KV        *mvccpb.KeyValue
+	Succeeded bool
+	Revision  int64
+}