mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
Refactor appendSuccesful and appendFailed methods to match
Signed-off-by: Marek Siarkowicz <siarkowicz@google.com>
This commit is contained in:
Marek Siarkowicz
parent
65ac859a1b
commit
229275d46e
@ -59,51 +59,33 @@ func (h *AppendableHistory) AppendRange(startKey, endKey string, revision, limit
|
||||
if resp != nil && resp.Header != nil {
|
||||
respRevision = resp.Header.Revision
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: staleRangeRequest(startKey, endKey, limit, revision),
|
||||
Call: start.Nanoseconds(),
|
||||
Output: rangeResponse(resp.Kvs, resp.Count, respRevision),
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
h.appendSuccessful(staleRangeRequest(startKey, endKey, limit, revision), start, end, rangeResponse(resp.Kvs, resp.Count, respRevision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) AppendPut(key, value string, start, end time.Duration, resp *clientv3.PutResponse, err error) {
|
||||
request := putRequest(key, value)
|
||||
if err != nil {
|
||||
h.appendFailed(request, start.Nanoseconds(), err)
|
||||
h.appendFailed(request, start, err)
|
||||
return
|
||||
}
|
||||
var revision int64
|
||||
if resp != nil && resp.Header != nil {
|
||||
revision = resp.Header.Revision
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: start.Nanoseconds(),
|
||||
Output: putResponse(revision),
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
h.appendSuccessful(request, start, end, putResponse(revision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) AppendPutWithLease(key, value string, leaseID int64, start, end time.Duration, resp *clientv3.PutResponse, err error) {
|
||||
request := putWithLeaseRequest(key, value, leaseID)
|
||||
if err != nil {
|
||||
h.appendFailed(request, start.Nanoseconds(), err)
|
||||
h.appendFailed(request, start, err)
|
||||
return
|
||||
}
|
||||
var revision int64
|
||||
if resp != nil && resp.Header != nil {
|
||||
revision = resp.Header.Revision
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: start.Nanoseconds(),
|
||||
Output: putResponse(revision),
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
h.appendSuccessful(request, start, end, putResponse(revision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) AppendLeaseGrant(start, end time.Duration, resp *clientv3.LeaseGrantResponse, err error) {
|
||||
@ -113,45 +95,33 @@ func (h *AppendableHistory) AppendLeaseGrant(start, end time.Duration, resp *cli
|
||||
}
|
||||
request := leaseGrantRequest(leaseID)
|
||||
if err != nil {
|
||||
h.appendFailed(request, start.Nanoseconds(), err)
|
||||
h.appendFailed(request, start, err)
|
||||
return
|
||||
}
|
||||
var revision int64
|
||||
if resp != nil && resp.ResponseHeader != nil {
|
||||
revision = resp.ResponseHeader.Revision
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: start.Nanoseconds(),
|
||||
Output: leaseGrantResponse(revision),
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
h.appendSuccessful(request, start, end, leaseGrantResponse(revision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) AppendLeaseRevoke(id int64, start, end time.Duration, resp *clientv3.LeaseRevokeResponse, err error) {
|
||||
request := leaseRevokeRequest(id)
|
||||
if err != nil {
|
||||
h.appendFailed(request, start.Nanoseconds(), err)
|
||||
h.appendFailed(request, start, err)
|
||||
return
|
||||
}
|
||||
var revision int64
|
||||
if resp != nil && resp.Header != nil {
|
||||
revision = resp.Header.Revision
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: start.Nanoseconds(),
|
||||
Output: leaseRevokeResponse(revision),
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
h.appendSuccessful(request, start, end, leaseRevokeResponse(revision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) AppendDelete(key string, start, end time.Duration, resp *clientv3.DeleteResponse, err error) {
|
||||
request := deleteRequest(key)
|
||||
if err != nil {
|
||||
h.appendFailed(request, start.Nanoseconds(), err)
|
||||
h.appendFailed(request, start, err)
|
||||
return
|
||||
}
|
||||
var revision int64
|
||||
@ -160,13 +130,7 @@ func (h *AppendableHistory) AppendDelete(key string, start, end time.Duration, r
|
||||
revision = resp.Header.Revision
|
||||
deleted = resp.Deleted
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: start.Nanoseconds(),
|
||||
Output: deleteResponse(deleted, revision),
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
h.appendSuccessful(request, start, end, deleteResponse(deleted, revision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) AppendTxn(cmp []clientv3.Cmp, clientOnSuccessOps, clientOnFailure []clientv3.Op, start, end time.Duration, resp *clientv3.TxnResponse, err error) {
|
||||
@ -184,7 +148,7 @@ func (h *AppendableHistory) AppendTxn(cmp []clientv3.Cmp, clientOnSuccessOps, cl
|
||||
}
|
||||
request := txnRequest(conds, modelOnSuccess, modelOnFailure)
|
||||
if err != nil {
|
||||
h.appendFailed(request, start.Nanoseconds(), err)
|
||||
h.appendFailed(request, start, err)
|
||||
return
|
||||
}
|
||||
var revision int64
|
||||
@ -195,16 +159,17 @@ func (h *AppendableHistory) AppendTxn(cmp []clientv3.Cmp, clientOnSuccessOps, cl
|
||||
for _, resp := range resp.Responses {
|
||||
results = append(results, toEtcdOperationResult(resp))
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
h.appendSuccessful(request, start, end, txnResponse(results, resp.Succeeded, revision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) appendSuccessful(request EtcdRequest, start, end time.Duration, response MaybeEtcdResponse) {
|
||||
op := porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: start.Nanoseconds(),
|
||||
Output: txnResponse(results, resp.Succeeded, revision),
|
||||
Output: response,
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) appendSuccessful(op porcupine.Operation) {
|
||||
}
|
||||
if op.Call >= op.Return {
|
||||
panic(fmt.Sprintf("Invalid operation, call(%d) >= return(%d)", op.Call, op.Return))
|
||||
}
|
||||
@ -298,45 +263,40 @@ func toEtcdOperationResult(resp *etcdserverpb.ResponseOp) EtcdOperationResult {
|
||||
func (h *AppendableHistory) AppendDefragment(start, end time.Duration, resp *clientv3.DefragmentResponse, err error) {
|
||||
request := defragmentRequest()
|
||||
if err != nil {
|
||||
h.appendFailed(request, start.Nanoseconds(), err)
|
||||
h.appendFailed(request, start, err)
|
||||
return
|
||||
}
|
||||
var revision int64
|
||||
if resp != nil && resp.Header != nil {
|
||||
revision = resp.Header.Revision
|
||||
}
|
||||
h.appendSuccessful(porcupine.Operation{
|
||||
h.appendSuccessful(request, start, end, defragmentResponse(revision))
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) appendFailed(request EtcdRequest, start time.Duration, err error) {
|
||||
op := porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: start.Nanoseconds(),
|
||||
Output: defragmentResponse(revision),
|
||||
Return: end.Nanoseconds(),
|
||||
})
|
||||
}
|
||||
|
||||
func (h *AppendableHistory) appendFailed(request EtcdRequest, call int64, err error) {
|
||||
Output: failedResponse(err),
|
||||
Return: 0, // For failed writes we don't know when request has really finished.
|
||||
}
|
||||
if len(h.successful) > 0 {
|
||||
prevSuccessful := h.successful[len(h.successful)-1]
|
||||
if call <= prevSuccessful.Call {
|
||||
panic(fmt.Sprintf("Out of order append, new.call(%d) <= prev.call(%d)", call, prevSuccessful.Call))
|
||||
if op.Call <= prevSuccessful.Call {
|
||||
panic(fmt.Sprintf("Out of order append, new.call(%d) <= prev.call(%d)", op.Call, prevSuccessful.Call))
|
||||
}
|
||||
if call <= prevSuccessful.Return {
|
||||
panic(fmt.Sprintf("Overlapping operations, new.call(%d) <= prev.return(%d)", call, prevSuccessful.Return))
|
||||
if op.Call <= prevSuccessful.Return {
|
||||
panic(fmt.Sprintf("Overlapping operations, new.call(%d) <= prev.return(%d)", op.Call, prevSuccessful.Return))
|
||||
}
|
||||
}
|
||||
if len(h.failed) > 0 {
|
||||
prevFailed := h.failed[len(h.failed)-1]
|
||||
if call <= prevFailed.Call {
|
||||
panic(fmt.Sprintf("Out of order append, new.call(%d) <= prev.call(%d)", call, prevFailed.Call))
|
||||
if op.Call <= prevFailed.Call {
|
||||
panic(fmt.Sprintf("Out of order append, new.call(%d) <= prev.call(%d)", op.Call, prevFailed.Call))
|
||||
}
|
||||
}
|
||||
h.failed = append(h.failed, porcupine.Operation{
|
||||
ClientId: h.streamID,
|
||||
Input: request,
|
||||
Call: call,
|
||||
Output: failedResponse(err),
|
||||
Return: 0, // For failed writes we don't know when request has really finished.
|
||||
})
|
||||
h.failed = append(h.failed, op)
|
||||
// Operations of single client needs to be sequential.
|
||||
// As we don't know return time of failed operations, all new writes need to be done with new stream id.
|
||||
h.streamID = h.idProvider.NewStreamID()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user