Review handling of status code in HTTP requests

internal/benchrunner/runners/rally/metrics.go

@@ -112,9 +112,14 @@ func (c *collector) stop() {
 }
 
 func (c *collector) collectMetricsBeforeRallyRun() {
-	_, err := c.esAPI.Indices.Refresh(c.esAPI.Indices.Refresh.WithIndex(c.datastream))
+	resp, err := c.esAPI.Indices.Refresh(c.esAPI.Indices.Refresh.WithIndex(c.datastream))
 	if err != nil {
-		logger.Errorf("unable to refresh data stream at the beginning of rally run")
+		logger.Errorf("unable to refresh data stream at the beginning of rally run: %s", err)
+		return
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		logger.Errorf("unable to refresh data stream at the beginning of rally run: %s", resp.String())
 		return
 	}
 
@@ -157,19 +162,13 @@ func (c *collector) publish(events [][]byte) {
 		logger.Errorf("error indexing event in metricstore: %w", err)
 		return
 	}
-
-	if resp.Body == nil {
-		logger.Errorf("empty index response body from metricstore: %w", err)
-		return
-	}
+	defer resp.Body.Close()
 
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		logger.Errorf("failed to read index response body from metricstore: %w", err)
 	}
 
-	resp.Body.Close()
-
 	if resp.StatusCode != 201 {
 		logger.Errorf("error indexing event in metricstore (%d): %s: %v", resp.StatusCode, resp.Status(), elasticsearch.NewError(body))
 	}
@@ -187,18 +186,18 @@ func (c *collector) createMetricsIndex() {
 
 	logger.Debugf("creating %s index in metricstore...", c.indexName())
 
-	createRes, err := c.metricsAPI.Indices.Create(
+	resp, err := c.metricsAPI.Indices.Create(
 		c.indexName(),
 		c.metricsAPI.Indices.Create.WithBody(reader),
 	)
 	if err != nil {
 		logger.Errorf("could not create index: %w", err)
 		return
 	}
-	createRes.Body.Close()
+	defer resp.Body.Close()
 
-	if createRes.IsError() {
-		logger.Errorf("got a response error while creating index")
+	if resp.IsError() {
+		logger.Errorf("got a response error while creating index: %s", resp.String())
 	}
 }
 
@@ -287,9 +286,14 @@ func (c *collector) collectDiskUsage() map[string]ingest.DiskUsage {
 }
 
 func (c *collector) collectMetricsAfterRallyRun() {
-	_, err := c.esAPI.Indices.Refresh(c.esAPI.Indices.Refresh.WithIndex(c.datastream))
+	resp, err := c.esAPI.Indices.Refresh(c.esAPI.Indices.Refresh.WithIndex(c.datastream))
 	if err != nil {
-		logger.Errorf("unable to refresh data stream at the end of rally run")
+		logger.Errorf("unable to refresh data stream at the end of rally run: %s", err)
+		return
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		logger.Errorf("unable to refresh data stream at the end of rally run: %s", resp.String())
 		return
 	}
 

internal/benchrunner/runners/rally/runner.go

@@ -12,6 +12,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -336,10 +337,20 @@ func (r *runner) collectAndSummarizeMetrics() (*metricsSummary, error) {
 
 func (r *runner) deleteDataStreamDocs(dataStream string) error {
 	body := strings.NewReader(`{ "query": { "match_all": {} } }`)
-	_, err := r.options.ESAPI.DeleteByQuery([]string{dataStream}, body)
+	resp, err := r.options.ESAPI.DeleteByQuery([]string{dataStream}, body)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to delete data stream docs for data stream %s: %w", dataStream, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusNotFound {
+		// Unavailable index is ok, this means that data is already not there.
+		return nil
 	}
+	if resp.IsError() {
+		return fmt.Errorf("failed to delete data stream docs for data stream %s: %s", dataStream, resp.String())
+	}
+
 	return nil
 }
 
@@ -665,6 +676,9 @@ func (r *runner) reindexData() error {
 		return fmt.Errorf("error getting mapping: %w", err)
 	}
 	defer mappingRes.Body.Close()
+	if mappingRes.IsError() {
+		return fmt.Errorf("error getting mapping: %s", mappingRes)
+	}
 
 	body, err := io.ReadAll(mappingRes.Body)
 	if err != nil {
@@ -709,7 +723,7 @@ func (r *runner) reindexData() error {
 	defer createRes.Body.Close()
 
 	if createRes.IsError() {
-		return errors.New("got a response error while creating index")
+		return fmt.Errorf("got a response error while creating index: %s", createRes)
 	}
 
 	bodyReader := strings.NewReader(`{"query":{"match_all":{}}}`)
@@ -725,21 +739,13 @@ func (r *runner) reindexData() error {
 		return fmt.Errorf("error executing search: %w", err)
 	}
 	defer res.Body.Close()
-
-	type searchRes struct {
-		Error *struct {
-			Reason string `json:"reson"`
-		} `json:"error"`
-		ScrollID string `json:"_scroll_id"`
-		Hits     []struct {
-			ID     string                 `json:"_id"`
-			Source map[string]interface{} `json:"_source"`
-		} `json:"hits"`
+	if res.IsError() {
+		return fmt.Errorf("error executing search: %s", res)
 	}
 
 	// Iterate through the search results using the Scroll API
 	for {
-		var sr searchRes
+		var sr searchResponse
 		if err := json.NewDecoder(res.Body).Decode(&sr); err != nil {
 			return fmt.Errorf("error decoding search response: %w", err)
 		}
@@ -752,40 +758,67 @@ func (r *runner) reindexData() error {
 			break
 		}
 
-		var bulkBodyBuilder strings.Builder
-		for _, hit := range sr.Hits {
-			bulkBodyBuilder.WriteString(fmt.Sprintf("{\"index\":{\"_index\":\"%s\",\"_id\":\"%s\"}}\n", indexName, hit.ID))
-			enriched := r.enrichEventWithBenchmarkMetadata(hit.Source)
-			src, err := json.Marshal(enriched)
-			if err != nil {
-				return fmt.Errorf("error decoding _source: %w", err)
-			}
-			bulkBodyBuilder.WriteString(fmt.Sprintf("%s\n", string(src)))
+		err := r.bulkMetrics(indexName, sr)
+		if err != nil {
+			return err
 		}
+	}
+
+	logger.Debug("reindexing operation finished")
+	return nil
+}
 
-		logger.Debugf("bulk request of %d events...", len(sr.Hits))
+type searchResponse struct {
+	Error *struct {
+		Reason string `json:"reson"`
+	} `json:"error"`
+	ScrollID string `json:"_scroll_id"`
+	Hits     []struct {
+		ID     string                 `json:"_id"`
+		Source map[string]interface{} `json:"_source"`
+	} `json:"hits"`
+}
 
-		bulkRes, err := r.options.ESMetricsAPI.Bulk(strings.NewReader(bulkBodyBuilder.String()))
+func (r *runner) bulkMetrics(indexName string, sr searchResponse) error {
+	var bulkBodyBuilder strings.Builder
+	for _, hit := range sr.Hits {
+		bulkBodyBuilder.WriteString(fmt.Sprintf("{\"index\":{\"_index\":\"%s\",\"_id\":\"%s\"}}\n", indexName, hit.ID))
+		enriched := r.enrichEventWithBenchmarkMetadata(hit.Source)
+		src, err := json.Marshal(enriched)
 		if err != nil {
-			return fmt.Errorf("error performing the bulk index request: %w", err)
+			return fmt.Errorf("error decoding _source: %w", err)
 		}
-		bulkRes.Body.Close()
+		bulkBodyBuilder.WriteString(fmt.Sprintf("%s\n", string(src)))
+	}
 
-		if sr.ScrollID == "" {
-			return errors.New("error getting scroll ID")
-		}
+	logger.Debugf("bulk request of %d events...", len(sr.Hits))
 
-		res, err = r.options.ESAPI.Scroll(
-			r.options.ESAPI.Scroll.WithScrollID(sr.ScrollID),
-			r.options.ESAPI.Scroll.WithScroll(time.Minute),
-		)
-		if err != nil {
-			return fmt.Errorf("error executing scroll: %s", err)
-		}
-		res.Body.Close()
+	resp, err := r.options.ESMetricsAPI.Bulk(strings.NewReader(bulkBodyBuilder.String()))
+	if err != nil {
+		return fmt.Errorf("error performing the bulk index request: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		return fmt.Errorf("error performing the bulk index request: %s", resp.String())
+	}
+
+	if sr.ScrollID == "" {
+		return errors.New("error getting scroll ID")
+	}
+
+	resp, err = r.options.ESAPI.Scroll(
+		r.options.ESAPI.Scroll.WithScrollID(sr.ScrollID),
+		r.options.ESAPI.Scroll.WithScroll(time.Minute),
+	)
+	if err != nil {
+		return fmt.Errorf("error executing scroll: %s", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.IsError() {
+		return fmt.Errorf("error executing scroll: %s", resp.String())
 	}
 
-	logger.Debug("reindexing operation finished")
 	return nil
 }
 
@@ -809,12 +842,17 @@ func (r *runner) enrichEventWithBenchmarkMetadata(e map[string]interface{}) map[
 func getTotalHits(esapi *elasticsearch.API, dataStream string) (int, error) {
 	resp, err := esapi.Count(
 		esapi.Count.WithIndex(dataStream),
+		esapi.Count.WithIgnoreUnavailable(true),
 	)
 	if err != nil {
 		return 0, fmt.Errorf("could not search data stream: %w", err)
 	}
 	defer resp.Body.Close()
 
+	if resp.IsError() {
+		return 0, fmt.Errorf("failed to get hits count: %s", resp.String())
+	}
+
 	var results struct {
 		Count int
 		Error *struct {

internal/benchrunner/runners/system/metrics.go

@@ -196,7 +196,7 @@ func (c *collector) createMetricsIndex() {
 	createRes.Body.Close()
 
 	if createRes.IsError() {
-		logger.Debug("got a response error while creating index")
+		logger.Debug("got a response error while creating index: %s", createRes)
 	}
 }