[AwsEcsMetricsReceiver] Add codes to read data from ECS endpoint

receiver/awsecscontainermetricsreceiver/awsecscontainermetrics/client.go

@@ -0,0 +1,115 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package awsecscontainermetrics
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+
+	"go.uber.org/zap"
+)
+
+// Client defines the rest client interface
+type Client interface {
+	Get(path string) ([]byte, error)
+}
+
+// NewClientProvider creates the default rest client provider
+func NewClientProvider(endpoint url.URL, logger *zap.Logger) ClientProvider {
+	return &defaultClientProvider{
+		endpoint: endpoint,
+		logger:   logger,
+	}
+}
+
+// ClientProvider defines
+type ClientProvider interface {
+	BuildClient() Client
+}
+
+type defaultClientProvider struct {
+	endpoint url.URL
+	logger   *zap.Logger
+}
+
+func (dcp *defaultClientProvider) BuildClient() Client {
+	return defaultClient(
+		dcp.endpoint,
+		dcp.logger,
+	)
+}
+
+// TODO: Try using config.HTTPClientSettings
+func defaultClient(
+	endpoint url.URL,
+	logger *zap.Logger,
+) *clientImpl {
+	tr := defaultTransport()
+	return &clientImpl{
+		baseURL:    endpoint,
+		httpClient: http.Client{Transport: tr},
+		logger:     logger,
+	}
+}
+
+func defaultTransport() *http.Transport {
+	return http.DefaultTransport.(*http.Transport).Clone()
+}
+
+var _ Client = (*clientImpl)(nil)
+
+type clientImpl struct {
+	baseURL    url.URL
+	httpClient http.Client
+	logger     *zap.Logger
+}
+
+func (c *clientImpl) Get(path string) ([]byte, error) {
+	req, err := c.buildReq(path)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := c.httpClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		closeErr := resp.Body.Close()
+		if closeErr != nil {
+			c.logger.Warn("Failed to close response body", zap.Error(closeErr))
+		}
+	}()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read response body %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("request GET %s failed - %q", req.URL.String(), resp.Status)
+	}
+	return body, nil
+}
+
+func (c *clientImpl) buildReq(path string) (*http.Request, error) {
+	url := c.baseURL.String() + path
+	req, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	return req, nil
+}

receiver/awsecscontainermetricsreceiver/awsecscontainermetrics/client_test.go

