feat: Harvest should support recording and replaying HTTP requests

cmd/collectors/keyperf/keyperf.go

@@ -16,12 +16,14 @@ import (
 )
 
 const (
-	latencyIoReqd = 10
+	latencyIoReqd    = 10
+	numRecordsToSave = 60 // Number of records to save when using the recorder
 )
 
 type KeyPerf struct {
-	*rest.Rest // provides: AbstractCollector, Client, Object, Query, TemplateFn, TemplateType
-	perfProp   *perfProp
+	*rest.Rest    // provides: AbstractCollector, Client, Object, Query, TemplateFn, TemplateType
+	perfProp      *perfProp
+	pollDataCalls uint8
 }
 
 type counter struct {
@@ -232,7 +234,16 @@ func (kp *KeyPerf) PollData() (map[string]*matrix.Matrix, error) {
 		return nil, errs.New(errs.ErrConfig, "empty url")
 	}
 
-	perfRecords, err = kp.GetRestData(href)
+	kp.pollDataCalls++
+	if kp.pollDataCalls > numRecordsToSave {
+		kp.pollDataCalls = 0
+	}
+
+	headers := map[string]string{
+		"From": strconv.Itoa(int(kp.pollDataCalls)),
+	}
+
+	perfRecords, err = kp.GetRestData(href, headers)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch href=%s %w", href, err)
 	}

cmd/collectors/rest/rest.go

@@ -677,13 +677,13 @@ func (r *Rest) HandleResults(mat *matrix.Matrix, result []gjson.Result, prop *pr
 	return count, numPartials
 }
 
