fix: handle batching in shelf plugin

cmd/collectors/commonutils.go

@@ -2,7 +2,6 @@ package collectors
 
 import (
 	"github.com/netapp/harvest/v2/cmd/tools/rest"
-	"github.com/netapp/harvest/v2/pkg/api/ontapi/zapi"
 	"github.com/netapp/harvest/v2/pkg/logging"
 	"github.com/netapp/harvest/v2/pkg/matrix"
 	"github.com/netapp/harvest/v2/pkg/tree/node"
@@ -11,6 +10,8 @@ import (
 	"time"
 )
 
+const DefaultBatchSize = "500"
+
 func InvokeRestCall(client *rest.Client, href string, logger *logging.Logger) ([]gjson.Result, error) {
 	result, err := rest.Fetch(client, href)
 	if err != nil {
@@ -25,45 +26,6 @@ func InvokeRestCall(client *rest.Client, href string, logger *logging.Logger) ([
 	return result, nil
 }
 
-func InvokeZapiCall(client *zapi.Client, request *node.Node, logger *logging.Logger, tag string) ([]*node.Node, string, error) {
-
-	var (
-		result   *node.Node
-		response []*node.Node
-		newTag   string
-		err      error
-	)
-
-	if tag != "" {
-		if result, newTag, err = client.InvokeBatchRequest(request, tag); err != nil {
-			return nil, "", err
-		}
-	} else {
-		if result, err = client.InvokeRequest(request); err != nil {
-			return nil, "", err
-		}
-	}
-
-	if result == nil {
-		return nil, "", nil
-	}
-
-	if x := result.GetChildS("attributes-list"); x != nil {
-		response = x.GetChildren()
-	} else if y := result.GetChildS("attributes"); y != nil {
-		// Check for non-list response
-		response = y.GetChildren()
-	}
-
-	if len(response) == 0 {
-		return nil, "", nil
-	}
-
-	logger.Trace().Int("object", len(response)).Msg("fetching")
-
-	return response, newTag, nil
-}
-
 func UpdateProtectedFields(instance *matrix.Instance) {
 
 	// check for group_type
@@ -161,7 +123,7 @@ func GetDataInterval(param *node.Node, defaultInterval time.Duration) float64 {
 	return defaultInterval.Seconds()
 }
 
-// timestamp in micro seconds
+// IsTimestampOlderThanDuration - timestamp units are micro seconds
 func IsTimestampOlderThanDuration(timestamp float64, duration time.Duration) bool {
 	return time.Since(time.UnixMicro(int64(timestamp))) > duration
 }

cmd/collectors/zapi/plugins/certificate/certificate.go

@@ -58,12 +58,11 @@ func (my *Certificate) Init() error {
 		return err
 	}
 
+	my.batchSize = BatchSize
 	if b := my.Params.GetChildContentS("batch_size"); b != "" {
 		if _, err := strconv.Atoi(b); err == nil {
 			my.batchSize = b
 		}
-	} else {
-		my.batchSize = BatchSize
 	}
 
 	return nil
@@ -204,7 +203,8 @@ func (my *Certificate) GetAdminVserver() (string, error) {
 	vserverInfo := query.NewChildS("vserver-info", "")
 	vserverInfo.NewChildS("vserver-type", "admin")
 
-	if result, _, err = collectors.InvokeZapiCall(my.client, request, my.Logger, ""); err != nil {
+	// Fetching only admin SVMs
+	if result, err = my.client.InvokeZapiCall(request); err != nil {
 		return "", err
 	}
 
@@ -214,6 +214,7 @@ func (my *Certificate) GetAdminVserver() (string, error) {
 	// This should be one iteration only as cluster can have one admin vserver
 	for _, svm := range result {
 		adminVserver = svm.GetChildContentS("vserver-name")
+		break
 	}
 	return adminVserver, nil
 }
@@ -233,7 +234,8 @@ func (my *Certificate) GetSecuritySsl(adminSvm string) (string, error) {
 	vserverInfo := query.NewChildS("vserver-ssl-info", "")
 	vserverInfo.NewChildS("vserver", adminSvm)
 
-	if result, _, err = collectors.InvokeZapiCall(my.client, request, my.Logger, ""); err != nil {
+	// fetching data of only admin vservers
+	if result, err = my.client.InvokeZapiCall(request); err != nil {
 		return "", err
 	}
 
@@ -243,6 +245,7 @@ func (my *Certificate) GetSecuritySsl(adminSvm string) (string, error) {
 	// This should be one iteration only as cluster can have one admin vserver
 	for _, ssl := range result {
 		certificateSerial = ssl.GetChildContentS("certificate-serial-number")
+		break
 	}
 	return certificateSerial, nil
 }

cmd/collectors/zapi/plugins/qtree/qtree.go

@@ -92,13 +92,12 @@ func (my *Qtree) Init() error {
 	my.Logger.Debug().Msgf("added data with %d metrics", len(my.data.GetMetrics()))
 
 	// setup batchSize for request
+	my.batchSize = BatchSize
 	if my.client.IsClustered() {
 		if b := my.Params.GetChildContentS("batch_size"); b != "" {
 			if _, err := strconv.Atoi(b); err == nil {
 				my.batchSize = b
 			}
-		} else {
-			my.batchSize = BatchSize
 		}
 	}
 

cmd/collectors/zapi/plugins/security/security.go

@@ -106,7 +106,8 @@ func (my *Security) getSecurityConfig() (string, error) {
 	request = node.NewXMLS("security-config-get")
 	request.NewChildS("interface", "ssl")
 
-	if result, _, err = collectors.InvokeZapiCall(my.client, request, my.Logger, ""); err != nil {
+	// fetching only ssl interface
+	if result, err = my.client.InvokeZapiCall(request); err != nil {
 		return "", err
 	}
 
@@ -116,6 +117,7 @@ func (my *Security) getSecurityConfig() (string, error) {
 
 	for _, securityConfig := range result {
 		fipsEnabled = securityConfig.GetChildContentS("is-fips-enabled")
+		break
 	}
 	return fipsEnabled, nil
 }
@@ -153,7 +155,8 @@ func (my *Security) getEnabledValue(request *node.Node) (string, error) {
 		err     error
 	)
 
-	if result, _, err = collectors.InvokeZapiCall(my.client, request, my.Logger, ""); err != nil {
+	// fetching only telnet/rsh protocols
+	if result, err = my.client.InvokeZapiCall(request); err != nil {
 		return "", err
 	}
 
@@ -163,6 +166,7 @@ func (my *Security) getEnabledValue(request *node.Node) (string, error) {
 
 	for _, securityConfig := range result {
 		enabled = securityConfig.GetChildContentS("enabled")
+		break
 	}
 
 	return enabled, nil

cmd/collectors/zapi/plugins/shelf/shelf.go

@@ -10,14 +10,18 @@ import (
 	"github.com/netapp/harvest/v2/pkg/matrix"
 	"github.com/netapp/harvest/v2/pkg/tree/node"
 	"github.com/netapp/harvest/v2/pkg/util"
+	"strconv"
 	"strings"
 )
 
+const BatchSize = "500"
+
 type Shelf struct {
 	*plugin.AbstractPlugin
 	data           map[string]*matrix.Matrix
 	instanceKeys   map[string]string
 	instanceLabels map[string]*dict.Dict
+	batchSize      string
 	client         *zapi.Client
 	query          string
 }
@@ -139,14 +143,24 @@ func (my *Shelf) Init() error {
 	}
 
 	my.Logger.Debug().Msgf("initialized with data [%d] objects", len(my.data))
+
+	// setup batchSize for request
+	my.batchSize = BatchSize
+	if my.client.IsClustered() {
+		if b := my.Params.GetChildContentS("batch_size"); b != "" {
+			if _, err := strconv.Atoi(b); err == nil {
+				my.batchSize = b
+			}
+		}
+	}
 	return nil
 }
 
 func (my *Shelf) Run(data *matrix.Matrix) ([]*matrix.Matrix, error) {
 
 	var (
-		result *node.Node
 		err    error
+		output []*matrix.Matrix
 	)
 
 	if !my.client.IsClustered() {
@@ -155,16 +169,18 @@ func (my *Shelf) Run(data *matrix.Matrix) ([]*matrix.Matrix, error) {
 		}
 	}
 
-	if result, err = my.client.InvokeRequestString(my.query); err != nil {
-		return nil, err
-	}
-
 	// Set all global labels from zapi.go if already not exist
 	for a := range my.instanceLabels {
 		my.data[a].SetGlobalLabels(data.GetGlobalLabels())
 	}
 
-	var output []*matrix.Matrix
+	request := node.NewXMLS(my.query)
+	request.NewChildS("max-records", my.batchSize)
+
+	result, err := my.client.InvokeZapiCall(request)
+	if err != nil {
+		return nil, err
+	}
 
 	if my.client.IsClustered() {
 		output, err = my.handleCMode(result)
@@ -176,14 +192,15 @@ func (my *Shelf) Run(data *matrix.Matrix) ([]*matrix.Matrix, error) {
 	}
 
 	if my.client.IsClustered() {
-		return my.calculateEnvironmentMetrics(output, data)
-	} else {
-		return output, nil
+		err := my.calculateEnvironmentMetrics(data)
+		if err != nil {
+			return nil, err
+		}
 	}
-
+	return output, nil
 }
 
-func (my *Shelf) calculateEnvironmentMetrics(output []*matrix.Matrix, data *matrix.Matrix) ([]*matrix.Matrix, error) {
+func (my *Shelf) calculateEnvironmentMetrics(data *matrix.Matrix) error {
 	var err error
 	shelfEnvironmentMetricMap := make(map[string]*shelfEnvironmentMetric, 0)
 	for _, o := range my.data {
@@ -326,25 +343,16 @@ func (my *Shelf) calculateEnvironmentMetrics(output []*matrix.Matrix, data *matr
 			}
 		}
 	}
-	return output, nil
+	return nil
 }
 
-func (my *Shelf) handleCMode(result *node.Node) ([]*matrix.Matrix, error) {
+func (my *Shelf) handleCMode(shelves []*node.Node) ([]*matrix.Matrix, error) {
 	var (
-		shelves []*node.Node
+		output []*matrix.Matrix
 	)
 
-	if x := result.GetChildS("attributes-list"); x != nil {
-		shelves = x.GetChildren()
-	}
-	if len(shelves) == 0 {
-		return nil, errs.New(errs.ErrNoInstance, "no shelf instances found")
-	}
-
 	my.Logger.Debug().Msgf("fetching %d shelf counters", len(shelves))
 
-	var output []*matrix.Matrix
-
 	// Purge and reset data
 	for _, data1 := range my.data {
 		data1.PurgeInstances()
@@ -399,7 +407,7 @@ func (my *Shelf) handleCMode(result *node.Node) ([]*matrix.Matrix, error) {
 						instance.SetLabel("shelf", shelfName)
 						instance.SetLabel("shelf_id", shelfID)
 
-						// Each child would have different possible values which is ugly way to write all of them,
+						// Each child would have different possible values which is an ugly way to write all of them,
 						// so normal value would be mapped to 1 and rest all are mapped to 0.
 						if instance.GetLabel("status") == "normal" {
 							_ = statusMetric.SetValueInt64(instance, 1)
@@ -431,20 +439,24 @@ func (my *Shelf) handleCMode(result *node.Node) ([]*matrix.Matrix, error) {
 	return output, nil
 }
 
-func (my *Shelf) handle7Mode(result *node.Node) ([]*matrix.Matrix, error) {
+func (my *Shelf) handle7Mode(result []*node.Node) ([]*matrix.Matrix, error) {
 	var (
 		shelves  []*node.Node
 		channels []*node.Node
+		output   []*matrix.Matrix
 	)
-	//fallback to 7mode
-	channels = result.SearchChildren([]string{"shelf-environ-channel-info"})
+
+	// Result would be the zapi response itself with only one record.
+	if len(result) != 1 {
+		return nil, errs.New(errs.ErrNoInstance, "no shelves found")
+	}
+	// fallback to 7mode
+	channels = result[0].SearchChildren([]string{"shelf-environ-channel-info"})
 
 	if len(channels) == 0 {
 		return nil, errs.New(errs.ErrNoInstance, "no channels found")
 	}
 
-	var output []*matrix.Matrix
-
 	// Purge and reset data
 	for _, data1 := range my.data {
 		data1.PurgeInstances()
@@ -504,7 +516,7 @@ func (my *Shelf) handle7Mode(result *node.Node) ([]*matrix.Matrix, error) {
 							instance.SetLabel("shelf_id", shelfID)
 							instance.SetLabel("channel", channelName)
 
-							// Each child would have different possible values which is ugly way to write all of them,
+							// Each child would have different possible values which is an ugly way to write all of them,
 							// so normal value would be mapped to 1 and rest all are mapped to 0.
 							if instance.GetLabel("status") == "normal" {
 								_ = statusMetric.SetValueInt64(instance, 1)

cmd/collectors/zapi/plugins/snapmirror/snapmirror.go

@@ -81,7 +81,7 @@ func (my *SnapMirror) Run(data *matrix.Matrix) ([]*matrix.Matrix, error) {
 	}
 
 	for _, instance := range data.GetInstances() {
-		// Zapi call with `expand=true` would gives all the constituent's relationships as well, which we don't want to export.
+		// Zapi call with `expand=true` returns all the constituent's relationships. We do not want to export them.
 		if match := flexgroupConstituentName.FindStringSubmatch(instance.GetLabel("destination_volume")); len(match) == 3 {
 			instance.SetExportable(false)
 			continue
@@ -185,6 +185,7 @@ func (my *SnapMirror) getSVMPeerData(cluster string) error {
 	)
 
 	request := node.NewXMLS("vserver-peer-get-iter")
+	request.NewChildS("max-records", collectors.DefaultBatchSize)
 	// Fetching only remote vserver-peer
 	query := request.NewChildS("query", "")
 	vserverPeerInfo := query.NewChildS("vserver-peer-info", "")
@@ -193,7 +194,8 @@ func (my *SnapMirror) getSVMPeerData(cluster string) error {
 	// Clean svmPeerMap map
 	my.svmPeerDataMap = make(map[string]Peer)
 
-	if result, _, err = collectors.InvokeZapiCall(my.client, request, my.Logger, ""); err != nil {
+	// fetching only remote vserver peer data
+	if result, err = my.client.InvokeZapiCall(request); err != nil {
 		return err
 	}