Cherry-pick #15712 to 7.x: [Metricbeat]Log prometheus errors instead …

…of parsing families (#15961)

* [Metricbeat]Log prometheus errors instead of parsing families (#15712)

CHANGELOG.next.asciidoc

@@ -113,6 +113,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix CPU count in docker/cpu in cases where no `online_cpus` are reported {pull}15070[15070]
 - Add dedot for tags in ec2 metricset and cloudwatch metricset. {issue}15843[15843] {pull}15844[15844]
 - Use RFC3339 format for timestamps collected using the SQL module. {pull}15847[15847]
+- Avoid parsing errors returned from prometheus endpoints. {pull}15712[15712]
 - Add dedot for cloudwatch metric name. {issue}15916[15916] {pull}15917[15917]
 
 *Packetbeat*

metricbeat/helper/prometheus/prometheus.go

@@ -20,13 +20,15 @@ package prometheus
 import (
 	"fmt"
 	"io"
+	"io/ioutil"
 	"net/http"
 
 	"github.com/pkg/errors"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/prometheus/common/expfmt"
 
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/helper"
 	"github.com/elastic/beats/metricbeat/mb"
 )
@@ -43,6 +45,7 @@ type Prometheus interface {
 
 type prometheus struct {
 	httpfetcher
+	logger *logp.Logger
 }
 
 type httpfetcher interface {
@@ -52,10 +55,11 @@ type httpfetcher interface {
 // NewPrometheusClient creates new prometheus helper
 func NewPrometheusClient(base mb.BaseMetricSet) (Prometheus, error) {
 	http, err := helper.NewHTTP(base)
+
 	if err != nil {
 		return nil, err
 	}
-	return &prometheus{http}, nil
+	return &prometheus{http, base.Logger()}, nil
 }
 
 // GetFamilies requests metric families from prometheus endpoint and returns them
@@ -66,6 +70,14 @@ func (p *prometheus) GetFamilies() ([]*dto.MetricFamily, error) {
 	}
 	defer resp.Body.Close()
 
+	if resp.StatusCode > 399 {
+		bodyBytes, err := ioutil.ReadAll(resp.Body)
+		if err == nil {
+			p.logger.Debug("error received from prometheus endpoint: ", string(bodyBytes))
+		}
+		return nil, fmt.Errorf("unexpected status code %d from server", resp.StatusCode)
+	}
+
 	format := expfmt.ResponseFormat(resp.Header)
 	if format == "" {
 		return nil, fmt.Errorf("Invalid format for response of response")

metricbeat/helper/prometheus/prometheus_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 )
 
@@ -185,14 +186,15 @@ var _ = httpfetcher(&mockFetcher{})
 // returns the mockFetcher.Response contents
 func (m mockFetcher) FetchResponse() (*http.Response, error) {
 	return &http.Response{
-		Header: make(http.Header),
-		Body:   ioutil.NopCloser(bytes.NewReader([]byte(m.response))),
+		StatusCode: 200,
+		Header:     make(http.Header),
+		Body:       ioutil.NopCloser(bytes.NewReader([]byte(m.response))),
 	}, nil
 }
 
 func TestPrometheus(t *testing.T) {
 
-	p := &prometheus{mockFetcher{response: promMetrics}}
+	p := &prometheus{mockFetcher{response: promMetrics}, logp.NewLogger("test")}
 
 	tests := []struct {
 		mapping  *MetricsMapping
@@ -933,7 +935,7 @@ func TestPrometheusKeyLabels(t *testing.T) {
 
 	for _, tc := range testCases {
 		r := &mbtest.CapturingReporterV2{}
-		p := &prometheus{mockFetcher{response: tc.prometheusResponse}}
+		p := &prometheus{mockFetcher{response: tc.prometheusResponse}, logp.NewLogger("test")}
 		p.ReportProcessedMetrics(tc.mapping, r)
 		if !assert.Nil(t, r.GetErrors(),
 			"error reporting/processing metrics, at %q", tc.testName) {