blob: ff59bac346ea0a82ac58cf2d3e08375bb39a3d1e [file] [log] [blame]
khenaidooab1f7bd2019-11-14 14:00:27 -05001// Copyright 2015 The etcd Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15// Package v3rpc implements etcd v3 RPC system based on gRPC.
16package v3rpc
17
18import (
19 "context"
20
21 "go.etcd.io/etcd/etcdserver"
22 "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
23 pb "go.etcd.io/etcd/etcdserver/etcdserverpb"
24 "go.etcd.io/etcd/pkg/adt"
25
26 "github.com/coreos/pkg/capnslog"
27)
28
29var (
30 plog = capnslog.NewPackageLogger("go.etcd.io/etcd", "etcdserver/api/v3rpc")
31)
32
33type kvServer struct {
34 hdr header
35 kv etcdserver.RaftKV
36 // maxTxnOps is the max operations per txn.
37 // e.g suppose maxTxnOps = 128.
38 // Txn.Success can have at most 128 operations,
39 // and Txn.Failure can have at most 128 operations.
40 maxTxnOps uint
41}
42
43func NewKVServer(s *etcdserver.EtcdServer) pb.KVServer {
44 return &kvServer{hdr: newHeader(s), kv: s, maxTxnOps: s.Cfg.MaxTxnOps}
45}
46
47func (s *kvServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
48 if err := checkRangeRequest(r); err != nil {
49 return nil, err
50 }
51
52 resp, err := s.kv.Range(ctx, r)
53 if err != nil {
54 return nil, togRPCError(err)
55 }
56
57 s.hdr.fill(resp.Header)
58 return resp, nil
59}
60
61func (s *kvServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {
62 if err := checkPutRequest(r); err != nil {
63 return nil, err
64 }
65
66 resp, err := s.kv.Put(ctx, r)
67 if err != nil {
68 return nil, togRPCError(err)
69 }
70
71 s.hdr.fill(resp.Header)
72 return resp, nil
73}
74
75func (s *kvServer) DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) {
76 if err := checkDeleteRequest(r); err != nil {
77 return nil, err
78 }
79
80 resp, err := s.kv.DeleteRange(ctx, r)
81 if err != nil {
82 return nil, togRPCError(err)
83 }
84
85 s.hdr.fill(resp.Header)
86 return resp, nil
87}
88
89func (s *kvServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) {
90 if err := checkTxnRequest(r, int(s.maxTxnOps)); err != nil {
91 return nil, err
92 }
93 // check for forbidden put/del overlaps after checking request to avoid quadratic blowup
94 if _, _, err := checkIntervals(r.Success); err != nil {
95 return nil, err
96 }
97 if _, _, err := checkIntervals(r.Failure); err != nil {
98 return nil, err
99 }
100
101 resp, err := s.kv.Txn(ctx, r)
102 if err != nil {
103 return nil, togRPCError(err)
104 }
105
106 s.hdr.fill(resp.Header)
107 return resp, nil
108}
109
110func (s *kvServer) Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error) {
111 resp, err := s.kv.Compact(ctx, r)
112 if err != nil {
113 return nil, togRPCError(err)
114 }
115
116 s.hdr.fill(resp.Header)
117 return resp, nil
118}
119
120func checkRangeRequest(r *pb.RangeRequest) error {
121 if len(r.Key) == 0 {
122 return rpctypes.ErrGRPCEmptyKey
123 }
124 return nil
125}
126
127func checkPutRequest(r *pb.PutRequest) error {
128 if len(r.Key) == 0 {
129 return rpctypes.ErrGRPCEmptyKey
130 }
131 if r.IgnoreValue && len(r.Value) != 0 {
132 return rpctypes.ErrGRPCValueProvided
133 }
134 if r.IgnoreLease && r.Lease != 0 {
135 return rpctypes.ErrGRPCLeaseProvided
136 }
137 return nil
138}
139
140func checkDeleteRequest(r *pb.DeleteRangeRequest) error {
141 if len(r.Key) == 0 {
142 return rpctypes.ErrGRPCEmptyKey
143 }
144 return nil
145}
146
147func checkTxnRequest(r *pb.TxnRequest, maxTxnOps int) error {
148 opc := len(r.Compare)
149 if opc < len(r.Success) {
150 opc = len(r.Success)
151 }
152 if opc < len(r.Failure) {
153 opc = len(r.Failure)
154 }
155 if opc > maxTxnOps {
156 return rpctypes.ErrGRPCTooManyOps
157 }
158
159 for _, c := range r.Compare {
160 if len(c.Key) == 0 {
161 return rpctypes.ErrGRPCEmptyKey
162 }
163 }
164 for _, u := range r.Success {
165 if err := checkRequestOp(u, maxTxnOps-opc); err != nil {
166 return err
167 }
168 }
169 for _, u := range r.Failure {
170 if err := checkRequestOp(u, maxTxnOps-opc); err != nil {
171 return err
172 }
173 }
174
175 return nil
176}
177
178// checkIntervals tests whether puts and deletes overlap for a list of ops. If
179// there is an overlap, returns an error. If no overlap, return put and delete
180// sets for recursive evaluation.
181func checkIntervals(reqs []*pb.RequestOp) (map[string]struct{}, adt.IntervalTree, error) {
182 dels := adt.NewIntervalTree()
183
184 // collect deletes from this level; build first to check lower level overlapped puts
185 for _, req := range reqs {
186 tv, ok := req.Request.(*pb.RequestOp_RequestDeleteRange)
187 if !ok {
188 continue
189 }
190 dreq := tv.RequestDeleteRange
191 if dreq == nil {
192 continue
193 }
194 var iv adt.Interval
195 if len(dreq.RangeEnd) != 0 {
196 iv = adt.NewStringAffineInterval(string(dreq.Key), string(dreq.RangeEnd))
197 } else {
198 iv = adt.NewStringAffinePoint(string(dreq.Key))
199 }
200 dels.Insert(iv, struct{}{})
201 }
202
203 // collect children puts/deletes
204 puts := make(map[string]struct{})
205 for _, req := range reqs {
206 tv, ok := req.Request.(*pb.RequestOp_RequestTxn)
207 if !ok {
208 continue
209 }
210 putsThen, delsThen, err := checkIntervals(tv.RequestTxn.Success)
211 if err != nil {
212 return nil, dels, err
213 }
214 putsElse, delsElse, err := checkIntervals(tv.RequestTxn.Failure)
215 if err != nil {
216 return nil, dels, err
217 }
218 for k := range putsThen {
219 if _, ok := puts[k]; ok {
220 return nil, dels, rpctypes.ErrGRPCDuplicateKey
221 }
222 if dels.Intersects(adt.NewStringAffinePoint(k)) {
223 return nil, dels, rpctypes.ErrGRPCDuplicateKey
224 }
225 puts[k] = struct{}{}
226 }
227 for k := range putsElse {
228 if _, ok := puts[k]; ok {
229 // if key is from putsThen, overlap is OK since
230 // either then/else are mutually exclusive
231 if _, isSafe := putsThen[k]; !isSafe {
232 return nil, dels, rpctypes.ErrGRPCDuplicateKey
233 }
234 }
235 if dels.Intersects(adt.NewStringAffinePoint(k)) {
236 return nil, dels, rpctypes.ErrGRPCDuplicateKey
237 }
238 puts[k] = struct{}{}
239 }
240 dels.Union(delsThen, adt.NewStringAffineInterval("\x00", ""))
241 dels.Union(delsElse, adt.NewStringAffineInterval("\x00", ""))
242 }
243
244 // collect and check this level's puts
245 for _, req := range reqs {
246 tv, ok := req.Request.(*pb.RequestOp_RequestPut)
247 if !ok || tv.RequestPut == nil {
248 continue
249 }
250 k := string(tv.RequestPut.Key)
251 if _, ok := puts[k]; ok {
252 return nil, dels, rpctypes.ErrGRPCDuplicateKey
253 }
254 if dels.Intersects(adt.NewStringAffinePoint(k)) {
255 return nil, dels, rpctypes.ErrGRPCDuplicateKey
256 }
257 puts[k] = struct{}{}
258 }
259 return puts, dels, nil
260}
261
262func checkRequestOp(u *pb.RequestOp, maxTxnOps int) error {
263 // TODO: ensure only one of the field is set.
264 switch uv := u.Request.(type) {
265 case *pb.RequestOp_RequestRange:
266 return checkRangeRequest(uv.RequestRange)
267 case *pb.RequestOp_RequestPut:
268 return checkPutRequest(uv.RequestPut)
269 case *pb.RequestOp_RequestDeleteRange:
270 return checkDeleteRequest(uv.RequestDeleteRange)
271 case *pb.RequestOp_RequestTxn:
272 return checkTxnRequest(uv.RequestTxn, maxTxnOps)
273 default:
274 // empty op / nil entry
275 return rpctypes.ErrGRPCKeyNotFound
276 }
277}