Skip to content

K
k3s
Commit bc60def8 authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #40664 from smarterclayton/etcd_data_wrapper

Automatic merge from submit-queue (batch tested with PRs 41061, 40888, 40664, 41020, 41085) Allow values to be wrapped prior to serialization in etcd This adds a new value transformer to the etcd2 store that can transform the value from etcd on read and write. This will allow the store to implement encryption at rest or otherwise transform the value prior to persistence. * [x] etcd3 store * [x] example of transformation * [x] partial error handling This is in support of #12742
parents b58a2074 bc4b5064
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 395 additions and 75 deletions
staging/src/k8s.io/apiserver/pkg/storage/errors/storage.go
View file @ bc60def8
......@@ -30,6 +30,8 @@ func InterpretListError(err error, qualifiedResource schema.GroupResource) error
return errors.NewNotFound(qualifiedResource, "")
case storage.IsUnreachable(err):
return errors.NewServerTimeout(qualifiedResource, "list", 2) // TODO: make configurable or handled at a higher level
case storage.IsInternalError(err):
return errors.NewInternalError(err)
default:
return err
}
......@@ -43,6 +45,8 @@ func InterpretGetError(err error, qualifiedResource schema.GroupResource, name s
return errors.NewNotFound(qualifiedResource, name)
case storage.IsUnreachable(err):
return errors.NewServerTimeout(qualifiedResource, "get", 2) // TODO: make configurable or handled at a higher level
case storage.IsInternalError(err):
return errors.NewInternalError(err)
default:
return err
}
......@@ -56,6 +60,8 @@ func InterpretCreateError(err error, qualifiedResource schema.GroupResource, nam
return errors.NewAlreadyExists(qualifiedResource, name)
case storage.IsUnreachable(err):
return errors.NewServerTimeout(qualifiedResource, "create", 2) // TODO: make configurable or handled at a higher level
case storage.IsInternalError(err):
return errors.NewInternalError(err)
default:
return err
}
......@@ -102,6 +108,8 @@ func InterpretWatchError(err error, resource schema.GroupResource, name string)
case storage.IsInvalidError(err):
invalidError, _ := err.(storage.InvalidError)
return errors.NewInvalid(schema.GroupKind{Group: resource.Group, Kind: resource.Resource}, name, invalidError.Errs)
case storage.IsInternalError(err):
return errors.NewInternalError(err)
default:
return err
}
......
staging/src/k8s.io/apiserver/pkg/storage/etcd/etcd_helper_test.go
View file @ bc60def8
......@@ -17,8 +17,10 @@ limitations under the License.
package etcd
import (
"fmt"
"path"
"reflect"
"strings"
"sync"
"testing"
"time"
......@@ -33,6 +35,7 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/apimachinery/pkg/util/diff"
"k8s.io/apiserver/pkg/apis/example"
examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
"k8s.io/apiserver/pkg/storage"
......@@ -42,6 +45,34 @@ import (
storagetests "k8s.io/apiserver/pkg/storage/tests"
)
// prefixTransformer adds and verifies that all data has the correct prefix on its way in and out.
type prefixTransformer struct {
prefix string
stale bool
err error
}
func (p prefixTransformer) TransformStringFromStorage(s string) (string, bool, error) {
if !strings.HasPrefix(s, p.prefix) {
return "", false, fmt.Errorf("value does not have expected prefix: %s", s)
}
return strings.TrimPrefix(s, p.prefix), p.stale, p.err
}
func (p prefixTransformer) TransformStringToStorage(s string) (string, error) {
if len(s) > 0 {
return p.prefix + s, p.err
}
return s, p.err
}
func defaultPrefix(s string) string {
return "test!" + s
}
func defaultPrefixValue(value []byte) string {
return defaultPrefix(string(value))
}
func testScheme(t *testing.T) (*runtime.Scheme, serializer.CodecFactory) {
scheme := runtime.NewScheme()
scheme.Log(t)
......@@ -66,7 +97,7 @@ func testScheme(t *testing.T) (*runtime.Scheme, serializer.CodecFactory) {
}
func newEtcdHelper(client etcd.Client, scheme *runtime.Scheme, codec runtime.Codec, prefix string) etcdHelper {
return *NewEtcdStorage(client, codec, prefix, false, etcdtest.DeserializationCacheSize, scheme).(*etcdHelper)
return *NewEtcdStorage(client, codec, prefix, false, etcdtest.DeserializationCacheSize, scheme, prefixTransformer{prefix: "test!"}).(*etcdHelper)
}
// Returns an encoded version of example.Pod with the given name.
......@@ -135,6 +166,61 @@ func TestList(t *testing.T) {
}
}
func TestTransformationFailure(t *testing.T) {
scheme, codecs := testScheme(t)
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
server := etcdtesting.NewEtcdTestClientServer(t)
defer server.Terminate(t)
helper := newEtcdHelper(server.Client, scheme, codec, etcdtest.PathPrefix())
pods := []example.Pod{
{
ObjectMeta: metav1.ObjectMeta{Name: "bar"},
Spec: storagetests.DeepEqualSafePodSpec(),
},
{
ObjectMeta: metav1.ObjectMeta{Name: "baz"},
Spec: storagetests.DeepEqualSafePodSpec(),
},
}
createPodList(t, helper, &example.PodList{Items: pods[:1]})
// create a second resource with an invalid prefix
oldTransformer := helper.transformer
helper.transformer = prefixTransformer{prefix: "otherprefix!"}
createPodList(t, helper, &example.PodList{Items: pods[1:]})
helper.transformer = oldTransformer
// only the first item is returned, and no error
var got example.PodList
if err := helper.List(context.TODO(), "/", "", storage.Everything, &got); err != nil {
t.Errorf("Unexpected error %v", err)
}
if e, a := pods[:1], got.Items; !reflect.DeepEqual(e, a) {
t.Errorf("Unexpected: %s", diff.ObjectReflectDiff(e, a))
}
// Get should fail
if err := helper.Get(context.TODO(), "/baz", "", &example.Pod{}, false); !storage.IsInternalError(err) {
t.Errorf("Unexpected error: %v", err)
}
// GuaranteedUpdate should return an error
if err := helper.GuaranteedUpdate(context.TODO(), "/baz", &example.Pod{}, false, nil, func(input runtime.Object, res storage.ResponseMeta) (output runtime.Object, ttl *uint64, err error) {
return input, nil, nil
}, &pods[1]); !storage.IsInternalError(err) {
t.Errorf("Unexpected error: %v", err)
}
// Delete succeeds but reports an error because we cannot access the body
if err := helper.Delete(context.TODO(), "/baz", &example.Pod{}, nil); !storage.IsInternalError(err) {
t.Errorf("Unexpected error: %v", err)
}
if err := helper.Get(context.TODO(), "/baz", "", &example.Pod{}, false); !storage.IsNotFound(err) {
t.Errorf("Unexpected error: %v", err)
}
}
func TestListFiltered(t *testing.T) {
scheme, codecs := testScheme(t)
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
......@@ -364,7 +450,8 @@ func TestGuaranteedUpdateNoChange(t *testing.T) {
helper := newEtcdHelper(server.Client, scheme, codec, etcdtest.PathPrefix())
obj := &storagetesting.TestResource{ObjectMeta: metav1.ObjectMeta{Name: "foo"}, Value: 1}
err := helper.GuaranteedUpdate(context.TODO(), key, &storagetesting.TestResource{}, true, nil, storage.SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
original := &storagetesting.TestResource{}
err := helper.GuaranteedUpdate(context.TODO(), key, original, true, nil, storage.SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
return obj, nil
}))
if err != nil {
......@@ -373,7 +460,27 @@ func TestGuaranteedUpdateNoChange(t *testing.T) {
// Update an existing node with the same data
callbackCalled := false
objUpdate := &storagetesting.TestResource{ObjectMeta: metav1.ObjectMeta{Name: "foo"}, Value: 1}
objUpdate := &storagetesting.TestResource{ObjectMeta: metav1.ObjectMeta{Name: "foo", ResourceVersion: original.ResourceVersion}, Value: 1}
result := &storagetesting.TestResource{}
err = helper.GuaranteedUpdate(context.TODO(), key, result, true, nil, storage.SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
callbackCalled = true
return objUpdate, nil
}))
if err != nil {
t.Fatalf("Unexpected error %#v", err)
}
if !callbackCalled {
t.Errorf("tryUpdate callback should have been called.")
}
if result.ResourceVersion != original.ResourceVersion {
t.Fatalf("updated the object resource version")
}
// Update an existing node with the same data but return stale
helper.transformer = prefixTransformer{prefix: "test!", stale: true}
callbackCalled = false
result = &storagetesting.TestResource{}
objUpdate = &storagetesting.TestResource{ObjectMeta: metav1.ObjectMeta{Name: "foo"}, Value: 1}
err = helper.GuaranteedUpdate(context.TODO(), key, &storagetesting.TestResource{}, true, nil, storage.SimpleUpdate(func(in runtime.Object) (runtime.Object, error) {
callbackCalled = true
return objUpdate, nil
......@@ -384,6 +491,9 @@ func TestGuaranteedUpdateNoChange(t *testing.T) {
if !callbackCalled {
t.Errorf("tryUpdate callback should have been called.")
}
if result.ResourceVersion == original.ResourceVersion {
t.Errorf("did not update the object resource version")
}
}
func TestGuaranteedUpdateKeyNotFound(t *testing.T) {
......@@ -540,7 +650,7 @@ func TestDeleteWithRetry(t *testing.T) {
// party has updated the object.
fakeGet := func(ctx context.Context, key string, opts *etcd.GetOptions) (*etcd.Response, error) {
data, _ := runtime.Encode(codec, obj)
return &etcd.Response{Node: &etcd.Node{Value: string(data), ModifiedIndex: 99}}, nil
return &etcd.Response{Node: &etcd.Node{Value: defaultPrefixValue(data), ModifiedIndex: 99}}, nil
}
expectedRetries := 3
helper := newEtcdHelper(server.Client, scheme, codec, prefix)
......
staging/src/k8s.io/apiserver/pkg/storage/etcd/etcd_watcher.go
View file @ bc60def8
......@@ -71,8 +71,9 @@ type etcdWatcher struct {
// Note that versioner is required for etcdWatcher to work correctly.
// There is no public constructor of it, so be careful when manipulating
// with it manually.
versioner storage.Versioner
transform TransformFunc
versioner storage.Versioner
transform TransformFunc
valueTransformer ValueTransformer
list bool // If we're doing a recursive watch, should be true.
quorum bool // If we enable quorum, shoule be true
......@@ -107,15 +108,18 @@ const watchWaitDuration = 100 * time.Millisecond
func newEtcdWatcher(
list bool, quorum bool, include includeFunc, filter storage.FilterFunc,
encoding runtime.Codec, versioner storage.Versioner, transform TransformFunc,
valueTransformer ValueTransformer,
cache etcdCache) *etcdWatcher {
w := &etcdWatcher{
encoding: encoding,
versioner: versioner,
transform: transform,
list: list,
quorum: quorum,
include: include,
filter: filter,
encoding: encoding,
versioner: versioner,
transform: transform,
valueTransformer: valueTransformer,
list: list,
quorum: quorum,
include: include,
filter: filter,
// Buffer this channel, so that the etcd client is not forced
// to context switch with every object it gets, and so that a
// long time spent decoding an object won't block the *next*
......@@ -309,12 +313,18 @@ func (w *etcdWatcher) translate() {
}
}
// decodeObject extracts an object from the provided etcd node or returns an error.
func (w *etcdWatcher) decodeObject(node *etcd.Node) (runtime.Object, error) {
if obj, found := w.cache.getFromCache(node.ModifiedIndex, storage.SimpleFilter(storage.Everything)); found {
return obj, nil
}
obj, err := runtime.Decode(w.encoding, []byte(node.Value))
body, _, err := w.valueTransformer.TransformStringFromStorage(node.Value)
if err != nil {
return nil, err
}
obj, err := runtime.Decode(w.encoding, []byte(body))
if err != nil {
return nil, err
}
......
staging/src/k8s.io/apiserver/pkg/storage/etcd/etcd_watcher_test.go
View file @ bc60def8
......@@ -133,7 +133,7 @@ func TestWatchInterpretations(t *testing.T) {
}
for name, item := range table {
for _, action := range item.actions {
w := newEtcdWatcher(true, false, nil, firstLetterIsB, codec, versioner, nil, &fakeEtcdCache{})
w := newEtcdWatcher(true, false, nil, firstLetterIsB, codec, versioner, nil, prefixTransformer{prefix: "test!"}, &fakeEtcdCache{})
emitCalled := false
w.emit = func(event watch.Event) {
emitCalled = true
......@@ -150,10 +150,10 @@ func TestWatchInterpretations(t *testing.T) {
var n, pn *etcd.Node
if item.nodeValue != "" {
n = &etcd.Node{Value: item.nodeValue}
n = &etcd.Node{Value: defaultPrefix(item.nodeValue)}
}
if item.prevNodeValue != "" {
pn = &etcd.Node{Value: item.prevNodeValue}
pn = &etcd.Node{Value: defaultPrefix(item.prevNodeValue)}
}
w.sendResult(&etcd.Response{
......@@ -173,7 +173,7 @@ func TestWatchInterpretations(t *testing.T) {
func TestWatchInterpretation_ResponseNotSet(t *testing.T) {
_, codecs := testScheme(t)
codec := codecs.LegacyCodec(schema.GroupVersion{Version: "v1"})
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, &fakeEtcdCache{})
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, prefixTransformer{prefix: "test!"}, &fakeEtcdCache{})
w.emit = func(e watch.Event) {
t.Errorf("Unexpected emit: %v", e)
}
......@@ -189,7 +189,7 @@ func TestWatchInterpretation_ResponseNoNode(t *testing.T) {
codec := codecs.LegacyCodec(schema.GroupVersion{Version: "v1"})
actions := []string{"create", "set", "compareAndSwap", "delete"}
for _, action := range actions {
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, &fakeEtcdCache{})
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, prefixTransformer{prefix: "test!"}, &fakeEtcdCache{})
w.emit = func(e watch.Event) {
t.Errorf("Unexpected emit: %v", e)
}
......@@ -205,20 +205,20 @@ func TestWatchInterpretation_ResponseBadData(t *testing.T) {
codec := codecs.LegacyCodec(schema.GroupVersion{Version: "v1"})
actions := []string{"create", "set", "compareAndSwap", "delete"}
for _, action := range actions {
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, &fakeEtcdCache{})
w := newEtcdWatcher(false, false, nil, storage.SimpleFilter(storage.Everything), codec, versioner, nil, prefixTransformer{prefix: "test!"}, &fakeEtcdCache{})
w.emit = func(e watch.Event) {
t.Errorf("Unexpected emit: %v", e)
}
w.sendResult(&etcd.Response{
Action: action,
Node: &etcd.Node{
Value: "foobar",
Value: defaultPrefix("foobar"),
},
})
w.sendResult(&etcd.Response{
Action: action,
PrevNode: &etcd.Node{
Value: "foobar",
Value: defaultPrefix("foobar"),
},
})
w.Stop()
......@@ -231,7 +231,7 @@ func TestSendResultDeleteEventHaveLatestIndex(t *testing.T) {
filter := func(obj runtime.Object) bool {
return obj.(*example.Pod).Name != "bar"
}
w := newEtcdWatcher(false, false, nil, filter, codec, versioner, nil, &fakeEtcdCache{})
w := newEtcdWatcher(false, false, nil, filter, codec, versioner, nil, prefixTransformer{prefix: "test!"}, &fakeEtcdCache{})
eventChan := make(chan watch.Event, 1)
w.emit = func(e watch.Event) {
......@@ -259,7 +259,7 @@ func TestSendResultDeleteEventHaveLatestIndex(t *testing.T) {
ModifiedIndex: 2,
},
PrevNode: &etcd.Node{
Value: string(fooBytes),
Value: defaultPrefixValue(fooBytes),
ModifiedIndex: 1,
},
},
......@@ -268,11 +268,11 @@ func TestSendResultDeleteEventHaveLatestIndex(t *testing.T) {
response: &etcd.Response{
Action: EtcdSet,
Node: &etcd.Node{
Value: string(barBytes),
Value: defaultPrefixValue(barBytes),
ModifiedIndex: 2,
},
PrevNode: &etcd.Node{
Value: string(fooBytes),
Value: defaultPrefixValue(fooBytes),
ModifiedIndex: 1,
},
},
......
staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go
View file @ bc60def8
......@@ -33,21 +33,42 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/conversion"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/apiserver/pkg/storage"
utiltrace "k8s.io/apiserver/pkg/util/trace"
"k8s.io/apiserver/pkg/storage/etcd"
utiltrace "k8s.io/apiserver/pkg/util/trace"
)
// ValueTransformer allows a string value to be transformed before being read from or written to the underlying store. The methods
// must be able to undo the transformation caused by the other.
type ValueTransformer interface {
// TransformFromStorage may transform the provided data from its underlying storage representation or return an error.
// Stale is true if the object on disk is stale and a write to etcd should be issued, even if the contents of the object
// have not changed.
TransformFromStorage([]byte) (data []byte, stale bool, err error)
// TransformToStorage may transform the provided data into the appropriate form in storage or return an error.
TransformToStorage([]byte) (data []byte, err error)
}
type identityTransformer struct{}
func (identityTransformer) TransformFromStorage(b []byte) ([]byte, bool, error) { return b, false, nil }
func (identityTransformer) TransformToStorage(b []byte) ([]byte, error) { return b, nil }
// IdentityTransformer performs no transformation on the provided values.
var IdentityTransformer ValueTransformer = identityTransformer{}
type store struct {
client *clientv3.Client
// getOpts contains additional options that should be passed
// to all Get() calls.
getOps []clientv3.OpOption
codec runtime.Codec
versioner storage.Versioner
pathPrefix string
watcher *watcher
getOps []clientv3.OpOption
codec runtime.Codec
versioner storage.Versioner
transformer ValueTransformer
pathPrefix string
watcher *watcher
}
type elemForDecode struct {
......@@ -56,31 +77,33 @@ type elemForDecode struct {
}
type objState struct {
obj runtime.Object
meta *storage.ResponseMeta
rev int64
data []byte
obj runtime.Object
meta *storage.ResponseMeta
rev int64
data []byte
stale bool
}
// New returns an etcd3 implementation of storage.Interface.
func New(c *clientv3.Client, codec runtime.Codec, prefix string) storage.Interface {
return newStore(c, true, codec, prefix)
func New(c *clientv3.Client, codec runtime.Codec, prefix string, transformer ValueTransformer) storage.Interface {
return newStore(c, true, codec, prefix, transformer)
}
// NewWithNoQuorumRead returns etcd3 implementation of storage.Interface
// where Get operations don't require quorum read.
func NewWithNoQuorumRead(c *clientv3.Client, codec runtime.Codec, prefix string) storage.Interface {
return newStore(c, false, codec, prefix)
func NewWithNoQuorumRead(c *clientv3.Client, codec runtime.Codec, prefix string, transformer ValueTransformer) storage.Interface {
return newStore(c, false, codec, prefix, transformer)
}
func newStore(c *clientv3.Client, quorumRead bool, codec runtime.Codec, prefix string) *store {
func newStore(c *clientv3.Client, quorumRead bool, codec runtime.Codec, prefix string, transformer ValueTransformer) *store {
versioner := etcd.APIObjectVersioner{}
result := &store{
client: c,
versioner: versioner,
codec: codec,
pathPrefix: prefix,
watcher: newWatcher(c, codec, versioner),
client: c,
codec: codec,
versioner: versioner,
transformer: transformer,
pathPrefix: prefix,
watcher: newWatcher(c, codec, versioner, transformer),
}
if !quorumRead {
// In case of non-quorum reads, we can set WithSerializable()
......@@ -110,7 +133,13 @@ func (s *store) Get(ctx context.Context, key string, resourceVersion string, out
return storage.NewKeyNotFoundError(key, 0)
}
kv := getResp.Kvs[0]
return decode(s.codec, s.versioner, kv.Value, out, kv.ModRevision)
data, _, err := s.transformer.TransformFromStorage(kv.Value)
if err != nil {
return storage.NewInternalError(err.Error())
}
return decode(s.codec, s.versioner, data, out, kv.ModRevision)
}
// Create implements storage.Interface.Create.
......@@ -129,10 +158,15 @@ func (s *store) Create(ctx context.Context, key string, obj, out runtime.Object,
return err
}
newData, err := s.transformer.TransformToStorage(data)
if err != nil {
return storage.NewInternalError(err.Error())
}
txnResp, err := s.client.KV.Txn(ctx).If(
notFound(key),
).Then(
clientv3.OpPut(key, string(data), opts...),
clientv3.OpPut(key, string(newData), opts...),
).Commit()
if err != nil {
return err
......@@ -177,7 +211,11 @@ func (s *store) unconditionalDelete(ctx context.Context, key string, out runtime
}
kv := getResp.Kvs[0]
return decode(s.codec, s.versioner, kv.Value, out, kv.ModRevision)
data, _, err := s.transformer.TransformFromStorage(kv.Value)
if err != nil {
return storage.NewInternalError(err.Error())
}
return decode(s.codec, s.versioner, data, out, kv.ModRevision)
}
func (s *store) conditionalDelete(ctx context.Context, key string, out runtime.Object, v reflect.Value, precondtions *storage.Preconditions) error {
......@@ -257,10 +295,15 @@ func (s *store) GuaranteedUpdate(
if err != nil {
return err
}
if bytes.Equal(data, origState.data) {
if !origState.stale && bytes.Equal(data, origState.data) {
return decode(s.codec, s.versioner, origState.data, out, origState.rev)
}
newData, err := s.transformer.TransformToStorage(data)
if err != nil {
return storage.NewInternalError(err.Error())
}
opts, err := s.ttlOpts(ctx, int64(ttl))
if err != nil {
return err
......@@ -270,7 +313,7 @@ func (s *store) GuaranteedUpdate(
txnResp, err := s.client.KV.Txn(ctx).If(
clientv3.Compare(clientv3.ModRevision(key), "=", origState.rev),
).Then(
clientv3.OpPut(key, string(data), opts...),
clientv3.OpPut(key, string(newData), opts...),
).Else(
clientv3.OpGet(key),
).Commit()
......@@ -289,6 +332,7 @@ func (s *store) GuaranteedUpdate(
continue
}
putResp := txnResp.Responses[0].GetResponsePut()
return decode(s.codec, s.versioner, data, out, putResp.Header.Revision)
}
}
......@@ -308,8 +352,12 @@ func (s *store) GetToList(ctx context.Context, key string, resourceVersion strin
if len(getResp.Kvs) == 0 {
return nil
}
data, _, err := s.transformer.TransformFromStorage(getResp.Kvs[0].Value)
if err != nil {
return storage.NewInternalError(err.Error())
}
elems := []*elemForDecode{{
data: getResp.Kvs[0].Value,
data: data,
rev: uint64(getResp.Kvs[0].ModRevision),
}}
if err := decodeList(elems, storage.SimpleFilter(pred), listPtr, s.codec, s.versioner); err != nil {
......@@ -337,12 +385,18 @@ func (s *store) List(ctx context.Context, key, resourceVersion string, pred stor
return err
}
elems := make([]*elemForDecode, len(getResp.Kvs))
for i, kv := range getResp.Kvs {
elems[i] = &elemForDecode{
data: kv.Value,
rev: uint64(kv.ModRevision),
elems := make([]*elemForDecode, 0, len(getResp.Kvs))
for _, kv := range getResp.Kvs {
data, _, err := s.transformer.TransformFromStorage(kv.Value)
if err != nil {
utilruntime.HandleError(fmt.Errorf("unable to transform key %q: %v", key, err))
continue
}
elems = append(elems, &elemForDecode{
data: data,
rev: uint64(kv.ModRevision),
})
}
if err := decodeList(elems, storage.SimpleFilter(pred), listPtr, s.codec, s.versioner); err != nil {
return err
......@@ -383,9 +437,14 @@ func (s *store) getState(getResp *clientv3.GetResponse, key string, v reflect.Va
return nil, err
}
} else {
data, stale, err := s.transformer.TransformFromStorage(getResp.Kvs[0].Value)
if err != nil {
return nil, storage.NewInternalError(err.Error())
}
state.rev = getResp.Kvs[0].ModRevision
state.meta.ResourceVersion = uint64(state.rev)
state.data = getResp.Kvs[0].Value
state.data = data
state.stale = stale
if err := decode(s.codec, s.versioner, state.data, state.obj, state.rev); err != nil {
return nil, err
}
......
staging/src/k8s.io/apiserver/pkg/storage/etcd3/store_test.go
View file @ bc60def8
......@@ -17,6 +17,7 @@ limitations under the License.
package etcd3
import (
"bytes"
"fmt"
"reflect"
"sync"
......@@ -28,10 +29,12 @@ import (
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/apimachinery/pkg/util/diff"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/apiserver/pkg/apis/example"
examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
"k8s.io/apiserver/pkg/storage"
storagetests "k8s.io/apiserver/pkg/storage/tests"
"github.com/coreos/etcd/integration"
"golang.org/x/net/context"
......@@ -46,6 +49,26 @@ func init() {
examplev1.AddToScheme(scheme)
}
// prefixTransformer adds and verifies that all data has the correct prefix on its way in and out.
type prefixTransformer struct {
prefix []byte
stale bool
err error
}
func (p prefixTransformer) TransformFromStorage(b []byte) ([]byte, bool, error) {
if !bytes.HasPrefix(b, p.prefix) {
return nil, false, fmt.Errorf("value does not have expected prefix: %s", string(b))
}
return bytes.TrimPrefix(b, p.prefix), p.stale, p.err
}
func (p prefixTransformer) TransformToStorage(b []byte) ([]byte, error) {
if len(b) > 0 {
return append(append([]byte{}, p.prefix...), b...), p.err
}
return b, p.err
}
func TestCreate(t *testing.T) {
ctx, store, cluster := testSetup(t)
defer cluster.Terminate(t)
......@@ -294,6 +317,7 @@ func TestGuaranteedUpdate(t *testing.T) {
expectNotFoundErr bool
expectInvalidObjErr bool
expectNoUpdate bool
transformStale bool
}{{ // GuaranteedUpdate on non-existing key with ignoreNotFound=false
key: "/non-existing",
ignoreNotFound: false,
......@@ -322,6 +346,14 @@ func TestGuaranteedUpdate(t *testing.T) {
expectNotFoundErr: false,
expectInvalidObjErr: false,
expectNoUpdate: true,
}, { // GuaranteedUpdate with same data but stale
key: key,
ignoreNotFound: false,
precondition: nil,
expectNotFoundErr: false,
expectInvalidObjErr: false,
expectNoUpdate: false,
transformStale: true,
}, { // GuaranteedUpdate with UID match
key: key,
ignoreNotFound: false,
......@@ -344,6 +376,12 @@ func TestGuaranteedUpdate(t *testing.T) {
if tt.expectNoUpdate {
name = storeObj.Name
}
originalTransformer := store.transformer.(prefixTransformer)
if tt.transformStale {
transformer := originalTransformer
transformer.stale = true
store.transformer = transformer
}
version := storeObj.ResourceVersion
err := store.GuaranteedUpdate(ctx, tt.key, out, tt.ignoreNotFound, tt.precondition,
storage.SimpleUpdate(func(obj runtime.Object) (runtime.Object, error) {
......@@ -356,6 +394,7 @@ func TestGuaranteedUpdate(t *testing.T) {
pod.Name = name
return &pod, nil
}))
store.transformer = originalTransformer
if tt.expectNotFoundErr {
if err == nil || !storage.IsNotFound(err) {
......@@ -460,11 +499,91 @@ func TestGuaranteedUpdateWithConflict(t *testing.T) {
}
}
func TestTransformationFailure(t *testing.T) {
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
cluster := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
defer cluster.Terminate(t)
store := newStore(cluster.RandClient(), false, codec, "", prefixTransformer{prefix: []byte("test!")})
ctx := context.Background()
preset := []struct {
key string
obj *example.Pod
storedObj *example.Pod
}{{
key: "/one-level/test",
obj: &example.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "bar"},
Spec: storagetests.DeepEqualSafePodSpec(),
},
}, {
key: "/two-level/1/test",
obj: &example.Pod{
ObjectMeta: metav1.ObjectMeta{Name: "baz"},
Spec: storagetests.DeepEqualSafePodSpec(),
},
}}
for i, ps := range preset[:1] {
preset[i].storedObj = &example.Pod{}
err := store.Create(ctx, ps.key, ps.obj, preset[:1][i].storedObj, 0)
if err != nil {
t.Fatalf("Set failed: %v", err)
}
}
// create a second resource with an invalid prefix
oldTransformer := store.transformer
store.transformer = prefixTransformer{prefix: []byte("otherprefix!")}
for i, ps := range preset[1:] {
preset[1:][i].storedObj = &example.Pod{}
err := store.Create(ctx, ps.key, ps.obj, preset[1:][i].storedObj, 0)
if err != nil {
t.Fatalf("Set failed: %v", err)
}
}
store.transformer = oldTransformer
// only the first item is returned, and no error
var got example.PodList
if err := store.List(ctx, "/", "", storage.Everything, &got); err != nil {
t.Errorf("Unexpected error %v", err)
}
if e, a := []example.Pod{*preset[0].storedObj}, got.Items; !reflect.DeepEqual(e, a) {
t.Errorf("Unexpected: %s", diff.ObjectReflectDiff(e, a))
}
// Get should fail
if err := store.Get(ctx, preset[1].key, "", &example.Pod{}, false); !storage.IsInternalError(err) {
t.Errorf("Unexpected error: %v", err)
}
// GuaranteedUpdate without suggestion should return an error
if err := store.GuaranteedUpdate(ctx, preset[1].key, &example.Pod{}, false, nil, func(input runtime.Object, res storage.ResponseMeta) (output runtime.Object, ttl *uint64, err error) {
return input, nil, nil
}); !storage.IsInternalError(err) {
t.Errorf("Unexpected error: %v", err)
}
// GuaranteedUpdate with suggestion should not return an error if we don't change the object
if err := store.GuaranteedUpdate(ctx, preset[1].key, &example.Pod{}, false, nil, func(input runtime.Object, res storage.ResponseMeta) (output runtime.Object, ttl *uint64, err error) {
return input, nil, nil
}, preset[1].obj); err != nil {
t.Errorf("Unexpected error: %v", err)
}
// Delete succeeds but reports an error because we cannot access the body
if err := store.Delete(ctx, preset[1].key, &example.Pod{}, nil); !storage.IsInternalError(err) {
t.Errorf("Unexpected error: %v", err)
}
if err := store.Get(ctx, preset[1].key, "", &example.Pod{}, false); !storage.IsNotFound(err) {
t.Errorf("Unexpected error: %v", err)
}
}
func TestList(t *testing.T) {
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
cluster := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
defer cluster.Terminate(t)
store := newStore(cluster.RandClient(), false, codec, "")
store := newStore(cluster.RandClient(), false, codec, "", prefixTransformer{prefix: []byte("test!")})
ctx := context.Background()
// Setup storage with the following structure:
......@@ -552,7 +671,7 @@ func TestList(t *testing.T) {
func testSetup(t *testing.T) (context.Context, *store, *integration.ClusterV3) {
codec := apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)
cluster := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
store := newStore(cluster.RandClient(), false, codec, "")
store := newStore(cluster.RandClient(), false, codec, "", prefixTransformer{prefix: []byte("test!")})
ctx := context.Background()
return ctx, store, cluster
}
......
staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go
View file @ bc60def8
......@@ -40,9 +40,10 @@ const (
)
type watcher struct {
client *clientv3.Client
codec runtime.Codec
versioner storage.Versioner
client *clientv3.Client
codec runtime.Codec
versioner storage.Versioner
transformer ValueTransformer
}
// watchChan implements watch.Interface.
......@@ -59,11 +60,12 @@ type watchChan struct {
errChan chan error
}
func newWatcher(client *clientv3.Client, codec runtime.Codec, versioner storage.Versioner) *watcher {
func newWatcher(client *clientv3.Client, codec runtime.Codec, versioner storage.Versioner, transformer ValueTransformer) *watcher {
return &watcher{
client: client,
codec: codec,
versioner: versioner,
client: client,
codec: codec,
versioner: versioner,
transformer: transformer,
}
}
......@@ -341,7 +343,11 @@ func (wc *watchChan) sendEvent(e *event) {
func (wc *watchChan) prepareObjs(e *event) (curObj runtime.Object, oldObj runtime.Object, err error) {
if !e.isDeleted {
curObj, err = decodeObj(wc.watcher.codec, wc.watcher.versioner, e.value, e.rev)
data, _, err := wc.watcher.transformer.TransformFromStorage(e.value)
if err != nil {
return nil, nil, err
}
curObj, err = decodeObj(wc.watcher.codec, wc.watcher.versioner, data, e.rev)
if err != nil {
return nil, nil, err
}
......@@ -352,9 +358,13 @@ func (wc *watchChan) prepareObjs(e *event) (curObj runtime.Object, oldObj runtim
// we need the object only to compute whether it was filtered out
// before).
if len(e.prevValue) > 0 && (e.isDeleted || !wc.acceptAll()) {
data, _, err := wc.watcher.transformer.TransformFromStorage(e.prevValue)
if err != nil {
return nil, nil, err
}
// Note that this sends the *old* object with the etcd revision for the time at
// which it gets deleted.
oldObj, err = decodeObj(wc.watcher.codec, wc.watcher.versioner, e.prevValue, e.rev)
oldObj, err = decodeObj(wc.watcher.codec, wc.watcher.versioner, data, e.rev)
if err != nil {
return nil, nil, err
}
......
staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher_test.go
View file @ bc60def8
......@@ -227,13 +227,13 @@ func TestWatchError(t *testing.T) {
codec := &testCodec{apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion)}
cluster := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
defer cluster.Terminate(t)
invalidStore := newStore(cluster.RandClient(), false, codec, "")
invalidStore := newStore(cluster.RandClient(), false, codec, "", prefixTransformer{prefix: []byte("test!")})
ctx := context.Background()
w, err := invalidStore.Watch(ctx, "/abc", "0", storage.Everything)
if err != nil {
t.Fatalf("Watch failed: %v", err)
}
validStore := newStore(cluster.RandClient(), false, codec, "")
validStore := newStore(cluster.RandClient(), false, codec, "", prefixTransformer{prefix: []byte("test!")})
validStore.GuaranteedUpdate(ctx, "/abc", &example.Pod{}, true, nil, storage.SimpleUpdate(
func(runtime.Object) (runtime.Object, error) {
return &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "foo"}}, nil
......
staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/etcd2.go
View file @ bc60def8
......@@ -39,7 +39,7 @@ func newETCD2Storage(c storagebackend.Config) (storage.Interface, DestroyFunc, e
if err != nil {
return nil, nil, err
}
s := etcd.NewEtcdStorage(client, c.Codec, c.Prefix, c.Quorum, c.DeserializationCacheSize, c.Copier)
s := etcd.NewEtcdStorage(client, c.Codec, c.Prefix, c.Quorum, c.DeserializationCacheSize, c.Copier, etcd.IdentityTransformer)
return s, tr.CloseIdleConnections, nil
}
......
staging/src/k8s.io/apiserver/pkg/storage/storagebackend/factory/etcd3.go
View file @ bc60def8
......@@ -56,7 +56,7 @@ func newETCD3Storage(c storagebackend.Config) (storage.Interface, DestroyFunc, e
client.Close()
}
if c.Quorum {
return etcd3.New(client, c.Codec, c.Prefix), destroyFunc, nil
return etcd3.New(client, c.Codec, c.Prefix, etcd3.IdentityTransformer), destroyFunc, nil
}
return etcd3.NewWithNoQuorumRead(client, c.Codec, c.Prefix), destroyFunc, nil
return etcd3.NewWithNoQuorumRead(client, c.Codec, c.Prefix, etcd3.IdentityTransformer), destroyFunc, nil
}
staging/src/k8s.io/apiserver/pkg/storage/tests/cacher_test.go
View file @ bc60def8
......@@ -93,7 +93,7 @@ func AddObjectMetaFieldsSet(source fields.Set, objectMeta *metav1.ObjectMeta, ha
func newEtcdTestStorage(t *testing.T, prefix string) (*etcdtesting.EtcdTestServer, storage.Interface) {
server, _ := etcdtesting.NewUnsecuredEtcd3TestClientServer(t, scheme)
storage := etcd3.New(server.V3Client, apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion), prefix)
storage := etcd3.New(server.V3Client, apitesting.TestCodec(codecs, examplev1.SchemeGroupVersion), prefix, etcd3.IdentityTransformer)
return server, storage
}
......
vendor/BUILD
View file @ bc60def8
......@@ -14930,6 +14930,7 @@ go_test(
"//vendor:k8s.io/apimachinery/pkg/runtime",
"//vendor:k8s.io/apimachinery/pkg/runtime/schema",
"//vendor:k8s.io/apimachinery/pkg/runtime/serializer",
"//vendor:k8s.io/apimachinery/pkg/util/diff",
"//vendor:k8s.io/apimachinery/pkg/watch",
"//vendor:k8s.io/apiserver/pkg/apis/example",
"//vendor:k8s.io/apiserver/pkg/apis/example/v1",
......@@ -15048,11 +15049,13 @@ go_test(
"//vendor:k8s.io/apimachinery/pkg/runtime",
"//vendor:k8s.io/apimachinery/pkg/runtime/schema",
"//vendor:k8s.io/apimachinery/pkg/runtime/serializer",
"//vendor:k8s.io/apimachinery/pkg/util/diff",
"//vendor:k8s.io/apimachinery/pkg/util/wait",
"//vendor:k8s.io/apimachinery/pkg/watch",
"//vendor:k8s.io/apiserver/pkg/apis/example",
"//vendor:k8s.io/apiserver/pkg/apis/example/v1",
"//vendor:k8s.io/apiserver/pkg/storage",
"//vendor:k8s.io/apiserver/pkg/storage/tests",
],
)
......@@ -15075,6 +15078,7 @@ go_library(
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
"//vendor:k8s.io/apimachinery/pkg/conversion",
"//vendor:k8s.io/apimachinery/pkg/runtime",
"//vendor:k8s.io/apimachinery/pkg/util/runtime",
"//vendor:k8s.io/apimachinery/pkg/watch",
"//vendor:k8s.io/apiserver/pkg/storage",
"//vendor:k8s.io/apiserver/pkg/storage/etcd",
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment