Skip to content

Commit

Permalink
Review handling of status code in HTTP requests (#1549)
Browse files Browse the repository at this point in the history 
In a number of places we were not handling HTTP status codes,
what may hide issues or complicate debugging.
Also ensure that body is closed in some cases where it was not
being closed.
  • Loading branch information
@jsoriano
jsoriano authored Nov 13, 2023
1 parent 7befcd5 commit 2c241f1
Show file tree
Hide file tree
Showing 13 changed files with 272 additions and 108 deletions.
34 changes: 19 additions & 15 deletions internal/benchrunner/runners/rally/metrics.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down Expand Up @@ -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))
}
Expand All @@ -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())
}
}

Expand Down Expand Up @@ -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
}

Expand Down
116 changes: 77 additions & 39 deletions internal/benchrunner/runners/rally/runner.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (
"errors"
"fmt"
"io"
"net/http"
"os"
"os/exec"
"path/filepath"
Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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":{}}}`)
Expand 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)
}
Expand All @@ -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
}

Expand All @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion internal/benchrunner/runners/system/metrics.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}

Expand Down
Loading

0 comments on commit 2c241f1

Please sign in to comment.