Combine common txn arguments into TxnContext

Signed-off-by: Marek Siarkowicz <[email protected]>

Author: serathius

server/etcdserver/txn/txn.go

@@ -32,24 +32,31 @@ import (
 	"go.etcd.io/etcd/server/v3/storage/mvcc"
 )
 
-func Put(ctx context.Context, lg *zap.Logger, lessor lease.Lessor, kv mvcc.KV, p *pb.PutRequest) (resp *pb.PutResponse, trace *traceutil.Trace, err error) {
+type TxnContext struct {
+	Logger                       *zap.Logger
+	Lessor                       lease.Lessor
+	KV                           mvcc.KV
+	TxnModeWriteWithSharedBuffer bool
+}
+
+func Put(ctx context.Context, txn TxnContext, p *pb.PutRequest) (resp *pb.PutResponse, trace *traceutil.Trace, err error) {
 	trace = traceutil.Get(ctx)
 	// create put tracing if the trace in context is empty
 	if trace.IsEmpty() {
 		trace = traceutil.New("put",
-			lg,
+			txn.Logger,
 			traceutil.Field{Key: "key", Value: string(p.Key)},
 			traceutil.Field{Key: "req_size", Value: p.Size()},
 		)
 		ctx = context.WithValue(ctx, traceutil.TraceKey{}, trace)
 	}
 	leaseID := lease.LeaseID(p.Lease)
 	if leaseID != lease.NoLease {
-		if l := lessor.Lookup(leaseID); l == nil {
+		if l := txn.Lessor.Lookup(leaseID); l == nil {
 			return nil, nil, lease.ErrLeaseNotFound
 		}
 	}
-	txnWrite := kv.Write(trace)
+	txnWrite := txn.KV.Write(trace)
 	defer txnWrite.End()
 	resp, err = put(ctx, txnWrite, p)
 	return resp, trace, err
@@ -94,18 +101,18 @@ func put(ctx context.Context, txnWrite mvcc.TxnWrite, p *pb.PutRequest) (resp *p
 	return resp, nil
 }
 
-func DeleteRange(ctx context.Context, lg *zap.Logger, kv mvcc.KV, dr *pb.DeleteRangeRequest) (resp *pb.DeleteRangeResponse, trace *traceutil.Trace, err error) {
+func DeleteRange(ctx context.Context, txn TxnContext, dr *pb.DeleteRangeRequest) (resp *pb.DeleteRangeResponse, trace *traceutil.Trace, err error) {
 	trace = traceutil.Get(ctx)
 	// create delete tracing if the trace in context is empty
 	if trace.IsEmpty() {
 		trace = traceutil.New("delete_range",
-			lg,
+			txn.Logger,
 			traceutil.Field{Key: "key", Value: string(dr.Key)},
 			traceutil.Field{Key: "range_end", Value: string(dr.RangeEnd)},
 		)
 		ctx = context.WithValue(ctx, traceutil.TraceKey{}, trace)
 	}
-	txnWrite := kv.Write(trace)
+	txnWrite := txn.KV.Write(trace)
 	defer txnWrite.End()
 	resp, err = deleteRange(ctx, txnWrite, dr)
 	return resp, trace, err
@@ -133,19 +140,19 @@ func deleteRange(ctx context.Context, txnWrite mvcc.TxnWrite, dr *pb.DeleteRange
 	return resp, nil
 }
 
-func Range(ctx context.Context, lg *zap.Logger, kv mvcc.KV, r *pb.RangeRequest) (resp *pb.RangeResponse, trace *traceutil.Trace, err error) {
+func Range(ctx context.Context, txn TxnContext, r *pb.RangeRequest) (resp *pb.RangeResponse, trace *traceutil.Trace, err error) {
 	trace = traceutil.Get(ctx)
 	if trace.IsEmpty() {
-		trace = traceutil.New("range", lg)
+		trace = traceutil.New("range", txn.Logger)
 		ctx = context.WithValue(ctx, traceutil.TraceKey{}, trace)
 	}
 	defer func(start time.Time) {
 		success := err == nil
 		RangeSecObserve(success, time.Since(start))
 	}(time.Now())
-	txnRead := kv.Read(mvcc.ConcurrentReadTxMode, trace)
+	txnRead := txn.KV.Read(mvcc.ConcurrentReadTxMode, trace)
 	defer txnRead.End()
-	resp, err = executeRange(ctx, lg, txnRead, r)
+	resp, err = executeRange(ctx, txn.Logger, txnRead, r)
 	return resp, trace, err
 }
 
@@ -249,22 +256,22 @@ func executeRange(ctx context.Context, lg *zap.Logger, txnRead mvcc.TxnRead, r *
 	return resp, nil
 }
 
-func Txn(ctx context.Context, lg *zap.Logger, rt *pb.TxnRequest, txnModeWriteWithSharedBuffer bool, kv mvcc.KV, lessor lease.Lessor) (*pb.TxnResponse, *traceutil.Trace, error) {
+func Txn(ctx context.Context, txn TxnContext, rt *pb.TxnRequest) (*pb.TxnResponse, *traceutil.Trace, error) {
 	trace := traceutil.Get(ctx)
 	if trace.IsEmpty() {
-		trace = traceutil.New("transaction", lg)
+		trace = traceutil.New("transaction", txn.Logger)
 		ctx = context.WithValue(ctx, traceutil.TraceKey{}, trace)
 	}
 	isWrite := !IsTxnReadonly(rt)
 	// When the transaction contains write operations, we use ReadTx instead of
 	// ConcurrentReadTx to avoid extra overhead of copying buffer.
 	var mode mvcc.ReadTxMode
-	if isWrite && txnModeWriteWithSharedBuffer /*a.s.Cfg.ServerFeatureGate.Enabled(features.TxnModeWriteWithSharedBuffer)*/ {
+	if isWrite && txn.TxnModeWriteWithSharedBuffer /*a.s.Cfg.ServerFeatureGate.Enabled(features.TxnModeWriteWithSharedBuffer)*/ {
 		mode = mvcc.SharedBufReadTxMode
 	} else {
 		mode = mvcc.ConcurrentReadTxMode
 	}
-	txnRead := kv.Read(mode, trace)
+	txnRead := txn.KV.Read(mode, trace)
 	var txnPath []bool
 	trace.StepWithFunction(
 		func() {
@@ -275,7 +282,7 @@ func Txn(ctx context.Context, lg *zap.Logger, rt *pb.TxnRequest, txnModeWriteWit
 	if isWrite {
 		trace.AddField(traceutil.Field{Key: "read_only", Value: false})
 	}
-	_, err := checkTxn(txnRead, rt, lessor, txnPath)
+	_, err := checkTxn(txnRead, rt, txn.Lessor, txnPath)
 	if err != nil {
 		txnRead.End()
 		return nil, nil, err
@@ -288,11 +295,11 @@ func Txn(ctx context.Context, lg *zap.Logger, rt *pb.TxnRequest, txnModeWriteWit
 	var txnWrite mvcc.TxnWrite
 	if isWrite {
 		txnRead.End()
-		txnWrite = kv.Write(trace)
+		txnWrite = txn.KV.Write(trace)
 	} else {
 		txnWrite = mvcc.NewReadOnlyTxnWrite(txnRead)
 	}
-	txnResp, err := txn(ctx, lg, txnWrite, rt, isWrite, txnPath)
+	txnResp, err := assembleTxnResponse(ctx, txn.Logger, txnWrite, rt, isWrite, txnPath)
 	txnWrite.End()
 
 	trace.AddField(
@@ -302,7 +309,7 @@ func Txn(ctx context.Context, lg *zap.Logger, rt *pb.TxnRequest, txnModeWriteWit
 	return txnResp, trace, err
 }
 
-func txn(ctx context.Context, lg *zap.Logger, txnWrite mvcc.TxnWrite, rt *pb.TxnRequest, isWrite bool, txnPath []bool) (*pb.TxnResponse, error) {
+func assembleTxnResponse(ctx context.Context, lg *zap.Logger, txnWrite mvcc.TxnWrite, rt *pb.TxnRequest, isWrite bool, txnPath []bool) (*pb.TxnResponse, error) {
 	txnResp, _ := newTxnResp(rt, txnPath)
 	_, err := executeTxn(ctx, lg, txnWrite, rt, txnPath, txnResp)
 	if err != nil {