public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Elasticsearch: Invalid response JSON parsing error should be downstream (#101506)

Browse Source 
* Elasticsearch: Invalid response JSON parsing error should be downstream

* Add test

* Update test
This commit is contained in:
Ivana Huckova
2025-03-03 18:14:43 +01:00
committed by GitHub
parent c7c68322b1
commit 66279081c6
2 changed files with 72 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tsdb/elasticsearch/client/client.go
+6 -1
View File
@@ -220,6 +220,10 @@ func (c *baseClientImpl) ExecuteMultisearch(r *MultiSearchRequest) (*MultiSearch
} else {
dec := json.NewDecoder(res.Body)
err = dec.Decode(&msr)
if err != nil {
// Invalid JSON response from Elasticsearch
err = backend.DownstreamError(err)
}
}
if err != nil {
c.logger.Error("Failed to decode response from Elasticsearch", "error", err, "duration", time.Since(start), "improvedParsingEnabled", improvedParsingEnabled)
@@ -239,7 +243,8 @@ func StreamMultiSearchResponse(body io.Reader, msr *MultiSearchResponse) error {
_, err := dec.Token() // reads the `{` opening brace
if err != nil {
return err
// Invalid JSON response from Elasticsearch
return backend.DownstreamError(err)
}
for dec.More() {
pkg/tsdb/elasticsearch/client/client_test.go
+66
View File
@@ -182,6 +182,32 @@ func TestClient_ExecuteMultisearch(t *testing.T) {
require.Contains(t, bodyString, "metrics-2018.05.15")
require.Contains(t, bodyString, "metrics-2018.05.17")
})
t.Run("Should return DownstreamError when decoding response fails", func(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
rw.Header().Set("Content-Type", "application/x-ndjson")
_, err := rw.Write([]byte(`{"invalid`))
require.NoError(t, err)
rw.WriteHeader(200)
}))
ds := &DatasourceInfo{
URL: ts.URL,
Database: "[metrics-]YYYY.MM.DD",
HTTPClient: ts.Client(),
}
c, err := NewClient(context.Background(), ds, log.NewNullLogger())
require.NoError(t, err)
t.Cleanup(func() {
ts.Close()
})
_, err = c.ExecuteMultisearch(&MultiSearchRequest{})
require.Error(t, err)
require.True(t, backend.IsDownstreamError(err))
})
}
func TestClient_Index(t *testing.T) {
@@ -410,6 +436,46 @@ func TestStreamMultiSearchResponse_InvalidHitElement(t *testing.T) {
}
}
func TestStreamMultiSearchResponse_ErrorHandling(t *testing.T) {
t.Run("Given invalid elasticsearch responses", func(t *testing.T) {
t.Run("When response is invalid JSON", func(t *testing.T) {
msr := &MultiSearchResponse{}
err := StreamMultiSearchResponse(strings.NewReader(`abc`), msr)
require.Error(t, err)
require.True(t, backend.IsDownstreamError(err))
})
})
t.Run("Given a client with invalid response", func(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
rw.Header().Set("Content-Type", "application/x-ndjson")
_, err := rw.Write([]byte(`{"invalid`))
require.NoError(t, err)
rw.WriteHeader(200)
}))
ds := &DatasourceInfo{
URL: ts.URL,
Database: "[metrics-]YYYY.MM.DD",
HTTPClient: ts.Client(),
}
c, err := NewClient(context.Background(), ds, log.NewNullLogger())
require.NoError(t, err)
t.Cleanup(func() {
ts.Close()
})
t.Run("When executing multi search with invalid response", func(t *testing.T) {
_, err = c.ExecuteMultisearch(&MultiSearchRequest{})
require.Error(t, err)
require.True(t, backend.IsDownstreamError(err))
})
})
}
func createMultisearchForTest(t *testing.T, c Client, timeRange backend.TimeRange) (*MultiSearchRequest, error) {
t.Helper()