mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
api: Annotate proto messages and use it to detect etcd version that generated wal
Co-authored-by: Lili Cosic <cosiclili@gmail.com>
This commit is contained in:
Marek Siarkowicz
158
server/storage/schema/wal.go
Normal file
158
server/storage/schema/wal.go
Normal file
@@ -0,0 +1,158 @@
|
||||
// Copyright 2021 The etcd Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package schema
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/coreos/go-semver/semver"
|
||||
"github.com/golang/protobuf/proto"
|
||||
"google.golang.org/protobuf/reflect/protoreflect"
|
||||
"google.golang.org/protobuf/types/descriptorpb"
|
||||
|
||||
"go.etcd.io/etcd/api/v3/etcdserverpb"
|
||||
"go.etcd.io/etcd/pkg/v3/pbutil"
|
||||
"go.etcd.io/etcd/raft/v3/raftpb"
|
||||
"go.etcd.io/etcd/server/v3/storage/wal"
|
||||
)
|
||||
|
||||
// MinimalStorageVersionFromWAL returns minimal etcd storage able to interpret provided WAL log,
|
||||
// determined by looking at entries since the last snapshot and returning the highest
|
||||
// etcd version annotation from used messages, fields, enums and their values.
|
||||
func MinimalStorageVersionFromWAL(wal *wal.WAL) *semver.Version {
|
||||
_, _, ents, err := wal.ReadAll()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
var maxVer *semver.Version
|
||||
for _, ent := range ents {
|
||||
maxVer = maxVersion(maxVer, etcdVersionFromEntry(ent))
|
||||
}
|
||||
return maxVer
|
||||
}
|
||||
|
||||
func etcdVersionFromEntry(ent raftpb.Entry) *semver.Version {
|
||||
msgVer := etcdVersionFromMessage(proto.MessageReflect(&ent))
|
||||
dataVer := etcdVersionFromData(ent.Type, ent.Data)
|
||||
return maxVersion(msgVer, dataVer)
|
||||
}
|
||||
|
||||
func etcdVersionFromData(entryType raftpb.EntryType, data []byte) *semver.Version {
|
||||
var msg protoreflect.Message
|
||||
switch entryType {
|
||||
case raftpb.EntryNormal:
|
||||
var raftReq etcdserverpb.InternalRaftRequest
|
||||
err := pbutil.Unmarshaler(&raftReq).Unmarshal(data)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
msg = proto.MessageReflect(&raftReq)
|
||||
case raftpb.EntryConfChange:
|
||||
var confChange raftpb.ConfChange
|
||||
err := pbutil.Unmarshaler(&confChange).Unmarshal(data)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
msg = proto.MessageReflect(&confChange)
|
||||
case raftpb.EntryConfChangeV2:
|
||||
var confChange raftpb.ConfChangeV2
|
||||
err := pbutil.Unmarshaler(&confChange).Unmarshal(data)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
msg = proto.MessageReflect(&confChange)
|
||||
default:
|
||||
panic("unhandled")
|
||||
}
|
||||
return etcdVersionFromMessage(msg)
|
||||
}
|
||||
|
||||
func etcdVersionFromMessage(m protoreflect.Message) *semver.Version {
|
||||
var maxVer *semver.Version
|
||||
md := m.Descriptor()
|
||||
opts := md.Options().(*descriptorpb.MessageOptions)
|
||||
if opts != nil {
|
||||
maxVer = maxVersion(maxVer, etcdVersionFromOptionsString(opts.String()))
|
||||
}
|
||||
|
||||
m.Range(func(field protoreflect.FieldDescriptor, value protoreflect.Value) bool {
|
||||
fd := md.Fields().Get(field.Index())
|
||||
maxVer = maxVersion(maxVer, etcdVersionFromField(fd))
|
||||
switch m := value.Interface().(type) {
|
||||
case protoreflect.Message:
|
||||
maxVer = maxVersion(maxVer, etcdVersionFromMessage(m))
|
||||
case protoreflect.EnumNumber:
|
||||
maxVer = maxVersion(maxVer, etcdVersionFromEnum(field.Enum(), m))
|
||||
}
|
||||
return true
|
||||
})
|
||||
return maxVer
|
||||
}
|
||||
|
||||
func etcdVersionFromEnum(enum protoreflect.EnumDescriptor, value protoreflect.EnumNumber) *semver.Version {
|
||||
var maxVer *semver.Version
|
||||
enumOpts := enum.Options().(*descriptorpb.EnumOptions)
|
||||
if enumOpts != nil {
|
||||
maxVer = maxVersion(maxVer, etcdVersionFromOptionsString(enumOpts.String()))
|
||||
}
|
||||
valueDesc := enum.Values().Get(int(value))
|
||||
valueOpts := valueDesc.Options().(*descriptorpb.EnumValueOptions)
|
||||
if valueOpts != nil {
|
||||
maxVer = maxVersion(maxVer, etcdVersionFromOptionsString(valueOpts.String()))
|
||||
}
|
||||
return maxVer
|
||||
}
|
||||
|
||||
func maxVersion(a *semver.Version, b *semver.Version) *semver.Version {
|
||||
if a != nil && (b == nil || b.LessThan(*a)) {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func etcdVersionFromField(fd protoreflect.FieldDescriptor) *semver.Version {
|
||||
opts := fd.Options().(*descriptorpb.FieldOptions)
|
||||
if opts == nil {
|
||||
return nil
|
||||
}
|
||||
return etcdVersionFromOptionsString(opts.String())
|
||||
}
|
||||
|
||||
func etcdVersionFromOptionsString(opts string) *semver.Version {
|
||||
// TODO: Use proto.GetExtention when gogo/protobuf is usable with protoreflect
|
||||
msgs := []string{"[versionpb.etcd_version_msg]:", "[versionpb.etcd_version_field]:", "[versionpb.etcd_version_enum]:", "[versionpb.etcd_version_enum_value]:"}
|
||||
var end, index int
|
||||
for _, msg := range msgs {
|
||||
index = strings.Index(opts, msg)
|
||||
end = index + len(msg)
|
||||
if index != -1 {
|
||||
break
|
||||
}
|
||||
}
|
||||
if index == -1 {
|
||||
return nil
|
||||
}
|
||||
var verStr string
|
||||
_, err := fmt.Sscanf(opts[end:], "%q", &verStr)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
ver, err := semver.NewVersion(verStr + ".0")
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return ver
|
||||
}
|
||||
268
server/storage/schema/wal_test.go
Normal file
268
server/storage/schema/wal_test.go
Normal file
@@ -0,0 +1,268 @@
|
||||
// Copyright 2021 The etcd Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package schema
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/coreos/go-semver/semver"
|
||||
"github.com/golang/protobuf/proto"
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"go.etcd.io/etcd/api/v3/etcdserverpb"
|
||||
"go.etcd.io/etcd/api/v3/membershippb"
|
||||
"go.etcd.io/etcd/pkg/v3/pbutil"
|
||||
"go.etcd.io/etcd/raft/v3/raftpb"
|
||||
)
|
||||
|
||||
var (
|
||||
V3_0 = semver.Version{Major: 3, Minor: 0}
|
||||
V3_1 = semver.Version{Major: 3, Minor: 1}
|
||||
V3_3 = semver.Version{Major: 3, Minor: 3}
|
||||
V3_4 = semver.Version{Major: 3, Minor: 4}
|
||||
)
|
||||
|
||||
func TestEtcdVersionFromEntry(t *testing.T) {
|
||||
raftReq := etcdserverpb.InternalRaftRequest{Header: &etcdserverpb.RequestHeader{AuthRevision: 1}}
|
||||
normalRequestData := pbutil.MustMarshal(&raftReq)
|
||||
|
||||
confChange := raftpb.ConfChange{Type: raftpb.ConfChangeAddLearnerNode}
|
||||
confChangeData := pbutil.MustMarshal(&confChange)
|
||||
|
||||
confChangeV2 := raftpb.ConfChangeV2{Transition: raftpb.ConfChangeTransitionJointExplicit}
|
||||
confChangeV2Data := pbutil.MustMarshal(&confChangeV2)
|
||||
|
||||
tcs := []struct {
|
||||
name string
|
||||
input raftpb.Entry
|
||||
expect *semver.Version
|
||||
}{
|
||||
{
|
||||
name: "Using RequestHeader AuthRevision in NormalEntry implies v3.1",
|
||||
input: raftpb.Entry{
|
||||
Term: 1,
|
||||
Index: 2,
|
||||
Type: raftpb.EntryNormal,
|
||||
Data: normalRequestData,
|
||||
},
|
||||
expect: &V3_1,
|
||||
},
|
||||
{
|
||||
name: "Using ConfigChange implies v3.4",
|
||||
input: raftpb.Entry{
|
||||
Term: 1,
|
||||
Index: 2,
|
||||
Type: raftpb.EntryConfChange,
|
||||
Data: confChangeData,
|
||||
},
|
||||
expect: &V3_0,
|
||||
},
|
||||
{
|
||||
name: "Using ConfigChangeV2 implies v3.4",
|
||||
input: raftpb.Entry{
|
||||
Term: 1,
|
||||
Index: 2,
|
||||
Type: raftpb.EntryConfChangeV2,
|
||||
Data: confChangeV2Data,
|
||||
},
|
||||
expect: &V3_4,
|
||||
},
|
||||
}
|
||||
for _, tc := range tcs {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ver := etcdVersionFromEntry(tc.input)
|
||||
assert.Equal(t, tc.expect, ver)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEtcdVersionFromMessage(t *testing.T) {
|
||||
tcs := []struct {
|
||||
name string
|
||||
input proto.Message
|
||||
expect *semver.Version
|
||||
}{
|
||||
{
|
||||
name: "Empty RequestHeader impies v3.0",
|
||||
input: &etcdserverpb.RequestHeader{},
|
||||
expect: &V3_0,
|
||||
},
|
||||
{
|
||||
name: "RequestHeader AuthRevision field set implies v3.5",
|
||||
input: &etcdserverpb.RequestHeader{AuthRevision: 1},
|
||||
expect: &V3_1,
|
||||
},
|
||||
{
|
||||
name: "RequestHeader Username set implies v3.0",
|
||||
input: &etcdserverpb.RequestHeader{Username: "Alice"},
|
||||
expect: &V3_0,
|
||||
},
|
||||
{
|
||||
name: "When two fields are set take higher version",
|
||||
input: &etcdserverpb.RequestHeader{AuthRevision: 1, Username: "Alice"},
|
||||
expect: &V3_1,
|
||||
},
|
||||
{
|
||||
name: "Setting a RequestHeader AuthRevision in subfield implies v3.1",
|
||||
input: &etcdserverpb.InternalRaftRequest{Header: &etcdserverpb.RequestHeader{AuthRevision: 1}},
|
||||
expect: &V3_1,
|
||||
},
|
||||
{
|
||||
name: "Setting a DowngradeInfoSetRequest implies v3.5",
|
||||
input: &etcdserverpb.InternalRaftRequest{DowngradeInfoSet: &membershippb.DowngradeInfoSetRequest{}},
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
name: "Enum CompareResult set to EQUAL implies v3.0",
|
||||
input: &etcdserverpb.Compare{Result: etcdserverpb.Compare_EQUAL},
|
||||
expect: &V3_0,
|
||||
},
|
||||
{
|
||||
name: "Enum CompareResult set to NOT_EQUAL implies v3.1",
|
||||
input: &etcdserverpb.Compare{Result: etcdserverpb.Compare_NOT_EQUAL},
|
||||
expect: &V3_1,
|
||||
},
|
||||
{
|
||||
name: "Oneof Compare version set implies v3.1",
|
||||
input: &etcdserverpb.Compare{TargetUnion: &etcdserverpb.Compare_Version{}},
|
||||
expect: &V3_0,
|
||||
},
|
||||
{
|
||||
name: "Oneof Compare lease set implies v3.3",
|
||||
input: &etcdserverpb.Compare{TargetUnion: &etcdserverpb.Compare_Lease{}},
|
||||
expect: &V3_3,
|
||||
},
|
||||
}
|
||||
for _, tc := range tcs {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ver := etcdVersionFromMessage(proto.MessageReflect(tc.input))
|
||||
assert.Equal(t, tc.expect, ver)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEtcdVersionFromFieldOptionsString(t *testing.T) {
|
||||
tcs := []struct {
|
||||
input string
|
||||
expect *semver.Version
|
||||
}{
|
||||
{
|
||||
input: "65001:0",
|
||||
},
|
||||
{
|
||||
input: `65001:0 65004:"NodeID"`,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.XXX]:"3.5"`,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.etcd_version_msg]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.etcd_version_enum]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.etcd_version_field]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.etcd_version_enum_value]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `65001:0 [versionpb.etcd_version_msg]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `65004:"NodeID" [versionpb.etcd_version_msg]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `65004:"NodeID" [versionpb.etcd_version_enum]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.other_field]:"NodeID" [versionpb.etcd_version_msg]:"3.5"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.etcd_version_msg]:"3.5" 65001:0`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.etcd_version_msg]:"3.5" 65004:"NodeID"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.etcd_version_msg]:"3.5" [versionpb.other_field]:"NodeID"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `[versionpb.other_field]:"NodeID" [versionpb.etcd_version_msg]:"3.5" [versionpb.another_field]:"NodeID"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
input: `65001:0 [versionpb.etcd_version_msg]:"3.5" 65001:0"`,
|
||||
expect: &V3_5,
|
||||
},
|
||||
}
|
||||
for _, tc := range tcs {
|
||||
t.Run(tc.input, func(t *testing.T) {
|
||||
ver := etcdVersionFromOptionsString(tc.input)
|
||||
assert.Equal(t, ver, tc.expect)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaxVersion(t *testing.T) {
|
||||
tcs := []struct {
|
||||
a, b, expect *semver.Version
|
||||
}{
|
||||
{
|
||||
a: nil,
|
||||
b: nil,
|
||||
expect: nil,
|
||||
},
|
||||
{
|
||||
a: &V3_5,
|
||||
b: nil,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
a: nil,
|
||||
b: &V3_5,
|
||||
expect: &V3_5,
|
||||
},
|
||||
{
|
||||
a: &V3_6,
|
||||
b: &V3_5,
|
||||
expect: &V3_6,
|
||||
},
|
||||
{
|
||||
a: &V3_5,
|
||||
b: &V3_6,
|
||||
expect: &V3_6,
|
||||
},
|
||||
}
|
||||
for _, tc := range tcs {
|
||||
t.Run(fmt.Sprintf("%v %v %v", tc.a, tc.b, tc.expect), func(t *testing.T) {
|
||||
got := maxVersion(tc.a, tc.b)
|
||||
assert.Equal(t, got, tc.expect)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user