@@ -0,0 +1,200 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package awsecscontainermetrics
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func TestClient(t *testing.T) {
+	tr := &fakeRoundTripper{}
+	baseURL, _ := url.Parse("http://localhost:8080")
+	client := &clientImpl{
+		baseURL:    *baseURL,
+		httpClient: http.Client{Transport: tr},
+	}
+	require.False(t, tr.closed)
+	resp, err := client.Get("/stats")
+	require.NoError(t, err)
+	require.Equal(t, "hello", string(resp))
+	require.True(t, tr.closed)
+	require.Equal(t, baseURL.String()+"/stats", tr.url)
+	require.Equal(t, 1, len(tr.header))
+	require.Equal(t, "application/json", tr.header["Content-Type"][0])
+	require.Equal(t, "GET", tr.method)
+}
+
+func TestNewClientProvider(t *testing.T) {
+	baseURL, _ := url.Parse("http://localhost:8080")
+	provider := NewClientProvider(*baseURL, zap.NewNop())
+	require.NotNil(t, provider)
+	_, ok := provider.(*defaultClientProvider)
+	require.True(t, ok)
+
+	client := provider.BuildClient()
+	require.Equal(t, "http://localhost:8080", client.(*clientImpl).baseURL.String())
+}
+
+func TestDefaultClient(t *testing.T) {
+	endpoint, _ := url.Parse("http://localhost:8080")
+	client := defaultClient(*endpoint, zap.NewNop())
+	require.NotNil(t, client.httpClient.Transport)
+	require.Equal(t, "http://localhost:8080", client.baseURL.String())
+}
+
+func TestBuildReq(t *testing.T) {
+	endpoint, _ := url.Parse("http://localhost:8080")
+	p := &defaultClientProvider{
+		endpoint: *endpoint,
+		logger:   zap.NewNop(),
+	}
+	cl := p.BuildClient()
+	req, err := cl.(*clientImpl).buildReq("/test")
+	require.NoError(t, err)
+	require.NotNil(t, req)
+	require.Equal(t, "application/json", req.Header["Content-Type"][0])
+}
+
+func TestBuildBadReq(t *testing.T) {
+	endpoint, _ := url.Parse("http://localhost:8080")
+	p := &defaultClientProvider{
+		endpoint: *endpoint,
+		logger:   zap.NewNop(),
+	}
+	cl := p.BuildClient()
+	_, err := cl.(*clientImpl).buildReq(" ")
+	require.Error(t, err)
+}
+
+func TestGetBad(t *testing.T) {
+	endpoint, _ := url.Parse("http://localhost:8080")
+	p := &defaultClientProvider{
+		endpoint: *endpoint,
+		logger:   zap.NewNop(),
+	}
+	cl := p.BuildClient()
+	resp, err := cl.(*clientImpl).Get(" ")
+	require.Error(t, err)
+	require.Nil(t, resp)
+}
+
+func TestFailedRT(t *testing.T) {
+	tr := &fakeRoundTripper{failOnRT: true}
+	endpoint, _ := url.Parse("http://localhost:8080")
+	client := &clientImpl{
+		baseURL:    *endpoint,
+		httpClient: http.Client{Transport: tr},
+	}
+	_, err := client.Get("/test")
+	require.Error(t, err)
+}
+
+func TestErrOnRead(t *testing.T) {
+	tr := &fakeRoundTripper{errOnRead: true}
+	endpoint, _ := url.Parse("http://localhost:8080")
+	client := &clientImpl{
+		baseURL:    *endpoint,
+		httpClient: http.Client{Transport: tr},
+		logger:     zap.NewNop(),
+	}
+	resp, err := client.Get("/foo")
+	require.Error(t, err)
+	require.Nil(t, resp)
+}
+
+func TestErrCode(t *testing.T) {
+	tr := &fakeRoundTripper{errCode: true}
+	endpoint, _ := url.Parse("http://localhost:8080")
+	client := &clientImpl{
+		baseURL:    *endpoint,
+		httpClient: http.Client{Transport: tr},
+		logger:     zap.NewNop(),
+	}
+	resp, err := client.Get("/foo")
+	require.Error(t, err)
+	require.Nil(t, resp)
+}
+
+var _ http.RoundTripper = (*fakeRoundTripper)(nil)
+
+type fakeRoundTripper struct {
+	closed     bool
+	header     http.Header
+	method     string
+	url        string
+	failOnRT   bool
+	errOnClose bool
+	errOnRead  bool
+	errCode    bool
+}
+
+func (f *fakeRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	if f.failOnRT {
+		return nil, fmt.Errorf("failOnRT == true")
+	}
+	f.header = req.Header
+	f.method = req.Method
+	f.url = req.URL.String()
+	var reader io.Reader
+	if f.errOnRead {
+		reader = &failingReader{}
+	} else {
+		reader = strings.NewReader("hello")
+	}
+	statusCode := 200
+	if f.errCode {
+		statusCode = 503
+	}
+	return &http.Response{
+		StatusCode: statusCode,
+		Body: &fakeReadCloser{
+			Reader: reader,
+			onClose: func() error {
+				f.closed = true
+				if f.errOnClose {
+					return fmt.Errorf("error on close")
+				}
+				return nil
+			},
+		},
+	}, nil
+}
+
+var _ io.Reader = (*failingReader)(nil)
+
+type failingReader struct{}
+
+func (f *failingReader) Read([]byte) (n int, err error) {
+	return 0, fmt.Errorf("error on read")
+}
+
+var _ io.ReadCloser = (*fakeReadCloser)(nil)
+
+type fakeReadCloser struct {
+	io.Reader
+	onClose func() error
+}
+
+func (f *fakeReadCloser) Close() error {
+	return f.onClose()
+}

receiver/awsecscontainermetricsreceiver/awsecscontainermetrics/constant.go

@@ -30,6 +30,10 @@ const (
 	ContainerPrefix    = "container."
 	MetricResourceType = "aoc.ecs"
 
+	EndpointEnvKey   = "ECS_CONTAINER_METADATA_URI_V4"
+	TaskStatsPath    = "/task/stats"
+	TaskMetadataPath = "/task"
+
 	AttributeMemoryUsage    = "memory.usage"
 	AttributeMemoryMaxUsage = "memory.usage.max"
 	AttributeMemoryLimit    = "memory.usage.limit"

receiver/awsecscontainermetricsreceiver/awsecscontainermetrics/metrics.go

@@ -0,0 +1,26 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package awsecscontainermetrics
+
+import (
+	"go.opentelemetry.io/collector/consumer/consumerdata"
+)
+
+func MetricsData(containerStatsMap map[string]ContainerStats, metadata TaskMetadata) []*consumerdata.MetricsData {
+	acc := &metricDataAccumulator{}
+	acc.getMetricsData(containerStatsMap, metadata)
+
+	return acc.md
+}

receiver/awsecscontainermetricsreceiver/awsecscontainermetrics/metrics_helper_test.go

@@ -109,6 +109,11 @@ func TestExtractStorageUsage(t *testing.T) {
 
 	require.EqualValues(t, v, read)
 	require.EqualValues(t, v, write)
+
+	read, write = extractStorageUsage(nil)
+	v = uint64(0)
+	require.EqualValues(t, v, read)
+	require.EqualValues(t, v, write)
 }
 
 func TestGetNetworkStats(t *testing.T) {