Add support for metrics

batch_command.go

@@ -74,6 +74,9 @@ func (cmd *batchCommand) retryBatch(ifc batcher, cluster *Cluster, deadline time
 	for _, batchNode := range batchNodes {
 		command := ifc.cloneBatchCommand(batchNode)
 		command.setSequence(cmd.sequenceAP, cmd.sequenceSC)
+
+		cluster.addRetry()
+
 		if err := command.executeAt(command, cmd.policy.GetBasePolicy(), deadline, iteration, commandWasSent); err != nil {
 			ferr = chainErrors(err, ferr)
 			if !cmd.policy.AllowPartialResults {

batch_command_delete.go

@@ -28,13 +28,15 @@ type batchCommandDelete struct {
 }
 
 func newBatchCommandDelete(
+	cluster *Cluster,
 	node *Node,
 	batch *batchNode,
 	policy *BatchPolicy,
 	keys []*Key,
 	records []*BatchRecord,
 	attr *batchAttr,
 ) *batchCommandDelete {
+	cluster.addTran()
 	res := &batchCommandDelete{
 		batchCommand: batchCommand{
 			baseMultiCommand: *newMultiCommand(node, nil, false),

batch_command_exists.go

@@ -27,12 +27,15 @@ type batchCommandExists struct {
 }
 
 func newBatchCommandExists(
+	cluster *Cluster,
 	node *Node,
 	batch *batchNode,
 	policy *BatchPolicy,
 	keys []*Key,
 	existsArray []bool,
 ) *batchCommandExists {
+	cluster.addTran()
+
 	res := &batchCommandExists{
 		batchCommand: batchCommand{
 			baseMultiCommand: *newMultiCommand(node, nil, false),

batch_command_get.go

@@ -55,6 +55,7 @@ var batchObjectParser func(
 ) Error
 
 func newBatchCommandGet(
+	cluster *Cluster,
 	node *Node,
 	batch *batchNode,
 	policy *BatchPolicy,
@@ -65,6 +66,8 @@ func newBatchCommandGet(
 	readAttr int,
 	isOperation bool,
 ) *batchCommandGet {
+	cluster.addTran()
+
 	res := &batchCommandGet{
 		batchCommand: batchCommand{
 			baseMultiCommand: *newMultiCommand(node, nil, isOperation),

batch_command_operate.go

@@ -37,11 +37,14 @@ type batchCommandOperate struct {
 }
 
 func newBatchCommandOperate(
+	cluster *Cluster,
 	node *Node,
 	batch *batchNode,
 	policy *BatchPolicy,
 	records []BatchRecordIfc,
 ) *batchCommandOperate {
+	cluster.addTran()
+
 	res := &batchCommandOperate{
 		batchCommand: batchCommand{
 			baseMultiCommand: *newMultiCommand(node, nil, true),

batch_command_udf.go

@@ -31,6 +31,7 @@ type batchCommandUDF struct {
 }
 
 func newBatchCommandUDF(
+	cluster *Cluster,
 	node *Node,
 	batch *batchNode,
 	policy *BatchPolicy,
@@ -41,6 +42,8 @@ func newBatchCommandUDF(
 	records []*BatchRecord,
 	attr *batchAttr,
 ) *batchCommandUDF {
+	cluster.addTran()
+
 	res := &batchCommandUDF{
 		batchCommand: batchCommand{
 			baseMultiCommand: *newMultiCommand(node, nil, false),

batch_index_command_get.go

@@ -19,6 +19,7 @@ type batchIndexCommandGet struct {
 }
 
 func newBatchIndexCommandGet(
+	cluster *Cluster,
 	batch *batchNode,
 	policy *BatchPolicy,
 	records []*BatchRead,
@@ -29,6 +30,8 @@ func newBatchIndexCommandGet(
 		node = batch.Node
 	}
 
+	cluster.addTran()
+
 	res := &batchIndexCommandGet{
 		batchCommandGet{
 			batchCommand: batchCommand{

client.go

@@ -110,6 +110,26 @@ func NewClientWithPolicyAndHost(policy *ClientPolicy, hosts ...*Host) (*Client,
 
 }
 
+// Metrics
+
+// Enable extended periodic cluster and node latency metrics.
+func (clnt *Client) EnableMetrics(policy *MetricsPolicy) error {
+	err := clnt.cluster.enableMetrics(policy)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// Disable extended periodic cluster and node latency metrics.
+func (clnt *Client) DisableMetrics() error {
+	err := clnt.cluster.disableMetrics()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 //-------------------------------------------------------
 // Policy methods
 //-------------------------------------------------------
@@ -434,7 +454,8 @@ 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)
+	cmd := newBatchCommandExists(clnt.cluster, nil, nil, policy, keys, existsArray)
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if filteredOut > 0 {
 		err = chainErrors(ErrFilteredOut.err(), err)
@@ -507,7 +528,8 @@ func (clnt *Client) BatchGet(policy *BatchPolicy, keys []*Key, binNames ...strin
 		return nil, err
 	}
 
-	cmd := newBatchCommandGet(nil, nil, policy, keys, binNames, nil, records, _INFO1_READ, false)
+	cmd := newBatchCommandGet(clnt.cluster, nil, nil, policy, keys, binNames, nil, records, _INFO1_READ, false)
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if err != nil && !policy.AllowPartialResults {
 		return nil, err
@@ -537,7 +559,8 @@ func (clnt *Client) BatchGetOperate(policy *BatchPolicy, keys []*Key, ops ...*Op
 		return nil, err
 	}
 
-	cmd := newBatchCommandGet(nil, nil, policy, keys, nil, ops, records, _INFO1_READ, true)
+	cmd := newBatchCommandGet(clnt.cluster, nil, nil, policy, keys, nil, ops, records, _INFO1_READ, true)
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if err != nil && !policy.AllowPartialResults {
 		return nil, err
@@ -559,7 +582,7 @@ func (clnt *Client) BatchGetOperate(policy *BatchPolicy, keys []*Key, ops ...*Op
 func (clnt *Client) BatchGetComplex(policy *BatchPolicy, records []*BatchRead) Error {
 	policy = clnt.getUsableBatchPolicy(policy)
 
-	cmd := newBatchIndexCommandGet(nil, policy, records, true)
+	cmd := newBatchIndexCommandGet(clnt.cluster, nil, policy, records, true)
 
 	batchNodes, err := newBatchIndexNodeList(clnt.cluster, policy, records)
 	if err != nil {
@@ -595,7 +618,8 @@ func (clnt *Client) BatchGetHeader(policy *BatchPolicy, keys []*Key) ([]*Record,
 		return nil, err
 	}
 
-	cmd := newBatchCommandGet(nil, nil, policy, keys, nil, nil, records, _INFO1_READ|_INFO1_NOBINDATA, false)
+	cmd := newBatchCommandGet(clnt.cluster, nil, nil, policy, keys, nil, nil, records, _INFO1_READ|_INFO1_NOBINDATA, false)
+
 	filteredOut, err := clnt.batchExecute(policy, batchNodes, cmd)
 	if err != nil && !policy.AllowPartialResults {
 		return nil, err
@@ -631,7 +655,8 @@ func (clnt *Client) BatchDelete(policy *BatchPolicy, deletePolicy *BatchDeletePo
 		return nil, err
 	}
 
-	cmd := newBatchCommandDelete(nil, nil, policy, keys, records, attr)
+	cmd := newBatchCommandDelete(clnt.cluster, nil, nil, policy, keys, records, attr)
+
 	_, err = clnt.batchExecute(policy, batchNodes, cmd)
 	return records, err
 }
@@ -651,7 +676,8 @@ func (clnt *Client) BatchOperate(policy *BatchPolicy, records []BatchRecordIfc)
 		return err
 	}
 
-	cmd := newBatchCommandOperate(nil, nil, policy, records)
+	cmd := newBatchCommandOperate(clnt.cluster, nil, nil, policy, records)
+
 	_, err = clnt.batchExecute(policy, batchNodes, cmd)
 	return err
 }
@@ -682,7 +708,8 @@ func (clnt *Client) BatchExecute(policy *BatchPolicy, udfPolicy *BatchUDFPolicy,
 		return nil, err
 	}
 
-	cmd := newBatchCommandUDF(nil, nil, policy, keys, packageName, functionName, args, records, attr)
+	cmd := newBatchCommandUDF(clnt.cluster, nil, nil, policy, keys, packageName, functionName, args, records, attr)
+
 	_, err = clnt.batchExecute(policy, batchNodes, cmd)
 	return records, err
 }

cluster.go

@@ -70,6 +70,13 @@ type Cluster struct {
 
 	// Password in hashed format in bytes.
 	password iatomic.SyncVal // []byte
+
+	metricsPolicy   *MetricsPolicy
+	metricsEnabled  bool
+	metricsListener MetricsListener
+
+	tranCount  iatomic.Int
+	retryCount iatomic.Int
 }
 
 // NewCluster generates a Cluster instance.
@@ -164,6 +171,53 @@ func NewCluster(policy *ClientPolicy, hosts []*Host) (*Cluster, Error) {
 	return newCluster, err
 }
 
+func (clstr *Cluster) enableMetrics(policy *MetricsPolicy) error {
+	if clstr.metricsEnabled {
+		// Disable the old metrics listener
+		err := clstr.metricsListener.onDisable(clstr)
+		if err != nil {
+			return err
+		}
+	}
+
+	var listener MetricsListener = policy.listener
+
+	if listener == nil {
+		listener = &MetricsWriter{dir: policy.reportDir}
+	}
+
+	clstr.metricsListener = listener
+	clstr.metricsPolicy = policy
+
+	for _, node := range clstr.GetNodes() {
+		node.enableMetrics(policy)
+	}
+
+	listener.onEnable(clstr, policy)
+	clstr.metricsEnabled = true
+}
+
+func (clstr *Cluster) disableMetrics() error {
+	if clstr.metricsEnabled {
+		clstr.metricsEnabled = false
+		err := clstr.metricsListener.onDisable(clstr)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (clstr *Cluster) addTran() {
+	if clstr.metricsEnabled {
+		clstr.tranCount.GetAndIncrement()
+	}
+}
+
+func (clstr *Cluster) addRetry() {
+	clstr.retryCount.GetAndIncrement()
+}
+
 // String implements the stringer interface
 func (clstr *Cluster) String() string {
 	return fmt.Sprintf("%v", clstr.GetNodes())
@@ -409,6 +463,10 @@ func (clstr *Cluster) tend() Error {
 		}
 	}
 
+	if clstr.metricsEnabled && clstr.tendCount%clstr.metricsPolicy.interval == 0 {
+		clstr.metricsListener.onSnapshot(clstr)
+	}
+
 	return nil
 }
 
@@ -783,6 +841,13 @@ func (clstr *Cluster) removeNodes(nodesToRemove []*Node) {
 			return nodesMap, nil
 		})
 
+		if clstr.metricsEnabled {
+			err := clstr.metricsListener.onNodeClose(node)
+			if err != nil {
+				logger.Logger.Warn("Write metrics failed on " + node.name + ": " + err.Error())
+			}
+		}
+
 		node.Close()
 	}
 

command.go

@@ -2614,6 +2614,8 @@ func (cmd *baseCommand) executeAt(ifc command, policy *BasePolicy, deadline time
 					}
 				}
 			}
+
+			cmd.node.cluster.addRetry()
 		}
 
 		// NOTE: This is important to be after the prepareRetry block above

connection.go

@@ -134,6 +134,7 @@ func newConnection(address string, timeout time.Duration) (*Connection, Error) {
 		logger.Logger.Debug("Connection to address `%s` failed to establish with error: %s", address, err.Error())
 		return nil, errToAerospikeErr(nil, err)
 	}
+
 	newConn.conn = conn
 	newConn.limitReader = &io.LimitedReader{R: conn, N: 0}
 

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,

go.mod

@@ -13,14 +13,22 @@ require (
 
 require (
 	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
-	github.com/google/go-cmp v0.5.9 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 // indirect
 	github.com/kr/pretty v0.1.0 // indirect
-	github.com/stretchr/testify v1.8.1 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
+	github.com/shirou/gopsutil/v3 v3.23.12 // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/stretchr/testify v1.8.4 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.3 // indirect
 	golang.org/x/net v0.8.0 // indirect
-	golang.org/x/sys v0.6.0 // indirect
+	golang.org/x/sys v0.15.0 // indirect
 	golang.org/x/text v0.8.0 // indirect
 	golang.org/x/tools v0.7.0 // indirect
 	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect