Measure transactions

client.go

@@ -445,6 +445,9 @@ func (clnt *Client) BatchExists(policy *BatchPolicy, keys []*Key) ([]bool, Error
 
 	// pass nil to make sure it will be cloned and prepared
 	cmd := newBatchCommandExists(nil, nil, policy, keys, existsArray)
+
+	clnt.cluster.addTran()
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if filteredOut > 0 {
 		err = chainErrors(ErrFilteredOut.err(), err)
@@ -518,6 +521,9 @@ func (clnt *Client) BatchGet(policy *BatchPolicy, keys []*Key, binNames ...strin
 	}
 
 	cmd := newBatchCommandGet(nil, nil, policy, keys, binNames, nil, records, _INFO1_READ, false)
+
+	clnt.cluster.addTran()
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if err != nil && !policy.AllowPartialResults {
 		return nil, err
@@ -548,6 +554,9 @@ func (clnt *Client) BatchGetOperate(policy *BatchPolicy, keys []*Key, ops ...*Op
 	}
 
 	cmd := newBatchCommandGet(nil, nil, policy, keys, nil, ops, records, _INFO1_READ, true)
+
+	clnt.cluster.addTran()
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if err != nil && !policy.AllowPartialResults {
 		return nil, err
@@ -571,6 +580,8 @@ func (clnt *Client) BatchGetComplex(policy *BatchPolicy, records []*BatchRead) E
 
 	cmd := newBatchIndexCommandGet(nil, policy, records, true)
 
+	clnt.cluster.addTran()
+
 	batchNodes, err := newBatchIndexNodeList(clnt.cluster, policy, records)
 	if err != nil {
 		return err
@@ -606,6 +617,9 @@ func (clnt *Client) BatchGetHeader(policy *BatchPolicy, keys []*Key) ([]*Record,
 	}
 
 	cmd := newBatchCommandGet(nil, nil, policy, keys, nil, nil, records, _INFO1_READ|_INFO1_NOBINDATA, false)
+
+	clnt.cluster.addTran()
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if err != nil && !policy.AllowPartialResults {
 		return nil, err
@@ -642,6 +656,9 @@ func (clnt *Client) BatchDelete(policy *BatchPolicy, deletePolicy *BatchDeletePo
 	}
 
 	cmd := newBatchCommandDelete(nil, nil, policy, keys, records, attr)
+
+	clnt.cluster.addTran()
+
 	_, err = clnt.batchExecute(policy, batchNodes, cmd)
 	return records, err
 }
@@ -662,6 +679,9 @@ func (clnt *Client) BatchOperate(policy *BatchPolicy, records []BatchRecordIfc)
 	}
 
 	cmd := newBatchCommandOperate(nil, nil, policy, records)
+
+	clnt.cluster.addTran()
+
 	_, err = clnt.batchExecute(policy, batchNodes, cmd)
 	return err
 }
@@ -693,6 +713,9 @@ func (clnt *Client) BatchExecute(policy *BatchPolicy, udfPolicy *BatchUDFPolicy,
 	}
 
 	cmd := newBatchCommandUDF(nil, nil, policy, keys, packageName, functionName, args, records, attr)
+
+	clnt.cluster.addTran()
+
 	_, err = clnt.batchExecute(policy, batchNodes, cmd)
 	return records, err
 }
@@ -750,6 +773,7 @@ func (clnt *Client) ScanPartitions(apolicy *ScanPolicy, partitionFilter *Partiti
 
 	// result recordset
 	res := newRecordset(policy.RecordQueueSize, 1)
+	clnt.cluster.addTran()
 	go clnt.scanPartitions(&policy, tracker, namespace, setName, res, binNames...)
 
 	return res, nil

cluster.go

@@ -179,6 +179,12 @@ func (clstr *Cluster) enableMetrics(policy *MetricsPolicy) {
 	}
 }
 
+func (clstr *Cluster) addTran() {
+	if clstr.metricsEnabled {
+		clstr.tranCount.GetAndIncrement()
+	}
+}
+
 // String implements the stringer interface
 func (clstr *Cluster) String() string {
 	return fmt.Sprintf("%v", clstr.GetNodes())

delete_command.go

@@ -48,6 +48,8 @@ func newDeleteCommand(cluster *Cluster, policy *WritePolicy, key *Key) (*deleteC
 		policy:        policy,
 	}
 
+	cluster.addTran()
+
 	return newDeleteCmd, nil
 }
 

exists_command.go

@@ -43,6 +43,8 @@ func newExistsCommand(cluster *Cluster, policy *BasePolicy, key *Key) (*existsCo
 		}
 	}
 
+	cluster.addTran()
+
 	return &existsCommand{
 		singleCommand: newSingleCommand(cluster, key, partition),
 		policy:        policy,

operate_command.go

@@ -28,6 +28,8 @@ type operateCommand struct {
 }
 
 func newOperateCommand(cluster *Cluster, policy *WritePolicy, key *Key, args operateArgs) (operateCommand, Error) {
+	cluster.addTran()
+
 	rdCommand, err := newReadCommand(cluster, &policy.BasePolicy, key, nil, args.partition)
 	if err != nil {
 		return operateCommand{}, err

read_command.go

@@ -60,6 +60,8 @@ func newReadCommand(cluster *Cluster, policy *BasePolicy, key *Key, binNames []s
 		}
 	}
 
+	cluster.addTran()
+
 	return readCommand{
 		singleCommand: newSingleCommand(cluster, key, partition),
 		binNames:      binNames,

read_header_command.go

@@ -45,6 +45,8 @@ func newReadHeaderCommand(cluster *Cluster, policy *BasePolicy, key *Key) (readH
 		policy:        policy,
 	}
 
+	cluster.addTran()
+
 	return newReadHeaderCmd, nil
 }
 

scan_executor.go

@@ -25,6 +25,9 @@ func (clnt *Client) scanPartitions(policy *ScanPolicy, tracker *partitionTracker
 	interval := policy.SleepBetweenRetries
 
 	var errs Error
+
+	clnt.cluster.addTran()
+
 	for {
 		list, err := tracker.assignPartitionsToNodes(clnt.Cluster(), namespace)
 		if err != nil {

touch_command.go

@@ -49,6 +49,8 @@ func newTouchCommand(cluster *Cluster, policy *WritePolicy, key *Key) (touchComm
 		policy:        policy,
 	}
 
+	cluster.addTran()
+
 	return newTouchCmd, nil
 }
 

write_command.go

@@ -59,6 +59,8 @@ func newWriteCommand(cluster *Cluster,
 		operation:     operation,
 	}
 
+	cluster.addTran()
+
 	return newWriteCmd, nil
 }