1// 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	"github.com/coreos/etcd/etcdserver"
22	"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
23	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
24	"github.com/coreos/etcd/pkg/adt"
25
26	"github.com/coreos/pkg/capnslog"
27)
28
29var (
30	plog = capnslog.NewPackageLogger("github.com/coreos/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}
278