-func (r *Rest) GetRestData(href string) ([]gjson.Result, error) {
+func (r *Rest) GetRestData(href string, headers ...map[string]string) ([]gjson.Result, error) {
 	r.Logger.Debug("Fetching data", slog.String("href", href))
 	if href == "" {
 		return nil, errs.New(errs.ErrConfig, "empty url")
 	}
 
-	result, err := rest.FetchAll(r.Client, href)
+	result, err := rest.FetchAll(r.Client, href, headers...)
 	if err != nil {
 		return r.handleError(err)
 	}

cmd/collectors/restperf/restperf.go

@@ -42,20 +42,19 @@ const (
 	objWorkloadClass       = "user_defined|system_defined"
 	objWorkloadVolumeClass = "autovolume"
 	timestampMetricName    = "timestamp"
+	numRecordsToSave       = 60 // Number of records to save when using the recorder
 )
 
 var (
-	constituentRegex = regexp.MustCompile(`^(.*)__(\d{4})$`)
+	constituentRegex      = regexp.MustCompile(`^(.*)__(\d{4})$`)
+	qosQuery              = "api/cluster/counter/tables/qos"
+	qosVolumeQuery        = "api/cluster/counter/tables/qos_volume"
+	qosDetailQuery        = "api/cluster/counter/tables/qos_detail"
+	qosDetailVolumeQuery  = "api/cluster/counter/tables/qos_detail_volume"
+	qosWorkloadQuery      = "api/storage/qos/workloads"
+	workloadDetailMetrics = []string{"resource_latency"}
 )
 
-var qosQuery = "api/cluster/counter/tables/qos"
-var qosVolumeQuery = "api/cluster/counter/tables/qos_volume"
-var qosDetailQuery = "api/cluster/counter/tables/qos_detail"
-var qosDetailVolumeQuery = "api/cluster/counter/tables/qos_detail_volume"
-var qosWorkloadQuery = "api/storage/qos/workloads"
-
-var workloadDetailMetrics = []string{"resource_latency"}
-
 var qosQueries = map[string]string{
 	qosQuery:       qosQuery,
 	qosVolumeQuery: qosVolumeQuery,
@@ -70,6 +69,8 @@ type RestPerf struct {
 	perfProp            *perfProp
 	archivedMetrics     map[string]*rest2.Metric // Keeps metric definitions that are not found in the counter schema. These metrics may be available in future ONTAP versions.
 	hasInstanceSchedule bool
+	pollInstanceCalls   uint8
+	pollDataCalls       uint8
 }
 
 type counter struct {
@@ -720,7 +721,15 @@ func (r *RestPerf) PollData() (map[string]*matrix.Matrix, error) {
 		return nil, errs.New(errs.ErrConfig, "empty url")
 	}
 
-	err = rest.FetchRestPerfData(r.Client, href, &perfRecords)
+	r.pollDataCalls++
+	if r.pollDataCalls > numRecordsToSave {
+		r.pollDataCalls = 0
+	}
+
+	headers := map[string]string{
+		"From": strconv.Itoa(int(r.pollDataCalls)),
+	}
+	err = rest.FetchRestPerfData(r.Client, href, &perfRecords, headers)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch href=%s %w", href, err)
 	}
@@ -1475,6 +1484,15 @@ func (r *RestPerf) PollInstance() (map[string]*matrix.Matrix, error) {
 		}
 	}
 
+	r.pollInstanceCalls++
+	if r.pollInstanceCalls > numRecordsToSave/3 {
+		r.pollInstanceCalls = 0
+	}
+
+	headers := map[string]string{
+		"From": strconv.Itoa(int(r.pollInstanceCalls)),
+	}
+
 	href := rest.NewHrefBuilder().
 		APIPath(dataQuery).
 		Fields([]string{fields}).
@@ -1490,7 +1508,7 @@ func (r *RestPerf) PollInstance() (map[string]*matrix.Matrix, error) {
 
 	apiT := time.Now()
 	r.Client.Metadata.Reset()
-	records, err = rest.FetchAll(r.Client, href)
+	records, err = rest.FetchAll(r.Client, href, headers)
 	if err != nil {
 		return r.handleError(err, href)
 	}

cmd/collectors/storagegrid/rest/client.go

@@ -81,7 +81,7 @@ func New(poller *conf.Poller, timeout time.Duration, c *auth.Credentials) (*Clie
 	var (
 		client     Client
 		httpclient *http.Client
-		transport  *http.Transport
+		transport  http.RoundTripper
 		addr       string
 		href       string
 		err        error
@@ -102,7 +102,7 @@ func New(poller *conf.Poller, timeout time.Duration, c *auth.Credentials) (*Clie
 	client.baseURL = href
 	client.Timeout = timeout
 
-	transport, err = c.Transport(nil)
+	transport, err = c.Transport(nil, poller)
 	if err != nil {
 		return nil, err
 	}

cmd/collectors/zapiperf/zapiperf.go

@@ -46,6 +46,8 @@ import (
 	"github.com/netapp/harvest/v2/pkg/slogx"
 	"github.com/netapp/harvest/v2/pkg/tree/node"
 	"log/slog"
+	"maps"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -68,25 +70,28 @@ const (
 	objWorkloadVolumeClass  = "autovolume"
 	BILLION                 = 1_000_000_000
 	timestampMetricName     = "timestamp"
+	numRecordsToSave        = 60 // Number of records to save when using the recorder
 )
 
 var workloadDetailMetrics = []string{"resource_latency"}
 
 type ZapiPerf struct {
-	*zapi.Zapi      // provides: AbstractCollector, Client, Object, Query, TemplateFn, TemplateType
-	object          string
-	filter          string
-	batchSize       int
-	latencyIoReqd   int
-	instanceKeys    []string
-	instanceLabels  map[string]string
-	histogramLabels map[string][]string
-	scalarCounters  []string
-	qosLabels       map[string]string
-	isCacheEmpty    bool
-	keyName         string
-	keyNameIndex    int
-	testFilePath    string // Used only from unit test
+	*zapi.Zapi        // provides: AbstractCollector, Client, Object, Query, TemplateFn, TemplateType
+	object            string
+	filter            string
+	batchSize         int
+	latencyIoReqd     int
+	instanceKeys      []string
+	instanceLabels    map[string]string
+	histogramLabels   map[string][]string
+	scalarCounters    []string
+	qosLabels         map[string]string
+	isCacheEmpty      bool
+	keyName           string
+	keyNameIndex      int
+	testFilePath      string // Used only from unit test
+	pollDataCalls     uint8
+	pollInstanceCalls uint8
 }
 
 func init() {
@@ -416,18 +421,27 @@ func (z *ZapiPerf) PollData() (map[string]*matrix.Matrix, error) {
 	// load requested counters (metrics + labels)
 	requestCounters := request.NewChildS("counters", "")
 	// load scalar metrics
+
+	// Sort the counters and instanceKeys so they are deterministic
+
 	for _, key := range z.scalarCounters {
 		requestCounters.NewChildS("counter", key)
 	}
+
 	// load histograms
-	for key := range z.histogramLabels {
+	sortedHistogramKeys := slices.Sorted(maps.Keys(z.histogramLabels))
+	for _, key := range sortedHistogramKeys {
 		requestCounters.NewChildS("counter", key)
 	}
+
 	// load instance labels
-	for key := range z.instanceLabels {
+	sortedLabels := slices.Sorted(maps.Keys(z.instanceLabels))
+	for _, key := range sortedLabels {
 		requestCounters.NewChildS("counter", key)
 	}
 
+	slices.Sort(instanceKeys)
+
 	// batch indices
 	startIndex := 0
 	endIndex := 0
@@ -457,7 +471,7 @@ func (z *ZapiPerf) PollData() (map[string]*matrix.Matrix, error) {
 						key = v[z.keyNameIndex]
 					}
 				}
-				// avoid adding duplicate keys. It can happen for flex-cache case
+				// Avoid adding duplicate keys. It can happen for flex-cache case
 				if !addedKeys[key] {
 					requestInstances.NewChildS(z.keyName, key)
 					addedKeys[key] = true
@@ -472,7 +486,17 @@ func (z *ZapiPerf) PollData() (map[string]*matrix.Matrix, error) {
 			return nil, err
 		}
 
-		response, rd, pd, err := z.Client.InvokeWithTimers(z.testFilePath)
+		z.pollDataCalls++
+		if z.pollDataCalls > numRecordsToSave {
+			z.pollDataCalls = 0
+		}
+
+		headers := map[string]string{
+			"From": strconv.Itoa(int(z.pollDataCalls)),
+		}
+
+		response, rd, pd, err := z.Client.InvokeWithTimers(z.testFilePath, headers)
+
 		if err != nil {
 			errMsg := strings.ToLower(err.Error())
 			// if ONTAP complains about batch size, use a smaller batch size
@@ -927,6 +951,7 @@ func (z *ZapiPerf) getParentOpsCounters(data *matrix.Matrix, keyAttr string) (ti
 	}
 
 	instanceKeys = data.GetInstanceKeys()
+	slices.Sort(instanceKeys)
 
 	// build ZAPI request
 	request := node.NewXMLS("perf-object-get-instances")
@@ -1324,6 +1349,7 @@ func (z *ZapiPerf) PollCounter() (map[string]*matrix.Matrix, error) {
 		return nil, errs.New(errs.ErrNoMetric, "")
 	}
 
+	slices.Sort(z.scalarCounters)
 	return nil, nil
 }
 
@@ -1608,7 +1634,17 @@ func (z *ZapiPerf) PollInstance() (map[string]*matrix.Matrix, error) {
 
 	for {
 		apiT = time.Now()
-		responseData, err := z.Client.InvokeBatchRequest(request, batchTag, z.testFilePath)
+
+		z.pollInstanceCalls++
+		if z.pollInstanceCalls > numRecordsToSave/3 {
+			z.pollInstanceCalls = 0
+		}
+
+		headers := map[string]string{
+			"From": strconv.Itoa(int(z.pollInstanceCalls)),
+		}
+
+		responseData, err := z.Client.InvokeBatchRequest(request, batchTag, z.testFilePath, headers)
 
 		if err != nil {
 			if errors.Is(err, errs.ErrAPIRequestRejected) {

cmd/tools/rest/client.go

@@ -15,7 +15,6 @@ import (
 	"github.com/tidwall/gjson"
 	"io"
 	"log/slog"
-	"net"
 	"net/http"
 	"net/http/httputil"
 	"os"
@@ -27,11 +26,9 @@ import (
 const (
 	// DefaultTimeout should be > than ONTAP's default REST timeout, which is 15 seconds for GET requests
 	DefaultTimeout = "30s"
-	// DefaultDialerTimeout limits the time spent establishing a TCP connection
-	DefaultDialerTimeout = 10 * time.Second
-	Message              = "message"
-	Code                 = "code"
-	Target               = "target"
+	Message        = "message"
+	Code           = "code"
+	Target         = "target"
 )
 
 type Client struct {
@@ -59,7 +56,7 @@ func New(poller *conf.Poller, timeout time.Duration, credentials *auth.Credentia
 	var (
 		client     Client
 		httpclient *http.Client
-		transport  *http.Transport
+		transport  http.RoundTripper
 		addr       string
 		url        string
 		err        error
@@ -83,11 +80,11 @@ func New(poller *conf.Poller, timeout time.Duration, credentials *auth.Credentia
 	client.baseURL = url
 	client.Timeout = timeout
 
-	transport, err = credentials.Transport(nil)
+	transport, err = credentials.Transport(nil, poller)
 	if err != nil {
 		return nil, err
 	}
-	transport.DialContext = (&net.Dialer{Timeout: DefaultDialerTimeout}).DialContext
+
 	httpclient = &http.Client{Transport: transport, Timeout: timeout}
 	client.client = httpclient
 
@@ -116,7 +113,7 @@ func (c *Client) printRequestAndResponse(req string, response []byte) {
 }
 
 // GetPlainRest makes a REST request to the cluster and returns a json response as a []byte
-func (c *Client) GetPlainRest(request string, encodeURL bool) ([]byte, error) {
+func (c *Client) GetPlainRest(request string, encodeURL bool, headers ...map[string]string) ([]byte, error) {
 	var err error
 	if strings.Index(request, "/") == 0 {
 		request = request[1:]
@@ -134,6 +131,13 @@ func (c *Client) GetPlainRest(request string, encodeURL bool) ([]byte, error) {
 		return nil, err
 	}
 	c.request.Header.Set("Accept", "application/json")
+
+	for _, hs := range headers {
+		for k, v := range hs {
+			c.request.Header.Set(k, v)
+		}
+	}
+
 	pollerAuth, err := c.auth.GetPollerAuth()
 	if err != nil {
 		return nil, err
@@ -159,8 +163,8 @@ func (c *Client) GetPlainRest(request string, encodeURL bool) ([]byte, error) {
 }
 
 // GetRest makes a REST request to the cluster and returns a json response as a []byte
-func (c *Client) GetRest(request string) ([]byte, error) {
-	return c.GetPlainRest(request, true)
+func (c *Client) GetRest(request string, headers ...map[string]string) ([]byte, error) {
+	return c.GetPlainRest(request, true, headers...)
 }
 
 func (c *Client) invokeWithAuthRetry() ([]byte, error) {
@@ -296,7 +300,7 @@ func downloadSwagger(poller *conf.Poller, path string, url string, verbose bool)
 
 	timeout, _ := time.ParseDuration(DefaultTimeout)
 	credentials := auth.NewCredentials(poller, slog.Default())
-	transport, err := credentials.Transport(request)
+	transport, err := credentials.Transport(request, poller)
 	if err != nil {
 		return 0, err
 	}